[音频底座重构]:完成AudioHub和WebRTC音频处理,包含AEC参考流和环形缓冲测试
This commit is contained in:
@@ -78,25 +78,57 @@ class CliAcceptanceTests(unittest.TestCase):
|
||||
self.assertIn("run-agent-live", buffer.getvalue())
|
||||
|
||||
def test_run_agent_live_check_config_reports_reserved_entrypoint(self) -> None:
|
||||
code, data = self.call("run-agent-live", "--check-config")
|
||||
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.assertTrue(data["full_duplex_runtime_ready"])
|
||||
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.assertEqual(data["turn_based_entry"], "run-live")
|
||||
|
||||
def test_run_agent_live_once_invokes_agent_runtime(self) -> None:
|
||||
class FakeRuntime:
|
||||
def run_agent(self, *, once: bool = False):
|
||||
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})()
|
||||
|
||||
fake_runtime = FakeRuntime()
|
||||
with patch("owner_voice_pet.cli.build_live_runtime", return_value=fake_runtime):
|
||||
code = main(["run-agent-live", "--once"])
|
||||
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.assertTrue(fake_runtime.once)
|
||||
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")
|
||||
|
||||
Reference in New Issue
Block a user