import asyncio import json import logging from types import SimpleNamespace import pytest from device_gateway.speech_diagnostics import SpeechDiagnostics from test_speech import manager def records(caplog): return [json.loads(r.getMessage().split('[SPEECH_DIAG] ', 1)[1]) for r in caplog.records if '[SPEECH_DIAG] ' in r.getMessage()] def test_transmitted_is_not_played_and_late_ack_is_traced(tmp_path, caplog): caplog.set_level(logging.INFO, logger='uvicorn.error.device_gateway.speech') async def check(): m, d, out, _, _ = manager(tmp_path, 25) d.capabilities.add('speech-buffer-v1') assert m.start(d, 'reply', 'speech_body_marker api_key=private-key') req = m.active['c'] await req.task assert not m.active end = records(caplog)[-1] assert end['status'] == 'sent' and end['sent_frames'] == 25 assert end['played_frames'] == 0 and end['audio_ms'] == 2000 before = len(out) # These messages used to be ignored after end; diagnostics must not # create credits, send packets, or reopen playback while observing them. await m.handle(d, dict(type='speech_progress', speech_id=req.speech_id, played_chunks=8)) assert records(caplog)[-1]['event'] == 'codec_progress' assert records(caplog)[-1]['played_frames'] == 8 for sid, count in [({}, 16), ([], 16), ('stale', 16), (req.speech_id, True), (req.speech_id, 26), (req.speech_id, 8)]: await m.handle(d, dict(type='speech_progress', speech_id=sid, played_chunks=count)) assert m.diagnostics.get(d, req.speech_id).played_frames == 8 await m.handle(d, dict(type='speech_stop', speech_id=req.speech_id)) assert records(caplog)[-1]['event'] == 'device_stop' assert not m.active and len(out) == before m.cancel(d, notify=False, forget=True, reason='disconnect') assert records(caplog)[-1]['event'] == 'disconnect' and not m.diagnostics.traces await m.close() asyncio.run(check()) wire = caplog.text assert all(secret not in wire for secret in ('speech_body_marker', 'private-key', 'agent:pi:s')) assert all('audio' not in row and 'text' not in row and 'target' not in row for row in records(caplog)) @pytest.mark.parametrize('stage', ['synthesis', 'send_chunk', 'credit_wait']) def test_failure_stage_is_logged_without_provider_body(tmp_path, caplog, monkeypatch, stage): caplog.set_level(logging.INFO, logger='uvicorn.error.device_gateway.speech') async def check(): m, d, out, _, _ = manager(tmp_path, 65) d.capabilities.add('speech-buffer-v1') if stage == 'synthesis': async def synth(*args): raise RuntimeError('raw-provider-secret') m.provider.synthesize = synth elif stage == 'send_chunk': original = m.send async def send(device, payload): if payload['type'] == 'speech_chunk': raise RuntimeError('raw-provider-secret') return await original(device, payload) m.send = send else: original = asyncio.wait_for async def wait(awaitable, timeout): if timeout == 30: awaitable.close() raise TimeoutError('raw-provider-secret') return await original(awaitable, timeout) monkeypatch.setattr(asyncio, 'wait_for', wait) assert m.start(d, 'reply', 'Private body marker.') await m.active['c'].task end = records(caplog)[-1] assert end['event'] == 'tx_end' and end['status'] == 'failed' and end['stage'] == stage assert end['error'] == ('timeout' if stage == 'credit_wait' else 'unavailable') assert end['sent_frames'] == (64 if stage == 'credit_wait' else 0) assert end['credit_waits'] == (1 if stage == 'credit_wait' else 0) assert [p['type'] for _, p in out][-2:] == ['speech_cancel', 'speech_error'] await m.close() asyncio.run(check()) assert 'raw-provider-secret' not in caplog.text and 'Private body' not in caplog.text def test_cancel_has_fixed_reason_and_does_not_emit_later_audio(tmp_path, caplog): caplog.set_level(logging.INFO, logger='uvicorn.error.device_gateway.speech') async def check(): m, d, out, _, _ = manager(tmp_path) entered = asyncio.Event() async def synth(*args): entered.set() await asyncio.Event().wait() m.provider.synthesize = synth assert m.start(d, 'reply', 'Secret body.') req = m.active['c'] await entered.wait() m.cancel(d, reason='approval') await asyncio.gather(req.task, return_exceptions=True) assert any(row['event'] == 'cancel' and row['reason'] == 'approval' for row in records(caplog)) assert records(caplog)[-1]['status'] == 'cancelled' assert not any(p['type'] in ('speech_start', 'speech_chunk') for _, p in out) await m.close() asyncio.run(check()) def test_metadata_limits_and_timing_do_not_log_per_frame(caplog): caplog.set_level(logging.INFO, logger='uvicorn.error.device_gateway.speech') now = [0.0] diag = SpeechDiagnostics(lambda: now[0]) d = SimpleNamespace(connection_id='untrusted\nconnection-secret') req = SimpleNamespace(speech_id='generated-id') diag.begin(d, req, 'untrusted\nsource-secret') for index in range(100): now[0] += .08 diag.sent(d, req, 'speech_chunk', now[0] - .05, (.02, .03)) diag.progress(d, dict(speech_id=req.speech_id, played_chunks=index + 1)) trace = diag.get(d, req.speech_id) assert trace.max_send_ms == 50 and trace.max_lock_wait_ms == 20 and trace.max_write_ms == 30 assert trace.played_frames == trace.sent_frames == 100 assert len(records(caplog)) <= 10 assert 'connection-secret' not in caplog.text and 'source-secret' not in caplog.text for index in range(40): diag.begin(d, SimpleNamespace(speech_id=str(index)), 'screen') assert len(diag.traces) == diag.limit now[0] += diag.ttl + 1 diag.prune() assert not diag.traces def test_shared_send_lock_wait_is_separate_from_socket_write(monkeypatch): from device_gateway import app async def check(): now = [10.0] monkeypatch.setattr(app.time, 'monotonic', lambda: now[0]) class Lock: async def __aenter__(self): now[0] += .4 async def __aexit__(self, *args): pass class Socket: async def send_json(self, payload): now[0] += .1 d = SimpleNamespace(online=True, websocket=Socket(), send_lock=Lock()) lock_wait, write = await app.send_to_device(d, {'type':'speech_chunk'}) assert lock_wait == pytest.approx(.4) and write == pytest.approx(.1) asyncio.run(check())