themajesticmagician
3a0959f55d
Latency: replies are spoken sentence-by-sentence off the desk API's NDJSON endpoint, so the wait is time-to-first-sentence rather than the whole model call, and Deepgram's live websocket transcribes while you're still talking instead of uploading the WAV afterwards. Both fall back invisibly — a stream that fails before anything was said drops to converse(), and a socket that never opens just means the old one-shot path. Speaking lived in four near-copies in the controller (a reply, a holding line, a streamed sentence, a dialogue scene) that had already drifted: one didn't arm barge-in, another skipped the follow-up rule. It's now speech.Speaker plus an Utterance describing the policy differences, with collaborators injected so the whole of it tests without Qt or audio. The mouth follows the audio rather than a timer: tts.level_of reduces each PCM frame to a 0..1 loudness on a sqrt curve (speech sits well below peak, and a linear map leaves the mouth barely open during normal talking) and that indexes the talking frames, which the sprite script now draws as an openness ramp. Offline pyttsx3 has no waveform, so stale levels hand control back to the timed loop instead of freezing the mouth mid-syllable. Also: the pet starts where you left it (ignoring positions on monitors that are no longer connected, since restoring those faithfully is how it ends up somewhere unreachable), and `python -m bolt_pet --doctor` is a preflight that says what to do about each problem rather than only what's wrong. tests/test_pipeline_smoke.py breaks the pure-logic rule on purpose. Every unit test passed all week while notifications sat unspoken for minutes, the pet said things twice and [laughing] got read aloud — each an interaction between two individually-correct units. It drives whole turns against a real HTTP server on a loopback port, faking only the mic and the speakers. It found a NameError in the paint path that would have fired on every repaint while talking. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
204 lines
8.1 KiB
Python
204 lines
8.1 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(autouse=True)
|
|
def _no_streaming(monkeypatch):
|
|
"""Most tests drive the plain request/response path; leaving streaming on
|
|
would have them attempt a real HTTP call, fail, and fall back — passing,
|
|
slowly, for the wrong reason. The streaming path has its own tests."""
|
|
monkeypatch.setattr(controller_mod.config, "STREAMING_REPLIES", 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, on_say=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, on_level=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, on_say=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, on_say=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, on_level=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
|