196 lines
7.7 KiB
Python
196 lines
7.7 KiB
Python
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from PySide6.QtWidgets import QApplication
|
|
|
|
from bolt_pet import controller as controller_mod
|
|
from bolt_pet.state import PetState
|
|
|
|
# A QApplication is required before any QObject with signals can be built.
|
|
_app = QApplication.instance() or QApplication(["test"])
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def no_screen_probes(monkeypatch):
|
|
# The real ones shell out to xprop/osascript — irrelevant here, and slow
|
|
# (or hung) on a headless box.
|
|
monkeypatch.setattr(controller_mod.screen_context, "context_for", lambda text: text)
|
|
monkeypatch.setattr(controller_mod.screen_context, "is_fullscreen_active", lambda: False)
|
|
|
|
|
|
@pytest.fixture
|
|
def ctrl():
|
|
return controller_mod.PetController()
|
|
|
|
|
|
def _capture(signal):
|
|
events = []
|
|
signal.connect(lambda *a: events.append(a[0] if len(a) == 1 else a))
|
|
return events
|
|
|
|
|
|
# ── conversation turn ────────────────────────────────────────────────────────
|
|
|
|
def test_full_turn_happy_path(monkeypatch, ctrl):
|
|
states = _capture(ctrl.state_changed)
|
|
said = _capture(ctrl.said)
|
|
|
|
monkeypatch.setattr(controller_mod.mic, "record_utterance", lambda *a, **k: np.zeros(10, dtype=np.int16))
|
|
monkeypatch.setattr(controller_mod.stt, "transcribe", lambda pcm: "what's the weather")
|
|
monkeypatch.setattr(controller_mod.server_client, "converse", lambda text, on_command=None: controller_mod.server_client.Reply("sunny and 72"))
|
|
spoken = []
|
|
monkeypatch.setattr(controller_mod.tts, "speak",
|
|
lambda text, on_error=None, should_stop=None, voice_id=None: (spoken.append(text), True)[1])
|
|
|
|
ctrl._handle_conversation_turn()
|
|
|
|
assert states == ["listening", "thinking", "talking", "idle"]
|
|
assert said == ["sunny and 72"]
|
|
assert spoken == ["sunny and 72"]
|
|
assert ctrl._state.state == PetState.IDLE
|
|
|
|
|
|
def test_turn_with_nothing_heard_returns_to_idle_without_calling_server(monkeypatch, ctrl):
|
|
states = _capture(ctrl.state_changed)
|
|
monkeypatch.setattr(controller_mod.mic, "record_utterance", lambda *a, **k: None)
|
|
called = {"n": 0}
|
|
monkeypatch.setattr(controller_mod.server_client, "converse",
|
|
lambda text, on_command=None: called.__setitem__("n", called["n"] + 1))
|
|
|
|
ctrl._handle_conversation_turn()
|
|
|
|
assert states == ["listening", "idle"]
|
|
assert called["n"] == 0
|
|
|
|
|
|
def test_turn_with_empty_transcript_returns_to_idle(monkeypatch, ctrl):
|
|
states = _capture(ctrl.state_changed)
|
|
monkeypatch.setattr(controller_mod.mic, "record_utterance", lambda *a, **k: np.zeros(10, dtype=np.int16))
|
|
monkeypatch.setattr(controller_mod.stt, "transcribe", lambda pcm: "")
|
|
|
|
ctrl._handle_conversation_turn()
|
|
|
|
assert states == ["listening", "thinking", "idle"]
|
|
|
|
|
|
def test_turn_with_stt_error_flashes_error_then_idle(monkeypatch, ctrl):
|
|
states = _capture(ctrl.state_changed)
|
|
logs = _capture(ctrl.log)
|
|
monkeypatch.setattr(controller_mod.mic, "record_utterance", lambda *a, **k: np.zeros(10, dtype=np.int16))
|
|
|
|
def boom(pcm):
|
|
raise controller_mod.stt.SttError("deepgram is down")
|
|
monkeypatch.setattr(controller_mod.stt, "transcribe", boom)
|
|
|
|
ctrl._handle_conversation_turn()
|
|
|
|
assert states == ["listening", "thinking", "error", "idle"]
|
|
assert any("deepgram is down" in msg for msg in logs)
|
|
|
|
|
|
def test_turn_with_server_error_flashes_error_then_idle(monkeypatch, ctrl):
|
|
states = _capture(ctrl.state_changed)
|
|
monkeypatch.setattr(controller_mod.mic, "record_utterance", lambda *a, **k: np.zeros(10, dtype=np.int16))
|
|
monkeypatch.setattr(controller_mod.stt, "transcribe", lambda pcm: "hello")
|
|
|
|
def boom(text, on_command=None):
|
|
raise controller_mod.server_client.ServerError("server is down")
|
|
monkeypatch.setattr(controller_mod.server_client, "converse", boom)
|
|
|
|
ctrl._handle_conversation_turn()
|
|
|
|
assert states == ["listening", "thinking", "error", "idle"]
|
|
|
|
|
|
# ── mute / talk-now / wake-or-click disambiguation ──────────────────────────
|
|
|
|
def test_toggle_mute_flips_and_returns_new_state(ctrl):
|
|
assert ctrl.toggle_mute() is True
|
|
assert ctrl.toggle_mute() is False
|
|
|
|
|
|
def test_wait_for_wake_or_click_true_on_phrase_detection(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.wake_word, "listen_for_wake_word", lambda *a, **k: True)
|
|
ctrl._stream = object()
|
|
assert ctrl._wait_for_wake_or_click() is True
|
|
|
|
|
|
def test_wait_for_wake_or_click_true_on_manual_trigger(monkeypatch, ctrl):
|
|
# listen_for_wake_word returns False because should_continue() went
|
|
# false (the talk_now flag got set) — controller must still recognize
|
|
# this as "proceed", not "spurious wakeup".
|
|
def fake_listen(stream, should_continue, on_tick=None, **kwargs):
|
|
ctrl.request_talk_now()
|
|
should_continue() # simulate the loop noticing the flag
|
|
return False
|
|
|
|
monkeypatch.setattr(controller_mod.wake_word, "listen_for_wake_word", fake_listen)
|
|
ctrl._stream = object()
|
|
assert ctrl._wait_for_wake_or_click() is True
|
|
assert not ctrl._talk_now.is_set() # cleared after being consumed
|
|
|
|
|
|
def test_wait_for_wake_or_click_false_on_shutdown(monkeypatch, ctrl):
|
|
def fake_listen(stream, should_continue, on_tick=None, **kwargs):
|
|
ctrl.stop()
|
|
return False
|
|
monkeypatch.setattr(controller_mod.wake_word, "listen_for_wake_word", fake_listen)
|
|
ctrl._stream = object()
|
|
assert ctrl._wait_for_wake_or_click() is False
|
|
|
|
|
|
# ── heartbeat / proactive announcements ─────────────────────────────────────
|
|
|
|
def test_heartbeat_speaks_a_pending_announcement_when_idle(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "HEARTBEAT_INTERVAL_SECONDS", 0)
|
|
monkeypatch.setattr(controller_mod.server_client, "report_status", lambda: "don't forget your 3pm")
|
|
spoken = []
|
|
monkeypatch.setattr(controller_mod.tts, "speak",
|
|
lambda text, on_error=None, should_stop=None, voice_id=None: (spoken.append(text), True)[1])
|
|
said = _capture(ctrl.said)
|
|
|
|
ctrl._maybe_heartbeat()
|
|
|
|
assert spoken == ["don't forget your 3pm"]
|
|
assert said == ["don't forget your 3pm"]
|
|
assert ctrl._state.state == PetState.IDLE
|
|
|
|
|
|
def test_heartbeat_does_nothing_when_not_idle(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "HEARTBEAT_INTERVAL_SECONDS", 0)
|
|
ctrl._state.transition(PetState.LISTENING)
|
|
called = {"n": 0}
|
|
monkeypatch.setattr(controller_mod.server_client, "report_status",
|
|
lambda: called.__setitem__("n", called["n"] + 1))
|
|
|
|
ctrl._maybe_heartbeat()
|
|
|
|
assert called["n"] == 0
|
|
|
|
|
|
def test_heartbeat_respects_the_interval(monkeypatch, ctrl):
|
|
# Deterministic fake clock — time.monotonic()'s absolute value is
|
|
# arbitrary (often system uptime), so asserting behavior relative to it
|
|
# without controlling it would be flaky.
|
|
fake_now = {"t": 1000.0}
|
|
monkeypatch.setattr(controller_mod.time, "monotonic", lambda: fake_now["t"])
|
|
monkeypatch.setattr(controller_mod.config, "HEARTBEAT_INTERVAL_SECONDS", 60)
|
|
called = {"n": 0}
|
|
monkeypatch.setattr(controller_mod.server_client, "report_status",
|
|
lambda: called.__setitem__("n", called["n"] + 1))
|
|
|
|
ctrl._maybe_heartbeat() # last_heartbeat starts at 0.0 -> elapsed is huge -> runs
|
|
assert called["n"] == 1
|
|
|
|
fake_now["t"] += 10 # only 10s later — inside the 60s interval
|
|
ctrl._maybe_heartbeat()
|
|
assert called["n"] == 1 # skipped
|
|
|
|
fake_now["t"] += 60 # now well past the interval
|
|
ctrl._maybe_heartbeat()
|
|
assert called["n"] == 2
|