sirmrak/wyoming-satellite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Don't use asyncio.timeout

Browse Source
This commit is contained in:
Michael Hansen
2024-01-09 16:25:34 -06:00
parent b164bafc9f
commit 350eee6cb2
1 changed files with 3 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_wake_streaming.py
+3 -5
View File
@@ -92,17 +92,15 @@ async def test_multiple_wakeups(tmp_path: Path) -> None:
satellite_task = asyncio.create_task(satellite.run(), name="satellite") satellite_task = asyncio.create_task(satellite.run(), name="satellite")
await satellite.event_from_server(RunSatellite().event()) await satellite.event_from_server(RunSatellite().event())
async with asyncio.timeout(1): await asyncio.wait_for(event_client.wake_event.wait(), timeout=1)
await event_client.wake_event.wait() event_client.wake_event.clear()
event_client.wake_event.clear()
# Stop streaming # Stop streaming
await satellite.event_from_server(Transcript("test").event()) await satellite.event_from_server(Transcript("test").event())
# Should not trigger again within refractory period (default: 5 sec) # Should not trigger again within refractory period (default: 5 sec)
with pytest.raises(asyncio.TimeoutError): with pytest.raises(asyncio.TimeoutError):
async with asyncio.timeout(0.15): await asyncio.wait_for(event_client.wake_event.wait(), timeout=0.15)
await event_client.wake_event.wait()
await satellite.stop() await satellite.stop()
await satellite_task await satellite_task