Files
unshackle/tests/remote/unit/test_handler_event_loop.py
T
imSp4rky dffb8b27e3 fix(serve): lock shared session state and throttle progress writes
Handlers share one service instance, so a per-session lock keeps two requests out of it at once. Track threads merge and write job state under one lock, and the progress sink coalesces ticks that say the same thing.
2026-09-06 20:01:38 -06:00

102 lines
3.5 KiB
Python

"""Blocking service calls must not run on the aiohttp event loop.
get_titles and get_tracks do network work that can take a whole manifest round trip. Run on
the loop, they stop the server: the progress pump and the SSE keep-alive stall with them. An
interactive prompt from the service can never be answered either, because the endpoint that
answers it needs the same loop. Each test here holds the service call in a two-event
handshake that only completes while the loop still runs another coroutine.
"""
from __future__ import annotations
import asyncio
import json
import threading
from types import SimpleNamespace
from typing import Any, Awaitable, Callable, List
import pytest
from aiohttp import web
from unshackle.core.api import handlers
from unshackle.core.api.session_store import get_session_store
from unshackle.core.titles.movie import Movie
pytestmark = pytest.mark.unit
# Bounds the wait when the loop is blocked, so a regression fails the test instead of hanging.
HANDSHAKE_TIMEOUT = 5.0
class Handshake:
"""A blocking call that reports whether the event loop kept running during it."""
def __init__(self) -> None:
self.entered = threading.Event()
self.resume = threading.Event()
self.loop_was_blocked = False
def block(self) -> None:
self.entered.set()
self.loop_was_blocked = not self.resume.wait(timeout=HANDSHAKE_TIMEOUT)
async def drive(self, call: Callable[[], Awaitable[web.Response]]) -> web.Response:
task = asyncio.create_task(call())
# Reached only while the loop is free.
await asyncio.to_thread(self.entered.wait, HANDSHAKE_TIMEOUT)
self.resume.set()
return await asyncio.wait_for(task, HANDSHAKE_TIMEOUT)
def a_movie() -> Movie:
return Movie(id_="movie-0001", service=object, name="Film", year=2024, language=None)
class BlockingTitles:
def __init__(self, handshake: Handshake) -> None:
self.handshake = handshake
def get_titles(self) -> List[Movie]:
self.handshake.block()
return [a_movie()]
class BlockingTracks:
def __init__(self, handshake: Handshake) -> None:
self.handshake = handshake
def get_titles(self) -> List[Movie]:
return [a_movie()]
def get_tracks(self, title: Any) -> Any:
self.handshake.block()
return SimpleNamespace(videos=[], audio=[], subtitles=[])
@pytest.mark.asyncio
async def test_session_titles_leaves_the_loop_free() -> None:
handshake = Handshake()
store = get_session_store()
await store.create("EXAMPLE", BlockingTitles(handshake), session_id="loop-test-titles")
try:
response = await handshake.drive(lambda: handlers.session_titles_handler("loop-test-titles"))
finally:
await store.delete("loop-test-titles")
assert not handshake.loop_was_blocked, "get_titles blocked the event loop"
assert [t["name"] for t in json.loads(response.body)["titles"]] == ["Film"]
@pytest.mark.asyncio
async def test_list_tracks_leaves_the_loop_free(monkeypatch: pytest.MonkeyPatch) -> None:
handshake = Handshake()
service = BlockingTracks(handshake)
monkeypatch.setattr(handlers, "validate_service", lambda tag, request=None: "EXAMPLE")
monkeypatch.setattr(handlers, "setup_list_service", lambda *args, **kwargs: service)
data = {"service": "EXAMPLE", "title_id": "movie-0001"}
response = await handshake.drive(lambda: handlers.list_tracks_handler(data))
assert not handshake.loop_was_blocked, "get_tracks blocked the event loop"
assert json.loads(response.body)["title"]["name"] == "Film"