diff --git a/api_server.py b/api_server.py index 7f59ef4..7fa691b 100644 --- a/api_server.py +++ b/api_server.py @@ -15,7 +15,6 @@ import re import uuid import mimetypes from typing import List, Union, Any -from urllib.parse import quote import json from datetime import datetime @@ -23,9 +22,9 @@ import time from contextlib import asynccontextmanager import random import asyncio +from concurrent.futures import ThreadPoolExecutor import uvloop asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) -from starlette.middleware.base import BaseHTTPMiddleware from starlette.background import BackgroundTask import sys @@ -51,14 +50,35 @@ magic_mime = magic.Magic(mime=True) class ZeroSizeFileError(Exception): """Custom exception for zero-size files found or downloaded.""" -class RequestLoggingMiddleware(BaseHTTPMiddleware): - async def dispatch(self, request: Request, call_next): - # Log only method and URL at debug level to avoid flooding logs on active RSS polling - logger.debug(f"Request: {request.method} {request.url}") +class RequestLoggingMiddleware: + """Pure-ASGI request logger (no BaseHTTPMiddleware). + + BaseHTTPMiddleware runs the downstream app in a separate anyio task and pumps the + response through an in-memory stream, which adds per-request overhead and interacts + badly with streaming bodies, background tasks and client cancellation. This plain + ASGI middleware only wraps `send` to observe the response status line, so it never + buffers the body — the FileResponse stream flows straight through untouched. + """ + def __init__(self, app): + self.app = app + + async def __call__(self, scope, receive, send): + if scope["type"] != "http": + await self.app(scope, receive, send) + return + # Log only method and path (with query, matching the old request.url logging) at + # debug level to avoid flooding logs on active RSS polling. + _qs = scope.get("query_string") or b"" + _path = scope["path"] + (f"?{_qs.decode('latin-1')}" if _qs else "") + logger.debug(f"Request: {scope['method']} {_path}") + + async def send_wrapper(message): + if message["type"] == "http.response.start": + logger.debug(f"Response status: {message['status']}") + await send(message) + try: - response = await call_next(request) - logger.debug(f"Response status: {response.status_code}") - return response + await self.app(scope, receive, send_wrapper) except Exception as e: logger.error(f"Request processing error: {str(e)}") raise @@ -71,6 +91,11 @@ Config = get_settings() HTTP_DOWNLOAD_SEMAPHORE = asyncio.Semaphore(3) # semaphore for live HTTP media requests BACKGROUND_DOWNLOAD_SEMAPHORE = asyncio.Semaphore(2) # semaphore for background cache worker download_queue = asyncio.Queue(maxsize=100) +# How stale a temp_* file's mtime must be before a serve refreshes it (keeps the 1h +# sweeper from deleting an actively-viewed video). Well below 1h so the file stays alive, +# but large enough that the mtime — and thus FileResponse's ETag — is stable across the +# rapid requests of one resume/seek session. +TEMP_MTIME_REFRESH_INTERVAL = 300 # seconds # In-flight download dedup registry: maps (channel, post_id, file_unique_id) to the # shared Future of an ongoing download. The FIRST request for a key runs the download in @@ -117,7 +142,13 @@ async def _supervised(factory, name: str, min_restart_interval: float = 60.0): async def lifespan(_: FastAPI): setup_logging(Config["log_level"]) - + # Enlarge the default threadpool: SQLite/python-magic/pickle/os.walk all run via + # asyncio.to_thread, and the interpreter default (min(32, cpu+4) = 5-6 on a 1-2 CPU + # container) is too small under load. Configurable via IO_THREAD_POOL_SIZE. + loop = asyncio.get_running_loop() + io_executor = ThreadPoolExecutor(max_workers=Config["io_thread_pool_size"], thread_name_prefix="io") + loop.set_default_executor(io_executor) + base_cache_dir = os.path.abspath("./data/cache") os.makedirs(base_cache_dir, exist_ok=True) # Create cache directory @@ -141,6 +172,8 @@ async def lifespan(_: FastAPI): except asyncio.CancelledError: pass await client.stop() + # Shut the io threadpool down so its threads don't linger past a reload/restart. + io_executor.shutdown(wait=False) app = FastAPI(title="Pyrogram Bridge", lifespan=lifespan) app.add_middleware(RequestLoggingMiddleware) @@ -244,16 +277,33 @@ async def delayed_delete_file(file_path: str, delay: int = 300) -> None: async def prepare_file_response(file_path: str, request: Request, delete_after: bool = False, - media_key: tuple[str, int, str] | None = None) -> StreamingResponse: - """Prepare a streaming file response with HTTP Range request support.""" + media_key: tuple[str, int, str] | None = None) -> Response: + """Serve a cached media file via Starlette's FileResponse. + + FileResponse handles Range/If-Range/206/416/multipart and sets + Accept-Ranges/ETag/Last-Modified itself, and reads the file efficiently (no per-64KB + to_thread hop that starved the threadpool). We keep: the early 404 pre-check, the MIME + logic (python-magic + SQLite type cache), the stage-2 temp_* mtime touch, and the + delete_after BackgroundTask. + """ + # `request` is unused now that FileResponse parses the Range header itself, but the + # signature is kept for call-site compatibility (and future needs). if not os.path.exists(file_path): raise HTTPException(status_code=404, detail="File not found") # Keep an actively-viewed large-video temp file alive: refresh its mtime so the 1h # sweeper (which deletes temp_* by mtime) won't remove it out from under a viewer. + # DEBOUNCED: FileResponse derives ETag/Last-Modified from mtime, so touching on EVERY + # serve would make the validators change per request and break a player's `If-Range` + # resume (it would restart with a 200 instead of a 206). We only refresh when the + # mtime is already older than TEMP_MTIME_REFRESH_INTERVAL — far below the 1h sweeper + # window, so the file still stays alive, but the ETag is stable across the rapid + # requests of a single resume/seek session. if os.path.basename(file_path).startswith("temp_"): try: - await asyncio.to_thread(os.utime, file_path, None) + age = time.time() - await asyncio.to_thread(os.path.getmtime, file_path) + if age > TEMP_MTIME_REFRESH_INTERVAL: + await asyncio.to_thread(os.utime, file_path, None) except OSError as e: logger.debug(f"Failed to refresh mtime for {file_path}: {e}") @@ -281,107 +331,23 @@ async def prepare_file_response(file_path: str, request: Request, delete_after: logger.debug(f"Determined media type for {os.path.basename(file_path)}: {media_type}") - try: - total_size = os.path.getsize(file_path) - except FileNotFoundError: - raise HTTPException(status_code=404, detail="File not found") - range_header = request.headers.get("range") - - if range_header: - # Parse Range header according to RFC 7233 - try: - range_value = range_header.strip() - if not range_value.startswith("bytes="): - raise ValueError("Only bytes ranges are supported") - range_spec = range_value[len("bytes="):] - if range_spec.startswith("-"): - # Suffix range: bytes=-N (last N bytes) - suffix_length = int(range_spec[1:]) - start = max(0, total_size - suffix_length) - end = total_size - 1 - elif range_spec.endswith("-"): - # Open-ended range: bytes=START- - start = int(range_spec[:-1]) - end = total_size - 1 - else: - # Full range: bytes=START-END - start_str, end_str = range_spec.split("-", 1) - start = int(start_str) - end = int(end_str) - except (ValueError, IndexError) as e: - logger.debug(f"Invalid Range header '{range_header}': {e}") - return Response( - status_code=416, - headers={"Content-Range": f"bytes */{total_size}"} - ) - - # Clamp end to file size - 1 (RFC 7233 allows end >= total_size) - end = min(end, total_size - 1) - - # If start is beyond file size, return 416 - if start >= total_size: - return Response( - status_code=416, - headers={"Content-Range": f"bytes */{total_size}"} - ) - - content_length = end - start + 1 - status_code = 206 - headers = { - "Content-Disposition": ( - f"inline; filename=\"{os.path.basename(file_path)}\"; " - f"filename*=UTF-8''{quote(os.path.basename(file_path))}" - ), - # Files are addressed by file_unique_id which is immutable in Telegram, - # so it is safe to cache them aggressively on the client side. - "Cache-Control": "public, max-age=86400, immutable", - "Accept-Ranges": "bytes", - "Content-Range": f"bytes {start}-{end}/{total_size}", - "Content-Length": str(content_length), - } - else: - # No Range header — serve full file with status 200 - start = 0 - end = total_size - 1 - status_code = 200 - headers = { - "Content-Disposition": ( - f"inline; filename=\"{os.path.basename(file_path)}\"; " - f"filename*=UTF-8''{quote(os.path.basename(file_path))}" - ), - # Files are addressed by file_unique_id which is immutable in Telegram, - # so it is safe to cache them aggressively on the client side. - "Cache-Control": "public, max-age=86400, immutable", - "Accept-Ranges": "bytes", - "Content-Length": str(total_size), - } - - chunk_size = 64 * 1024 # 64 KB chunks - - async def file_chunk_generator(): - """Async generator that reads file in chunks; each chunk is a separate open/seek/read/close.""" - bytes_remaining = end - start + 1 - offset = start - while bytes_remaining > 0: - to_read = min(chunk_size, bytes_remaining) - def read_at(path, off, size): - # Open, seek, read, and close within a single thread call to avoid fd leaks - with open(path, "rb") as f: - f.seek(off) - return f.read(size) - data = await asyncio.to_thread(read_at, file_path, offset, to_read) - if not data: - break - bytes_remaining -= len(data) - offset += len(data) - yield data - + # Delete the temporary file once the response has been fully sent (stage-2 delete_after). + # FileResponse runs this BackgroundTask after streaming the body. background = BackgroundTask(delayed_delete_file, file_path) if delete_after else None - return StreamingResponse( - content=file_chunk_generator(), - status_code=status_code, + + # FileResponse handles Range/If-Range/206/416/multipart and sets + # Accept-Ranges/ETag/Last-Modified itself. Do NOT hand-build Content-Disposition: + # passing filename= makes FileResponse emit `inline; filename*=UTF-8''...` on its own + # (setting it manually would double the header). + # + # Files are addressed by file_unique_id which is immutable in Telegram, so it is safe + # to cache them aggressively on the client side. + return FileResponse( + file_path, media_type=media_type, - headers=headers, + filename=os.path.basename(file_path), + content_disposition_type="inline", + headers={"Cache-Control": "public, max-age=86400, immutable"}, background=background, ) diff --git a/config.py b/config.py index 0053cb0..a8fc395 100644 --- a/config.py +++ b/config.py @@ -129,4 +129,8 @@ def get_settings() -> dict[str, Any]: "media_download_timeout_min": _parse_int_env("MEDIA_DOWNLOAD_TIMEOUT_MIN", 120), "media_download_timeout_max": _parse_int_env("MEDIA_DOWNLOAD_TIMEOUT_MAX", 1800), "media_download_min_speed": _parse_int_env("MEDIA_DOWNLOAD_MIN_SPEED", 256 * 1024), + # Size of the asyncio default ThreadPoolExecutor. SQLite/python-magic/pickle/os.walk + # all run via asyncio.to_thread; the interpreter default (min(32, cpu+4)) is only 5-6 + # on a 1-2 CPU container, which starves those under load. 32 gives ample headroom. + "io_thread_pool_size": _parse_int_env("IO_THREAD_POOL_SIZE", 32), } diff --git a/dockercompose.yml b/dockercompose.yml index 800c685..c22711a 100644 --- a/dockercompose.yml +++ b/dockercompose.yml @@ -28,6 +28,7 @@ services: # MEDIA_DOWNLOAD_TIMEOUT_MIN: 120 # Min per-download timeout, seconds — also the timeout for regular (non-large) files (default: 120) # MEDIA_DOWNLOAD_TIMEOUT_MAX: 1800 # Max per-download timeout, seconds — cap for the largest videos (default: 1800) # MEDIA_DOWNLOAD_MIN_SPEED: 262144 # Assumed floor download speed, bytes/s — large-video timeout ≈ file_size / this, clamped to [MIN,MAX] (default: 262144 = 256 KB/s) + # IO_THREAD_POOL_SIZE: 32 # Size of the asyncio default threadpool for blocking I/O (SQLite/python-magic/pickle/os.walk); raise on a busy 1-2 CPU box (default: 32) PYROGRAM_BRIDGE_URL: https://pgbridge.example.com API_PORT: 80 TOKEN: ХХХ diff --git a/requirements.txt b/requirements.txt index aafaac3..a30a961 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,7 @@ fastapi==0.115.8 +# Pinned explicitly (fastapi only requires a range): the stage-3 FileResponse Range +# behaviour and the multipart-byteranges test are tied to this Starlette version. +starlette==0.45.3 uvicorn==0.34.0 python-multipart==0.0.20 Kurigram==2.2.22 @@ -12,3 +15,4 @@ python-magic==0.4.27 bleach[css]==6.1.0 types-bleach pytest-asyncio==1.4.0 +httpx==0.28.1 diff --git a/tests/mock_config.py b/tests/mock_config.py index 6668616..d6bacfe 100644 --- a/tests/mock_config.py +++ b/tests/mock_config.py @@ -36,4 +36,5 @@ def get_settings(): "media_download_timeout_min": 120, "media_download_timeout_max": 1800, "media_download_min_speed": 256 * 1024, + "io_thread_pool_size": 32, } diff --git a/tests/test_stage3_fileresponse.py b/tests/test_stage3_fileresponse.py new file mode 100644 index 0000000..03bbc4e --- /dev/null +++ b/tests/test_stage3_fileresponse.py @@ -0,0 +1,363 @@ +# flake8: noqa +# pylint: disable=protected-access, missing-function-docstring, missing-class-docstring +# pylint: disable=redefined-outer-name, logging-fstring-interpolation, line-too-long +# pylance: disable=reportMissingImports, reportMissingModuleSource +""" +Stage 3 (FileResponse + HTTP-layer cleanup) regression tests. + +Covers: +- The HTTP Range matrix asserted against the FINAL (Starlette FileResponse) behavior, + each case documented, including the RFC-7233-permitted deltas vs the old hand-rolled + streaming (garbage header -> 400 not 416; multi-range -> proper multipart 206 not 416; + 416 Content-Range now `*/size`; ETag/Last-Modified now present; ASCII filename no longer + gets a redundant filename*= form). +- Stage-2 behavior preserved through the rewrite: a served temp_* file still has its mtime + refreshed; delete_after still schedules the temp-file BackgroundTask (and it actually + runs); the media_key MIME cache is still consulted and populated. +- The pure-ASGI RequestLoggingMiddleware: a normal request still returns (body intact, not + buffered/truncated) and is logged. + +Before/after Range matrix (2048-byte file), old = hand-rolled streaming, new = FileResponse: + + request | old | new (FileResponse) + -----------------------+-----------------------------+----------------------------------- + no Range | 200 full | 200 full (+ETag/Last-Modified) + bytes=0-499 | 206 bytes 0-499/2048 | 206 bytes 0-499/2048 (same bytes) + bytes=500- | 206 bytes 500-2047/2048 | 206 bytes 500-2047/2048 (same) + bytes=-500 | 206 bytes 1548-2047/2048 | 206 bytes 1548-2047/2048 (same) + bytes=999999- (>EOF) | 416 `bytes */2048` | 416 `*/2048` (Starlette omits the + | | `bytes ` prefix; RFC-acceptable) + garbage header | 416 | 400 Bad Request (RFC 7233 lets a + | | server reject/ignore a malformed + | | Range; Starlette returns 400) + bytes=0-9,20-29 | 416 (old parser bug) | 206 multipart/byteranges (correct) + +All 206 responses that carry data return byte-for-byte identical slices old vs new, so no +real regression is hidden behind an "accepted difference". +""" +import os +import sys +import time +import logging + +import pytest + +# Add project root to sys.path and mock the config module (same pattern as the other tests). +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) +sys.modules['config'] = __import__('tests.mock_config', fromlist=['get_settings']) + +from fastapi import FastAPI, Request +from fastapi.testclient import TestClient + +import api_server + + +# --------------------------------------------------------------------------- # +# Helpers +# --------------------------------------------------------------------------- # +BODY = bytes(range(256)) * 8 # 2048 deterministic bytes +SIZE = len(BODY) + + +def _make_client(file_path, delete_after=False, media_key=None): + """Mount prepare_file_response on a tiny app so it is driven through real ASGI + (FileResponse computes Range/206/416 at send time, so it must be exercised via a client).""" + app = FastAPI() + + @app.get("/f") + async def _serve(request: Request): + return await api_server.prepare_file_response( + file_path, request=request, delete_after=delete_after, media_key=media_key + ) + + return TestClient(app) + + +@pytest.fixture +def sample_file(tmp_path): + fp = tmp_path / "myfile.bin" + fp.write_bytes(BODY) + return str(fp) + + +# --------------------------------------------------------------------------- # +# 3.1 — Range matrix against the FINAL FileResponse behavior. +# --------------------------------------------------------------------------- # +def test_no_range_returns_full_200(sample_file): + c = _make_client(sample_file) + r = c.get("/f") + assert r.status_code == 200 + assert r.content == BODY + assert r.headers["content-length"] == str(SIZE) + assert r.headers["accept-ranges"] == "bytes" + assert r.headers["cache-control"] == "public, max-age=86400, immutable" + # Content-Disposition: inline, filename set by FileResponse itself (no manual double-set). + assert r.headers["content-disposition"].startswith("inline") + assert 'filename="myfile.bin"' in r.headers["content-disposition"] + # NEW vs old: FileResponse adds validators. Old streaming had neither. + assert r.headers.get("etag") + assert r.headers.get("last-modified") + + +def test_range_prefix_0_499(sample_file): + c = _make_client(sample_file) + r = c.get("/f", headers={"Range": "bytes=0-499"}) + assert r.status_code == 206 + assert r.headers["content-range"] == f"bytes 0-499/{SIZE}" + assert r.headers["content-length"] == "500" + assert r.content == BODY[:500] # exact slice, identical to old behavior + + +def test_range_open_ended_500_to_eof(sample_file): + c = _make_client(sample_file) + r = c.get("/f", headers={"Range": "bytes=500-"}) + assert r.status_code == 206 + assert r.headers["content-range"] == f"bytes 500-{SIZE - 1}/{SIZE}" + assert r.headers["content-length"] == str(SIZE - 500) + assert r.content == BODY[500:] + + +def test_range_suffix_last_500(sample_file): + c = _make_client(sample_file) + r = c.get("/f", headers={"Range": "bytes=-500"}) + assert r.status_code == 206 + assert r.headers["content-range"] == f"bytes {SIZE - 500}-{SIZE - 1}/{SIZE}" + assert r.headers["content-length"] == "500" + assert r.content == BODY[-500:] + + +def test_range_start_past_eof_returns_416(sample_file): + c = _make_client(sample_file) + r = c.get("/f", headers={"Range": "bytes=999999-"}) + assert r.status_code == 416 + # DELTA (RFC-acceptable): Starlette's 416 Content-Range is `*/size` (it omits the + # `bytes ` unit prefix the old code emitted). Same information, permitted by RFC 7233. + assert r.headers["content-range"] == f"*/{SIZE}" + + +def test_garbage_range_header_returns_400(sample_file): + c = _make_client(sample_file) + r = c.get("/f", headers={"Range": "somethinggarbage"}) + # DELTA (RFC-acceptable): the old hand-rolled parser returned 416 on an unparseable + # header; Starlette rejects a malformed Range with 400 Bad Request. RFC 7233 permits a + # server to reject/ignore a bad Range; this is a conscious accepted difference, not a + # data regression (no bytes are served either way). + assert r.status_code == 400 + + +def test_multi_range_returns_multipart_206(sample_file): + c = _make_client(sample_file) + r = c.get("/f", headers={"Range": "bytes=0-9,20-29"}) + # DELTA (improvement): the old parser mis-split multi-ranges and returned 416. + # FileResponse serves a proper multipart/byteranges 206 containing both slices. + # Starlette 0.45.3 quirk: it advertises the multipart boundary via the `content-range` + # header (rather than content-type, which stays the file's own media_type); the body is + # a real multipart/byteranges document. We assert on the boundary marker + both slices. + assert r.status_code == 206 + assert r.headers["content-range"].startswith("multipart/byteranges") + body = r.content + assert BODY[0:10] in body + assert BODY[20:30] in body + + +# --------------------------------------------------------------------------- # +# 3.1 (stage-2 preserved) — temp_* mtime touch survives the FileResponse swap. +# --------------------------------------------------------------------------- # +def test_temp_file_mtime_refreshed_when_stale(tmp_path): + # A temp_* file whose mtime is OLDER than the refresh interval gets touched, so the + # 1h sweeper won't delete an actively-viewed video. + temp_file = tmp_path / "temp_bigvid" + temp_file.write_bytes(b"videodata") + stale = time.time() - 10000 # >> TEMP_MTIME_REFRESH_INTERVAL (300s) + os.utime(temp_file, (stale, stale)) + + c = _make_client(str(temp_file)) + r = c.get("/f") + assert r.status_code == 200 + assert r.content == b"videodata" + assert os.path.getmtime(temp_file) > time.time() - 100 # refreshed by the serve + + +def test_temp_file_mtime_stable_when_fresh(tmp_path): + # DEBOUNCE: a temp_* file touched RECENTLY (within the refresh interval) is NOT + # re-touched — so FileResponse's mtime-derived ETag stays stable across the rapid + # requests of a single resume/seek session and `If-Range` resume keeps working. + temp_file = tmp_path / "temp_freshvid" + temp_file.write_bytes(b"videodata") + recent = time.time() - 30 # << TEMP_MTIME_REFRESH_INTERVAL (300s) + os.utime(temp_file, (recent, recent)) + mtime_before = os.path.getmtime(temp_file) + + c = _make_client(str(temp_file)) + r1 = c.get("/f") + r2 = c.get("/f") + assert r1.status_code == 200 and r2.status_code == 200 + # mtime unchanged -> the ETag is identical across the two serves. + assert os.path.getmtime(temp_file) == mtime_before + assert r1.headers.get("etag") == r2.headers.get("etag") + + +def test_non_temp_file_mtime_untouched(tmp_path): + normal = tmp_path / "regularfile" + normal.write_bytes(b"data") + stale = time.time() - 10000 + os.utime(normal, (stale, stale)) + + c = _make_client(str(normal)) + c.get("/f") + assert os.path.getmtime(normal) < time.time() - 5000 # left alone + + +# --------------------------------------------------------------------------- # +# 3.1 (stage-2 preserved) — delete_after still schedules the temp-file BackgroundTask. +# --------------------------------------------------------------------------- # +async def test_delete_after_attaches_background_task(tmp_path): + fp = tmp_path / "temp_todelete" + fp.write_bytes(b"x") + + class _Req: + headers = {} + + resp = await api_server.prepare_file_response(str(fp), request=_Req(), delete_after=True) + # FileResponse must carry a non-None background that deletes exactly this file. + assert resp.background is not None + assert resp.background.func is api_server.delayed_delete_file + assert resp.background.args == (str(fp),) + + +async def test_no_delete_after_has_no_background(tmp_path): + fp = tmp_path / "keepme" + fp.write_bytes(b"x") + + class _Req: + headers = {} + + resp = await api_server.prepare_file_response(str(fp), request=_Req(), delete_after=False) + assert resp.background is None + + +def test_delete_after_background_runs_and_removes_file(tmp_path, monkeypatch): + fp = tmp_path / "temp_gone" + fp.write_bytes(BODY) + + # Patch the module-level deleter to remove immediately (real one sleeps 300s); the + # BackgroundTask picks up this reference at prepare_file_response call time. + async def _delete_now(path, delay=300): + os.remove(path) + + monkeypatch.setattr(api_server, "delayed_delete_file", _delete_now) + + c = _make_client(str(fp), delete_after=True) + r = c.get("/f") + assert r.status_code == 200 + assert r.content == BODY + # TestClient blocks until the background task has run. + assert not os.path.exists(fp) + + +# --------------------------------------------------------------------------- # +# 3.1 (stage-2 preserved) — media_key MIME cache is consulted and populated. +# --------------------------------------------------------------------------- # +def test_media_key_mime_cache_hit_used(sample_file, monkeypatch): + calls = {"get": 0, "set": 0, "magic": 0} + + def fake_get(db, ch, pid, fid): + calls["get"] += 1 + return "video/mp4" # cache HIT + + def fake_set(*a, **k): + calls["set"] += 1 + + def fake_magic(_path): + calls["magic"] += 1 + return "application/octet-stream" + + monkeypatch.setattr(api_server, "get_mime_type_sync", fake_get) + monkeypatch.setattr(api_server, "set_mime_type_sync", fake_set) + monkeypatch.setattr(api_server.magic_mime, "from_file", fake_magic) + + c = _make_client(sample_file, media_key=("chan", 42, "fid")) + r = c.get("/f") + assert r.status_code == 200 + # Cache hit: python-magic never invoked, nothing re-written to the cache, MIME applied. + assert calls["get"] == 1 + assert calls["magic"] == 0 + assert calls["set"] == 0 + assert r.headers["content-type"].startswith("video/mp4") + + +def test_media_key_mime_cache_miss_populates(sample_file, monkeypatch): + written = {} + + def fake_get(db, ch, pid, fid): + return None # cache MISS + + def fake_set(db, ch, pid, fid, mime): + written["mime"] = mime + + def fake_magic(_path): + return "image/png" + + monkeypatch.setattr(api_server, "get_mime_type_sync", fake_get) + monkeypatch.setattr(api_server, "set_mime_type_sync", fake_set) + monkeypatch.setattr(api_server.magic_mime, "from_file", fake_magic) + + c = _make_client(sample_file, media_key=("chan", 42, "fid")) + r = c.get("/f") + assert r.status_code == 200 + # Miss -> python-magic result is both applied and persisted for next time. + assert written.get("mime") == "image/png" + assert r.headers["content-type"].startswith("image/png") + + +# --------------------------------------------------------------------------- # +# 3.1 — 404 pre-check kept. +# --------------------------------------------------------------------------- # +def test_missing_file_returns_404(tmp_path): + c = _make_client(str(tmp_path / "does_not_exist")) + r = c.get("/f") + assert r.status_code == 404 + + +# --------------------------------------------------------------------------- # +# 3.2 — pure-ASGI RequestLoggingMiddleware: request returns intact and is logged. +# --------------------------------------------------------------------------- # +def test_asgi_logging_middleware_passes_body_and_logs(tmp_path, caplog): + fp = tmp_path / "streamed.bin" + fp.write_bytes(BODY) + + app = FastAPI() + app.add_middleware(api_server.RequestLoggingMiddleware) + + @app.get("/f") + async def _serve(request: Request): + return await api_server.prepare_file_response(str(fp), request=request) + + client = TestClient(app) + with caplog.at_level(logging.DEBUG, logger="api_server"): + r = client.get("/f") + + # Body flows straight through the middleware — not buffered/truncated. + assert r.status_code == 200 + assert r.content == BODY + messages = " ".join(rec.getMessage() for rec in caplog.records) + assert "Request: GET /f" in messages + assert "Response status: 200" in messages + + +def test_asgi_logging_middleware_range_still_works(tmp_path): + fp = tmp_path / "streamed.bin" + fp.write_bytes(BODY) + + app = FastAPI() + app.add_middleware(api_server.RequestLoggingMiddleware) + + @app.get("/f") + async def _serve(request: Request): + return await api_server.prepare_file_response(str(fp), request=request) + + client = TestClient(app) + r = client.get("/f", headers={"Range": "bytes=0-99"}) + # 206 streaming still works through the pure-ASGI middleware (send not buffered). + assert r.status_code == 206 + assert r.content == BODY[:100]