mkbk
257 lines
11 KiB
Python
257 lines
11 KiB
Python
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"], ["第一问"])
|
|
|
|
def test_agent_self_test_runs_full_duplex_profile(self) -> None:
|
|
code, data = self.call("agent-self-test", "--profile", "full-duplex", "--turns", "3")
|
|
|
|
self.assertEqual(code, 0)
|
|
self.assertTrue(data["success"])
|
|
self.assertTrue(data["checks"]["barge_in"])
|
|
self.assertTrue(data["checks"]["tool_router"])
|
|
|
|
def test_audio_self_test_reports_missing_required_apm(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
code, data = self.call(
|
|
"--env-file",
|
|
str(Path(tmp) / ".env"),
|
|
"audio-self-test",
|
|
"--duration",
|
|
"1",
|
|
"--check-echo",
|
|
)
|
|
|
|
self.assertEqual(code, 1)
|
|
self.assertFalse(data["success"])
|
|
self.assertEqual(data["apm"]["error_code"], "AUDIO_APM_UNAVAILABLE")
|
|
|
|
def test_audio_self_test_can_use_fake_apm_for_deterministic_check(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
env_file = Path(tmp) / ".env"
|
|
env_file.write_text(
|
|
"OWNER_AUDIO_APM_PROVIDER=fake\nOWNER_AUDIO_APM_REQUIRED=0\n",
|
|
encoding="utf-8",
|
|
)
|
|
code, data = self.call(
|
|
"--env-file",
|
|
str(env_file),
|
|
"audio-self-test",
|
|
"--duration",
|
|
"1",
|
|
"--check-echo",
|
|
)
|
|
|
|
self.assertEqual(code, 0)
|
|
self.assertTrue(data["success"])
|
|
self.assertTrue(data["echo"]["suppressed"])
|
|
self.assertEqual(data["apm"]["provider"], "fake_webrtc")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|