from __future__ import annotations import io import json import tempfile import unittest from contextlib import redirect_stdout from pathlib import Path from unittest.mock import patch from owner_voice_pet.cli import main from owner_voice_pet.speech_models import REQUIRED_MODEL_FILES, write_default_manifest class CliAcceptanceTests(unittest.TestCase): def call(self, *args: str) -> tuple[int, dict[str, object]]: buffer = io.StringIO() with redirect_stdout(buffer): code = main(list(args)) return code, json.loads(buffer.getvalue()) def test_acceptance_command_runs_end_to_end(self) -> None: code, data = self.call("acceptance") self.assertEqual(code, 0) self.assertTrue(data["success"]) self.assertEqual(data["transcript"], "你是谁") self.assertGreaterEqual(data["played_segments"], 1) def test_validate_assets_command(self) -> None: code, data = self.call("validate-assets") self.assertEqual(code, 0) self.assertTrue(data["valid"]) def test_show_config_reads_env_file_without_printing_secret(self) -> None: import tempfile with tempfile.TemporaryDirectory() as tmp: path = f"{tmp}/.env" with open(path, "w", encoding="utf-8") as handle: handle.write("OWNER_LLM_API_KEY=secret-value\nOWNER_LLM_MODEL=file-model\n") code, data = self.call("--env-file", path, "--show-config") self.assertEqual(code, 0) self.assertTrue(data["llm_api_key_present"]) self.assertEqual(data["llm_model"], "file-model") self.assertEqual(data["assistant_mode"], "turn_based_voice_pet") self.assertEqual(data["pipeline_mode"], "live_turn_based") self.assertEqual(data["endpoint_mode"], "primary_speaker") self.assertEqual(data["realtime_transcript_idle_timeout_ms"], 1500) self.assertEqual(data["context_mode"], "session_memory") self.assertTrue(data["end_chime_enabled"]) self.assertEqual(data["end_chime_file"], "assets/sounds/codex-notification.wav") self.assertEqual(data["end_chime_duration_ms"], 140) self.assertEqual(data["audio_apm_provider"], "webrtc") self.assertTrue(data["audio_aec_enabled"]) self.assertTrue(data["audio_ns_enabled"]) self.assertTrue(data["audio_agc_enabled"]) self.assertEqual(data["audio_frame_ms"], 20) self.assertEqual(data["audio_ring_buffer_ms"], 3000) self.assertTrue(data["interrupt_enabled"]) self.assertEqual(data["interrupt_target_latency_ms"], 200) self.assertEqual(data["streaming_stt_provider"], "faster_whisper") self.assertEqual(data["streaming_stt_product_candidate"], "sensevoice") self.assertEqual(data["streaming_tts_provider"], "cosyvoice") self.assertTrue(data["memory_enabled"]) self.assertEqual(data["memory_provider"], "faiss_sqlite") self.assertTrue(data["tool_router_enabled"]) self.assertFalse(data["openinterpreter_enabled"]) self.assertFalse(data["browser_playwright_enabled"]) self.assertFalse(data["computer_control_enabled"]) self.assertNotIn("secret-value", str(data)) def test_help_lists_full_duplex_agent_entrypoint(self) -> None: buffer = io.StringIO() with redirect_stdout(buffer), self.assertRaises(SystemExit) as raised: main(["--help"]) self.assertEqual(raised.exception.code, 0) self.assertIn("run-agent-live", buffer.getvalue()) def test_run_agent_live_check_config_reports_reserved_entrypoint(self) -> None: with tempfile.TemporaryDirectory() as tmp: env_file = str(Path(tmp) / ".env") code, data = self.call("--env-file", env_file, "run-agent-live", "--check-config") self.assertEqual(code, 0) self.assertTrue(data["success"]) self.assertEqual(data["command"], "run-agent-live") self.assertEqual(data["assistant_mode"], "full_duplex_agent") self.assertFalse(data["full_duplex_runtime_ready"]) self.assertEqual(data["audio_apm_provider"], "webrtc") self.assertEqual(data["audio_apm_error_code"], "AUDIO_APM_UNAVAILABLE") self.assertTrue(data["llm_streaming_enabled"]) self.assertEqual(data["turn_based_entry"], "run-live") def test_run_agent_live_once_invokes_agent_runtime(self) -> None: class FakeRuntime: def __init__(self, *, config): self.config = config self.once = None def run(self, *, once: bool = False): self.once = once return type("Summary", (), {"completed_turns": 1, "interrupted": False})() created: list[FakeRuntime] = [] def make_runtime(*, config): runtime = FakeRuntime(config=config) created.append(runtime) return runtime with ( patch("owner_voice_pet.cli.build_audio_processing_provider"), patch("owner_voice_pet.cli.FullDuplexAgentRuntime", side_effect=make_runtime), ): with tempfile.TemporaryDirectory() as tmp: code = main(["--env-file", str(Path(tmp) / ".env"), "run-agent-live", "--once"]) self.assertEqual(code, 0) self.assertEqual(len(created), 1) self.assertTrue(created[0].once) def test_run_agent_live_fails_when_required_apm_is_unavailable(self) -> None: with patch("owner_voice_pet.full_duplex_audio.webrtc_apm_probe") as probe: probe.return_value = type( "Health", (), {"available": False, "message": "missing binding"}, )() with tempfile.TemporaryDirectory() as tmp: code, data = self.call("--env-file", str(Path(tmp) / ".env"), "run-agent-live", "--once") self.assertEqual(code, 1) self.assertEqual(data["code"], "AUDIO_APM_UNAVAILABLE") def test_security_check_command_has_no_leaks(self) -> None: code, data = self.call("security-check") self.assertEqual(code, 0) self.assertEqual(data["secret_leaks"], []) def test_model_check_reports_manifest_and_files(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) write_default_manifest(root) for relative in REQUIRED_MODEL_FILES: path = root / relative path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(b"placeholder") with ( patch("importlib.util.find_spec", return_value=object()), patch("owner_voice_pet.cli.SherpaOnnxKeywordWakeWordProvider") as wake_cls, patch("owner_voice_pet.cli.SherpaOnnxVadProvider") as vad_cls, patch("owner_voice_pet.cli.SherpaOnnxSttProvider") as stt_cls, patch("owner_voice_pet.cli.SherpaOnnxDenoiserPreprocessor") as denoiser_cls, ): wake_cls.return_value.load.return_value = None vad_cls.return_value.load.return_value = None stt_cls.return_value.load.return_value = None denoiser_cls.return_value.load.return_value = None code, data = self.call("model-check", "--models-dir", str(root)) self.assertEqual(code, 0) self.assertTrue(data["ok"]) self.assertEqual(data["missing_files"], []) self.assertTrue(data["provider_load_checked"]) def test_model_check_reports_missing_files(self) -> None: with tempfile.TemporaryDirectory() as tmp: with patch("importlib.util.find_spec", return_value=object()): code, data = self.call("model-check", "--models-dir", tmp) self.assertEqual(code, 1) self.assertFalse(data["ok"]) self.assertTrue(data["missing_files"]) def test_device_check_reports_sounddevice_status(self) -> None: with patch("owner_voice_pet.cli.sounddevice_device_report", return_value={"ok": True, "devices": []}): code, data = self.call("device-check") self.assertEqual(code, 0) self.assertTrue(data["ok"]) def test_run_live_once_invokes_runtime(self) -> None: class FakeRuntime: def run(self, *, once: bool = False): self.once = once return type("Summary", (), {"completed_turns": 1, "interrupted": False})() fake_runtime = FakeRuntime() with patch("owner_voice_pet.cli.build_live_runtime", return_value=fake_runtime): code = main(["run-live", "--once"]) self.assertEqual(code, 0) self.assertTrue(fake_runtime.once) def test_simulate_live_command_runs_two_turns(self) -> None: code, data = self.call("simulate-live", "--turns", "2") self.assertEqual(code, 0) self.assertTrue(data["success"]) self.assertEqual(data["completed_turns"], 2) self.assertEqual(data["final_transcripts"], ["第1轮模拟问题", "第2轮模拟问题"]) self.assertTrue(data["checks"]["partial_noise_filtered"]) def test_real_live_check_command_invokes_real_fixture_runner(self) -> None: expected = {"success": True, "completed_turns": 2} with patch("owner_voice_pet.cli.run_real_live_check", return_value=expected) as runner: code, data = self.call("real-live-check", "--turns", "2", "--no-playback", "--question", "第一问") self.assertEqual(code, 0) self.assertEqual(data, expected) kwargs = runner.call_args.kwargs self.assertEqual(kwargs["turns"], 2) self.assertFalse(kwargs["play_audio"]) self.assertEqual(kwargs["questions"], ["第一问"]) if __name__ == "__main__": unittest.main()