themajesticmagician
3ee67cb4d6
- Refactor `run_local_command` to manage subprocesses more effectively, ensuring child processes are terminated on timeout. - Introduce `_terminate` function to handle process group termination and capture output. - Implement `_command_output` to format command results with a character limit. - Add local intent recognition in `intents.py` to handle commands like "stop", "go to sleep", and "come here" without server interaction. - Normalize user input to match local intents while stripping filler words. - Update tests to cover new local intent functionality and ensure proper command handling. - Enhance speech processing to handle abbreviations and improve spoken output clarity.
1005 lines
42 KiB
Python
1005 lines
42 KiB
Python
"""Controller-level wiring for the newer behaviours: petctl routing,
|
|
barge-in follow-up, quiet hours, screen context, notification forwarding and
|
|
the live wake threshold.
|
|
|
|
Needs a QApplication (signals), so run with QT_QPA_PLATFORM=offscreen.
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from PySide6.QtWidgets import QApplication
|
|
|
|
from bolt_pet import controller as controller_mod
|
|
from bolt_pet.notifications import Notification
|
|
from bolt_pet.server_client import Reply
|
|
from bolt_pet.state import PetState
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
_app = QApplication.instance() or QApplication(["test"])
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def no_screen_probes(monkeypatch):
|
|
monkeypatch.setattr(controller_mod.screen_context, "is_fullscreen_active", lambda: False)
|
|
monkeypatch.setattr(controller_mod.screen_context, "context_for", lambda text: text)
|
|
|
|
|
|
@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
|
|
|
|
|
|
# ── petctl routing ──────────────────────────────────────────────────────────
|
|
|
|
def test_petctl_commands_never_reach_the_shell(monkeypatch, ctrl):
|
|
ran = []
|
|
monkeypatch.setattr(controller_mod.server_client, "run_local_command", lambda cmd: ran.append(cmd))
|
|
actions = _capture(ctrl.action)
|
|
|
|
output = ctrl._handle_command("petctl move top-left")
|
|
|
|
assert ran == []
|
|
assert actions == [{"action": "move", "anchor": "top-left"}]
|
|
assert "top-left" in output
|
|
|
|
|
|
def test_ordinary_commands_still_run_locally(monkeypatch, ctrl):
|
|
ran = []
|
|
monkeypatch.setattr(controller_mod.server_client, "run_local_command",
|
|
lambda cmd: ran.append(cmd) or "[exit 0]\n")
|
|
actions = _capture(ctrl.action)
|
|
|
|
ctrl._handle_command("df -h /")
|
|
|
|
assert ran == ["df -h /"]
|
|
assert actions == []
|
|
|
|
|
|
def test_bad_petctl_syntax_is_reported_back_not_executed(monkeypatch, ctrl):
|
|
ran = []
|
|
monkeypatch.setattr(controller_mod.server_client, "run_local_command", lambda cmd: ran.append(cmd))
|
|
output = ctrl._handle_command("petctl move sideways")
|
|
assert ran == []
|
|
assert "[pet]" in output
|
|
|
|
|
|
def test_petctl_nap_also_flips_the_controller_state(ctrl):
|
|
napping = _capture(ctrl.napping)
|
|
ctrl._handle_command("petctl nap on")
|
|
assert napping == [True]
|
|
assert ctrl._napping is True
|
|
|
|
|
|
# ── filectl routing ──────────────────────────────────────────────────────────
|
|
# filectl's wire format is a single-line JSON envelope (see file_ops.py's
|
|
# module docstring for why) — build commands with json.dumps so the tests
|
|
# don't hardcode escaping by hand.
|
|
|
|
def _filectl(payload: dict) -> str:
|
|
return "filectl " + json.dumps(payload)
|
|
|
|
|
|
def test_filectl_commands_never_reach_the_shell(monkeypatch, ctrl, tmp_path):
|
|
ran = []
|
|
monkeypatch.setattr(controller_mod.server_client, "run_local_command", lambda cmd: ran.append(cmd))
|
|
target = tmp_path / "a.txt"
|
|
|
|
output = ctrl._handle_command(_filectl({"op": "write", "path": str(target), "content": "hello"}))
|
|
|
|
assert ran == []
|
|
assert target.read_text() == "hello"
|
|
assert str(target) in output
|
|
|
|
|
|
def test_filectl_edit_round_trips_through_the_relay(monkeypatch, ctrl, tmp_path):
|
|
monkeypatch.setattr(controller_mod.server_client, "run_local_command",
|
|
lambda cmd: (_ for _ in ()).throw(AssertionError("should not shell out")))
|
|
target = tmp_path / "a.py"
|
|
target.write_text("x = 1\n")
|
|
|
|
output = ctrl._handle_command(
|
|
_filectl({"op": "edit", "path": str(target), "old": "x = 1", "new": "x = 2"})
|
|
)
|
|
|
|
assert target.read_text() == "x = 2\n"
|
|
assert str(target) in output
|
|
|
|
|
|
def test_bad_filectl_syntax_is_reported_back_not_executed(monkeypatch, ctrl):
|
|
ran = []
|
|
monkeypatch.setattr(controller_mod.server_client, "run_local_command", lambda cmd: ran.append(cmd))
|
|
output = ctrl._handle_command("filectl {not valid json")
|
|
assert ran == []
|
|
assert "[filectl]" in output
|
|
|
|
|
|
def test_filectl_execution_failure_is_reported_back_not_raised(monkeypatch, ctrl):
|
|
output = ctrl._handle_command(_filectl({"op": "read", "path": "/no/such/file.txt"}))
|
|
assert "[filectl]" in output
|
|
|
|
|
|
def test_ordinary_commands_still_run_locally_alongside_filectl(monkeypatch, ctrl):
|
|
ran = []
|
|
monkeypatch.setattr(controller_mod.server_client, "run_local_command",
|
|
lambda cmd: ran.append(cmd) or "[exit 0]\n")
|
|
ctrl._handle_command("df -h /")
|
|
assert ran == ["df -h /"]
|
|
|
|
|
|
# ── barge-in ────────────────────────────────────────────────────────────────
|
|
|
|
def test_the_interrupt_log_reports_what_fired_not_the_reset_counters(monkeypatch, ctrl):
|
|
"""_speak resets the detector after playback so the pet's own voice
|
|
doesn't linger in the wake model's window. Reading the stats after that
|
|
reset reports 0.000 at frame 0 for every interruption, which is worse
|
|
than no instrumentation — it looks like hard evidence and isn't."""
|
|
from bolt_pet.audio import barge_in as barge_in_mod
|
|
|
|
class Detector(barge_in_mod.WakeWordBargeIn):
|
|
def __init__(self):
|
|
self._frames, self._peak, self._last, self._last_threshold = 0, 0.0, 0.0, 0.5
|
|
self._frame_len = 1280
|
|
self.reset_calls = 0
|
|
|
|
def reset(self):
|
|
self.reset_calls += 1
|
|
self._frames, self._peak, self._last = 0, 0.0, 0.0
|
|
|
|
detector = Detector()
|
|
ctrl._barge_in = detector
|
|
logs = _capture(ctrl.log)
|
|
|
|
def interrupted_playback(text, on_error=None, should_stop=None, voice_id=None):
|
|
# What really happens: frames get scored during playback, then one
|
|
# clears the threshold and playback aborts.
|
|
detector._frames, detector._peak, detector._last = 7, 0.81, 0.81
|
|
return False
|
|
|
|
monkeypatch.setattr(controller_mod.tts, "speak", interrupted_playback)
|
|
|
|
ctrl._speak("a very long explanation")
|
|
|
|
interrupted = next(m for m in logs if "Interrupted" in m)
|
|
assert "0.810" in interrupted and "frame 7" in interrupted
|
|
# Once before playback (clear the window) and once after (drop the pet's
|
|
# own voice) — the point is that the *read* happens between them.
|
|
assert detector.reset_calls == 2
|
|
|
|
|
|
def test_interrupted_playback_queues_an_immediate_next_turn(monkeypatch, ctrl):
|
|
logs = _capture(ctrl.log)
|
|
monkeypatch.setattr(controller_mod.tts, "speak",
|
|
lambda text, on_error=None, should_stop=None, voice_id=None: False) # interrupted
|
|
|
|
ctrl._speak("a very long explanation")
|
|
|
|
assert ctrl._talk_now.is_set() # loop picks the next turn up without a wake word
|
|
assert any("Interrupted" in message for message in logs)
|
|
|
|
|
|
def test_uninterrupted_playback_does_not_queue_a_turn(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.tts, "speak",
|
|
lambda text, on_error=None, should_stop=None, voice_id=None: True)
|
|
ctrl._speak("short answer")
|
|
assert not ctrl._talk_now.is_set()
|
|
|
|
|
|
# ── follow-up listening ─────────────────────────────────────────────────────
|
|
|
|
@pytest.fixture
|
|
def spoke(monkeypatch):
|
|
"""Playback that always completes, so only the follow-up rule decides
|
|
whether another turn is queued."""
|
|
monkeypatch.setattr(controller_mod.tts, "speak",
|
|
lambda text, on_error=None, should_stop=None, voice_id=None: True)
|
|
|
|
|
|
def test_a_reply_ending_in_a_question_keeps_listening(spoke, ctrl):
|
|
logs = _capture(ctrl.log)
|
|
|
|
ctrl._speak("You're still in ~/Documents/bolt-pet. Ready to run a command?")
|
|
|
|
assert ctrl._talk_now.is_set() # no wake word needed for the answer
|
|
assert ctrl._pending_follow_up # and the next turn knows it's an answer
|
|
assert any("listening for your answer" in message for message in logs)
|
|
|
|
|
|
def test_a_statement_does_not_keep_listening(spoke, ctrl):
|
|
ctrl._speak("It's 7:15 AM on July 23, 2026.")
|
|
assert not ctrl._talk_now.is_set()
|
|
assert not ctrl._pending_follow_up
|
|
|
|
|
|
def test_a_question_anywhere_in_the_reply_keeps_listening(spoke, ctrl):
|
|
"""Bolt often asks and then keeps talking ("Want me to fix it? I'd start
|
|
with the config."), so the question mark doesn't have to be last. An
|
|
unwanted extra listen ends itself on VAD_GRACE_SECONDS of silence; a missed
|
|
one costs you a wake word, which is the more expensive mistake."""
|
|
ctrl._speak("Want me to restart it? It's been up for 40 days.")
|
|
assert ctrl._talk_now.is_set()
|
|
|
|
|
|
def test_follow_ups_stop_at_the_cap(spoke, monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "FOLLOW_UP_MAX_TURNS", 2)
|
|
logs = _capture(ctrl.log)
|
|
|
|
for _ in range(2):
|
|
ctrl._speak("Want me to keep going?")
|
|
ctrl._talk_now.clear()
|
|
assert ctrl._follow_ups == 2
|
|
|
|
ctrl._speak("Want me to keep going?")
|
|
|
|
assert not ctrl._talk_now.is_set() # chain broken until you re-trigger it
|
|
assert any("Follow-up limit reached" in message for message in logs)
|
|
|
|
|
|
def test_the_cap_is_not_announced_on_ordinary_replies(spoke, monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "FOLLOW_UP_MAX_TURNS", 1)
|
|
ctrl._follow_ups = 1
|
|
logs = _capture(ctrl.log)
|
|
|
|
ctrl._speak("Done — the file is saved.")
|
|
|
|
assert not any("Follow-up limit" in message for message in logs)
|
|
|
|
|
|
def test_starting_a_turn_yourself_resets_the_chain(spoke, monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.mic, "record_utterance",
|
|
lambda *a, **kw: None) # you said nothing
|
|
ctrl._follow_ups = 3
|
|
|
|
ctrl._handle_conversation_turn()
|
|
|
|
assert ctrl._follow_ups == 0
|
|
|
|
|
|
def test_an_answered_question_gets_a_longer_grace_period(spoke, monkeypatch, ctrl):
|
|
"""You were just asked something — you get longer to think than when you
|
|
deliberately said the wake word."""
|
|
grace = []
|
|
monkeypatch.setattr(controller_mod.mic, "record_utterance",
|
|
lambda *a, **kw: grace.append(kw.get("grace_s")) or None)
|
|
|
|
ctrl._handle_conversation_turn() # you started this one
|
|
ctrl._pending_follow_up = True
|
|
ctrl._handle_conversation_turn() # this one answers a question
|
|
|
|
assert grace == [None, controller_mod.config.FOLLOW_UP_GRACE_SECONDS]
|
|
|
|
|
|
def test_muting_stops_follow_ups(spoke, ctrl):
|
|
ctrl._muted = True
|
|
ctrl._speak("Shall I continue?")
|
|
assert not ctrl._talk_now.is_set() # mute means don't listen, question or not
|
|
|
|
|
|
def test_follow_up_can_be_turned_off(spoke, monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "FOLLOW_UP_LISTEN", False)
|
|
ctrl._speak("Shall I continue?")
|
|
assert not ctrl._talk_now.is_set()
|
|
|
|
|
|
def test_being_interrupted_restarts_the_chain(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.tts, "speak",
|
|
lambda text, on_error=None, should_stop=None, voice_id=None: False)
|
|
ctrl._follow_ups = 3
|
|
|
|
ctrl._speak("a very long explanation")
|
|
|
|
assert ctrl._follow_ups == 0 # you're clearly engaged
|
|
assert ctrl._talk_now.is_set()
|
|
|
|
|
|
def test_speech_is_recorded_in_the_history(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.tts, "speak",
|
|
lambda text, on_error=None, should_stop=None, voice_id=None: True)
|
|
ctrl._speak("**bold** reply")
|
|
assert ctrl.history.last().text == "**bold** reply" # raw, for copy/paste
|
|
|
|
|
|
# ── screen context ──────────────────────────────────────────────────────────
|
|
|
|
def test_the_active_window_rides_along_with_the_utterance(monkeypatch, ctrl):
|
|
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 this error?")
|
|
monkeypatch.setattr(controller_mod.screen_context, "context_for",
|
|
lambda text: f"{text}\n\n[on screen right now: app.py]")
|
|
monkeypatch.setattr(controller_mod.tts, "speak",
|
|
lambda text, on_error=None, should_stop=None, voice_id=None: True)
|
|
sent = []
|
|
monkeypatch.setattr(controller_mod.server_client, "converse",
|
|
lambda text, on_command=None: sent.append(text) or Reply("that's a KeyError"))
|
|
|
|
ctrl._handle_conversation_turn()
|
|
|
|
assert "[on screen right now: app.py]" in sent[0]
|
|
# ...but the *history* keeps what you actually said, not the annotation.
|
|
assert ctrl.history.entries()[0].text == "what's this error?"
|
|
|
|
|
|
# ── quiet hours / do-not-disturb ────────────────────────────────────────────
|
|
|
|
def test_quiet_hours_suppress_the_heartbeat(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "HEARTBEAT_INTERVAL_SECONDS", 0)
|
|
monkeypatch.setattr(controller_mod.config, "QUIET_HOURS", "00:00-23:59")
|
|
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
|
|
assert ctrl._napping is True
|
|
|
|
|
|
def test_fullscreen_triggers_do_not_disturb(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "QUIET_HOURS", "")
|
|
monkeypatch.setattr(controller_mod.config, "DND_ON_FULLSCREEN", True)
|
|
monkeypatch.setattr(controller_mod.screen_context, "is_fullscreen_active", lambda: True)
|
|
napping = _capture(ctrl.napping)
|
|
|
|
ctrl._refresh_nap_state()
|
|
|
|
assert napping == [True]
|
|
|
|
|
|
def test_a_manual_nap_overrides_the_schedule(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "QUIET_HOURS", "")
|
|
ctrl.set_napping(True)
|
|
ctrl._last_nap_check = 0.0
|
|
ctrl._refresh_nap_state()
|
|
assert ctrl._napping is True # the schedule doesn't wake it back up
|
|
|
|
ctrl.set_napping(None) # back on schedule
|
|
ctrl._last_nap_check = 0.0
|
|
ctrl._refresh_nap_state()
|
|
assert ctrl._napping is False
|
|
|
|
|
|
def test_napping_still_answers_when_spoken_to(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.mic, "record_utterance",
|
|
lambda *a, **k: np.zeros(10, dtype=np.int16))
|
|
monkeypatch.setattr(controller_mod.stt, "transcribe", lambda pcm: "you awake?")
|
|
monkeypatch.setattr(controller_mod.server_client, "converse",
|
|
lambda text, on_command=None: Reply("always"))
|
|
spoken = []
|
|
monkeypatch.setattr(controller_mod.tts, "speak",
|
|
lambda text, on_error=None, should_stop=None, voice_id=None: spoken.append(text) or True)
|
|
ctrl.set_napping(True)
|
|
|
|
ctrl._handle_conversation_turn()
|
|
|
|
assert spoken == ["always"]
|
|
|
|
|
|
# ── local intents ───────────────────────────────────────────────────────────
|
|
|
|
@pytest.fixture
|
|
def heard(monkeypatch):
|
|
"""A turn where you said something, with the server and TTS recorded.
|
|
Returns (utterance_setter, sent, spoken)."""
|
|
said = {"text": ""}
|
|
sent, spoken = [], []
|
|
monkeypatch.setattr(controller_mod.mic, "record_utterance",
|
|
lambda *a, **k: np.zeros(10, dtype=np.int16))
|
|
monkeypatch.setattr(controller_mod.stt, "transcribe", lambda pcm: said["text"])
|
|
monkeypatch.setattr(controller_mod.server_client, "converse",
|
|
lambda text, on_command=None: sent.append(text) or Reply("from the server"))
|
|
monkeypatch.setattr(controller_mod.tts, "speak",
|
|
lambda text, on_error=None, should_stop=None, voice_id=None:
|
|
spoken.append(text) or True)
|
|
return said, sent, spoken
|
|
|
|
|
|
def test_a_local_intent_never_reaches_the_server(heard, ctrl):
|
|
said, sent, spoken = heard
|
|
said["text"] = "come here"
|
|
actions = _capture(ctrl.action)
|
|
|
|
ctrl._handle_conversation_turn()
|
|
|
|
assert sent == [] # no round trip at all
|
|
assert actions == [{"action": "move", "anchor": "cursor"}]
|
|
assert spoken == [] # walking over is the reply
|
|
assert ctrl._state.state == PetState.IDLE
|
|
|
|
|
|
def test_stop_is_answered_with_silence(heard, ctrl):
|
|
said, sent, spoken = heard
|
|
said["text"] = "be quiet"
|
|
ctrl._handle_conversation_turn()
|
|
assert (sent, spoken) == ([], [])
|
|
|
|
|
|
def test_a_request_that_merely_starts_with_an_intent_word_goes_to_the_server(heard, ctrl):
|
|
said, sent, spoken = heard
|
|
said["text"] = "stop the docker container"
|
|
ctrl._handle_conversation_turn()
|
|
assert sent and "stop the docker container" in sent[0]
|
|
assert spoken == ["from the server"]
|
|
|
|
|
|
def test_local_intents_are_skipped_while_answering_a_question(heard, ctrl):
|
|
"""Bolt asked something; "never mind" is an answer to him, not a body
|
|
command. Swallowing it locally would leave the server holding a question it
|
|
never got a reply to."""
|
|
said, sent, spoken = heard
|
|
said["text"] = "never mind"
|
|
ctrl._pending_follow_up = True
|
|
|
|
ctrl._handle_conversation_turn()
|
|
|
|
assert sent and "never mind" in sent[0]
|
|
|
|
|
|
def test_local_intents_can_be_turned_off(monkeypatch, heard, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "LOCAL_INTENTS", False)
|
|
said, sent, spoken = heard
|
|
said["text"] = "come here"
|
|
ctrl._handle_conversation_turn()
|
|
assert sent and "come here" in sent[0]
|
|
|
|
|
|
def test_say_that_again_replays_the_last_line_without_duplicating_history(heard, ctrl):
|
|
said, sent, spoken = heard
|
|
ctrl.history.add(controller_mod.history_mod.PET, "it's 7:15 AM", 0.0)
|
|
said["text"] = "what did you say?"
|
|
|
|
ctrl._handle_conversation_turn()
|
|
|
|
assert spoken == ["it's 7:15 AM"]
|
|
assert sent == []
|
|
pet_lines = [e.text for e in ctrl.history.entries()
|
|
if e.role == controller_mod.history_mod.PET]
|
|
assert pet_lines == ["it's 7:15 AM"] # replayed, not re-recorded
|
|
|
|
|
|
def test_repeat_with_nothing_to_repeat_says_so(heard, ctrl):
|
|
said, sent, spoken = heard
|
|
said["text"] = "say that again"
|
|
ctrl._handle_conversation_turn()
|
|
assert spoken == ["I haven't said anything yet."]
|
|
|
|
|
|
def test_going_back_to_the_normal_voice_needs_no_server_prompt_support(heard, ctrl):
|
|
"""The server can only offer `petctl voice reset` if its prompt happens to
|
|
advertise the verb; recognising the phrase here works regardless."""
|
|
said, sent, spoken = heard
|
|
ctrl._voice_id, ctrl._voice_name = "voice-123", "Brian"
|
|
changed = _capture(ctrl.voice_changed)
|
|
said["text"] = "go back to your normal voice"
|
|
|
|
ctrl._handle_conversation_turn()
|
|
|
|
assert ctrl._voice_id == ""
|
|
assert changed == [""]
|
|
assert spoken == ["Back to my own voice."]
|
|
assert sent == []
|
|
|
|
|
|
def test_go_to_sleep_overrides_the_quiet_hours_schedule(heard, ctrl):
|
|
said, sent, spoken = heard
|
|
said["text"] = "go to sleep"
|
|
|
|
ctrl._handle_conversation_turn()
|
|
|
|
assert ctrl._napping is True
|
|
assert ctrl._nap_forced is True # not undone by the next schedule check
|
|
assert spoken == ["Night."]
|
|
|
|
|
|
# ── failure containment ─────────────────────────────────────────────────────
|
|
|
|
def test_one_bad_turn_does_not_end_the_session(monkeypatch, ctrl):
|
|
"""A turn raising something unforeseen used to unwind _loop and kill the
|
|
thread — the pet would go deaf until it was restarted by hand."""
|
|
logs = _capture(ctrl.log)
|
|
|
|
def explode():
|
|
raise RuntimeError("numpy said no")
|
|
|
|
assert ctrl._guarded(explode, "conversation turn") is False
|
|
assert ctrl._state.state == PetState.IDLE
|
|
assert any("Recovered from a conversation turn failure" in m for m in logs)
|
|
|
|
|
|
def test_a_command_handler_crash_is_reported_up_the_relay(monkeypatch, ctrl):
|
|
"""The server is blocked on /desk/tool_result while this runs. Raising would
|
|
leave it waiting out its own timeout on a turn that can never finish."""
|
|
monkeypatch.setattr(controller_mod.pet_actions, "parse",
|
|
lambda command: (_ for _ in ()).throw(KeyError("boom")))
|
|
|
|
output = ctrl._handle_command("petctl move top-left")
|
|
|
|
assert output.startswith("[error]") and "boom" in output
|
|
|
|
|
|
# ── notification bridge ─────────────────────────────────────────────────────
|
|
|
|
def test_notifications_are_forwarded_and_spoken(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "NOTIFICATION_MIN_INTERVAL_SECONDS", 0)
|
|
ctrl._notification_gate = controller_mod.notifications.NotificationGate("", 0)
|
|
sent, spoken = [], []
|
|
monkeypatch.setattr(controller_mod.server_client, "converse",
|
|
lambda text, on_command=None: sent.append(text) or Reply("your build is green"))
|
|
monkeypatch.setattr(controller_mod.tts, "speak",
|
|
lambda text, on_error=None, should_stop=None, voice_id=None: spoken.append(text) or True)
|
|
|
|
ctrl._queue_notification(Notification(app="CI", summary="Build finished", body=""))
|
|
ctrl._drain_notifications()
|
|
|
|
assert "CI: Build finished" in sent[0]
|
|
assert spoken == ["your build is green"]
|
|
assert ctrl._state.state == PetState.IDLE
|
|
|
|
|
|
def test_filtered_out_notifications_are_never_queued(ctrl):
|
|
ctrl._notification_gate = controller_mod.notifications.NotificationGate("deploy", 0)
|
|
ctrl._queue_notification(Notification(app="Chat", summary="lunch?", body=""))
|
|
assert not ctrl._pending_notifications
|
|
|
|
|
|
def test_the_notification_queue_is_bounded(monkeypatch, ctrl):
|
|
"""An overnight nap can't grow the queue without limit — the drain only runs
|
|
from the heartbeat, and the heartbeat doesn't run while napping."""
|
|
monkeypatch.setattr(controller_mod.config, "NOTIFICATION_QUEUE_LIMIT", 3)
|
|
ctrl = controller_mod.PetController()
|
|
ctrl._notification_gate = controller_mod.notifications.NotificationGate("", 0)
|
|
|
|
for index in range(10):
|
|
ctrl._queue_notification(Notification(app="CI", summary=f"build {index}", body=""))
|
|
|
|
queued = [notification.summary for _stamp, notification in ctrl._pending_notifications]
|
|
assert queued == ["build 7", "build 8", "build 9"] # oldest dropped
|
|
|
|
|
|
def test_stale_notifications_are_dropped_instead_of_read_out(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "NOTIFICATION_MAX_AGE_SECONDS", 900)
|
|
sent = []
|
|
monkeypatch.setattr(controller_mod.server_client, "converse",
|
|
lambda text, on_command=None: sent.append(text) or Reply(""))
|
|
ctrl._notification_gate = controller_mod.notifications.NotificationGate("", 0)
|
|
ctrl._queue_notification(Notification(app="CI", summary="fresh", body=""))
|
|
# Backdate it past the age limit, as an overnight backlog would be.
|
|
stamp, notification = ctrl._pending_notifications.pop()
|
|
ctrl._pending_notifications.append((stamp - 4000, notification))
|
|
|
|
ctrl._drain_notifications()
|
|
|
|
assert sent == []
|
|
|
|
|
|
def test_a_nap_starting_mid_drain_keeps_the_rest_queued(monkeypatch, ctrl):
|
|
"""The old code swapped the queue out and returned, losing the remainder."""
|
|
def converse(text, on_command=None):
|
|
ctrl._napping = True # e.g. quiet hours began, or a fullscreen app opened
|
|
return Reply("")
|
|
|
|
monkeypatch.setattr(controller_mod.server_client, "converse", converse)
|
|
ctrl._notification_gate = controller_mod.notifications.NotificationGate("", 0)
|
|
for index in range(3):
|
|
ctrl._queue_notification(Notification(app="CI", summary=f"build {index}", body=""))
|
|
|
|
ctrl._drain_notifications()
|
|
|
|
remaining = [notification.summary for _stamp, notification in ctrl._pending_notifications]
|
|
assert remaining == ["build 1", "build 2"]
|
|
|
|
|
|
def test_notifications_are_not_forwarded_while_napping(monkeypatch, ctrl):
|
|
ctrl._notification_gate = controller_mod.notifications.NotificationGate("", 0)
|
|
called = {"n": 0}
|
|
monkeypatch.setattr(controller_mod.server_client, "converse",
|
|
lambda text, on_command=None: called.__setitem__("n", called["n"] + 1))
|
|
ctrl.set_napping(True)
|
|
|
|
ctrl._queue_notification(Notification(app="CI", summary="Build finished", body=""))
|
|
ctrl._drain_notifications()
|
|
|
|
assert called["n"] == 0
|
|
|
|
|
|
# ── wake threshold ──────────────────────────────────────────────────────────
|
|
|
|
def test_threshold_is_live_and_clamped(ctrl):
|
|
ctrl.set_wake_threshold(0.72)
|
|
assert ctrl.wake_threshold() == pytest.approx(0.72)
|
|
ctrl.set_wake_threshold(5)
|
|
assert ctrl.wake_threshold() == pytest.approx(0.99)
|
|
ctrl.set_wake_threshold(-1)
|
|
assert ctrl.wake_threshold() == pytest.approx(0.01)
|
|
|
|
|
|
def test_near_misses_are_recorded_for_the_tuner(ctrl):
|
|
ctrl.set_wake_threshold(0.5)
|
|
ctrl._observe_wake_score(0.42, 0.5) # near miss
|
|
ctrl._observe_wake_score(0.01, 0.5) # background noise, not interesting
|
|
stats = ctrl.wake_stats()
|
|
assert len(stats["near_misses"]) == 1
|
|
assert stats["peak"] == pytest.approx(0.42)
|
|
|
|
ctrl.reset_wake_stats()
|
|
assert ctrl.wake_stats()["near_misses"] == []
|
|
|
|
|
|
# ── file delivery ────────────────────────────────────────────────────────────
|
|
|
|
def test_check_deliveries_downloads_and_saves_queued_files(monkeypatch, ctrl, tmp_path):
|
|
monkeypatch.setattr(controller_mod.config, "DELIVERED_FILES_DIR", tmp_path)
|
|
monkeypatch.setattr(controller_mod.server_client, "list_outbox_files",
|
|
lambda: [{"id": "abc", "name": "report.pdf", "size": 5}])
|
|
monkeypatch.setattr(controller_mod.server_client, "download_outbox_file",
|
|
lambda file_id: b"hello" if file_id == "abc" else b"")
|
|
|
|
ctrl._check_deliveries()
|
|
|
|
assert (tmp_path / "report.pdf").read_bytes() == b"hello"
|
|
|
|
|
|
def test_check_deliveries_is_a_noop_when_nothing_queued(monkeypatch, ctrl, tmp_path):
|
|
monkeypatch.setattr(controller_mod.config, "DELIVERED_FILES_DIR", tmp_path)
|
|
monkeypatch.setattr(controller_mod.server_client, "list_outbox_files", lambda: [])
|
|
downloaded = []
|
|
monkeypatch.setattr(controller_mod.server_client, "download_outbox_file",
|
|
lambda file_id: downloaded.append(file_id))
|
|
|
|
ctrl._check_deliveries()
|
|
|
|
assert downloaded == []
|
|
|
|
|
|
def test_check_deliveries_can_be_disabled(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "RECEIVE_FILES", False)
|
|
called = {"n": 0}
|
|
monkeypatch.setattr(controller_mod.server_client, "list_outbox_files",
|
|
lambda: called.__setitem__("n", called["n"] + 1))
|
|
|
|
ctrl._check_deliveries()
|
|
|
|
assert called["n"] == 0
|
|
|
|
|
|
def test_check_deliveries_logs_and_continues_on_download_failure(monkeypatch, ctrl, tmp_path):
|
|
monkeypatch.setattr(controller_mod.config, "DELIVERED_FILES_DIR", tmp_path)
|
|
monkeypatch.setattr(controller_mod.server_client, "list_outbox_files", lambda: [
|
|
{"id": "bad", "name": "a.txt", "size": 1},
|
|
{"id": "good", "name": "b.txt", "size": 1},
|
|
])
|
|
|
|
def fake_download(file_id):
|
|
if file_id == "bad":
|
|
raise controller_mod.server_client.ServerError("gone")
|
|
return b"ok"
|
|
|
|
monkeypatch.setattr(controller_mod.server_client, "download_outbox_file", fake_download)
|
|
logs = _capture(ctrl.log)
|
|
|
|
ctrl._check_deliveries()
|
|
|
|
assert (tmp_path / "b.txt").read_bytes() == b"ok"
|
|
assert not (tmp_path / "a.txt").exists()
|
|
assert any("bad" in msg or "a.txt" in msg for msg in logs)
|
|
|
|
|
|
def test_check_deliveries_runs_after_a_conversation_turn(monkeypatch, ctrl, tmp_path):
|
|
monkeypatch.setattr(controller_mod.mic, "record_utterance",
|
|
lambda *a, **k: np.zeros(10, dtype=np.int16))
|
|
monkeypatch.setattr(controller_mod.stt, "transcribe", lambda pcm: "send me that file")
|
|
monkeypatch.setattr(controller_mod.server_client, "converse",
|
|
lambda text, on_command=None: Reply("it's on the way"))
|
|
monkeypatch.setattr(controller_mod.tts, "speak",
|
|
lambda text, on_error=None, should_stop=None, voice_id=None: True)
|
|
monkeypatch.setattr(controller_mod.config, "DELIVERED_FILES_DIR", tmp_path)
|
|
monkeypatch.setattr(controller_mod.server_client, "list_outbox_files",
|
|
lambda: [{"id": "abc", "name": "notes.txt", "size": 2}])
|
|
monkeypatch.setattr(controller_mod.server_client, "download_outbox_file",
|
|
lambda file_id: b"hi")
|
|
|
|
ctrl._handle_conversation_turn()
|
|
|
|
assert (tmp_path / "notes.txt").read_bytes() == b"hi"
|
|
|
|
|
|
# ── server-picked voice (the desk API's speak_as marker) ────────────────────
|
|
|
|
def _voice_turn(monkeypatch, ctrl, reply, said="talk like a pirate"):
|
|
"""Run one full conversation turn whose reply is *reply*, returning the
|
|
voice_id each tts.speak() call was given."""
|
|
voices = []
|
|
monkeypatch.setattr(controller_mod.mic, "record_utterance",
|
|
lambda *a, **k: np.zeros(10, dtype=np.int16))
|
|
monkeypatch.setattr(controller_mod.stt, "transcribe", lambda pcm: said)
|
|
monkeypatch.setattr(controller_mod.server_client, "converse",
|
|
lambda text, on_command=None: reply)
|
|
monkeypatch.setattr(
|
|
controller_mod.tts, "speak",
|
|
lambda text, on_error=None, should_stop=None, voice_id=None:
|
|
voices.append(voice_id) or True,
|
|
)
|
|
ctrl._handle_conversation_turn()
|
|
return voices
|
|
|
|
|
|
def test_a_speak_as_reply_is_spoken_in_that_voice(monkeypatch, ctrl):
|
|
voices = _voice_turn(monkeypatch, ctrl, Reply("Ahoy.", "VOICE1", "Terence"))
|
|
assert voices == ["VOICE1"]
|
|
assert ctrl.current_voice() == "Terence"
|
|
|
|
|
|
def test_the_picked_voice_sticks_for_later_replies(monkeypatch, ctrl):
|
|
"""The server tags one reply and doesn't keep the id in its history, so
|
|
it can't re-request the voice when you say "keep talking like that"."""
|
|
monkeypatch.setattr(controller_mod.config, "VOICE_STICKY", True)
|
|
_voice_turn(monkeypatch, ctrl, Reply("Ahoy.", "VOICE1", "Terence"))
|
|
voices = _voice_turn(monkeypatch, ctrl, Reply("Still me."), said="and now?")
|
|
assert voices == ["VOICE1"]
|
|
|
|
|
|
def test_voice_stickiness_can_be_turned_off(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "VOICE_STICKY", False)
|
|
_voice_turn(monkeypatch, ctrl, Reply("Ahoy.", "VOICE1", "Terence"))
|
|
voices = _voice_turn(monkeypatch, ctrl, Reply("Back to normal."), said="and now?")
|
|
assert voices == [None]
|
|
assert ctrl.current_voice() == ""
|
|
|
|
|
|
def test_a_new_pick_replaces_the_old_one(monkeypatch, ctrl):
|
|
changes = _capture(ctrl.voice_changed)
|
|
_voice_turn(monkeypatch, ctrl, Reply("Ahoy.", "VOICE1", "Terence"))
|
|
voices = _voice_turn(monkeypatch, ctrl, Reply("こんにちは。", "VOICE2", "Asahi"),
|
|
said="say that in Japanese")
|
|
assert voices == ["VOICE2"]
|
|
assert changes == ["Terence", "Asahi"]
|
|
|
|
|
|
def test_resetting_the_voice_goes_back_to_the_default(monkeypatch, ctrl):
|
|
_voice_turn(monkeypatch, ctrl, Reply("Ahoy.", "VOICE1", "Terence"))
|
|
changes = _capture(ctrl.voice_changed)
|
|
|
|
ctrl.reset_voice()
|
|
|
|
assert ctrl.current_voice() == ""
|
|
assert changes == [""] # the tray's menu entry follows this signal
|
|
voices = _voice_turn(monkeypatch, ctrl, Reply("Normal again."), said="hi")
|
|
assert voices == [None]
|
|
|
|
|
|
def test_an_unnamed_voice_still_reports_something_resettable(monkeypatch, ctrl):
|
|
"""voice_name is optional server-side — falling back to the id keeps the
|
|
tray entry from reading "now: " with nothing after it."""
|
|
_voice_turn(monkeypatch, ctrl, Reply("Ahoy.", "VOICE1"))
|
|
assert ctrl.current_voice() == "VOICE1"
|
|
|
|
|
|
def test_petctl_voice_reset_returns_bolt_to_his_own_voice(monkeypatch, ctrl):
|
|
"""The server can pick a voice but can't ask for the default back — it
|
|
was never told what Bolt's own voice id is. This is how it asks."""
|
|
ran = []
|
|
monkeypatch.setattr(controller_mod.server_client, "run_local_command", lambda cmd: ran.append(cmd))
|
|
_voice_turn(monkeypatch, ctrl, Reply("Ahoy.", "VOICE1", "Terence"))
|
|
|
|
output = ctrl._handle_command("petctl voice reset")
|
|
|
|
assert ran == [] # never reaches a shell, like every other petctl verb
|
|
assert "Terence" in output # the server can't see the voice; tell it what changed
|
|
assert ctrl.current_voice() == ""
|
|
assert _voice_turn(monkeypatch, ctrl, Reply("Normal again."), said="hi") == [None]
|
|
|
|
|
|
def test_petctl_voice_reset_says_so_when_there_was_nothing_to_reset(ctrl):
|
|
assert "already" in ctrl._handle_command("petctl voice reset")
|
|
|
|
|
|
# ── dialoguectl (multi-voice scenes) ────────────────────────────────────────
|
|
|
|
def _dialogue_command(*lines):
|
|
import json
|
|
return "dialoguectl " + json.dumps({"lines": list(lines)})
|
|
|
|
|
|
def test_dialoguectl_never_reaches_the_shell(monkeypatch, ctrl):
|
|
ran = []
|
|
monkeypatch.setattr(controller_mod.server_client, "run_local_command", lambda cmd: ran.append(cmd))
|
|
monkeypatch.setattr(controller_mod.config, "ELEVENLABS_VOICE_ID", "aaorr6ZHIL88gEexu7dC")
|
|
monkeypatch.setattr(controller_mod.tts, "synthesize_dialogue",
|
|
lambda inputs, model_id=None, stability=None: (np.zeros(4, dtype=np.int16), 24000))
|
|
monkeypatch.setattr(controller_mod.tts, "play_pcm",
|
|
lambda pcm, rate, should_stop=None: True)
|
|
|
|
output = ctrl._handle_command(_dialogue_command({"voice": "self", "text": "[cheerfully] hi"}))
|
|
|
|
assert ran == []
|
|
assert "[dialogue] played 1 line" in output
|
|
|
|
|
|
def test_a_scene_shows_in_the_bubble_with_the_delivery_tags_stripped(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "ELEVENLABS_VOICE_ID", "aaorr6ZHIL88gEexu7dC")
|
|
monkeypatch.setattr(controller_mod.config, "DIALOGUE_VOICES", "narrator:9BWtsMINqrJLrRacOk9x")
|
|
monkeypatch.setattr(controller_mod.tts, "synthesize_dialogue",
|
|
lambda inputs, model_id=None, stability=None: (np.zeros(4, dtype=np.int16), 24000))
|
|
monkeypatch.setattr(controller_mod.tts, "play_pcm", lambda pcm, rate, should_stop=None: True)
|
|
said = _capture(ctrl.said)
|
|
|
|
ctrl._handle_command(_dialogue_command(
|
|
{"voice": "self", "text": "[cheerfully] Hello there!"},
|
|
{"voice": "narrator", "text": "[whispering] He is lying."},
|
|
))
|
|
|
|
assert said == ["Hello there! He is lying."]
|
|
assert ctrl.history.last().text == "Hello there! He is lying."
|
|
|
|
|
|
def test_a_mid_turn_scene_returns_to_thinking_not_idle(monkeypatch, ctrl):
|
|
"""The server is still waiting on the tool result, so the pet talks and
|
|
goes back to waiting — dropping to IDLE would look like the turn ended."""
|
|
monkeypatch.setattr(controller_mod.config, "ELEVENLABS_VOICE_ID", "aaorr6ZHIL88gEexu7dC")
|
|
monkeypatch.setattr(controller_mod.tts, "synthesize_dialogue",
|
|
lambda inputs, model_id=None, stability=None: (np.zeros(4, dtype=np.int16), 24000))
|
|
monkeypatch.setattr(controller_mod.tts, "play_pcm", lambda pcm, rate, should_stop=None: True)
|
|
ctrl._state.transition(PetState.LISTENING)
|
|
ctrl._state.transition(PetState.THINKING)
|
|
states = _capture(ctrl.state_changed)
|
|
|
|
ctrl._handle_command(_dialogue_command({"voice": "self", "text": "hi"}))
|
|
|
|
assert states == ["talking", "thinking"]
|
|
assert ctrl._state.state == PetState.THINKING
|
|
|
|
|
|
def test_the_scene_uses_a_voice_the_server_picked_with_speak_as(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "ELEVENLABS_VOICE_ID", "aaorr6ZHIL88gEexu7dC")
|
|
seen = {}
|
|
|
|
def capture(inputs, model_id=None, stability=None):
|
|
seen["inputs"] = inputs
|
|
return np.zeros(4, dtype=np.int16), 24000
|
|
|
|
monkeypatch.setattr(controller_mod.tts, "synthesize_dialogue", capture)
|
|
monkeypatch.setattr(controller_mod.tts, "play_pcm", lambda pcm, rate, should_stop=None: True)
|
|
ctrl._apply_voice(controller_mod.server_client.Reply("ok", "PICKEDvoice123456789", "Terence"))
|
|
|
|
ctrl._handle_command(_dialogue_command({"voice": "self", "text": "hi"}))
|
|
|
|
assert seen["inputs"][0]["voice_id"] == "PICKEDvoice123456789"
|
|
|
|
|
|
def test_a_synthesis_failure_is_reported_back_for_bolt_to_retry(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "ELEVENLABS_VOICE_ID", "aaorr6ZHIL88gEexu7dC")
|
|
|
|
def boom(inputs, model_id=None, stability=None):
|
|
raise controller_mod.tts.TtsError("voice_id not found")
|
|
|
|
monkeypatch.setattr(controller_mod.tts, "synthesize_dialogue", boom)
|
|
|
|
output = ctrl._handle_command(_dialogue_command({"voice": "self", "text": "hi"}))
|
|
|
|
assert "couldn't synthesize" in output and "voice_id not found" in output
|
|
assert ctrl._state.state == PetState.IDLE # nothing left half-transitioned
|
|
|
|
|
|
def test_a_bad_voice_name_comes_back_as_advice_not_an_exception(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "DIALOGUE_VOICES", "narrator:9BWtsMINqrJLrRacOk9x")
|
|
output = ctrl._handle_command(_dialogue_command({"voice": "wizard", "text": "hi"}))
|
|
assert "unknown voice" in output and "narrator" in output
|
|
|
|
|
|
def test_dialogue_can_be_switched_off_on_this_device(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "DIALOGUE", False)
|
|
called = []
|
|
monkeypatch.setattr(controller_mod.tts, "synthesize_dialogue",
|
|
lambda *a, **k: called.append(1))
|
|
output = ctrl._handle_command(_dialogue_command({"voice": "self", "text": "hi"}))
|
|
assert "disabled" in output and called == []
|
|
|
|
|
|
def test_talking_over_a_scene_is_reported_up_the_relay(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "ELEVENLABS_VOICE_ID", "aaorr6ZHIL88gEexu7dC")
|
|
monkeypatch.setattr(controller_mod.tts, "synthesize_dialogue",
|
|
lambda inputs, model_id=None, stability=None: (np.zeros(4, dtype=np.int16), 24000))
|
|
monkeypatch.setattr(controller_mod.tts, "play_pcm",
|
|
lambda pcm, rate, should_stop=None: False) # barge-in
|
|
|
|
output = ctrl._handle_command(_dialogue_command({"voice": "self", "text": "hi"}))
|
|
|
|
assert "interrupted" in output
|
|
|
|
|
|
# ── petctl self_restart ─────────────────────────────────────────────────────
|
|
|
|
def test_self_restart_arms_after_the_turn_rather_than_dying_mid_relay(monkeypatch, ctrl, tmp_path):
|
|
"""Restarting inline would kill the HTTP tool relay before the result was
|
|
posted, and the server would wait out its timeout on a turn that can never
|
|
finish. So the command returns, the turn completes, *then* the pet dies."""
|
|
monkeypatch.setattr(controller_mod.self_restart, "DEFAULT_STATE_PATH", tmp_path / "ctx.json")
|
|
monkeypatch.setattr(controller_mod.self_restart, "preflight", lambda *a, **k: None)
|
|
restarts = _capture(ctrl.restart_requested)
|
|
|
|
output = ctrl._handle_command("petctl self_restart check the new dialogue code")
|
|
|
|
assert "restarting as soon as this turn finishes" in output
|
|
assert restarts == [] # nothing has happened yet
|
|
|
|
assert ctrl._maybe_self_restart() is True
|
|
assert restarts and "check the new dialogue code" in restarts[0]
|
|
|
|
|
|
def test_a_broken_edit_is_reported_instead_of_leaving_nothing_running(monkeypatch, ctrl, tmp_path):
|
|
monkeypatch.setattr(controller_mod.self_restart, "DEFAULT_STATE_PATH", tmp_path / "ctx.json")
|
|
|
|
def boom(*args, **kwargs):
|
|
raise controller_mod.self_restart.RestartError(
|
|
"the current code does not import, so restarting would leave you with "
|
|
"nothing running. Fix this first:\nSyntaxError: invalid syntax"
|
|
)
|
|
|
|
monkeypatch.setattr(controller_mod.self_restart, "preflight", boom)
|
|
restarts = _capture(ctrl.restart_requested)
|
|
|
|
output = ctrl._handle_command("petctl self_restart try the new code")
|
|
|
|
assert "SyntaxError" in output and "refused" in output
|
|
assert ctrl._maybe_self_restart() is False
|
|
assert restarts == []
|
|
|
|
|
|
def test_a_second_restart_request_in_one_turn_is_a_no_op(monkeypatch, ctrl, tmp_path):
|
|
monkeypatch.setattr(controller_mod.self_restart, "DEFAULT_STATE_PATH", tmp_path / "ctx.json")
|
|
monkeypatch.setattr(controller_mod.self_restart, "preflight", lambda *a, **k: None)
|
|
ctrl._handle_command("petctl self_restart first")
|
|
assert "already armed" in ctrl._handle_command("petctl self_restart second")
|
|
|
|
|
|
def test_self_restart_can_be_switched_off_on_this_device(monkeypatch, ctrl):
|
|
monkeypatch.setattr(controller_mod.config, "SELF_RESTART", False)
|
|
checked = []
|
|
monkeypatch.setattr(controller_mod.self_restart, "preflight",
|
|
lambda *a, **k: checked.append(1))
|
|
assert "disabled" in ctrl._handle_command("petctl self_restart go")
|
|
assert checked == []
|
|
|
|
|
|
def test_coming_back_up_reports_to_the_server_and_speaks_the_reply(monkeypatch, ctrl, tmp_path):
|
|
"""The half that makes it a loop: the new process tells Bolt it's back and
|
|
why, and his answer is spoken like any other turn."""
|
|
state = tmp_path / "ctx.json"
|
|
monkeypatch.setattr(controller_mod.self_restart, "DEFAULT_STATE_PATH", state)
|
|
controller_mod.self_restart.arm("check the walk cycle", version="0.2.3",
|
|
path=state, now=1000.0)
|
|
sent, spoken = [], []
|
|
monkeypatch.setattr(controller_mod.server_client, "converse",
|
|
lambda text, on_command=None: sent.append(text) or Reply("Good, it's up."))
|
|
monkeypatch.setattr(controller_mod.tts, "speak",
|
|
lambda text, on_error=None, should_stop=None, voice_id=None:
|
|
spoken.append(text) or True)
|
|
|
|
ctrl._report_self_restart()
|
|
|
|
assert "[pet self-restart]" in sent[0] and "check the walk cycle" in sent[0]
|
|
assert spoken == ["Good, it's up."]
|
|
# Consumed, so the next start doesn't announce the same restart again.
|
|
assert controller_mod.self_restart.load(state) is None
|
|
|
|
|
|
def test_an_ordinary_start_reports_nothing(monkeypatch, ctrl, tmp_path):
|
|
monkeypatch.setattr(controller_mod.self_restart, "DEFAULT_STATE_PATH", tmp_path / "none.json")
|
|
called = []
|
|
monkeypatch.setattr(controller_mod.server_client, "converse",
|
|
lambda text, on_command=None: called.append(text))
|
|
|
|
ctrl._report_self_restart()
|
|
|
|
assert called == []
|