test_tui_state.py 4.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138
  1. """Tests for the TUI view-state reducer."""
  2. from stt_daemon.tui.state import MAX_TRANSCRIPTS, ViewState
  3. def test_mode_event_updates_mode_and_style():
  4. state = ViewState()
  5. state.ingest({"type": "mode", "mode": "command_armed"})
  6. assert state.mode == "command_armed"
  7. assert state.mode_style() == "yellow"
  8. def test_heartbeat_self_corrects_missed_mode():
  9. state = ViewState()
  10. state.ingest(
  11. {
  12. "type": "heartbeat",
  13. "mode": "chat",
  14. "stt_ok": True,
  15. "in_flight": False,
  16. "worker_stall_s": 0.0,
  17. "queue_depth": 2,
  18. "device_ok": True,
  19. }
  20. )
  21. assert state.mode == "chat"
  22. assert state.queue_depth == 2
  23. def test_health_red_when_stt_not_ok():
  24. state = ViewState()
  25. state.ingest({"type": "heartbeat", "mode": "idle", "stt_ok": False})
  26. assert state.health_ok() is False
  27. def test_health_red_when_device_down():
  28. state = ViewState()
  29. state.ingest({"type": "heartbeat", "mode": "idle", "stt_ok": True, "device_ok": False})
  30. assert state.health_ok() is False
  31. def test_worker_stall_recorded_and_degrades():
  32. state = ViewState()
  33. state.ingest({"type": "heartbeat", "stt_ok": False, "worker_stall_s": 16.5})
  34. assert state.worker_stall_s == 16.5
  35. assert state.health_ok() is False
  36. def test_command_text_and_edited_flag():
  37. state = ViewState()
  38. state.ingest({"type": "command_text", "text": "hello", "edited": True, "phase": "ready"})
  39. assert state.command_text == "hello"
  40. assert state.command_edited is True
  41. assert state.phase == "ready"
  42. def test_status_populates_fields():
  43. state = ViewState()
  44. state.ingest(
  45. {
  46. "type": "status",
  47. "mode": "command_armed",
  48. "ready": True,
  49. "phase": "ready",
  50. "command_text": "x",
  51. "stt_ok": True,
  52. }
  53. )
  54. assert state.mode == "command_armed"
  55. assert state.ready is True
  56. assert state.command_text == "x"
  57. def test_transcripts_are_bounded():
  58. state = ViewState()
  59. for i in range(MAX_TRANSCRIPTS + 25):
  60. state.ingest({"type": "transcript", "text": f"t{i}", "lang": "en"})
  61. assert len(state.transcripts) == MAX_TRANSCRIPTS
  62. assert state.transcripts[-1]["text"] == f"t{MAX_TRANSCRIPTS + 24}"
  63. assert state.transcript_lines()[-1].startswith("[en]")
  64. def test_command_output_is_collected_for_display():
  65. state = ViewState()
  66. state.ingest({"type": "tool_call", "tool": "get_datetime", "args": {}})
  67. state.ingest(
  68. {"type": "tool_result", "tool": "get_datetime", "ok": True, "result": "2026-09-21T01:00:00"}
  69. )
  70. state.ingest({"type": "error", "code": "not_ready", "message": "command is not ready"})
  71. lines = state.lower_lines()
  72. assert any("tool get_datetime" in line for line in lines)
  73. assert any("ok:" in line and "2026-09-21" in line for line in lines)
  74. assert any("not_ready" in line for line in lines)
  75. def test_chat_turns_also_appear_in_lower_log():
  76. state = ViewState()
  77. state.ingest({"type": "chat_message", "role": "user", "text": "hello"})
  78. assert state.lower_lines() == ["user: hello"]
  79. def test_lower_log_is_bounded():
  80. from stt_daemon.tui.state import MAX_LOWER
  81. state = ViewState()
  82. for i in range(MAX_LOWER + 20):
  83. state.ingest({"type": "error", "code": f"e{i}", "message": "m"})
  84. assert len(state.lower_lines()) == MAX_LOWER
  85. def test_speech_activity_is_tracked():
  86. state = ViewState()
  87. assert state.speech_open is False
  88. state.ingest({"type": "speech_start"})
  89. assert state.speech_open is True
  90. state.ingest({"type": "speech_stop"})
  91. assert state.speech_open is False
  92. state.ingest({"type": "status", "speech_open": True})
  93. assert state.speech_open is True
  94. def test_chat_and_tool_events_recorded():
  95. state = ViewState()
  96. state.ingest({"type": "chat_message", "role": "user", "text": "hi"})
  97. state.ingest({"type": "chat_message", "role": "assistant", "text": "hello"})
  98. state.ingest({"type": "tool_call", "tool": "get_datetime", "args": {}})
  99. state.ingest({"type": "tool_result", "tool": "get_datetime", "ok": True, "result": "x"})
  100. state.ingest({"type": "error", "code": "no_viewer", "message": "m"})
  101. assert state.chat_lines() == ["user: hi", "assistant: hello"]
  102. assert state.tool is not None and state.tool["tool"] == "get_datetime"
  103. assert state.tool_result is not None and state.tool_result["ok"] is True
  104. assert state.last_error is not None and state.last_error["code"] == "no_viewer"