"""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()