from __future__ import annotations import asyncio import copy import gzip import json import tempfile import threading import time import unittest from collections import OrderedDict from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock, Mock, patch IMPORT_ERROR = None try: import aiohttp from backend import app from backend import upstream_http except ModuleNotFoundError as exc: # pragma: no cover app = None IMPORT_ERROR = exc @unittest.skipIf(app is None, f"proxy dependencies unavailable: {IMPORT_ERROR}") class FakeResponse: def __init__(self, status=200, headers=None, chunks=(b"ok",)): self.status = status self.headers = headers or {} self.content = self self.chunks = chunks self.released = False def release(self): self.released = True async def read(self, amount=-1): return b"".join(self.chunks)[:amount if amount >= 0 else None] def iter_chunked(self, size): async def generator(): for chunk in self.chunks: yield chunk return generator() async def __aenter__(self): return self async def __aexit__(self, *args): self.release() @unittest.skipIf(app is None, f"proxy dependencies unavailable: {IMPORT_ERROR}") class DownloadConnectionWarmTests(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): self.response = FakeResponse() self.state = SimpleNamespace( download_warm_enabled=True, download_warm_task=None, download_warm_at=float("-inf"), upstream_semaphore=asyncio.Semaphore(2), http_session=SimpleNamespace(head=Mock(return_value=self.response)), ) self.application = SimpleNamespace(state=self.state) async def asyncTearDown(self): if self.state.download_warm_task is not None: self.state.download_warm_task.cancel() await asyncio.gather(self.state.download_warm_task, return_exceptions=True) async def test_ping_returns_before_warmup_and_coalesces_repeated_visits(self): with patch.object(app, "app", self.application): first = await app.api_ping() task = self.state.download_warm_task second = await app.api_ping() self.assertEqual([first.status_code, second.status_code], [204, 204]) self.assertEqual(first.headers["cache-control"], "no-store") self.state.http_session.head.assert_not_called() self.assertIs(self.state.download_warm_task, task) await task self.state.http_session.head.assert_called_once() args, kwargs = self.state.http_session.head.call_args self.assertEqual(args, ("https://huggingface.co/robots.txt",)) self.assertFalse(kwargs["allow_redirects"]) self.assertEqual(kwargs["timeout"].total, 4) self.assertTrue(self.response.released) app.schedule_download_connection_warm(self.application) self.assertIs(self.state.download_warm_task, task) self.assertEqual(self.state.upstream_semaphore._value, 2) async def test_failed_warmup_backs_off_then_allows_retry(self): self.state.http_session.head.side_effect = aiohttp.ClientConnectionError() app.schedule_download_connection_warm(self.application) await self.state.download_warm_task app.schedule_download_connection_warm(self.application) self.assertEqual(self.state.http_session.head.call_count, 1) self.state.download_warm_at -= app.DOWNLOAD_CONNECTION_WARM_INTERVAL_SECONDS self.state.http_session.head.side_effect = None app.schedule_download_connection_warm(self.application) await self.state.download_warm_task self.assertEqual(self.state.http_session.head.call_count, 2) self.assertEqual(self.state.upstream_semaphore._value, 2) async def test_saturated_pool_skips_warmup_without_queuing(self): self.state.upstream_semaphore = asyncio.Semaphore(0) await app.warm_download_connection(self.application) self.state.http_session.head.assert_not_called() async def test_cancellation_releases_pool_and_disabled_state_stops_scheduling(self): entered = asyncio.Event() class StalledResponse(FakeResponse): async def __aenter__(self): entered.set() await asyncio.Future() self.state.http_session.head.return_value = StalledResponse() app.schedule_download_connection_warm(self.application) await entered.wait() task = self.state.download_warm_task task.cancel() with self.assertRaises(asyncio.CancelledError): await task self.assertEqual(self.state.upstream_semaphore._value, 2) self.state.download_warm_enabled = False self.state.download_warm_at = float("-inf") app.schedule_download_connection_warm(self.application) self.assertIs(self.state.download_warm_task, task) class FakeSession: def __init__(self, responses): self.responses = list(responses) self.urls = [] async def get(self, url, **kwargs): self.urls.append((url, kwargs)) response = self.responses.pop(0) if isinstance(response, Exception): raise response return response async def request(self, method, url, **kwargs): kwargs["method"] = method return await self.get(url, **kwargs) @unittest.skipIf(app is None, f"proxy dependencies unavailable: {IMPORT_ERROR}") class AsyncProxyTests(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): self._app_state = dict(app.app.state._state) self._assets_state = {name: getattr(app, name) for name in ( 'reader_assets_cache', 'reader_assets_cache_at', 'reader_assets_revision')} app.app.state.reader_assets_task = None app.app.state.reader_assets_lock = asyncio.Lock() app.app.state.http_session = FakeSession([]) app.app.state.upstream_semaphore = asyncio.Semaphore(1) app.initialize_reader_bucket_state(app.app) self._redirect_cache = dict(upstream_http.upstream_redirect_cache) upstream_http.upstream_redirect_cache.clear() async def asyncTearDown(self): task = getattr(app.app.state, 'reader_assets_task', None) if task is not None and task is not self._app_state.get('reader_assets_task'): task.cancel() await asyncio.gather(task, return_exceptions=True) await app.shutdown_reader_bucket_state(app.app) for name, value in self._assets_state.items(): setattr(app, name, value) app.app.state._state.clear() app.app.state._state.update(self._app_state) upstream_http.upstream_redirect_cache.clear() upstream_http.upstream_redirect_cache.update(self._redirect_cache) def test_reader_assets_decoder_filters_invalid_entries(self): valid_path = "objects/aa/" + "b" * 64 + "/document.pdf" valid_linearized_path = "objects/aa/" + "f" * 64 + "/linearized.pdf" valid_azw_path = "objects/aa/" + "e" * 64 + "/document.azw" page_manifest = "objects/aa/" + "d" * 64 + "/page-manifest.json" media_path = "objects/aa/" + "c" * 64 + "/ffmpeg-audio-mp3-v1/audio.mp3" raw = gzip.compress(json.dumps({"v": 1, "f": { "VoiceOfML/Test\0book.doc": {"s": 2, "m": "p", "p": valid_path}, "VoiceOfML/Test\0book-linearized.pdf": {"s": 2, "m": "p", "p": valid_linearized_path}, "VoiceOfML/Test\0book.azw": {"s": 2, "m": "e", "p": valid_azw_path}, "VoiceOfML/Test\0sound.wma": {"s": 2, "m": "a", "p": media_path}, "VoiceOfML/Test\0scan.pdf": {"s": 2, "m": "p", "p": page_manifest}, "bad-path": {"s": 2, "m": "p", "p": "objects/../../secret.pdf"}, "null-path": {"s": 2, "m": "p", "p": None}, "unknown-mode": {"s": 2, "m": "unknown", "p": valid_path}, "unhashable-mode": {"s": 2, "m": [], "p": valid_path}, "failed": {"s": 4}, }}).encode()) self.assertEqual(app.decode_reader_assets(raw), {"v": 1, "f": { "VoiceOfML/Test\0book.doc": {"s": 2, "m": "p", "p": valid_path}, "VoiceOfML/Test\0book-linearized.pdf": {"s": 2, "m": "p", "p": valid_linearized_path}, "VoiceOfML/Test\0book.azw": {"s": 2, "m": "e", "p": valid_azw_path}, "VoiceOfML/Test\0sound.wma": {"s": 2, "m": "a", "p": media_path}, "VoiceOfML/Test\0scan.pdf": {"s": 2, "m": "p", "p": page_manifest}, "failed": {"s": 4}, }}) async def test_reader_assets_refresh_is_cached_and_releases_response(self): payload = gzip.compress(json.dumps({"v": 1, "f": {"failed": {"s": 4}}}).encode()) response = FakeResponse(200, chunks=(payload,)) session = FakeSession([response]) app.app.state.http_session = session app.app.state.reader_assets_lock = asyncio.Lock() old_cache, old_at = app.reader_assets_cache, app.reader_assets_cache_at try: app.reader_assets_cache = {"v": 1, "f": {}} app.reader_assets_cache_at = 0 with patch.object(app, "load_reading_catalog", AsyncMock(return_value=None)): first = await app.get_reader_assets() second = await app.get_reader_assets() self.assertEqual(first, {"v": 1, "f": {"failed": {"s": 4}}}) self.assertIs(first, second) self.assertEqual(len(session.urls), 1) self.assertTrue(response.released) finally: app.reader_assets_cache, app.reader_assets_cache_at = old_cache, old_at async def test_search_returns_while_shared_reader_assets_refresh_is_stalled(self): started, release = asyncio.Event(), asyncio.Event() payload = gzip.compress(json.dumps({"v": 1, "f": {"new": {"s": 4}}}).encode()) response = FakeResponse(200, chunks=(payload,)) async def get(*args, **kwargs): started.set() await release.wait() return response app.app.state.http_session = SimpleNamespace(get=AsyncMock(side_effect=get)) app.app.state.reader_assets_lock = asyncio.Lock() app.app.state.reader_assets_task = None with patch.object(app, "reader_assets_cache", {"v": 1, "f": {}}), \ patch.object(app, "load_reading_catalog", AsyncMock(return_value=None)), \ patch.object(app, "reader_assets_cache_at", 0), \ patch.object(app, "search", return_value={"results": [], "total": 0, "page": 1, "page_size": 100}): try: first = await asyncio.wait_for(app.api_search(app.SearchRequest()), 1) await asyncio.wait_for(started.wait(), 1) task = app.app.state.reader_assets_task second = await asyncio.wait_for(app.api_search(app.SearchRequest()), 1) self.assertEqual([first.status_code, second.status_code], [200, 200]) self.assertIs(task, app.app.state.reader_assets_task) self.assertFalse(task.done()) release.set() await task self.assertEqual(app.reader_assets_cache["f"], {"new": {"s": 4}}) self.assertTrue(response.released) self.assertEqual(app.app.state.http_session.get.call_count, 1) finally: release.set() if app.app.state.reader_assets_task: await app.app.state.reader_assets_task async def test_reader_assets_refresh_failure_backs_off_with_existing_cache(self): response = FakeResponse(503) session = FakeSession([response]) app.app.state.http_session = session app.app.state.reader_assets_lock = asyncio.Lock() old_cache, old_at = app.reader_assets_cache, app.reader_assets_cache_at try: app.reader_assets_cache = {"v": 1, "f": {"old": {"s": 4}}} app.reader_assets_cache_at = time.monotonic() - app.READER_ASSETS_CACHE_TTL_SECONDS - 1 first = await app.get_reader_assets() second = await app.get_reader_assets() self.assertIs(first, second) self.assertEqual(len(session.urls), 1) self.assertTrue(response.released) finally: app.reader_assets_cache, app.reader_assets_cache_at = old_cache, old_at async def test_initial_reader_assets_failure_is_not_http_cached_and_recovers(self): payload = gzip.compress(json.dumps({"v": 1, "f": {"ready": {"s": 4}}}).encode()) session = FakeSession([FakeResponse(503), FakeResponse(200, chunks=(payload,))]) app.app.state.http_session = session app.app.state.reader_assets_lock = asyncio.Lock() with patch.object(app, "reader_assets_cache", {"v": 1, "f": {}}), \ patch.object(app, "load_reading_catalog", AsyncMock(return_value=None)), \ patch.object(app, "reader_assets_cache_at", 0), \ patch.object(app, 'time', SimpleNamespace(monotonic=Mock(return_value=100), time=time.time, perf_counter=time.perf_counter)) as clock: failed = await app.api_reader_assets() self.assertEqual(failed.headers["cache-control"], "no-store") await app.get_reader_assets() self.assertEqual(len(session.urls), 1) clock.monotonic.return_value += 6 recovered = await app.api_reader_assets() self.assertEqual(json.loads(recovered.body)["f"], {"ready": {"s": 4}}) self.assertEqual(len(session.urls), 2) async def test_reader_bucket_cache_hit_preserves_success_cache_control(self): path = f"objects/aa/{'a' * 64}/{'b' * 16}/page-manifest.json" with patch.object(app, "read_reader_bucket_resource", return_value=b"manifest") as read: first = await app.api_reader_bucket_resource(path) second = await app.api_reader_bucket_resource(path) self.assertEqual(first.body, b"manifest") self.assertEqual(second.body, b"manifest") self.assertEqual(read.call_count, 1) self.assertEqual(second.headers["cache-control"], "public, max-age=3600") async def test_random_reader_uses_complete_snapshot_without_waiting_for_refresh(self): snapshot = {"v": 1, "f": {"ready": {"s": 4}}} entered, release = asyncio.Event(), asyncio.Event() async def refresh(): entered.set() await release.wait() return snapshot with patch.object(app, "reader_assets_cache", snapshot), \ patch.object(app, "reader_assets_cache_at", time.monotonic() - app.READER_ASSETS_CACHE_TTL_SECONDS - 1), \ patch.object(app, "refresh_reader_assets", side_effect=refresh), \ patch.object(app, "ensure_reader_indexes_async", AsyncMock()), \ patch.object(app, "converted_reader_indices", return_value={}), \ patch.object(app, "reader_record_indices", [0]), \ patch.object(app, "records", [{"Repo": "VoiceOfML/Test", "File": "book", "Extension": "txt", "Folder": [], "Size": 1}]): result = await asyncio.wait_for(app.api_random_reader(None), .5) await entered.wait() self.assertFalse(release.is_set()) self.assertEqual(result.status_code, 200) self.assertEqual(json.loads(result.body)["File"], "book") status = await asyncio.wait_for(app.api_random_reader_status(None), .5) self.assertEqual(json.loads(status.body), {"available": True, "count": 1}) release.set() await app.app.state.reader_assets_task async def test_cold_random_reader_waits_for_first_complete_asset_snapshot(self): entered, release = asyncio.Event(), asyncio.Event() async def refresh(): entered.set() await release.wait() return {"v": 1, "f": {}} with patch.object(app, "reader_assets_cache", {"v": 1, "f": {}}), \ patch.object(app, "reader_assets_cache_at", 0), \ patch.object(app, "refresh_reader_assets", side_effect=refresh), \ patch.object(app, "ensure_reader_indexes_async", AsyncMock()), \ patch.object(app, "converted_reader_indices", return_value={}), \ patch.object(app, "reader_record_indices", []): task = asyncio.create_task(app.api_random_reader(None)) await entered.wait() self.assertFalse(task.done()) release.set() self.assertEqual((await task).status_code, 404) async def test_reader_asset_decode_does_not_block_ping(self): entered, release = threading.Event(), threading.Event() def decode(raw): entered.set() release.wait(3) return {"v": 1, "f": {}} app.app.state.http_session = FakeSession([FakeResponse(200, chunks=(b"sidecar",))]) with patch.object(app, "reader_assets_cache_at", 0), \ patch.object(app, "decode_reader_assets", side_effect=decode), \ patch.object(app, "load_reading_catalog", AsyncMock(return_value=None)): task = asyncio.create_task(app.refresh_reader_assets()) try: self.assertTrue(await asyncio.to_thread(entered.wait, 1)) self.assertEqual((await asyncio.wait_for(app.api_ping(), .2)).status_code, 204) self.assertFalse(task.done()) finally: release.set() await task async def test_v2_chapter_manifest_and_search_index_reuse_bounded_cache(self): prefix = "chapters/ebook/epub/" + "a" * 64 + "/" + "b" * 16 + "/" for name in ("chapter-manifest.json", "epub-search-index.json.gz"): with self.subTest(name=name), patch.object(app, "read_reader_bucket_resource", return_value=b"chapter metadata") as read: first = await app.api_reader_bucket_resource(prefix + name) second = await app.api_reader_bucket_resource(prefix + name) self.assertEqual(first.status_code, 200) self.assertEqual(second.body, first.body) self.assertEqual(read.call_count, 1) self.assertLessEqual(app.app.state.reader_bucket_cache_bytes, app.READER_BUCKET_CACHE_MAX_BYTES) async def test_bucket_shutdown_keeps_slot_owned_until_filesystem_thread_finishes(self): started, release = threading.Event(), threading.Event() path = "chapters/ebook/epub/" + "a" * 64 + "/" + "b" * 16 + "/chapter-manifest.json" semaphore = app.app.state.reader_bucket_semaphore = asyncio.Semaphore(1) def read(_path, _limit): started.set() release.wait(3) return b"manifest" with patch.object(app, "read_reader_bucket_resource", side_effect=read): loading = asyncio.create_task(app.get_reader_bucket_resource(path)) shutdown = None try: self.assertTrue(await asyncio.to_thread(started.wait, 1)) shutdown = asyncio.create_task(app.shutdown_reader_bucket_state(app.app)) await asyncio.sleep(0.02) self.assertEqual(semaphore._value, 0) self.assertEqual(len(app.app.state.reader_bucket_workers), 1) self.assertFalse(shutdown.done()) finally: release.set() await asyncio.gather(loading, return_exceptions=True) if shutdown is not None: await shutdown self.assertEqual(semaphore._value, 1) self.assertEqual(app.app.state.reader_bucket_workers, set()) self.assertEqual(app.app.state.reader_bucket_cache, OrderedDict()) async def test_reader_content_routes_both_v2_bucket_origins_to_proxy(self): request = SimpleNamespace(method="GET", headers={"Range": "bytes=0-1"}) paths = { "reader-assets-v2": "documents/pdf/ppt/" + "a" * 64 + "/document.pdf", "pdf-pages-v2": "derived/test/" + "a" * 32 + "/document.pdf", } for bucket, path in paths.items(): url = f"https://huggingface.co/buckets/vomebook/{bucket}/resolve/{path}" with self.subTest(bucket=bucket), patch.object(app, "api_reader_bucket_resource", AsyncMock(return_value=app.Response(b"ab", status_code=206))) as proxy: response = await app.api_reader_content(request, url=url, id=None) self.assertEqual(response.status_code, 206) proxy.assert_awaited_once_with(path, request) async def test_reader_bucket_page_uses_immutable_cache_control(self): path = f"objects/aa/{'a' * 64}/{'b' * 16}/pages/page-000002.webp" with patch.object(app, "read_reader_bucket_resource", return_value=b"page"): response = await app.api_reader_bucket_resource(path) self.assertEqual(response.headers["cache-control"], "public, max-age=31536000, immutable") async def test_bucket_pdf_streams_initial_get_head_and_range_without_buffering(self): path = "documents/pdf/test/" + "a" * 64 + "/document.pdf" for method, headers, status, upstream_headers in ( ("GET", {}, 200, {"Content-Length": "10", "Accept-Ranges": "bytes"}), ("HEAD", {}, 200, {"Content-Length": "10", "Accept-Ranges": "bytes"}), ("GET", {"Range": "bytes=0-1"}, 206, {"Content-Length": "2", "Content-Range": "bytes 0-1/10", "Accept-Ranges": "bytes"}), ): with self.subTest(method=method, headers=headers): upstream = FakeResponse(status, upstream_headers, chunks=(b"ab",)) session = FakeSession([upstream]) app.app.state.http_session = session semaphore = app.app.state.upstream_semaphore = asyncio.Semaphore(1) app.app.state.reader_range_cache = OrderedDict() app.app.state.reader_range_cache_bytes = 0 request = SimpleNamespace(method=method, headers=headers) with patch.object(app, "get_reader_bucket_resource", AsyncMock()) as whole_read: response = await app.api_reader_bucket_resource(path, request) whole_read.assert_not_awaited() self.assertEqual(response.status_code, status) self.assertEqual(session.urls[0][0], app.reader_bucket_source_url(path)) self.assertEqual(session.urls[0][1]["method"], method) self.assertEqual(session.urls[0][1]["headers"], {"Accept-Encoding": "identity", **headers}) if method != "HEAD": self.assertFalse(upstream.released) self.assertEqual([chunk async for chunk in response.body_iterator], [b"ab"]) self.assertTrue(upstream.released) self.assertEqual(semaphore._value, 1) async def test_bucket_media_streams_head_and_ranges_without_whole_file_reads(self): for kind, suffix, mime in (("audio", "flac", "audio/flac"), ("video", "mp4", "video/mp4")): path = f"media/{kind}/native/{'a' * 64}/{kind}.{suffix}" for method, forwarded, status, upstream_headers in ( ("GET", {}, 200, {"Content-Length": "10", "Accept-Ranges": "bytes"}), ("HEAD", {}, 200, {"Content-Length": "10", "Accept-Ranges": "bytes"}), ("GET", {"Range": "bytes=0-1"}, 206, {"Content-Length": "2", "Content-Range": "bytes 0-1/10", "Accept-Ranges": "bytes"}), ): with self.subTest(kind=kind, method=method, range=forwarded): upstream = FakeResponse(status, {"Content-Type": "application/octet-stream", **upstream_headers}, chunks=(b"ab",)) session = app.app.state.http_session = FakeSession([upstream]) semaphore = app.app.state.upstream_semaphore = asyncio.Semaphore(1) app.app.state.reader_range_cache = OrderedDict() app.app.state.reader_range_cache_bytes = 0 with patch.object(app, "get_reader_bucket_resource", AsyncMock()) as whole_read: response = await app.api_reader_bucket_resource(path, SimpleNamespace(method=method, headers=forwarded)) whole_read.assert_not_awaited() self.assertEqual(response.status_code, status) self.assertEqual(response.headers["content-type"], mime) self.assertNotIn("content-encoding", response.headers) self.assertEqual(session.urls[0][1]["headers"], {"Accept-Encoding": "identity", **forwarded}) if method != "HEAD": self.assertEqual([chunk async for chunk in response.body_iterator], [b"ab"]) if forwarded: cached = await app.api_reader_bucket_resource(path, SimpleNamespace(method=method, headers=forwarded)) self.assertEqual(cached.body, b"ab") self.assertNotIn("content-encoding", cached.headers) self.assertEqual(len(session.urls), 1) self.assertTrue(upstream.released) self.assertEqual(semaphore._value, 1) async def test_media_remains_unencoded_through_real_gzip_middleware_and_range_cache(self): path = f"media/audio/native/{'a' * 64}/audio.mp3" payload = b"media" * 400 for status in (200, 206): forwarded = {"Range": f"bytes=0-{len(payload) - 1}"} if status == 206 else {} headers = {"Content-Length": str(len(payload)), "Content-Encoding": "identity"} if status == 206: headers["Content-Range"] = f"bytes 0-{len(payload) - 1}/{len(payload) * 2}" upstream = FakeResponse(status, headers, chunks=(payload,)) app.app.state.http_session = FakeSession([upstream]) app.app.state.reader_range_cache = OrderedDict() app.app.state.reader_range_cache_bytes = 0 for cached in (False, True) if status == 206 else (False,): with self.subTest(status=status, cached=cached): response = await app.api_reader_bucket_resource(path, SimpleNamespace(method="GET", headers=forwarded)) messages = [] async def serve(scope, receive, send): await response(scope, receive, send) async def receive(): return {"type": "http.request", "body": b"", "more_body": False} async def send(message): messages.append(message) await app.GZipMiddleware(serve, minimum_size=1)( {"type": "http", "method": "GET", "asgi": {"spec_version": "2.4"}, "headers": [(b"accept-encoding", b"gzip")]}, receive, send) start = messages[0] self.assertEqual(start["status"], status) self.assertNotIn(b"content-encoding", dict(start["headers"])) self.assertEqual(dict(start["headers"])[b"content-length"], str(len(payload)).encode()) self.assertEqual(b"".join(message.get("body", b"") for message in messages[1:]), payload) self.assertTrue(upstream.released) self.assertEqual(app.app.state.upstream_semaphore._value, 1) async def test_complete_reader_text_negotiates_gzip_without_changing_decoded_bytes(self): payload = ("正文 UTF-8 literal \ufffd / unchanged\n" * 5000).encode() for extension in ("txt", "md", "html", "xhtml", "json"): for encoding in (b"gzip", b"identity"): with self.subTest(extension=extension, encoding=encoding): upstream = FakeResponse(200, {"Content-Length": str(len(payload)), "ETag": '"same-source"'}, chunks=tuple(payload[i:i + 65536] for i in range(0, len(payload), 65536))) app.app.state.http_session = FakeSession([upstream]) app.app.state.upstream_semaphore = asyncio.Semaphore(1) response = await app.stream_reader_document(SimpleNamespace(method="GET", headers={}), f"https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/book.{extension}") self.assertNotIn("content-encoding", response.headers) messages = [] async def serve(scope, receive, send): await response(scope, receive, send) async def receive(): return {"type": "http.request", "body": b"", "more_body": False} async def send(message): messages.append(message) await app.GZipMiddleware(serve, minimum_size=500)( {"type": "http", "method": "GET", "asgi": {"spec_version": "2.4"}, "headers": [(b"accept-encoding", encoding)]}, receive, send) headers = dict(messages[0]["headers"]) self.assertEqual(headers[b"etag"], b'W/"same-source"') body = b"".join(message.get("body", b"") for message in messages[1:]) self.assertEqual(gzip.decompress(body) if encoding == b"gzip" else body, payload) if encoding == b"gzip": self.assertEqual(headers[b"content-encoding"], b"gzip") self.assertIn(b"Accept-Encoding", headers[b"vary"]) self.assertLess(len(body), len(payload) // 2) else: self.assertNotIn(b"content-encoding", headers) self.assertTrue(upstream.released) self.assertEqual(app.app.state.upstream_semaphore._value, 1) async def test_text_compression_cannot_change_range_head_or_binary_policy(self): source = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/" for suffix in ("pdf", "epub", "docx", "jpg", "webp", "mp3", "mp4", "json.gz"): self.assertFalse(app.reader_body_allows_compression(source + "book." + suffix)) for method, headers, status in (("HEAD", {}, 200), ("GET", {"Range": "bytes=0-1"}, 200), ("GET", {"Range": "bytes=0-1"}, 206), ("GET", {"If-None-Match": '"A"'}, 200), ("GET", {}, 304), ("GET", {}, 416)): self.assertFalse(app.reader_body_allows_compression(source + "book.txt", SimpleNamespace(method=method, headers=headers), status)) async def test_large_media_streams_metadata_and_ranges_without_whole_file_limit(self): total = app.READER_DOCUMENT_MAX_BYTES + 100 for kind, suffix in (("audio", "mp3"), ("video", "mp4")): path = f"media/{kind}/native/{'a' * 64}/{kind}.{suffix}" for method, forwarded, status, headers in ( ("HEAD", {}, 200, {"Content-Length": str(total)}), ("GET", {"Range": "bytes=0-1"}, 206, {"Content-Length": "2", "Content-Range": f"bytes 0-1/{total}"}), ("GET", {"Range": "bytes=-2"}, 206, {"Content-Length": "2", "Content-Range": f"bytes {total - 2}-{total - 1}/{total}"}), ("GET", {"Range": f"bytes={total}-"}, 416, {"Content-Length": "0", "Content-Range": f"bytes */{total}"}), ): for source in ("bucket", "original"): with self.subTest(kind=kind, method=method, range=forwarded, source=source): upstream = FakeResponse(status, headers, chunks=() if status == 416 else (b"ab",)) session = app.app.state.http_session = FakeSession([upstream]) semaphore = app.app.state.upstream_semaphore = asyncio.Semaphore(1) app.app.state.reader_range_cache = OrderedDict() app.app.state.reader_range_cache_bytes = 0 request = SimpleNamespace(method=method, headers=forwarded) if source == "bucket": response = await app.api_reader_bucket_resource(path, request) else: url = f"https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/large.{suffix}" response = await app.api_reader_content(request, url=url, id=None) self.assertEqual(response.status_code, status) self.assertIsNone(session.urls[0][1]["timeout"].total) self.assertEqual(session.urls[0][1]["timeout"].sock_read, 60) if method != "HEAD": self.assertEqual(b"".join([chunk async for chunk in response.body_iterator]), b"" if status == 416 else b"ab") self.assertTrue(upstream.released) self.assertEqual(semaphore._value, 1) async def test_media_unknown_length_stream_does_not_use_document_body_limit(self): path = f"media/audio/native/{'a' * 64}/audio.mp3" upstream = FakeResponse(200, chunks=(b"abc", b"def")) app.app.state.http_session = FakeSession([upstream]) with patch.object(app, "READER_DOCUMENT_MAX_BYTES", 2), patch.object(app, "READER_RESOURCE_MAX_BYTES", 2): response = await app.api_reader_bucket_resource(path, SimpleNamespace(method="GET", headers={})) self.assertEqual(response.status_code, 200) self.assertEqual([chunk async for chunk in response.body_iterator], [b"abc", b"def"]) self.assertTrue(upstream.released) self.assertEqual(app.app.state.upstream_semaphore._value, 1) async def test_bucket_media_preserves_range_validation_and_cancellation_cleanup(self): path = f"media/video/native/{'a' * 64}/video.mp4" for status, headers in ( (206, {"Content-Length": "2", "Content-Range": "invalid"}), (206, {"Content-Length": "2", "Content-Range": "bytes 0-2/100"}), (206, {"Content-Length": "2", "Content-Range": "bytes 99-100/100"}), ): upstream = FakeResponse(status, headers) app.app.state.http_session = FakeSession([upstream]) semaphore = app.app.state.upstream_semaphore = asyncio.Semaphore(1) app.app.state.reader_range_cache = OrderedDict() app.app.state.reader_range_cache_bytes = 0 response = await app.api_reader_bucket_resource(path, SimpleNamespace(method="GET", headers={"Range": "bytes=0-1"})) self.assertEqual(response.status_code, 502) self.assertTrue(upstream.released) self.assertEqual(semaphore._value, 1) with patch.object(app, "open_download_response", AsyncMock(side_effect=asyncio.CancelledError)): with self.assertRaises(asyncio.CancelledError): await app.api_reader_bucket_resource(path, SimpleNamespace(method="GET", headers={})) self.assertEqual(semaphore._value, 1) async def test_bucket_pdf_validates_range_and_releases_on_failure_or_cancel(self): path = "derived/test/" + "a" * 32 + "/document.pdf" for status, headers, expected in ( (206, {"Content-Range": "invalid"}, 502), (206, {"Content-Range": f"bytes 0-1/{app.READER_DOCUMENT_MAX_BYTES + 1}"}, 413), (200, {"Content-Length": str(app.READER_DOCUMENT_MAX_BYTES + 1)}, 413), (200, {"Content-Encoding": "gzip"}, 502), (404, {}, 404), ): with self.subTest(status=status, headers=headers): upstream = FakeResponse(status, headers) app.app.state.http_session = FakeSession([upstream]) semaphore = app.app.state.upstream_semaphore = asyncio.Semaphore(1) response = await app.api_reader_bucket_resource(path, SimpleNamespace(method="GET", headers={})) self.assertEqual(response.status_code, expected) self.assertTrue(upstream.released) self.assertEqual(semaphore._value, 1) with patch.object(app, "open_download_response", AsyncMock(side_effect=asyncio.CancelledError)): with self.assertRaises(asyncio.CancelledError): await app.api_reader_bucket_resource(path, SimpleNamespace(method="GET", headers={})) self.assertEqual(semaphore._value, 1) async def test_bucket_pdf_redirect_cannot_switch_to_another_hf_source(self): path = "derived/test/" + "a" * 32 + "/document.pdf" redirect = FakeResponse(302, {"Location": "https://huggingface.co/datasets/Other/Private/resolve/main/book.pdf"}) session = app.app.state.http_session = FakeSession([redirect]) semaphore = app.app.state.upstream_semaphore = asyncio.Semaphore(1) response = await app.api_reader_bucket_resource(path, SimpleNamespace(method="GET", headers={})) self.assertEqual(response.status_code, 502) self.assertEqual(len(session.urls), 1) self.assertTrue(redirect.released) self.assertEqual(semaphore._value, 1) async def test_reader_content_accepts_resolved_v2_proxy_but_not_other_hosts(self): path = "derived/test/" + "a" * 32 + "/document.pdf" request = SimpleNamespace(method="GET", headers={"Range": "bytes=0-1"}) proxy_url = "https://voiceofml-search.hf.space/api/reader-bucket-resource?path=" + path with patch.object(app, "api_reader_bucket_resource", AsyncMock(return_value=app.Response(b"ab", status_code=206))) as bucket: with patch.object(app, "api_reader_resolve", AsyncMock(return_value=app.JSONResponse({"url": proxy_url}))): response = await app.api_reader_content(request, url=None, id="398vk0yyy8m29") self.assertEqual(response.status_code, 206) bucket.assert_awaited_once_with(path, request) bucket.reset_mock() response = await app.api_reader_content(request, url=proxy_url.replace("voiceofml-search.hf.space", "other.example"), id=None) self.assertEqual(response.status_code, 403) bucket.assert_not_awaited() async def test_reader_bucket_single_flight_coalesces_uncached_page(self): path = f"objects/aa/{'a' * 64}/{'b' * 16}/pages/page-000002.webp" started = threading.Event() release = threading.Event() def read(_path, max_bytes=app.READER_RESOURCE_MAX_BYTES): started.set() release.wait(1) return b"page" with patch.object(app, "read_reader_bucket_resource", side_effect=read) as bucket_read: first = asyncio.create_task(app.api_reader_bucket_resource(path)) self.assertTrue(await asyncio.to_thread(started.wait, 1)) second = asyncio.create_task(app.api_reader_bucket_resource(path)) await asyncio.sleep(0) self.assertEqual(bucket_read.call_count, 1) release.set() responses = await asyncio.gather(first, second) self.assertEqual([response.body for response in responses], [b"page", b"page"]) self.assertEqual(bucket_read.call_count, 1) self.assertNotIn(path, app.app.state.reader_bucket_cache) self.assertEqual(app.app.state.reader_bucket_inflight, {}) async def test_reader_bucket_semaphore_bounds_distinct_reads(self): app.app.state.reader_bucket_semaphore = asyncio.Semaphore(1) first_path = f"objects/aa/{'a' * 64}/{'b' * 16}/pages/page-000002.webp" second_path = f"objects/aa/{'a' * 64}/{'b' * 16}/pages/page-000003.webp" first_started = threading.Event() release = threading.Event() def read(path, max_bytes=app.READER_RESOURCE_MAX_BYTES): if path == first_path: first_started.set() release.wait(1) return path.encode() with patch.object(app, "read_reader_bucket_resource", side_effect=read) as bucket_read: first = asyncio.create_task(app.api_reader_bucket_resource(first_path)) self.assertTrue(await asyncio.to_thread(first_started.wait, 1)) second = asyncio.create_task(app.api_reader_bucket_resource(second_path)) await asyncio.sleep(0.05) self.assertEqual(bucket_read.call_count, 1) release.set() await asyncio.gather(first, second) self.assertEqual(bucket_read.call_count, 2) async def test_reader_bucket_failures_are_no_store_and_not_cached(self): path = f"objects/aa/{'a' * 64}/{'b' * 16}/page-manifest.json" invalid = await app.api_reader_bucket_resource("invalid") self.assertEqual(invalid.status_code, 400) self.assertEqual(invalid.headers["cache-control"], "no-store") for error, status in ((FileNotFoundError(), 404), (RuntimeError(), 502)): with self.subTest(status=status), patch.object( app, "read_reader_bucket_resource", side_effect=error ) as bucket_read: first = await app.api_reader_bucket_resource(path) second = await app.api_reader_bucket_resource(path) self.assertEqual(first.status_code, status) self.assertEqual(second.status_code, status) self.assertEqual(first.headers["cache-control"], "no-store") self.assertEqual(bucket_read.call_count, 2) async def test_reader_bucket_cache_evicts_by_bytes_and_skips_oversize_items(self): manifest = f"objects/aa/{'a' * 64}/{'b' * 16}/page-manifest.json" first_page = f"objects/aa/{'a' * 64}/{'b' * 16}/pages/page-000001.webp" other_manifest = f"objects/bb/{'c' * 64}/{'d' * 16}/page-manifest.json" old_total = app.READER_BUCKET_CACHE_MAX_BYTES old_item = app.READER_BUCKET_CACHE_MAX_ITEM_BYTES app.READER_BUCKET_CACHE_MAX_BYTES = 5 app.READER_BUCKET_CACHE_MAX_ITEM_BYTES = 4 payloads = {manifest: b"aaa", first_page: b"bbb", other_manifest: b"12345"} try: with patch.object(app, "read_reader_bucket_resource", side_effect=lambda path, limit: payloads.get(path)) as bucket_read: await app.api_reader_bucket_resource(manifest) await app.api_reader_bucket_resource(first_page) self.assertNotIn(manifest, app.app.state.reader_bucket_cache) self.assertEqual(app.app.state.reader_bucket_cache_bytes, 3) await app.api_reader_bucket_resource(manifest) await app.api_reader_bucket_resource(other_manifest) await app.api_reader_bucket_resource(other_manifest) self.assertEqual(bucket_read.call_count, 5) self.assertNotIn(other_manifest, app.app.state.reader_bucket_cache) self.assertLessEqual(app.app.state.reader_bucket_cache_bytes, 5) finally: app.READER_BUCKET_CACHE_MAX_BYTES = old_total app.READER_BUCKET_CACHE_MAX_ITEM_BYTES = old_item async def test_reader_bucket_shutdown_cancels_flights_and_clears_cache(self): task = asyncio.create_task(asyncio.sleep(10)) app.app.state.reader_bucket_inflight["path"] = task app.app.state.reader_bucket_cache["path"] = (time.monotonic(), b"data") app.app.state.reader_bucket_cache_bytes = 4 await app.shutdown_reader_bucket_state(app.app) self.assertTrue(task.cancelled()) self.assertEqual(app.app.state.reader_bucket_inflight, {}) self.assertEqual(app.app.state.reader_bucket_cache, {}) self.assertEqual(app.app.state.reader_bucket_cache_bytes, 0) async def test_relative_redirect_is_resolved_against_current_url(self): session = FakeSession([FakeResponse(302, {"Location": "next"}), FakeResponse(200)]) start = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/book.pdf" result = await app.open_download_response( session, start, aiohttp.ClientTimeout(total=1), url_validator=app.validate_voiceofml_source_url, ) self.assertEqual(result.status, 200) self.assertEqual([url for url, _ in session.urls], [start, "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/next"]) async def test_redirect_cannot_escape_voiceofml_dataset_scope(self): first = FakeResponse(302, {"Location": "https://huggingface.co/datasets/Other/Test/resolve/main/book.pdf"}) start = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/book.pdf" with self.assertRaises(ValueError): await app.open_download_response( FakeSession([first]), start, aiohttp.ClientTimeout(total=1), url_validator=app.validate_voiceofml_source_url, ) self.assertTrue(first.released) async def test_redirect_cannot_escape_reader_assets_scope(self): digest = "b" * 64 start = f"https://huggingface.co/datasets/vomebook/Reader-Assets/resolve/main/objects/aa/{digest}/document.pdf" first = FakeResponse(302, {"Location": "/datasets/VoiceOfML/Test/resolve/main/book.pdf"}) with self.assertRaises(ValueError): await app.open_download_response( FakeSession([first]), start, aiohttp.ClientTimeout(total=1), url_validator=app.validate_reader_source_url, ) self.assertTrue(first.released) async def test_multilayer_encoded_reader_assets_redirect_cannot_escape_scope(self): digest = "b" * 64 start = f"https://huggingface.co/datasets/%252576omebook/Reader-Assets/resolve/main/objects/aa/{digest}/document.pdf" first = FakeResponse(302, {"Location": "/datasets/VoiceOfML/Test/resolve/main/book.pdf"}) self.assertEqual(app.validate_reader_source_url(start), start) with self.assertRaises(ValueError): await app.open_download_response( FakeSession([first, FakeResponse(200)]), start, aiohttp.ClientTimeout(total=1), url_validator=app.validate_reader_source_url, ) self.assertTrue(first.released) async def test_redirect_to_disallowed_host_is_rejected_and_releases_first(self): first = FakeResponse(302, {"Location": "https://evil.test/x"}) with self.assertRaises(ValueError): await app.open_download_response(FakeSession([first]), "https://huggingface.co/x", aiohttp.ClientTimeout(total=1)) self.assertTrue(first.released) async def test_redirect_without_location_is_rejected_and_released(self): response = FakeResponse(302) with self.assertRaises(ValueError): await app.open_download_response(FakeSession([response]), "https://huggingface.co/x", aiohttp.ClientTimeout(total=1)) self.assertTrue(response.released) async def test_too_many_redirects_are_rejected(self): responses = [FakeResponse(302, {"Location": "/next"}) for _ in range(app.MAX_REDIRECTS + 1)] with self.assertRaises(ValueError): await app.open_download_response(FakeSession(responses), "https://huggingface.co/x", aiohttp.ClientTimeout(total=1)) async def test_resolved_redirect_target_is_reused_and_scoped(self): start = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/cached.pdf" final = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/actual.pdf" first = FakeSession([FakeResponse(302, {"Location": "actual.pdf"}), FakeResponse(200)]) result = await app.open_download_response( first, start, aiohttp.ClientTimeout(total=1), url_validator=app.validate_voiceofml_source_url, ) self.assertEqual(result.status, 200) self.assertEqual([url for url, _ in first.urls], [start, final]) second = FakeSession([FakeResponse(200)]) result = await app.open_download_response( second, start, aiohttp.ClientTimeout(total=1), url_validator=app.validate_voiceofml_source_url, ) self.assertEqual(result.status, 200) self.assertEqual([url for url, _ in second.urls], [final]) async def test_rejected_redirect_target_falls_back_to_fresh_walk(self): start = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/retry.pdf" stale = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/stale.pdf" fresh = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/fresh.pdf" upstream_http.upstream_redirect_cache[("GET", start)] = (time.monotonic(), stale) session = FakeSession([ FakeResponse(403), FakeResponse(302, {"Location": "fresh.pdf"}), FakeResponse(200), ]) result = await app.open_download_response( session, start, aiohttp.ClientTimeout(total=1), url_validator=app.validate_voiceofml_source_url, ) self.assertEqual(result.status, 200) self.assertEqual([url for url, _ in session.urls], [stale, start, fresh]) self.assertEqual(upstream_http.upstream_redirect_cache[("GET", start)][1], fresh) async def test_out_of_scope_cached_target_is_ignored(self): start = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/scope.pdf" escaped = "https://huggingface.co/datasets/vomebook/Reader-Assets/resolve/main/objects/aa/" + "b" * 64 + "/document.pdf" upstream_http.upstream_redirect_cache[("GET", start)] = (time.monotonic(), escaped) session = FakeSession([FakeResponse(200)]) result = await app.open_download_response( session, start, aiohttp.ClientTimeout(total=1), ) self.assertEqual(result.status, 200) self.assertEqual([url for url, _ in session.urls], [start]) async def test_non_redirect_response_is_returned_without_release(self): response = FakeResponse(404) result = await app.open_download_response(FakeSession([response]), "https://huggingface.co/x", aiohttp.ClientTimeout(total=1)) self.assertIs(result, response) self.assertFalse(response.released) async def test_download_rejects_external_source_before_session_call(self): session = FakeSession([]) app.app.state.http_session = session app.app.state.upstream_semaphore = asyncio.Semaphore(1) result = await app.api_download("x.pdf", "https://evil.test/x") self.assertEqual(result.status_code, 403) self.assertEqual(session.urls, []) async def test_download_upstream_error_releases_semaphore(self): session = FakeSession([FakeResponse(502)]) semaphore = asyncio.Semaphore(1) app.app.state.http_session = session app.app.state.upstream_semaphore = semaphore result = await app.api_download("x.pdf", "https://huggingface.co/datasets/VoiceOfML/A/x") self.assertEqual(result.status_code, 502) self.assertEqual(semaphore._value, 1) async def test_download_cancellation_releases_semaphore(self): semaphore = asyncio.Semaphore(1) app.app.state.http_session = FakeSession([]) app.app.state.upstream_semaphore = semaphore with patch.object(app, "open_download_response", AsyncMock(side_effect=asyncio.CancelledError)): with self.assertRaises(asyncio.CancelledError): await app.api_download("x.pdf", "https://huggingface.co/datasets/VoiceOfML/A/x") self.assertEqual(semaphore._value, 1) async def test_download_non_200_releases_upstream_and_semaphore(self): upstream = FakeResponse(404) app.app.state.http_session = FakeSession([upstream]) semaphore = asyncio.Semaphore(1) app.app.state.upstream_semaphore = semaphore result = await app.api_download("x.pdf", "https://huggingface.co/datasets/VoiceOfML/A/x") self.assertEqual(result.status_code, 502) self.assertTrue(upstream.released) self.assertEqual(semaphore._value, 1) async def test_download_response_stream_releases_resources_after_consumption(self): upstream = FakeResponse(200, chunks=(b"a", b"b")) app.app.state.http_session = FakeSession([upstream]) semaphore = asyncio.Semaphore(1) app.app.state.upstream_semaphore = semaphore response = await app.api_download("../book.pdf", "https://huggingface.co/datasets/VoiceOfML/A/x") chunks = [chunk async for chunk in response.body_iterator] self.assertEqual(chunks, [b"a", b"b"]) self.assertTrue(upstream.released) self.assertEqual(semaphore._value, 1) self.assertIn("filename*=UTF-8''book.pdf", response.headers["content-disposition"]) # Slow large transfers are bounded by inactivity, not a total deadline. timeout = app.app.state.http_session.urls[0][1]["timeout"] self.assertIsNone(timeout.total) self.assertEqual((timeout.connect, timeout.sock_read), (15, 60)) async def test_reader_content_forwards_range_and_preserves_partial_headers(self): upstream = FakeResponse(206, { "Content-Type": "application/pdf", "Content-Length": "2", "Content-Range": "bytes 0-1/10", "Accept-Ranges": "bytes", "ETag": '"reader"', "Set-Cookie": "secret=1", }, chunks=(b"ab",)) session = FakeSession([upstream]) semaphore = asyncio.Semaphore(1) app.app.state.http_session = session app.app.state.upstream_semaphore = semaphore request = SimpleNamespace(method="GET", headers={ "Range": "bytes=0-1", "If-None-Match": '"reader"', "If-Modified-Since": "Tue, 25 Aug 2026 00:00:00 GMT", "Cookie": "secret=1", }) url = "https://huggingface.co/datasets/vomebook/Reader-Assets/resolve/main/objects/aa/" + "b" * 64 + "/document.pdf" response = await app.api_reader_content(request, url) self.assertEqual(response.status_code, 206) self.assertEqual([chunk async for chunk in response.body_iterator], [b"ab"]) self.assertEqual(session.urls[0][1]["headers"], { "Accept-Encoding": "identity", "Range": "bytes=0-1", "If-None-Match": '"reader"', "If-Modified-Since": "Tue, 25 Aug 2026 00:00:00 GMT", }) self.assertEqual(response.headers["content-range"], "bytes 0-1/10") self.assertEqual(response.headers["content-length"], "2") self.assertEqual(response.headers["content-encoding"], "identity") self.assertEqual(response.headers["cache-control"], "no-store") self.assertIs(session.urls[0][1]["auto_decompress"], False) self.assertNotIn("set-cookie", response.headers) self.assertTrue(upstream.released) self.assertEqual(semaphore._value, 1) async def test_reader_content_reuses_completed_small_range_from_cache(self): upstream = FakeResponse(206, { "Content-Type": "application/pdf", "Content-Length": "2", "Content-Range": "bytes 0-1/10", "Accept-Ranges": "bytes", }, chunks=(b"ab",)) session = FakeSession([upstream]) app.app.state.http_session = session app.app.state.upstream_semaphore = asyncio.Semaphore(1) app.app.state.reader_range_cache = OrderedDict() app.app.state.reader_range_cache_bytes = 0 request = SimpleNamespace(method="GET", headers={"Range": "bytes=0-1"}) url = "https://huggingface.co/datasets/vomebook/Reader-Assets/resolve/main/objects/aa/" + "b" * 64 + "/document.pdf" first = await app.api_reader_content(request, url) self.assertEqual(first.headers["cache-control"], "no-store") self.assertEqual([chunk async for chunk in first.body_iterator], [b"ab"]) next(iter(app.app.state.reader_range_cache.values()))[2]["Cache-Control"] = "public, max-age=31536000" second = await app.api_reader_content(request, url) self.assertEqual(second.body, b"ab") self.assertEqual(second.headers["cache-control"], "no-store") self.assertEqual(len(session.urls), 1) async def test_reader_browser_cache_policy_only_changes_partial_pdf_responses(self): for suffix, mime, status, expected in ( ("pdf", "application/pdf", 206, "no-store"), ("PDF", "application/octet-stream", 206, "no-store"), ("bin", "application/pdf; charset=binary", 206, "no-store"), ("pdf", "application/pdf", 200, "public, max-age=3600"), ("pdf", "application/pdf", 304, "public, max-age=3600"), ("mp4", "video/mp4", 206, "public, max-age=3600"), ): with self.subTest(suffix=suffix, mime=mime, status=status): headers = {"Content-Type": mime, "Cache-Control": "public, max-age=3600", "ETag": '"same"'} result = app.reader_document_browser_headers( "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/book." + suffix, headers, status) self.assertEqual(result["Cache-Control"], expected) self.assertEqual(result["Content-Type"], mime) self.assertEqual(result["ETag"], '"same"') self.assertEqual(headers["Cache-Control"], "public, max-age=3600") async def test_reader_content_does_not_reuse_range_cache_for_conditional_request(self): first_upstream = FakeResponse(206, { "Content-Type": "application/pdf", "Content-Length": "2", "Content-Range": "bytes 0-1/10", "Accept-Ranges": "bytes", "ETag": '"reader"', }, chunks=(b"ab",)) second_upstream = FakeResponse(304, {"ETag": '"reader"'}) session = FakeSession([first_upstream, second_upstream]) app.app.state.http_session = session app.app.state.upstream_semaphore = asyncio.Semaphore(1) app.app.state.reader_range_cache = OrderedDict() app.app.state.reader_range_cache_bytes = 0 url = "https://huggingface.co/datasets/vomebook/Reader-Assets/resolve/main/objects/aa/" + "b" * 64 + "/document.pdf" first = await app.api_reader_content(SimpleNamespace(method="GET", headers={"Range": "bytes=0-1"}), url) self.assertEqual([chunk async for chunk in first.body_iterator], [b"ab"]) second = await app.api_reader_content(SimpleNamespace(method="GET", headers={ "Range": "bytes=0-1", "If-None-Match": '"reader"', }), url) self.assertEqual(second.status_code, 304) self.assertEqual(len(session.urls), 2) async def test_reader_range_cache_does_not_turn_origin_200_into_206(self): upstream = FakeResponse(200, {"Content-Type": "application/pdf", "Content-Length": "2"}, chunks=(b"ab",)) session = FakeSession([upstream]) app.app.state.http_session = session app.app.state.upstream_semaphore = asyncio.Semaphore(1) app.app.state.reader_range_cache = OrderedDict() app.app.state.reader_range_cache_bytes = 0 url = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/book.pdf" response = await app.api_reader_content(SimpleNamespace(method="GET", headers={"Range": "bytes=0-1"}), url) self.assertEqual(response.status_code, 200) self.assertEqual([chunk async for chunk in response.body_iterator], [b"ab"]) self.assertEqual(app.app.state.reader_range_cache, OrderedDict()) async def test_reader_range_cache_rejects_incomplete_body_and_retries_origin(self): headers = {"Content-Length": "2", "Content-Range": "bytes 0-1/10"} partial = FakeResponse(206, headers, chunks=(b"a",)) complete = FakeResponse(206, headers, chunks=(b"ab",)) session = app.app.state.http_session = FakeSession([partial, complete]) semaphore = app.app.state.upstream_semaphore = asyncio.Semaphore(1) app.app.state.reader_range_cache = OrderedDict() app.app.state.reader_range_cache_bytes = 0 request = SimpleNamespace(method="GET", headers={"Range": "bytes=0-1"}) url = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/incomplete.pdf" first = await app.api_reader_content(request, url) self.assertEqual([chunk async for chunk in first.body_iterator], [b"a"]) self.assertEqual(app.app.state.reader_range_cache, OrderedDict()) second = await app.api_reader_content(request, url) self.assertEqual([chunk async for chunk in second.body_iterator], [b"ab"]) third = await app.api_reader_content(request, url) self.assertEqual(third.body, b"ab") self.assertEqual(len(session.urls), 2) self.assertTrue(partial.released and complete.released) self.assertEqual(semaphore._value, 1) async def test_reader_content_rejects_oversized_content_length_before_streaming(self): upstream = FakeResponse(200, {"Content-Length": str(app.READER_DOCUMENT_MAX_BYTES + 1)}, chunks=(b"not-read",)) semaphore = asyncio.Semaphore(1) app.app.state.http_session = FakeSession([upstream]) app.app.state.upstream_semaphore = semaphore response = await app.api_reader_content( SimpleNamespace(method="GET", headers={}), "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/book.pdf", ) self.assertEqual(response.status_code, 413) self.assertTrue(upstream.released) self.assertEqual(semaphore._value, 1) async def test_reader_content_rejects_oversized_range_total(self): old_limit = app.READER_DOCUMENT_MAX_BYTES app.READER_DOCUMENT_MAX_BYTES = 3 upstream = FakeResponse(206, { "Content-Length": "1", "Content-Range": "bytes 0-0/4", }, chunks=(b"a",)) semaphore = asyncio.Semaphore(1) app.app.state.http_session = FakeSession([upstream]) app.app.state.upstream_semaphore = semaphore try: response = await app.api_reader_content( SimpleNamespace(method="GET", headers={}), "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/book.pdf", ) finally: app.READER_DOCUMENT_MAX_BYTES = old_limit self.assertEqual(response.status_code, 413) self.assertTrue(upstream.released) self.assertEqual(semaphore._value, 1) async def test_reader_content_enforces_actual_stream_limit_and_releases(self): old_limit = app.READER_DOCUMENT_MAX_BYTES app.READER_DOCUMENT_MAX_BYTES = 3 upstream = FakeResponse(200, {}, chunks=(b"ab", b"cd")) semaphore = asyncio.Semaphore(1) app.app.state.http_session = FakeSession([upstream]) app.app.state.upstream_semaphore = semaphore try: response = await app.api_reader_content( SimpleNamespace(method="GET", headers={}), "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/book.pdf", ) with self.assertRaises(app.ReaderDocumentLimitError): [chunk async for chunk in response.body_iterator] finally: app.READER_DOCUMENT_MAX_BYTES = old_limit self.assertTrue(upstream.released) self.assertEqual(semaphore._value, 1) async def test_reader_content_cancellation_releases_semaphore(self): semaphore = asyncio.Semaphore(1) app.app.state.upstream_semaphore = semaphore request = SimpleNamespace(method="GET", headers={}) with patch.object(app, "open_download_response", AsyncMock(side_effect=asyncio.CancelledError)): with self.assertRaises(asyncio.CancelledError): await app.api_reader_content( request, "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/book.pdf" ) self.assertEqual(semaphore._value, 1) async def test_reader_content_head_releases_resources_without_streaming(self): upstream = FakeResponse(200, {"Content-Length": "10", "Accept-Ranges": "bytes"}) session = FakeSession([upstream]) semaphore = asyncio.Semaphore(1) app.app.state.http_session = session app.app.state.upstream_semaphore = semaphore request = SimpleNamespace(method="HEAD", headers={}) response = await app.api_reader_content(request, "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/book.epub") self.assertEqual(response.status_code, 200) self.assertEqual(session.urls[0][1]["method"], "HEAD") self.assertIs(session.urls[0][1]["auto_decompress"], False) self.assertEqual(response.headers["content-length"], "10") self.assertTrue(upstream.released) self.assertEqual(semaphore._value, 1) async def test_reader_content_rejects_external_source_before_session_call(self): session = FakeSession([]) app.app.state.http_session = session app.app.state.upstream_semaphore = asyncio.Semaphore(1) result = await app.api_reader_content(SimpleNamespace(method="GET", headers={}), "https://evil.test/book.pdf") self.assertEqual(result.status_code, 403) self.assertEqual(session.urls, []) async def test_download_check_upstream_error_returns_502_and_releases(self): upstream = FakeResponse(503) app.app.state.http_session = FakeSession([upstream]) result = await app.api_download_check("https://huggingface.co/datasets/VoiceOfML/A/x") self.assertEqual(result.status_code, 502) self.assertTrue(upstream.released) async def test_download_check_uses_head_and_shared_semaphore(self): upstream = FakeResponse(200) session = FakeSession([upstream]) semaphore = asyncio.Semaphore(1) app.app.state.http_session = session app.app.state.upstream_semaphore = semaphore result = await app.api_download_check("https://huggingface.co/datasets/VoiceOfML/A/x") self.assertEqual(result.status_code, 200) self.assertEqual(session.urls[0][1]["method"], "HEAD") self.assertTrue(upstream.released) self.assertEqual(semaphore._value, 1) async def test_local_txt_file_is_served_without_remote_request(self): with tempfile.TemporaryDirectory() as directory: old_dir = app.TXT_DIR app.TXT_DIR = Path(directory) local = Path(directory) / "local.txt" local.write_text("local", encoding="utf-8") session = FakeSession([]) app.app.state.http_session = session try: response = await app.serve_txt_proxy("local.txt") self.assertEqual(response.path, local) self.assertEqual(session.urls, []) finally: app.TXT_DIR = old_dir async def test_remote_txt_success_streams_content(self): upstream = FakeResponse(200, chunks=(b"remote",)) app.app.state.http_session = FakeSession([upstream]) with tempfile.TemporaryDirectory() as directory, patch.object(app, "resolve_safe_txt_path", return_value=Path(directory) / "missing.txt"): response = await app.serve_txt_proxy("remote.txt") self.assertEqual([chunk async for chunk in response.body_iterator], [b"remote"]) self.assertTrue(upstream.released) async def test_txt_traversal_is_400_without_remote_request(self): session = FakeSession([]) app.app.state.http_session = session response = await app.serve_txt_proxy("../secret.txt") self.assertEqual(response.status_code, 400) self.assertEqual(session.urls, []) async def test_remote_txt_404_maps_to_404(self): upstream = FakeResponse(404) app.app.state.http_session = FakeSession([upstream]) with tempfile.TemporaryDirectory() as directory, patch.object(app, "resolve_safe_txt_path", return_value=Path(directory) / "missing.txt"): response = await app.serve_txt_proxy("missing.txt") self.assertEqual(response.status_code, 404) self.assertTrue(upstream.released) async def test_reader_cached_containing_range_serves_exact_slice_without_upstream(self): upstream = FakeResponse(206, { "Content-Type": "application/pdf", "Content-Length": "8", "Content-Range": "bytes 2-9/10", "ETag": '"same"', }, chunks=(b"2345", b"6789")) session = app.app.state.http_session = FakeSession([upstream]) app.app.state.upstream_semaphore = asyncio.Semaphore(1) app.app.state.reader_range_cache = OrderedDict() app.app.state.reader_range_cache_bytes = 0 url = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/book.pdf" first = await app.api_reader_content(SimpleNamespace(method="GET", headers={"Range": "bytes=2-9"}), url) self.assertIn('reader_cache;desc="miss"', first.headers['server-timing']) self.assertEqual(b"".join([part async for part in first.body_iterator]), b"23456789") for requested, body, returned in (("bytes=4-6", b"456", "bytes 4-6/10"), ("bytes=7-20", b"789", "bytes 7-9/10")): response = await app.api_reader_content(SimpleNamespace(method="GET", headers={"Range": requested}), url) self.assertEqual(response.body, body) self.assertEqual(response.headers["content-range"], returned) self.assertEqual(response.headers["content-length"], str(len(body))) self.assertEqual(response.headers["etag"], '"same"') self.assertEqual(response.headers["cache-control"], "no-store") self.assertIn('reader_cache;desc="slice"', response.headers['server-timing']) self.assertEqual(len(session.urls), 1) self.assertEqual(app.app.state.reader_range_cache_bytes, 8) self.assertEqual(len(app.app.state.reader_range_cache), 1) self.assertIsNone(app.reader_cached_subrange((url, 'bytes=0-3', app.READER_DOCUMENT_MAX_BYTES, None))) self.assertIsNone(app.reader_cached_subrange((url, 'bytes=4-6', 10, None))) self.assertIsNone(app.reader_cached_subrange((url, 'bytes=4-6', app.READER_DOCUMENT_MAX_BYTES, 'video/mp4'))) self.assertIsNone(app.reader_cached_subrange((url, 'bytes=-3', app.READER_DOCUMENT_MAX_BYTES, None))) key = next(iter(app.app.state.reader_range_cache)) _, payload, headers = app.app.state.reader_range_cache[key] app.app.state.reader_range_cache[key] = (float('-inf'), payload, headers) self.assertIsNone(app.reader_cached_subrange((url, 'bytes=4-6', app.READER_DOCUMENT_MAX_BYTES, None))) self.assertEqual(app.app.state.reader_range_cache_bytes, 0) async def test_reader_range_cache_entry_cap_evicts_oldest_completed_range(self): responses = [FakeResponse(206, {"Content-Length": "1", "Content-Range": f"bytes {i}-{i}/10"}, chunks=(str(i).encode(),)) for i in range(3)] app.app.state.http_session = FakeSession(responses) app.app.state.upstream_semaphore = asyncio.Semaphore(1) app.app.state.reader_range_cache = OrderedDict() app.app.state.reader_range_cache_bytes = 0 url = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/range-cap.pdf" with patch.object(app, "READER_RANGE_CACHE_MAX_ENTRIES", 2): for i in range(3): response = await app.api_reader_content(SimpleNamespace(method="GET", headers={"Range": f"bytes={i}-{i}"}), url) self.assertEqual(b"".join([part async for part in response.body_iterator]), str(i).encode()) self.assertEqual([key[1] for key in app.app.state.reader_range_cache], ["bytes=1-1", "bytes=2-2"]) self.assertEqual(app.app.state.reader_range_cache_bytes, 2) # Reader archive resources (/api/reader-resource). @staticmethod def zip_bytes(files=None): import io import zipfile buffer = io.BytesIO() with zipfile.ZipFile(buffer, "w") as archive: for name, body in (files or {"image.svg": b""}).items(): archive.writestr(name, body) return buffer.getvalue() async def test_reader_archive_rejects_non_archive_formats_before_request(self): app.initialize_reader_archive_state(app.app) session = app.app.state.http_session = FakeSession([]) for name in ("book.pdf", "book.mobi", "book.azw3", "book.fb2", "audio.mp3", "cover.jpg"): with self.subTest(name=name): response = await app.api_reader_resource( "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/" + name, "image.svg") self.assertEqual(response.status_code, 400) self.assertEqual(session.urls, []) async def test_reader_archive_rejects_declared_oversize_without_reading_body(self): app.initialize_reader_archive_state(app.app) source = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/huge.epub" reads = [] class LargeResponse(FakeResponse): def iter_chunked(self, size): reads.append(size) return super().iter_chunked(size) upstream = LargeResponse(200, {"Content-Length": str(app.READER_RESOURCE_MAX_ARCHIVE_BYTES + 1)}, chunks=(self.zip_bytes(),)) app.app.state.http_session = FakeSession([upstream]) response = await app.api_reader_resource(source, "image.svg") self.assertEqual(response.status_code, 413) self.assertEqual(reads, []) self.assertTrue(upstream.released) self.assertEqual(app.app.state.reader_archive_cache, {}) self.assertEqual(app.app.state.upstream_semaphore._value, 1) async def test_reader_archive_stops_after_non_archive_signature(self): app.initialize_reader_archive_state(app.app) source = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/fake.epub" consumed = [] class TextResponse(FakeResponse): def iter_chunked(self, size): async def generator(): for chunk in (b"plain text, not an archive", b"x" * 1024, b"y" * 1024): consumed.append(chunk) yield chunk return generator() upstream = TextResponse() app.app.state.http_session = FakeSession([upstream]) response = await app.api_reader_resource(source, "image.svg") self.assertEqual(response.status_code, 400) self.assertEqual(len(consumed), 1) self.assertTrue(upstream.released) async def test_reader_archive_split_signature_and_single_copy_roundtrip(self): app.initialize_reader_archive_state(app.app) source = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/split.epub" raw = self.zip_bytes() upstream = FakeResponse(200, {"Content-Length": str(len(raw))}, chunks=(raw[:1], raw[1:3], raw[3:])) app.app.state.http_session = FakeSession([upstream]) prepared_inputs = [] original_prepare = app.prepare_reader_archive def spy(value): prepared_inputs.append(type(value)) return original_prepare(value) with patch.object(app, "prepare_reader_archive", side_effect=spy): response = await app.api_reader_resource(source, "image.svg") self.assertEqual(response.status_code, 200) self.assertEqual(response.body, b"") # The worker receives the single joined bytes buffer, not a bytearray # that must be copied again while both are alive. self.assertEqual(prepared_inputs, [bytes]) cached = app.app.state.reader_archive_cache[source][1] self.assertIs(type(cached), bytes) self.assertIs(app.prepare_reader_archive(cached), cached) gzipped = FakeResponse(chunks=(gzip.compress(raw),)) app.app.state.http_session = FakeSession([gzipped]) response = await app.api_reader_resource(source.replace("split", "packed"), "image.svg") self.assertEqual(response.status_code, 200) # Redirect-cache recovery (upstream_http.open_download_response). async def test_cached_target_transport_failure_evicts_and_walks_fresh(self): start = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/conn.pdf" stale = "https://cdn-lfs.hf.co/stale-signed" fresh = "https://cdn-lfs.hf.co/fresh-signed" for error in (aiohttp.ClientConnectionError("reset"), asyncio.TimeoutError()): with self.subTest(error=type(error).__name__): upstream_http.upstream_redirect_cache.clear() upstream_http.upstream_redirect_cache[("GET", start)] = (time.monotonic(), stale) session = FakeSession([error, FakeResponse(302, {"Location": fresh}), FakeResponse(200)]) result = await app.open_download_response( session, start, aiohttp.ClientTimeout(total=1), url_validator=app.validate_voiceofml_redirect_url, ) self.assertEqual(result.status, 200) self.assertEqual([url for url, _ in session.urls], [stale, start, fresh]) self.assertEqual(upstream_http.upstream_redirect_cache[("GET", start)][1], fresh) async def test_cached_target_gone_status_evicts_and_walks_fresh(self): start = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/gone.pdf" stale = "https://cdn-lfs.hf.co/stale-signed" fresh = "https://cdn-lfs.hf.co/fresh-signed" for status in (404, 410): with self.subTest(status=status): upstream_http.upstream_redirect_cache.clear() upstream_http.upstream_redirect_cache[("GET", start)] = (time.monotonic(), stale) gone = FakeResponse(status) session = FakeSession([gone, FakeResponse(302, {"Location": fresh}), FakeResponse(200)]) result = await app.open_download_response( session, start, aiohttp.ClientTimeout(total=1), url_validator=app.validate_voiceofml_redirect_url, ) self.assertEqual(result.status, 200) self.assertTrue(gone.released) self.assertEqual([url for url, _ in session.urls], [stale, start, fresh]) async def test_fresh_walk_failure_is_not_walked_twice(self): start = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/denied.pdf" signed = "https://cdn-lfs.hf.co/signed" for status in (401, 403, 404): with self.subTest(status=status): upstream_http.upstream_redirect_cache.clear() denied = FakeResponse(status) session = FakeSession([FakeResponse(302, {"Location": signed}), denied, FakeResponse(302, {"Location": signed}), FakeResponse(200)]) result = await app.open_download_response( session, start, aiohttp.ClientTimeout(total=1), url_validator=app.validate_voiceofml_redirect_url, ) self.assertIs(result, denied) self.assertFalse(denied.released) self.assertEqual([url for url, _ in session.urls], [start, signed]) self.assertEqual(upstream_http.upstream_redirect_cache, {}) session = FakeSession([aiohttp.ClientConnectionError("down"), FakeResponse(200)]) with self.assertRaises(aiohttp.ClientConnectionError): await app.open_download_response(session, start, aiohttp.ClientTimeout(total=1), url_validator=app.validate_voiceofml_redirect_url) self.assertEqual(len(session.urls), 1) # Reader bucket path decoding, admission and redirect scope. async def test_reader_bucket_path_is_decoded_exactly_once(self): root = "chapters/ebook/epub/" + "a" * 64 + "/" + "b" * 16 literal = root + "/chapter-manifest.json" with patch.object(app, "read_reader_bucket_resource", return_value=b"{}") as read: response = await app.api_reader_bucket_resource(literal) self.assertEqual(response.status_code, 200) read.assert_called_once_with(literal, app.READER_RESOURCE_MAX_BYTES) resource = root + "/resources/image%41.svg" source_urls = [] async def stream(request, source_url, **kwargs): source_urls.append(source_url) return app.Response(b"svg") with patch.object(app, "stream_reader_document", side_effect=stream): response = await app.api_reader_bucket_resource(resource, SimpleNamespace(method="GET", headers={})) self.assertEqual(response.status_code, 200) self.assertEqual(source_urls, [app.reader_bucket_source_url(resource)]) self.assertTrue(source_urls[0].endswith("/resources/image%2541.svg")) for unsafe in (root + "/resources/%2e%2e/x.svg", root + "/resources/%252e%252e/x.svg", root + "/resources/../x.svg"): with self.subTest(path=unsafe), patch.object(app, "stream_reader_document", side_effect=stream): rejected = await app.api_reader_bucket_resource(unsafe, SimpleNamespace(method="GET", headers={})) self.assertEqual(rejected.status_code, 400) async def test_reader_bucket_admission_limit_is_retryable_503(self): path = f"objects/aa/{'a' * 64}/{'b' * 16}/page-manifest.json" with patch.object(app, "READER_BUCKET_MAX_INFLIGHT", 0), \ patch.object(app, "read_reader_bucket_resource", return_value=b"{}") as read: response = await app.api_reader_bucket_resource(path) self.assertEqual(response.status_code, 503) self.assertEqual(response.headers["retry-after"], "1") self.assertEqual(response.headers["cache-control"], "no-store") read.assert_not_called() async def test_bucket_pdf_redirect_rejects_unlisted_hf_storage_host(self): path = "derived/test/" + "a" * 32 + "/document.pdf" for location in ("https://attacker.hf.co/secret.pdf", "https://datasets.huggingface.co/x.pdf", "https://hf-mirror.com/buckets/vomebook/pdf-pages-v2/resolve/x.pdf"): with self.subTest(location=location): upstream_http.upstream_redirect_cache.clear() app.app.state.upstream_semaphore = asyncio.Semaphore(1) redirect = FakeResponse(302, {"Location": location}) session = app.app.state.http_session = FakeSession([redirect, FakeResponse(200)]) response = await asyncio.wait_for(app.api_reader_bucket_resource( path, SimpleNamespace(method="GET", headers={})), 2) self.assertEqual(response.status_code, 502) self.assertEqual(len(session.urls), 1) self.assertTrue(redirect.released) signed = "https://cas-bridge.xethub.hf.co/xet-bridge-us/signed" app.app.state.upstream_semaphore = asyncio.Semaphore(1) session = app.app.state.http_session = FakeSession([ FakeResponse(302, {"Location": signed}), FakeResponse(200, {"Content-Length": "3", "Content-Type": "application/pdf"}, chunks=(b"pdf",)), ]) response = await asyncio.wait_for(app.api_reader_bucket_resource( path, SimpleNamespace(method="GET", headers={})), 2) self.assertEqual(response.status_code, 200) self.assertEqual(b"".join([part async for part in response.body_iterator]), b"pdf") self.assertEqual([url for url, _ in session.urls][1], signed) # Reader asset sidecar and v3 catalog fetches. async def test_reader_assets_sidecar_redirect_hops_are_validated(self): payload = gzip.compress(json.dumps({"v": 1, "f": {"ready": {"s": 4}}}).encode()) signed = "https://cas-bridge.xethub.hf.co/xet-bridge-us/sidecar" cases = ( ([FakeResponse(302, {"Location": "https://attacker.hf.co/sidecar"})], {}), ([FakeResponse(302, {"Location": signed}), FakeResponse(200, chunks=(payload,))], {"ready": {"s": 4}}), ) for responses, expected in cases: with self.subTest(expected=expected): session = FakeSession(responses) app.app.state.http_session = session app.app.state.reader_assets_lock = asyncio.Lock() with patch.object(app, "reader_assets_cache", {"v": 1, "f": {}}), \ patch.object(app, "reader_assets_cache_at", 0), \ patch.object(app, "load_reading_catalog", AsyncMock(return_value=None)): assets = await app.refresh_reader_assets() self.assertEqual(assets["f"], expected) self.assertEqual(len(session.urls), len(responses)) self.assertTrue(all(not kwargs["allow_redirects"] for _, kwargs in session.urls)) self.assertTrue(all(response.released for response in responses)) self.assertEqual(upstream_http.upstream_redirect_cache, {}) async def test_v3_catalog_redirect_hops_are_validated_and_not_cached(self): from backend import reader_v3_catalog as v3 from tests.test_reader_v3_catalog import fixture pointer, raw = fixture() escaped = FakeResponse(302, {"Location": "https://attacker.hf.co/pointer"}) session = FakeSession([escaped]) with self.assertRaisesRegex(ValueError, "存储范围"): await v3.load_reading_catalog(session) self.assertEqual(len(session.urls), 1) self.assertTrue(escaped.released) signed = "https://cas-bridge.xethub.hf.co/xet-bridge-us/" session = FakeSession([ FakeResponse(302, {"Location": signed + "pointer"}), FakeResponse(200, chunks=(json.dumps(pointer).encode(),)), FakeResponse(302, {"Location": signed + "catalog"}), FakeResponse(200, chunks=(raw,)), ]) result = await v3.load_reading_catalog(session) self.assertEqual(result["generation"], pointer["generation"]) self.assertEqual([url for url, _ in session.urls], [ v3.BASE + v3.POINTER, signed + "pointer", v3.BASE + pointer["catalog"]["path"], signed + "catalog"]) self.assertTrue(all(not kwargs["allow_redirects"] for _, kwargs in session.urls)) self.assertEqual(session.urls[0][1]["headers"]["Cache-Control"], "no-cache") self.assertEqual(upstream_http.upstream_redirect_cache, {}) if __name__ == "__main__": unittest.main()