diff --git a/README.md b/README.md index 2e6a6db..98ed0ab 100644 --- a/README.md +++ b/README.md @@ -61,6 +61,7 @@ This file handles the DAV server logic and RD polling. | `poll_interval_secs` | `10` | How often Buzz polls Real-Debrid for changes. | | `server.bind` | `0.0.0.0` | IP address the DAV server binds to. | | `server.port` | `9999` | Port for the DAV server. | +| `server.stream_buffer_size` | `0` | Read-ahead buffer size in bytes for streaming media (e.g., 50MB: `52428800`). Set to `0` to disable. | | `state_dir` | `/app/data` | Path to store the SQLite DB and snapshots inside the container. | | `hooks.on_library_change` | `sh /app/scripts/media_update.sh` | Shell command executed when a change in the library is detected. | | `hooks.curator_url` | `http://buzz-curator:8400/rebuild` | Internal URL to trigger the Curator rebuild. | diff --git a/buzz.dist.yml b/buzz.dist.yml index efc47d3..3534f24 100644 --- a/buzz.dist.yml +++ b/buzz.dist.yml @@ -4,6 +4,10 @@ poll_interval_secs: 10 server: bind: 0.0.0.0 port: 9999 + # Read-ahead buffer size in bytes for streaming media (e.g., 52428800 for 50MB). + # Set to 0 to disable. When enabled, a background thread pre-fetches data from + # Real-Debrid into a bounded queue to smooth out network variations. + stream_buffer_size: 0 state_dir: /app/data hooks: on_library_change: "bash /app/scripts/media_update.sh" diff --git a/buzz/core/curator.py b/buzz/core/curator.py index cba27c7..2fde7db 100644 --- a/buzz/core/curator.py +++ b/buzz/core/curator.py @@ -16,6 +16,7 @@ YEAR_RE, ) from .events import record_event +from .state import is_internal_category from .media import ( is_sidecar_file, is_video_file, @@ -506,8 +507,8 @@ def trigger_jellyfin_selective_refresh( return categories = {root.split("/")[0] for root in changed_roots if "/" in root} - # Filter out internal/virtual categories like __unplayable__ that shouldn't trigger scans - categories = {cat for cat in categories if cat != "__unplayable__"} + # Filter out internal/virtual categories like __unplayable__ that shouldn't trigger scans. + categories = {cat for cat in categories if not is_internal_category(cat)} if not categories: return diff --git a/buzz/core/state.py b/buzz/core/state.py index 28be555..ea74a09 100644 --- a/buzz/core/state.py +++ b/buzz/core/state.py @@ -62,6 +62,10 @@ def canonical_snapshot(snapshot: dict[str, Any]) -> dict[str, Any]: } +def is_internal_category(name: str) -> bool: + return name.startswith("__") + + class LibraryBuilder: def __init__(self, config: DavConfig): self.config = config @@ -319,9 +323,9 @@ def _root_for_snapshot_path(self, path: str) -> str | None: parts = tuple(part for part in normalized.split("/") if part) if len(parts) < 2: return None - if parts[0] == "__all__": + if is_internal_category(parts[0]): return None - if parts[0] not in {"movies", "shows", "anime", "__unplayable__"}: + if parts[0] not in {"movies", "shows", "anime"}: return None return "/".join(parts[:2]) @@ -417,7 +421,9 @@ def sync(self, *, trigger_hook: bool = True) -> dict[str, Any]: self.snapshot_digest = digest self._write_json(self.snapshot_path, self.snapshot) self.snapshot_loaded = True - if trigger_hook and self.config.hook_command: + if trigger_hook and ( + self.config.hook_command or self.config.curator_url + ): hook_paths = changed_paths self.last_sync_at = report["timestamp"] @@ -598,8 +604,10 @@ def _trigger_curator(self, changed_roots: list[str]) -> None: def _run_hook(self, changed_roots: list[str]) -> None: if not self.config.hook_command: return - # Filter out internal/virtual categories like __unplayable__ - filtered_roots = [r for r in changed_roots if not r.startswith("__unplayable__")] + # Filter out internal/virtual categories like __unplayable__ and __all__. + filtered_roots = [ + r for r in changed_roots if not is_internal_category(r.split("/", 1)[0]) + ] if not filtered_roots: return @@ -607,8 +615,34 @@ def _run_hook(self, changed_roots: list[str]) -> None: try: cmd = shlex.split(self.config.hook_command) cmd.extend(filtered_roots) - subprocess.run(cmd, check=True, timeout=60) + subprocess.run( + cmd, + check=True, + timeout=60, + capture_output=True, + text=True, + ) self.verbose_log("Library update hook completed successfully") + except subprocess.TimeoutExpired as exc: + details = [f"Library update hook timed out after {exc.timeout}s: {exc.cmd}"] + stdout = (exc.stdout or "").strip() + stderr = (exc.stderr or "").strip() + if stdout: + details.append(f"stdout:\n{stdout}") + if stderr: + details.append(f"stderr:\n{stderr}") + record_event("\n".join(details), level="error") + except subprocess.CalledProcessError as exc: + details = [ + f"Library update hook failed with exit code {exc.returncode}: {exc.cmd}" + ] + stdout = (exc.stdout or "").strip() + stderr = (exc.stderr or "").strip() + if stdout: + details.append(f"stdout:\n{stdout}") + if stderr: + details.append(f"stderr:\n{stderr}") + record_event("\n".join(details), level="error") except Exception as exc: record_event(f"Library update hook failed: {exc}", level="error") @@ -821,9 +855,18 @@ def resolve_download_url(self, source_url: str, force_refresh: bool = False) -> if download_url: return download_url - download_url = self.client.unrestrict.link(source_url).json().get("download") + try: + res = self.client.unrestrict.link(source_url) + data = res.json() + except Exception as exc: + raise ValueError(f"Failed to unrestrict {source_url}: {exc}") from exc + + download_url = data.get("download") if not download_url: - raise ValueError(f"Failed to resolve download link for {source_url}") + error_msg = data.get("error") or "no download link in response" + raise ValueError( + f"Failed to resolve download link for {source_url}: {error_msg}" + ) with self.lock: self.resolved_urls[source_url] = {"download_url": download_url} @@ -845,6 +888,25 @@ def __init__(self, state: BuzzState): self.state = state self._stop_event = threading.Event() + def _format_change_message( + self, + added: list[str], + removed: list[str], + updated: list[str], + synced: int, + ) -> str: + lines = [f"Real-Debrid library changed ({synced} torrents):"] + if added: + lines.append(f" +{len(added)} added") + lines.extend(f" {path}" for path in added) + if removed: + lines.append(f" -{len(removed)} removed") + lines.extend(f" {path}" for path in removed) + if updated: + lines.append(f" ~{len(updated)} updated") + lines.extend(f" {path}" for path in updated) + return "\n".join(lines) + def run(self) -> None: while not self._stop_event.wait(self.state.config.poll_interval_secs): try: @@ -854,15 +916,10 @@ def run(self) -> None: removed = report.get("removed_paths", []) updated = report.get("updated_paths", []) synced = report.get("synced_torrents", 0) - parts = [] - if added: - parts.append(f"+{len(added)} added: {', '.join(added)}") - if removed: - parts.append(f"-{len(removed)} removed: {', '.join(removed)}") - if updated: - parts.append(f"~{len(updated)} updated: {', '.join(updated)}") + if not any((added, removed, updated)): + continue record_event( - f"Real-Debrid library changed: {'; '.join(parts)} ({synced} torrents)", + self._format_change_message(added, removed, updated, synced), event="realdebrid_update", ) except Exception as exc: # noqa: BLE001 diff --git a/buzz/dav_app.py b/buzz/dav_app.py index 5f1daa8..5b08fc3 100644 --- a/buzz/dav_app.py +++ b/buzz/dav_app.py @@ -1,6 +1,8 @@ import hashlib import json import os +import queue +import threading from contextlib import asynccontextmanager from http import HTTPStatus from typing import Any @@ -338,15 +340,77 @@ def stream_generator(): response, first_chunk = open_remote_media( self.state, node, range_header ) + + chunk_size = 64 * 1024 + buffer_size = self.config.stream_buffer_size + + if buffer_size < chunk_size: + try: + if first_chunk: + yield first_chunk + while True: + chunk = response.read(chunk_size) + if not chunk: + break + yield chunk + finally: + response.close() + return + + # Buffered path: background thread reads ahead into a bounded queue. + q = queue.Queue(maxsize=max(1, buffer_size // chunk_size)) + stop_event = threading.Event() + + def buffer_reader(): + try: + while not stop_event.is_set(): + chunk = response.read(chunk_size) + if not chunk: + break + while not stop_event.is_set(): + try: + q.put(chunk, timeout=1) + break + except queue.Full: + continue + except Exception as exc: + print( + json.dumps( + {"event": "buffer_reader_error", "error": str(exc)}, + sort_keys=True, + ), + flush=True, + ) + finally: + # Signal end-of-stream; use timeout to avoid hanging + # if the queue is full and the consumer is gone. + while not stop_event.is_set(): + try: + q.put(None, timeout=1) + break + except queue.Full: + continue + + t = threading.Thread(target=buffer_reader, daemon=True) + t.start() + try: if first_chunk: yield first_chunk + while True: - chunk = response.read(64 * 1024) - if not chunk: + try: + item = q.get(timeout=1) + except queue.Empty: + if not t.is_alive(): + break + continue + if item is None: break - yield chunk + yield item finally: + stop_event.set() + t.join(timeout=5) response.close() return StreamingResponse( diff --git a/buzz/dav_protocol.py b/buzz/dav_protocol.py index fb99b72..61c2324 100644 --- a/buzz/dav_protocol.py +++ b/buzz/dav_protocol.py @@ -54,10 +54,20 @@ def open_remote_media( if not source_url: raise ValueError("missing Real-Debrid source URL") last_error = "unable to resolve upstream media" + state.verbose_log(f"Opening remote media from {source_url!r}") for attempt in range(2): - download_url = state.resolve_download_url( - source_url, force_refresh=attempt == 1 - ) + try: + download_url = state.resolve_download_url( + source_url, force_refresh=attempt == 1 + ) + except Exception as exc: + last_error = str(exc) + state.verbose_log(f"Failed to resolve download URL: {exc}") + if attempt == 0: + continue + raise + + state.verbose_log(f"Resolved to {download_url!r} (attempt {attempt + 1}/2)") req = request.Request(download_url, method="GET") if range_header: start, end = range_header @@ -66,10 +76,21 @@ def open_remote_media( response = request.urlopen(req, timeout=60) except error.HTTPError as exc: state.invalidate_download_url(source_url) - last_error = f"upstream returned HTTP {exc.code}" + last_error = f"upstream returned HTTP {exc.code} for {download_url}" + state.verbose_log( + f"HTTP Error {exc.code} on attempt {attempt + 1}: {exc.reason}" + ) if attempt == 0: continue raise ValueError(last_error) from exc + except Exception as exc: + state.invalidate_download_url(source_url) + last_error = f"failed to connect to upstream: {exc}" + state.verbose_log(f"Connection error on attempt {attempt + 1}: {exc}") + if attempt == 0: + continue + raise ValueError(last_error) from exc + try: first_chunk = validate_remote_media_response(response, range_header) return response, first_chunk @@ -77,6 +98,7 @@ def open_remote_media( response.close() state.invalidate_download_url(source_url) last_error = str(exc) + state.verbose_log(f"Validation failed on attempt {attempt + 1}: {exc}") if attempt == 0: continue raise diff --git a/buzz/models.py b/buzz/models.py index 2a70e91..9c2f875 100644 --- a/buzz/models.py +++ b/buzz/models.py @@ -15,6 +15,7 @@ class DavConfig(BaseModel): poll_interval_secs: int = 10 bind: str = "0.0.0.0" port: int = 9999 + stream_buffer_size: int = 0 state_dir: str = "/app/data" hook_command: str = "" anime_patterns: tuple[str, ...] = (DEFAULT_ANIME_PATTERN,) @@ -52,6 +53,7 @@ def load(cls, path: str = DEFAULT_DAV_CONFIG_PATH) -> "DavConfig": poll_interval_secs=int(raw.get("poll_interval_secs", 10)), bind=str(server.get("bind", "0.0.0.0")), port=int(server.get("port", 9999)), + stream_buffer_size=int(server.get("stream_buffer_size", 0)), state_dir=str(raw.get("state_dir", "/app/data")), hook_command=str(hooks.get("on_library_change", "")).strip(), curator_url=str( diff --git a/scripts/jellyfin_update.sh b/scripts/jellyfin_update.sh deleted file mode 100644 index 6372e5f..0000000 --- a/scripts/jellyfin_update.sh +++ /dev/null @@ -1,32 +0,0 @@ -#!/bin/sh - -# JELLYFIN UPDATE script -# This script refreshes the Jellyfin library after the curation layer is ready. - -jellyfin_url="${JELLYFIN_URL:-http://jellyfin:8096}" -jellyfin_api_key="${JELLYFIN_API_KEY:-}" - -if [ -z "$jellyfin_api_key" ] || [ "$jellyfin_api_key" = "" ]; then - echo "JELLYFIN_API_KEY is not set, skipping library refresh" - exit 0 -fi - -if [ -z "$jellyfin_url" ]; then - echo "JELLYFIN_URL is not set, skipping library refresh" - exit 0 -fi - -echo "Triggering Jellyfin library scan at: $jellyfin_url" - -# Trigger a full library scan -response="$(curl --connect-timeout 5 --max-time 30 --fail-with-body -sS -X POST \ - "$jellyfin_url/Library/Refresh?api_key=$jellyfin_api_key" 2>&1)" -status=$? - -if [ "$status" -ne 0 ]; then - printf '%s\n' "$response" >&2 - echo "Jellyfin library refresh failed" >&2 - exit "$status" -fi - -echo "Jellyfin library refresh requested" diff --git a/scripts/media_update.sh b/scripts/media_update.sh index ce95d1e..c044599 100644 --- a/scripts/media_update.sh +++ b/scripts/media_update.sh @@ -6,9 +6,6 @@ case "$media_server" in plex) exec bash /app/scripts/plex_update.sh "$@" ;; -jellyfin) - exec bash /app/scripts/jellyfin_update.sh "$@" - ;; *) echo "Unsupported MEDIA_SERVER: $media_server" >&2 exit 1 diff --git a/tests/test_buzz.py b/tests/test_buzz.py index e20d78c..e948e21 100644 --- a/tests/test_buzz.py +++ b/tests/test_buzz.py @@ -1,10 +1,11 @@ import json +import subprocess import tempfile import threading import time import unittest from pathlib import Path -from unittest.mock import patch +from unittest.mock import MagicMock, patch from fastapi.testclient import TestClient @@ -14,6 +15,7 @@ from buzz.core.state import ( BuzzState, LibraryBuilder, + Poller, canonical_snapshot, dav_rel_path, normalize_posix_path, @@ -448,6 +450,87 @@ def test_identical_syncs_after_first_change_are_stable(self): self.assertFalse(second["changed"]) self.assertEqual(second["changed_paths"], []) + def test_sync_excludes_internal_roots_from_changed_paths(self): + with tempfile.TemporaryDirectory() as tmpdir: + config = Config( + token="token", + poll_interval_secs=10, + bind="127.0.0.1", + port=9999, + state_dir=tmpdir, + hook_command="", + anime_patterns=(r"\b[a-fA-F0-9]{8}\b",), + enable_all_dir=True, + enable_unplayable_dir=True, + request_timeout_secs=30, + user_agent="buzz-tests", + version_label="buzz/test", + rd_update_delay_secs=0, + curator_url="", + ) + client = self.FakeRD( + torrents_list=[ + { + "id": "BROKEN1", + "filename": "Broken Torrent", + "bytes": 42, + "progress": 0, + "status": "error", + "ended": "2026-01-01T00:00:00Z", + "links": [], + } + ], + torrent_infos={ + "BROKEN1": { + "id": "BROKEN1", + "status": "error", + "filename": "Broken Torrent", + "links": [], + "files": [ + { + "id": 1, + "path": "/Broken.Movie.mkv", + "bytes": 42, + "selected": 1, + } + ], + } + }, + ) + state = BuzzState(config, client=client) + + report = state.sync(trigger_hook=False) + + self.assertTrue(report["changed"]) + self.assertEqual(report["changed_paths"], []) + self.assertEqual(report["added_paths"], []) + + def test_poller_formats_change_log_across_multiple_lines(self): + state = MagicMock() + poller = Poller(state) + + message = poller._format_change_message( + [ + "movies/The.Lord.of.the.Rings.The.Fellowship.of.the.Ring.2001.EXTENDED.2160p.UHD.BluRay.x265-BOREDOR", + "movies/The.Lord.of.the.Rings.The.Return.Of.The.King.2003.EXTENDED.2160p.UHD.BluRay.x265-BOREDOR", + ], + [], + [], + 96, + ) + + self.assertEqual( + message, + "\n".join( + [ + "Real-Debrid library changed (96 torrents):", + " +2 added", + " movies/The.Lord.of.the.Rings.The.Fellowship.of.the.Ring.2001.EXTENDED.2160p.UHD.BluRay.x265-BOREDOR", + " movies/The.Lord.of.the.Rings.The.Return.Of.The.King.2003.EXTENDED.2160p.UHD.BluRay.x265-BOREDOR", + ] + ), + ) + def test_identical_syncs_do_not_enqueue_duplicate_hooks(self): with tempfile.TemporaryDirectory() as tmpdir: config = Config( @@ -477,6 +560,75 @@ def test_identical_syncs_do_not_enqueue_duplicate_hooks(self): self.assertFalse(second["changed"]) self.assertEqual(enqueued, [["movies/Movie 2026"]]) + def test_sync_enqueues_curator_rebuild_without_hook_command(self): + with tempfile.TemporaryDirectory() as tmpdir: + config = Config( + token="token", + poll_interval_secs=10, + bind="127.0.0.1", + port=9999, + state_dir=tmpdir, + hook_command="", + anime_patterns=(r"\b[a-fA-F0-9]{8}\b",), + enable_all_dir=True, + enable_unplayable_dir=True, + request_timeout_secs=30, + user_agent="buzz-tests", + version_label="buzz/test", + curator_url="http://curator.invalid/rebuild", + ) + state = BuzzState(config, client=self._create_fake_rd()) + enqueued = [] + state._enqueue_hook = lambda changed_roots: enqueued.append( + list(changed_roots) + ) + + report = state.sync() + + self.assertTrue(report["changed"]) + self.assertEqual(enqueued, [["movies/Movie 2026"]]) + + @patch("buzz.core.state.record_event") + @patch("buzz.core.state.subprocess.run") + def test_run_hook_logs_stdout_and_stderr_on_failure( + self, mock_run, mock_record_event + ): + config = Config( + token="token", + poll_interval_secs=10, + bind="127.0.0.1", + port=9999, + state_dir="/tmp/buzz-tests", + hook_command="sh /app/scripts/media_update.sh", + anime_patterns=(r"\b[a-fA-F0-9]{8}\b",), + enable_all_dir=True, + enable_unplayable_dir=True, + request_timeout_secs=30, + user_agent="buzz-tests", + version_label="buzz/test", + curator_url="", + ) + state = BuzzState(config, client=None) + mock_run.side_effect = subprocess.CalledProcessError( + 2, + ["sh", "/app/scripts/media_update.sh", "movies/Interstellar"], + output="hook stdout", + stderr="hook stderr", + ) + + state._run_hook(["movies/Interstellar"]) + + mock_record_event.assert_called_once_with( + "\n".join( + [ + "Library update hook failed with exit code 2: ['sh', '/app/scripts/media_update.sh', 'movies/Interstellar']", + "stdout:\nhook stdout", + "stderr:\nhook stderr", + ] + ), + level="error", + ) + def test_existing_snapshot_digest_stays_stable_across_restart(self): with tempfile.TemporaryDirectory() as tmpdir: config = Config( @@ -1049,5 +1201,192 @@ def close(self): open_remote_media(self.state, node, None) +class DavBufferedStreamingTests(unittest.TestCase): + """Thread-safety tests for the buffered streaming path (stream_buffer_size >= 64KB).""" + + # 256KB buffer — large enough to exercise the buffered code path. + BUFFER_SIZE = 256 * 1024 + CHUNK_SIZE = 64 * 1024 + + class FakeResponse: + """Streaming response backed by a memoryview; supports read() and close().""" + + def __init__(self, body: bytes, content_type: str = "video/x-matroska"): + self._stream = memoryview(body) + self.headers = {"Content-Type": content_type} + self.closed = False + + def read(self, amount=-1): + if amount is None or amount < 0: + amount = len(self._stream) + chunk = self._stream[:amount].tobytes() + self._stream = self._stream[amount:] + return chunk + + def close(self): + self.closed = True + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + self.close() + return False + + def _make_dav_app(self): + tmpdir = tempfile.mkdtemp() + self.addCleanup(__import__("shutil").rmtree, tmpdir) + state_dir = Path(tmpdir) + snapshot = { + "dirs": ["", "movies", "movies/Test Film"], + "files": { + "movies/Test Film/film.mkv": { + "type": "remote", + "size": str(self.BUFFER_SIZE * 2), # bigger than the buffer + "source_url": "https://example.invalid/source", + "mime_type": "video/x-matroska", + "modified": "2026-01-01T00:00:00Z", + "etag": "etag-buf-1", + }, + }, + } + (state_dir / "library_snapshot.json").write_text( + json.dumps(snapshot), encoding="utf-8" + ) + config = Config( + token="token", + poll_interval_secs=10, + bind="127.0.0.1", + port=9999, + state_dir=str(state_dir), + hook_command="", + anime_patterns=(r"\b[a-fA-F0-9]{8}\b",), + enable_all_dir=True, + enable_unplayable_dir=True, + request_timeout_secs=30, + user_agent="buzz-tests", + version_label="buzz/test", + rd_update_delay_secs=0, + stream_buffer_size=self.BUFFER_SIZE, + ) + rd_patcher = patch("buzz.dav_app.RD", return_value=DavAppTests.FakeRD()) + self.addCleanup(rd_patcher.stop) + rd_patcher.start() + return DavApp(config) + + def _get_serve_dav(self, dav_app): + """Return the serve_dav route endpoint directly for generator-level testing.""" + for route in dav_app.app.routes: + if ( + getattr(route, "path", None) == "/dav/{path:path}" + and "GET" in getattr(route, "methods", set()) + ): + return route.endpoint + raise AssertionError("serve_dav GET route not found") + + def _mock_request(self, url_path: str): + req = MagicMock() + req.method = "GET" + req.url.path = url_path + req.headers.get.return_value = None + return req + + # ------------------------------------------------------------------ + # Happy-path: all bytes flow through the buffered path correctly + # ------------------------------------------------------------------ + + def test_buffered_streaming_all_bytes_received(self): + dav_app = self._make_dav_app() + payload = bytes(range(256)) * (self.BUFFER_SIZE * 2 // 256) + fake_response = self.FakeResponse(payload) + + with patch( + "buzz.dav_app.open_remote_media", + return_value=(fake_response, b""), + ): + client = TestClient(dav_app.app) + r = client.get("/dav/movies/Test%20Film/film.mkv") + + self.assertEqual(r.status_code, 200) + self.assertEqual(r.content, payload) + self.assertTrue(fake_response.closed) + + # ------------------------------------------------------------------ + # Helpers for generator-level thread tests + # ------------------------------------------------------------------ + + def _start_patches(self, fake_response): + """ + Start persistent patches for open_remote_media and StreamingResponse. + Returns the captured raw sync generator after serve_dav is called. + Both patches remain active until tearDown via addCleanup, so the + open_remote_media mock is still in place when the generator runs. + """ + captured = {} + + def fake_streaming_response(content, **kwargs): + captured["gen"] = content + return MagicMock(status_code=200) + + orm_patch = patch("buzz.dav_app.open_remote_media", return_value=(fake_response, b"")) + sr_patch = patch("buzz.dav_app.StreamingResponse", side_effect=fake_streaming_response) + orm_patch.start() + sr_patch.start() + self.addCleanup(orm_patch.stop) + self.addCleanup(sr_patch.stop) + return captured + + # ------------------------------------------------------------------ + # Thread cleanup: background thread joins after normal generator exit + # ------------------------------------------------------------------ + + def test_background_thread_joins_after_normal_completion(self): + dav_app = self._make_dav_app() + payload = bytes(range(256)) * (self.BUFFER_SIZE * 2 // 256) + fake_response = self.FakeResponse(payload) + mock_req = self._mock_request("/dav/movies/Test%20Film/film.mkv") + + captured = self._start_patches(fake_response) + serve_dav = self._get_serve_dav(dav_app) + serve_dav(path="movies/Test%20Film/film.mkv", request=mock_req) + gen = captured["gen"] + + threads_before = threading.active_count() + # Exhaust the generator; the finally block runs when StopIteration is raised. + received = b"".join(gen) + threads_after = threading.active_count() + + self.assertEqual(received, payload) + # The background thread must have joined before the generator returned. + self.assertLessEqual(threads_after, threads_before) + self.assertTrue(fake_response.closed) + + # ------------------------------------------------------------------ + # Thread cleanup: background thread joins after premature generator close + # ------------------------------------------------------------------ + + def test_background_thread_joins_after_early_close(self): + dav_app = self._make_dav_app() + # Large payload so the background thread is still active when we close. + payload = bytes(range(256)) * (self.BUFFER_SIZE * 2 // 256) + fake_response = self.FakeResponse(payload) + mock_req = self._mock_request("/dav/movies/Test%20Film/film.mkv") + + captured = self._start_patches(fake_response) + serve_dav = self._get_serve_dav(dav_app) + serve_dav(path="movies/Test%20Film/film.mkv", request=mock_req) + gen = captured["gen"] + + threads_before = threading.active_count() + # Read one chunk then abandon the rest. + next(gen) + # gen.close() throws GeneratorExit into the generator, firing the finally block + # synchronously: stop_event.set() -> t.join(timeout=5) -> response.close() + gen.close() + threads_after = threading.active_count() + + # Background thread must have joined before gen.close() returned. + self.assertLessEqual(threads_after, threads_before) + self.assertTrue(fake_response.closed) if __name__ == "__main__": unittest.main()