mirror of
https://github.com/fusion44/blitz_api.git
synced 2026-08-16 12:20:14 +02:00
868457ffa5
get_event_loop() is deprecated on Python 3.11+ when there is no running loop and is slated to change behaviour further. - SSEManager.setup() ran at import time (app.api.utils) via get_event_loop(); this only worked because uvicorn imports the app inside its loop and would break when imported without a running loop (e.g. a Celery worker). Start the broadcast consumer lazily from within a running loop instead. - everywhere else the pattern was get_event_loop().create_task(x) inside a coroutine; replace with asyncio.create_task(x). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
25 lines
896 B
Python
25 lines
896 B
Python
"""
|
|
Regression test: SSEManager must not touch the event loop at import time.
|
|
|
|
setup() ran at module import via asyncio.get_event_loop(), which is
|
|
deprecated without a running loop and breaks when the module is imported
|
|
outside a running loop (e.g. in a Celery worker). The broadcast consumer
|
|
must instead start lazily, the first time it's needed from within a loop.
|
|
"""
|
|
|
|
|
|
async def test_broadcast_task_starts_lazily_within_a_loop():
|
|
from app.api.sse_manager import SSEManager
|
|
|
|
mgr = SSEManager()
|
|
assert mgr._broadcast_task is None, "must not start a task before it's needed"
|
|
|
|
mgr._ensure_broadcast_task()
|
|
assert mgr._broadcast_task is not None, "task should start within a running loop"
|
|
|
|
# a second call must not spawn another task
|
|
first = mgr._broadcast_task
|
|
mgr._ensure_broadcast_task()
|
|
assert mgr._broadcast_task is first
|
|
|
|
mgr._broadcast_task.cancel()
|