| """Audio receipt and preview safety, without models or network services.""" |
| import json,struct,unittest |
| from types import SimpleNamespace |
| from unittest.mock import AsyncMock,patch,Mock |
| import numpy as np |
| from aiohttp import WSMsgType |
| from server import RoboConversation,Conversation |
|
|
| class AudioInputTests(unittest.IsolatedAsyncioTestCase): |
| def conversation(self): |
| c=RoboConversation.__new__(RoboConversation) |
| c.ws=SimpleNamespace(send_json=AsyncMock()) |
| c.input_samples=0;c.input_blocks=0;c.last_input_notice=0. |
| c.pending_barge_proposal=None |
| c._send_turn=AsyncMock() |
| return c |
| async def test_receipt_ack_has_level_and_duration_but_no_audio(self): |
| c=self.conversation();pcm=np.full(320,3277,'<i2') |
| message=SimpleNamespace(type=WSMsgType.BINARY,data=struct.pack('<ii',0,-1)+pcm.tobytes()) |
| with patch.object(Conversation,'_on_browser',new_callable=AsyncMock) as parent: |
| await c._on_browser(message);parent.assert_awaited_once_with(message) |
| packet=c.ws.send_json.call_args.args[0] |
| self.assertEqual(packet['type'],'input_status');self.assertEqual(packet['seconds'],.02) |
| self.assertAlmostEqual(packet['rms_dbfs'],-20.,places=1) |
| self.assertEqual(set(packet),{'type','blocks','seconds','rms_dbfs'}) |
| async def test_shared_service_status_is_not_a_spoken_reply_or_identity_update(self): |
| c=self.conversation();c.speaker_ext=SimpleNamespace(metadata=AsyncMock()) |
| await c._on_text(json.dumps({'status_only':True,'stage':'waiting','text':'Waiting for shared voice model.'})) |
| message=c.ws.send_json.call_args.args[0] |
| self.assertEqual(message['type'],'service_status');self.assertEqual(message['stage'],'waiting') |
| c.speaker_ext.metadata.assert_not_awaited() |
| async def test_speaker_filter_can_show_asr_without_submitting_it(self): |
| c=self.conversation();c.identity_enabled=True;c.identity_mode='v4-long';c.speaking=False |
| c.ears=SimpleNamespace(sr=16000) |
| from unittest.mock import Mock |
| c.speaker_ext=SimpleNamespace(observe=AsyncMock(),last_activity=None,note_asr=Mock()) |
| events=[{'kind':'PARTIAL','text':'Can you hear me?'},{'kind':'ENDPOINT','transcript':'Can you hear me?'}] |
| with patch.object(Conversation,'_on_events',new_callable=AsyncMock) as parent: |
| await c._on_events(events,5120);parent.assert_awaited_once_with([],5120) |
| preview=c.ws.send_json.call_args.args[0] |
| self.assertEqual(preview['type'],'asr_preview');self.assertEqual(preview['status'],'not_submitted') |
| c._send_turn.assert_not_awaited() |
| async def test_filter_off_restores_native_asr_events(self): |
| c=self.conversation();c.identity_enabled=False;c.identity_mode='v4-long';c.speaker_ext=None |
| events=[{'kind':'PARTIAL','text':'Can you hear me?'},{'kind':'ENDPOINT','transcript':'Can you hear me?'}] |
| with patch.object(Conversation,'_on_events',new_callable=AsyncMock) as parent: |
| await c._on_events(events,5120);parent.assert_awaited_once_with(events,5120) |
| async def test_native_interruption_is_not_authorized_by_channel_alone(self): |
| c=self.conversation();c.identity_enabled=True;c.identity_mode='v4-long';c.speaking=True |
| c.ears=SimpleNamespace(sr=16000) |
| c.speaker_ext=SimpleNamespace(observe=AsyncMock(),last_activity={'id':'someone'},confirm_interrupt=AsyncMock(return_value=None)) |
| proposal={'kind':'BARGE','yield_prob':.7,'onset_s':1.} |
| with patch.object(Conversation,'_on_events',new_callable=AsyncMock) as parent: |
| await c._on_events([proposal],5120);parent.assert_awaited_once_with([],5120) |
| c.speaker_ext.confirm_interrupt.assert_awaited_once_with(proposal,5120) |
| c.speaker_ext.confirm_interrupt.return_value='verified-user' |
| with patch.object(Conversation,'_on_events',new_callable=AsyncMock) as parent: |
| await c._on_events([],10240);parent.assert_awaited_once_with([proposal],10240) |
| def test_native_settings_restored_when_speaker_filter_is_disabled(self): |
| c=self.conversation();c.session=object();c.native_barge_settings=(.5,2);c.ears=SimpleNamespace(set_threshold=Mock()) |
| c._speaker_barge_sensitivity(True) |
| self.assertEqual([call.args[2] for call in c.ears.set_threshold.call_args_list],[.2,1]) |
| c.ears.set_threshold.reset_mock();c._speaker_barge_sensitivity(False) |
| self.assertEqual([call.args[2] for call in c.ears.set_threshold.call_args_list],[.5,2]) |
| async def test_hop_keeps_strongest_native_proposal_not_last_weak_frame(self): |
| c=self.conversation();c.identity_enabled=True;c.identity_mode='v4-long';c.speaking=True;c.ears=SimpleNamespace(sr=16000) |
| c.speaker_ext=SimpleNamespace(observe=AsyncMock(),last_activity=None,confirm_interrupt=AsyncMock(return_value=None)) |
| strong={'kind':'BARGE','yield_prob':.9};weak={'kind':'BARGE','yield_prob':.25} |
| with patch.object(Conversation,'_on_events',new_callable=AsyncMock):await c._on_events([strong,weak],5120) |
| c.speaker_ext.confirm_interrupt.assert_awaited_once_with(strong,5120) |
|
|
| if __name__=='__main__':unittest.main() |
|
|
