| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495 |
- """Tests for the TUI view-state reducer."""
- from stt_daemon.tui.state import MAX_TRANSCRIPTS, ViewState
- def test_mode_event_updates_mode_and_style():
- state = ViewState()
- state.ingest({"type": "mode", "mode": "command_armed"})
- assert state.mode == "command_armed"
- assert state.mode_style() == "yellow"
- def test_heartbeat_self_corrects_missed_mode():
- state = ViewState()
- state.ingest(
- {
- "type": "heartbeat",
- "mode": "chat",
- "stt_ok": True,
- "in_flight": False,
- "worker_stall_s": 0.0,
- "queue_depth": 2,
- "device_ok": True,
- }
- )
- assert state.mode == "chat"
- assert state.queue_depth == 2
- def test_health_red_when_stt_not_ok():
- state = ViewState()
- state.ingest({"type": "heartbeat", "mode": "idle", "stt_ok": False})
- assert state.health_ok() is False
- def test_health_red_when_device_down():
- state = ViewState()
- state.ingest({"type": "heartbeat", "mode": "idle", "stt_ok": True, "device_ok": False})
- assert state.health_ok() is False
- def test_worker_stall_recorded_and_degrades():
- state = ViewState()
- state.ingest({"type": "heartbeat", "stt_ok": False, "worker_stall_s": 16.5})
- assert state.worker_stall_s == 16.5
- assert state.health_ok() is False
- def test_command_text_and_edited_flag():
- state = ViewState()
- state.ingest({"type": "command_text", "text": "hello", "edited": True, "phase": "ready"})
- assert state.command_text == "hello"
- assert state.command_edited is True
- assert state.phase == "ready"
- def test_status_populates_fields():
- state = ViewState()
- state.ingest(
- {
- "type": "status",
- "mode": "command_armed",
- "ready": True,
- "phase": "ready",
- "command_text": "x",
- "stt_ok": True,
- }
- )
- assert state.mode == "command_armed"
- assert state.ready is True
- assert state.command_text == "x"
- def test_transcripts_are_bounded():
- state = ViewState()
- for i in range(MAX_TRANSCRIPTS + 25):
- state.ingest({"type": "transcript", "text": f"t{i}", "lang": "en"})
- assert len(state.transcripts) == MAX_TRANSCRIPTS
- assert state.transcripts[-1]["text"] == f"t{MAX_TRANSCRIPTS + 24}"
- assert state.transcript_lines()[-1].startswith("[en]")
- def test_chat_and_tool_events_recorded():
- state = ViewState()
- state.ingest({"type": "chat_message", "role": "user", "text": "hi"})
- state.ingest({"type": "chat_message", "role": "assistant", "text": "hello"})
- state.ingest({"type": "tool_call", "tool": "get_datetime", "args": {}})
- state.ingest({"type": "tool_result", "tool": "get_datetime", "ok": True, "result": "x"})
- state.ingest({"type": "error", "code": "no_viewer", "message": "m"})
- assert state.chat_lines() == ["user: hi", "assistant: hello"]
- assert state.tool is not None and state.tool["tool"] == "get_datetime"
- assert state.tool_result is not None and state.tool_result["ok"] is True
- assert state.last_error is not None and state.last_error["code"] == "no_viewer"
|