File size: 2,242 Bytes
b21cfae 0efd0f1 b21cfae | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 | 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()
|