Spaces:
Sleeping
Sleeping
| """ProgressBus pubsub semantics.""" | |
| import asyncio | |
| import pytest | |
| from backend.app.services.progress import ProgressBus, ProgressEvent | |
| async def test_subscribe_replays_history(): | |
| bus = ProgressBus() | |
| bus.publish(ProgressEvent(video_id="v1", stage="scenes", message="hello")) | |
| bus.publish(ProgressEvent(video_id="v1", stage="embed", message="world")) | |
| q = await bus.subscribe("v1") | |
| received = [q.get_nowait().stage for _ in range(2)] | |
| assert received == ["scenes", "embed"] | |
| async def test_publish_pushes_to_live_subscriber(): | |
| bus = ProgressBus() | |
| q = await bus.subscribe("v1") | |
| bus.publish(ProgressEvent(video_id="v1", stage="done", message="ok", percent=100.0)) | |
| ev = await asyncio.wait_for(q.get(), timeout=1.0) | |
| assert ev.stage == "done" | |
| assert ev.percent == 100.0 | |
| async def test_unsubscribe_stops_delivery(): | |
| bus = ProgressBus() | |
| q = await bus.subscribe("v1") | |
| await bus.unsubscribe("v1", q) | |
| bus.publish(ProgressEvent(video_id="v1", stage="done", message="late")) | |
| with pytest.raises(asyncio.QueueEmpty): | |
| q.get_nowait() | |
| async def test_subscribers_isolated_per_video(): | |
| bus = ProgressBus() | |
| q_a = await bus.subscribe("v1") | |
| q_b = await bus.subscribe("v2") | |
| bus.publish(ProgressEvent(video_id="v1", stage="done", message="x")) | |
| assert q_a.get_nowait().video_id == "v1" | |
| with pytest.raises(asyncio.QueueEmpty): | |
| q_b.get_nowait() | |