| import asyncio |
| import json |
| import struct |
| import unittest |
| from unittest.mock import patch |
| from tts_bridge import MouthBridge |
| from app.wire import async_recv_msg |
|
|
|
|
| class FakeClient: |
| def __init__(self,*args,**kwargs):pass |
| async def __aenter__(self):return self |
| async def __aexit__(self,*args):pass |
| async def get(self,*args,**kwargs): |
| class Response: |
| status_code=200 |
| def json(self):return {'busy':False} |
| return Response() |
| def stream(self,*args,**kwargs): |
| class Waiting: |
| async def __aenter__(self):await asyncio.sleep(60) |
| async def __aexit__(self,*args):pass |
| return Waiting() |
|
|
|
|
| def frame(kind,payload): |
| raw=json.dumps(payload).encode();return kind+struct.pack('>I',len(raw))+raw |
|
|
|
|
| class BridgeTests(unittest.IsolatedAsyncioTestCase): |
| async def check_cancel(self,delay): |
| with patch('tts_bridge.httpx.AsyncClient',FakeClient): |
| bridge=MouthBridge() |
| server=await asyncio.start_server(bridge.accept,'127.0.0.1',0) |
| reader,writer=await asyncio.open_connection('127.0.0.1',server.sockets[0].getsockname()[1]) |
| try: |
| if delay is None: |
| writer.write(frame(b'T',{'transcript':'Synthetic test.'})+frame(b'B',{'played_s':0})) |
| else: |
| writer.write(frame(b'T',{'transcript':'Synthetic test.'}));await writer.drain();await asyncio.sleep(delay) |
| writer.write(frame(b'B',{'played_s':0})) |
| await writer.drain() |
| kind,raw=await asyncio.wait_for(async_recv_msg(reader),2) |
| self.assertEqual(kind,b'D');self.assertTrue(json.loads(raw)['barged']) |
| with self.assertRaises(asyncio.TimeoutError):await asyncio.wait_for(async_recv_msg(reader),.05) |
| self.assertEqual(bridge.active,0) |
| finally: |
| writer.close();await writer.wait_closed();server.close();await server.wait_closed();await asyncio.sleep(.01) |
|
|
| async def test_immediate_cancel_acknowledges_once(self):await self.check_cancel(None) |
| async def test_inflight_cancel_acknowledges_once(self):await self.check_cancel(.02) |
|
|
|
|
| if __name__=='__main__':unittest.main() |
|
|
