from __future__ import annotations

import asyncio
import contextlib
import gzip
import io
import json
import random
import re
import struct
import threading
import time
import unittest
import zipfile
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
from urllib.parse import quote

IMPORT_ERROR = None
try:
    import app
    import httpx
except ModuleNotFoundError as exc:
    app = None
    IMPORT_ERROR = exc

from tests.test_async_proxy import FakeResponse, FakeSession


SOURCE = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/book29.txt"
OBJECT = "objects/aa/" + "a" * 64
BUCKET_PATH = OBJECT + "/" + "b" * 16 + "/page-manifest.json"


def make_zip(files=None, comment=b"", zip64=False):
    output = io.BytesIO()
    with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED) as archive:
        for name, content in (files if files is not None else {"image.svg": b"<svg/>"}).items():
            archive.writestr(name, content)
        archive.comment = comment
    raw = output.getvalue()
    if not zip64:
        return raw
    offset = len(raw) - 22 - len(comment)
    end = list(struct.unpack("<4s4H2LH", raw[offset:offset + 22]))
    record64 = struct.pack("<4sQ2H2L4Q", b"PK\x06\x06", 44, 45, 45, 0, 0, end[4], end[4], end[5], end[6])
    locator = struct.pack("<4sLQL", b"PK\x06\x07", 0, offset, 1)
    end[3:7] = [0xffff, 0xffff, 0xffffffff, 0xffffffff]
    return raw[:offset] + record64 + locator + struct.pack("<4s4H2LH", *end) + comment


@unittest.skipIf(app is None, f"backend dependencies unavailable: {IMPORT_ERROR}")
class ReaderArchiveValidationTests(unittest.TestCase):
    def test_canonical_paths_preserve_character_encoder_semantics(self):
        prefix = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/"
        def reference(value):
            def replace(match):
                token = match.group()
                if re.fullmatch(r"%[0-9a-fA-F]{2}", token):
                    char = chr(int(token[1:], 16))
                    return char if re.fullmatch(r"[A-Za-z0-9._~-]", char) else token.upper()
                return quote(token, safe="")
            return prefix + re.sub(r"%[0-9a-fA-F]{2}|[^A-Za-z0-9._~/-]", replace, value)
        rng = random.Random(20260927)
        tokens = [chr(i) for i in range(128) if chr(i) not in "?#"]
        tokens += ["\u4e66", "\u00e9", "\U0001f600", "%2f", "%25", "%41", "%7e", "%G0", "%"]
        paths = ["%" + first + second for i in range(256)
                 for first in {f"{i:02x}"[0], f"{i:02X}"[0]}
                 for second in {f"{i:02x}"[1], f"{i:02X}"[1]}]
        paths += ["".join(rng.choices(tokens, k=40)) for _ in range(256)]
        for path in paths:
            self.assertEqual(app.canonical_reader_source_url(prefix + path), reference(path))
        for value in ("https://example.com/%ab", prefix + "a?raw=%ab#fragment"):
            self.assertEqual(app.canonical_reader_source_url(value), value)

    def test_base36_independent_fixed_vectors(self):
        for value, expected in ((0, "0"), (35, "z"), (36, "10"), (72, "20"), (1296, "100"), (2**64 - 1, "3w5e11264sgsf")):
            with self.subTest(value=value):
                self.assertEqual(app._base36(value), expected)
        self.assertEqual(app.reader_short_id(SOURCE), "0ym7e7wbcxau0")
        vectors = json.loads(Path(__file__).with_name("reader-contract-vectors.json").read_text())
        for value, expected in vectors["ids"]:
            self.assertEqual(app.reader_short_id(value), expected)
        prefix = "https://huggingface.co/datasets/VoiceOfML/VOMEBOOK/resolve/main/"
        for value, expected in vectors["canonicalPaths"]:
            self.assertEqual(app.canonical_reader_source_url(prefix + value), prefix + expected)
            self.assertEqual(app.canonical_reader_source_url(prefix + expected), prefix + expected)
            self.assertEqual(app.reader_short_id(prefix + value), app.reader_short_id(prefix + expected))
        self.assertEqual(app.reader_short_id(prefix + vectors["reportedBook"]["path"]), vectors["reportedBook"]["id"])
        for record, expected in vectors["ocr"]:
            self.assertEqual(app.build_txt_relative_path(record), expected)

    def test_valid_standard_zip64_comments_and_prepended_data(self):
        for zip64 in (False, True):
            for comment in (b"", b"archive comment"):
                for prefix in (b"", b"self-extracting prefix"):
                    with self.subTest(zip64=zip64, comment=comment, prefix=prefix):
                        raw = prefix + make_zip(comment=comment, zip64=zip64)
                        app.validate_reader_archive(raw)
                        self.assertEqual(app.read_reader_archive_member(raw, "image.svg"), b"<svg/>")
        app.validate_reader_archive(make_zip(files={}))

    def test_gzip_wrapper_has_bounded_expansion(self):
        raw = make_zip()
        self.assertEqual(app.prepare_reader_archive(gzip.compress(raw)), raw)
        with patch.object(app, "READER_RESOURCE_MAX_ARCHIVE_BYTES", len(raw) - 1):
            with self.assertRaises(app.ReaderArchiveLimitError):
                app.prepare_reader_archive(gzip.compress(raw))
        with self.assertRaises(app.ReaderArchiveInvalidError):
            app.prepare_reader_archive(b"\x1f\x8btruncated")

    def test_directory_limits_checked_before_zipfile_construction(self):
        for zip64 in (False, True):
            raw = make_zip(zip64=zip64)
            for limit in ("READER_ARCHIVE_MAX_CENTRAL_DIRECTORY_BYTES", "READER_ARCHIVE_MAX_ENTRIES"):
                with self.subTest(zip64=zip64, limit=limit), patch.object(app, limit, 0), patch.object(app.zipfile, "ZipFile") as parser:
                    with self.assertRaises(app.ReaderArchiveLimitError):
                        app.validate_reader_archive(raw)
                    parser.assert_not_called()

    def test_malformed_end_records_and_multidisk_archives(self):
        raw = make_zip()
        variants = [raw[:-1], b"PKnot-a-zip"]
        for offset, fmt, value in ((4, "H", 1), (10, "H", 2), (16, "L", len(raw) + 1), (20, "H", 8)):
            bad = bytearray(raw)
            struct.pack_into("<" + fmt, bad, len(raw) - 22 + offset, value)
            variants.append(bytes(bad))
        raw64 = make_zip(zip64=True)
        bad64 = bytearray(raw64)
        struct.pack_into("<L", bad64, len(raw64) - 22 - 20 + 4, 1)
        variants.extend((bytes(bad64), raw64[:-23] + raw64[-22:]))
        for index, bad in enumerate(variants):
            with self.subTest(index=index), self.assertRaises(app.ReaderArchiveInvalidError):
                app.validate_reader_archive(bad)

    def test_unsafe_members_and_compression_limits(self):
        for name in ("../escape", "/absolute", "a/../escape", "a//escape", "a\\..\\escape"):
            with self.subTest(name=name), self.assertRaises(app.ReaderArchiveInvalidError):
                app.validate_reader_archive(make_zip({name: b"content"}))
        with self.assertRaises(app.ReaderArchiveLimitError):
            app.validate_reader_archive(make_zip({"bomb": b"x" * 100000}))
        with patch.object(app, "READER_ARCHIVE_MAX_EXPANDED_BYTES", 3), self.assertRaises(app.ReaderArchiveLimitError):
            app.validate_reader_archive(make_zip({"small": b"1234"}))
        with self.assertRaises(app.ReaderArchiveInvalidError):
            app.validate_reader_archive(make_zip().replace(b"image.svg", b"im\x00ge.svg"))

    def test_accepted_sidecar_paths_match_the_delivery_validator(self):
        files = {}
        for mode, filenames in app.READER_ASSET_FILENAMES.items():
            for filename in filenames:
                for directory in ("", "converter-v1/", "b" * 16 + "/", "b" * 16 + "/converter-v1/"):
                    path = OBJECT + "/" + directory + filename
                    # Page manifests have only an optional version directory.
                    if filename == "page-manifest.json" and "converter" in directory:
                        continue
                    files[path] = {"s": 2, "m": mode, "p": path}
        files["bucket"] = {"s": 2, "m": "p", "p": BUCKET_PATH, "b": "vomebook/pdf-pages"}
        ebook_chapter_path = ("ebook-chapters/objects/aa/" + "c" * 64
                              + "/chapter-bundle/chapter-profile-epub-chapters-v7-bucket/"
                              "epub-chapters/chapter-manifest.json")
        files["ebook-chapters"] = {
            "s": 2, "m": "e", "p": OBJECT + "/document.epub",
            "c": ebook_chapter_path, "cb": "vomebook/pdf-pages",
        }
        files["spreadsheet-pages"] = {
            "s": 2, "m": "p", "p": OBJECT + "/1234567890abcdef/page-manifest.json",
            "b": "vomebook/pdf-pages",
        }
        files["native-audio-bucket"] = {
            "s": 2, "m": "a", "p": OBJECT + "/native-media-cdn-v1/audio.flac",
            "b": "vomebook/pdf-pages",
        }
        decoded = app.decode_reader_assets(gzip.compress(json.dumps({"v": 1, "f": files}).encode()))["f"]
        self.assertEqual(set(decoded), set(files))
        self.assertEqual(decoded["ebook-chapters"]["cb"], "vomebook/pdf-pages")
        self.assertEqual(
            app.reader_asset_url(decoded["spreadsheet-pages"]),
            "https://huggingface.co/buckets/vomebook/pdf-pages/resolve/"
            + OBJECT + "/1234567890abcdef/page-manifest.json",
        )
        self.assertEqual(
            app.reader_asset_url(decoded["native-audio-bucket"]),
            "https://huggingface.co/buckets/vomebook/pdf-pages/resolve/"
            + OBJECT + "/native-media-cdn-v1/audio.flac",
        )
        self.assertEqual(app.READER_RESOURCE_TYPES["flac"], "audio/flac")
        self.assertEqual(app.reader_extension_for_asset(decoded["native-audio-bucket"], "flac"), "audio")
        for entry in decoded.values():
            if entry.get("b"):
                self.assertIsNotNone(app.READER_BUCKET_PATH_RE.fullmatch(entry["p"]))
            else:
                app.validate_reader_source_url(app.reader_asset_url(entry))
        for bad in (
            {"s": 2, "m": "e", "p": OBJECT + "/document.pdf"},
            {"s": 2, "m": "p", "p": OBJECT + "/document.pdf", "b": "vomebook/pdf-pages"},
            {"s": 2, "m": "p", "p": OBJECT + "/page-manifest.json", "b": "vomebook/pdf-pages"},
        ):
            decoded = app.decode_reader_assets(gzip.compress(json.dumps({"v": 1, "f": {"bad": bad}}).encode()))
            self.assertEqual(decoded["f"], {})


@unittest.skipIf(app is None, f"backend dependencies unavailable: {IMPORT_ERROR}")
class ReaderBackendTests(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        self.saved_state = dict(app.app.state._state)
        app.initialize_reader_archive_state(app.app)
        app.initialize_reader_bucket_state(app.app)
        app.app.state.upstream_semaphore = asyncio.Semaphore(1)
        self.stack = contextlib.ExitStack()
        self.record = {"Repo": "VoiceOfML/Test", "File": "book29", "Extension": "txt", "Folder": []}
        self.assets = {"v": 1, "f": {}}
        self.stack.enter_context(patch.object(app, "get_reader_assets", AsyncMock(return_value=self.assets)))
        self.stack.enter_context(patch.object(app, "records", [self.record]))
        self.stack.enter_context(patch.object(app, "reader_record_key_indices", {app.build_reader_asset_key(self.record): 0}))
        self.stack.enter_context(patch.object(app, "reader_source_id_indices", {"0ym7e7wbcxau0": [0]}))
        self.stack.enter_context(patch("upstream_http.upstream_redirect_cache", {}))

    async def asyncTearDown(self):
        await app.shutdown_reader_archive_state(app.app)
        await app.shutdown_reader_bucket_state(app.app)
        self.stack.close()
        app.app.state._state.clear()
        app.app.state._state.update(self.saved_state)

    def session(self, *responses):
        session = FakeSession(responses)
        app.app.state.http_session = session
        return session

    async def test_bucket_audio_resource_uses_audio_mime_type(self):
        path = OBJECT + "/native-media-cdn-v1/audio.flac"
        with patch.object(app, "get_reader_bucket_resource", AsyncMock(return_value=b"flac")):
            response = await app.api_reader_bucket_resource(path)
        self.assertEqual(response.media_type, "audio/flac")
        self.assertEqual(response.body, b"flac")

    async def content(self, upstream, method="GET"):
        self.session(upstream)
        return await app.api_reader_content(SimpleNamespace(method=method, headers={}), SOURCE, None)

    def assert_isolated(self, response):
        policy = response.headers["content-security-policy"]
        self.assertIn("sandbox;", policy)
        self.assertIn("default-src 'none'", policy)
        self.assertNotIn("allow-scripts", policy)
        self.assertNotIn("allow-same-origin", policy)
        self.assertEqual(response.headers["x-content-type-options"], "nosniff")

    async def test_valid_resource_roundtrip_and_literal_http_member_names(self):
        files = {"literal%20name.svg": b"percent", "literal name.svg": b"space", "percent%25.svg": b"percent25", "\u4e66.svg": b"unicode"}
        upstream = FakeResponse(chunks=(make_zip(files),))
        session = self.session(upstream)
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app.app), base_url="http://test") as client:
            for name, body in files.items():
                response = await client.get("/api/reader-resource", params={"book": SOURCE, "path": name})
                self.assertEqual(response.status_code, 200, response.text)
                self.assertEqual(response.content, body)
                self.assert_isolated(response)
            for unsafe in ("../escape", "/absolute", "a//b", "a/../b", "nul\x00"):
                response = await client.get("/api/reader-resource", params={"book": SOURCE, "path": unsafe})
                self.assertEqual(response.status_code, 400)
        self.assertEqual(len(session.urls), 1)
        self.assertTrue(upstream.released)

    async def test_archive_failures_are_not_cached_and_member_limits_are_reported(self):
        invalid = FakeResponse(chunks=(b"not-a-zip",))
        valid = FakeResponse(chunks=(make_zip(),))
        self.session(invalid, valid)
        self.assertEqual((await app.api_reader_resource(SOURCE, "image.svg")).status_code, 400)
        self.assertEqual(app.app.state.reader_archive_cache, {})
        self.assertEqual((await app.api_reader_resource(SOURCE, "missing.svg")).status_code, 404)
        with patch.object(app, "READER_RESOURCE_MAX_BYTES", 2):
            self.assertEqual((await app.api_reader_resource(SOURCE, "image.svg")).status_code, 413)
        self.assertTrue(invalid.released and valid.released)
        self.assertEqual(app.app.state.upstream_semaphore._value, 1)

    async def test_archive_cold_requests_share_load_and_cancellation_preserves_waiter(self):
        started, release = asyncio.Event(), asyncio.Event()
        raw = make_zip()

        class SlowResponse(FakeResponse):
            def iter_chunked(self, _size):
                async def chunks():
                    started.set()
                    await release.wait()
                    yield raw
                return chunks()

        upstream = SlowResponse()
        session = self.session(upstream)
        first = asyncio.create_task(app.get_reader_archive(SOURCE))
        await asyncio.wait_for(started.wait(), 1)
        second = asyncio.create_task(app.get_reader_archive(SOURCE))
        await asyncio.sleep(0)
        first.cancel()
        with self.assertRaises(asyncio.CancelledError):
            await first
        release.set()
        self.assertEqual(await asyncio.wait_for(second, 2), raw)
        self.assertEqual(len(session.urls), 1)
        self.assertTrue(upstream.released)
        self.assertEqual(app.app.state.reader_archive_inflight, {})

    async def test_archive_cache_total_bytes_expiry_and_oversize_admission(self):
        raw = make_zip()
        session = self.session(*(FakeResponse(chunks=(raw,)) for _ in range(6)))
        with patch.object(app, "READER_ARCHIVE_CACHE_MAX_BYTES", len(raw) * 2):
            for index in range(3):
                await app.get_reader_archive(SOURCE + str(index))
            cache = app.app.state.reader_archive_cache
            self.assertNotIn(SOURCE + "0", cache)
            self.assertEqual(sum(len(item[1]) for item in cache.values()), len(raw) * 2)
            cache[SOURCE + "2"] = (time.monotonic() - app.READER_ARCHIVE_CACHE_TTL_SECONDS - 1, raw)
            await app.get_reader_archive(SOURCE + "2")
            self.assertEqual(len(session.urls), 4)
        with patch.object(app, "READER_ARCHIVE_CACHE_MAX_BYTES", len(raw) - 1):
            await app.get_reader_archive(SOURCE + "big")
            await app.get_reader_archive(SOURCE + "big")
            self.assertNotIn(SOURCE + "big", cache)
        self.assertEqual(len(session.urls), 6)

    async def test_archive_shutdown_releases_active_download(self):
        started = asyncio.Event()

        class BlockedResponse(FakeResponse):
            def iter_chunked(self, _size):
                async def chunks():
                    started.set()
                    await asyncio.Event().wait()
                    yield b""
                return chunks()

        upstream = BlockedResponse()
        self.session(upstream)
        request = asyncio.create_task(app.get_reader_archive(SOURCE))
        await asyncio.wait_for(started.wait(), 1)
        with patch.object(app, "READER_ARCHIVE_MAX_INFLIGHT", 1):
            rejected = await app.api_reader_resource(SOURCE + "other", "image.svg")
            self.assertEqual(rejected.status_code, 503)
        await app.shutdown_reader_archive_state(app.app)
        with self.assertRaises(asyncio.CancelledError):
            await request
        self.assertTrue(upstream.released)
        self.assertEqual(app.app.state.upstream_semaphore._value, 1)
        self.assertEqual(app.app.state.reader_archive_cache, {})

    async def test_worker_cancellation_retains_slot_until_thread_finishes(self):
        app.app.state.reader_archive_worker_semaphore = asyncio.Semaphore(1)
        started, release, second_started = threading.Event(), threading.Event(), threading.Event()

        def blocking():
            started.set()
            release.wait(2)

        first = asyncio.create_task(app.run_reader_archive_worker(blocking))
        self.assertTrue(await asyncio.to_thread(started.wait, 1))
        first.cancel()
        second = asyncio.create_task(app.run_reader_archive_worker(second_started.set))
        try:
            with self.assertRaises(asyncio.CancelledError):
                await asyncio.wait_for(first, 1)
            await asyncio.sleep(0.02)
            self.assertEqual(len(app.app.state.reader_archive_workers), 1)
            self.assertFalse(second_started.is_set())
        finally:
            release.set()
            results = await asyncio.gather(first, second, return_exceptions=True)
        self.assertIsInstance(results[0], asyncio.CancelledError)
        self.assertTrue(second_started.is_set())

    async def test_archive_shutdown_waits_for_abandoned_worker(self):
        started, release = threading.Event(), threading.Event()

        def blocking():
            started.set()
            release.wait(2)

        caller = asyncio.create_task(app.run_reader_archive_worker(blocking))
        self.assertTrue(await asyncio.to_thread(started.wait, 1))
        caller.cancel()
        with self.assertRaises(asyncio.CancelledError):
            await caller
        shutdown = asyncio.create_task(app.shutdown_reader_archive_state(app.app))
        try:
            await asyncio.sleep(0.02)
            self.assertFalse(shutdown.done())
        finally:
            release.set()
            await asyncio.wait_for(shutdown, 1)
        self.assertEqual(app.app.state.reader_archive_workers, set())

    async def test_content_isolation_for_active_document_mime_types(self):
        for mime in ("text/html", "application/xhtml+xml", "image/svg+xml"):
            with self.subTest(mime=mime):
                response = await self.content(FakeResponse(headers={"Content-Type": mime}, chunks=(b"active content",)))
                self.assert_isolated(response)
                self.assertEqual(b"".join([part async for part in response.body_iterator]), b"active content")

    async def test_compressed_upstream_is_rejected_before_body_and_on_head(self):
        for method in ("GET", "HEAD"):
            for status in (200, 206):
                with self.subTest(method=method, status=status):
                    upstream = FakeResponse(status, {"Content-Encoding": "gzip", "Content-Length": "35"}, chunks=(gzip.compress(b"reader text\n" * 20),))
                    with patch.object(upstream, "iter_chunked", side_effect=AssertionError("encoded body must not be read")):
                        response = await self.content(upstream, method)
                    self.assertEqual(response.status_code, 502)
                    self.assertTrue(upstream.released)
                    self.assertEqual(app.app.state.upstream_semaphore._value, 1)
                    self.assertIs(app.app.state.http_session.urls[0][1]["auto_decompress"], False)

    async def test_stream_cleanup_before_start_and_after_first_body(self):
        for fail_at in ("http.response.start", "http.response.body"):
            with self.subTest(fail_at=fail_at):
                upstream = FakeResponse(chunks=(b"one", b"two"))
                with patch.object(upstream, "release", wraps=upstream.release) as released:
                    response = await self.content(upstream)

                    async def send(message):
                        if message["type"] == fail_at:
                            raise OSError("client disconnected")

                    async def receive():
                        return {"type": "http.disconnect"}

                    with self.assertRaises(Exception):
                        await response({"type": "http", "asgi": {"spec_version": "2.4"}}, receive, send)
                    released.assert_called_once()
                self.assertEqual(app.app.state.upstream_semaphore._value, 1)

    async def test_stream_cleanup_for_legacy_asgi_disconnect_before_iteration(self):
        upstream = FakeResponse()
        response = await self.content(upstream)
        started = asyncio.Event()

        async def send(message):
            started.set()
            await asyncio.Event().wait()

        async def receive():
            await started.wait()
            return {"type": "http.disconnect"}

        await asyncio.wait_for(response({"type": "http", "asgi": {"spec_version": "2.0"}}, receive, send), 1)
        self.assertTrue(upstream.released)
        self.assertEqual(app.app.state.upstream_semaphore._value, 1)

    async def test_stream_cleanup_after_normal_asgi_completion_is_exactly_once(self):
        upstream = FakeResponse(chunks=(b"one", b"two"))
        messages = []
        with patch.object(upstream, "release", wraps=upstream.release) as released:
            response = await self.content(upstream)
            async def send(message):
                messages.append(message)
            await response({"type": "http", "asgi": {"spec_version": "2.4"}}, AsyncMock(), send)
            released.assert_called_once()
        self.assertEqual(b"".join(message.get("body", b"") for message in messages), b"onetwo")
        self.assertEqual(app.app.state.upstream_semaphore._value, 1)

    async def test_malformed_headers_and_not_modified_release_upstream(self):
        cases = (
            (200, {"Content-Length": "invalid"}, 502),
            (206, {"Content-Range": "bytes " + "9" * 5000 + "-1/2"}, 502),
            (206, {"Content-Range": "bytes 0-1/2", "Content-Length": "1"}, 502),
            (304, {"ETag": "unchanged"}, 304),
        )
        for status, headers, expected in cases:
            with self.subTest(status=status, expected=expected):
                upstream = FakeResponse(status, headers)
                response = await self.content(upstream)
                self.assertEqual(response.status_code, expected)
                self.assertTrue(upstream.released)
                self.assertEqual(app.app.state.upstream_semaphore._value, 1)

    async def test_bucket_get_head_direct_and_by_source_id(self):
        self.assets["f"][app.build_reader_asset_key(self.record)] = {"s": 2, "m": "p", "p": BUCKET_PATH, "b": "vomebook/pdf-pages"}
        body = b'{"kind":"pdf-pages"}'
        with patch.object(app, "read_reader_bucket_resource", return_value=body) as read:
            async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app.app), base_url="http://test") as client:
                cases = (("/api/reader-bucket-resource", {"path": BUCKET_PATH}), ("/api/reader-content", {"id": "0ym7e7wbcxau0"}))
                for route, params in cases:
                    for method in ("GET", "HEAD"):
                        with self.subTest(route=route, params=params, method=method):
                            response = await client.request(method, route, params=params)
                            self.assertEqual(response.status_code, 200, response.text)
                            self.assertEqual(response.content, body if method == "GET" else b"")
                            self.assertEqual(response.headers["content-length"], str(len(body)))
                            self.assertIn("application/json", response.headers["content-type"])
                            self.assert_isolated(response)
            self.assertEqual(read.call_count, 1)

    async def test_bucket_resource_falls_back_to_dataset_object_when_bucket_is_missing(self):
        class Stream:
            def __init__(self, payload):
                self.payload = payload

            def __enter__(self):
                return self

            def __exit__(self, *_args):
                return False

            def read(self, _limit):
                return self.payload

        class FileSystem:
            def __init__(self):
                self.uris = []

            def open(self, uri, _mode):
                self.uris.append(uri)
                if uri.startswith("hf://buckets/"):
                    raise FileNotFoundError(uri)
                return Stream(b"dataset fallback")

        filesystem = FileSystem()
        with patch.object(app, "HfFileSystem", return_value=filesystem):
            payload = app.read_reader_bucket_resource("objects/aa/" + "a" * 64 + "/document.html")
        self.assertEqual(payload, b"dataset fallback")
        self.assertEqual(filesystem.uris[0].split("/")[:4], ["hf:", "", "buckets", "vomebook"])
        self.assertIn("hf://datasets/vomebook/Reader-Assets/", filesystem.uris[1])

    async def test_bucket_chapter_sidecar_resolves_to_shared_bucket_endpoint(self):
        chapter_path = ("ebook-chapters/objects/aa/" + "c" * 64
                        + "/chapter-bundle/chapter-profile-epub-chapters-v7-bucket/"
                        "epub-chapters/chapter-manifest.json")
        self.assets["f"][app.build_reader_asset_key(self.record)] = {
            "s": 2, "m": "e", "p": OBJECT + "/document.epub",
            "c": chapter_path, "cb": "vomebook/pdf-pages",
        }
        resolved = json.loads((await app.api_reader_resolve("0ym7e7wbcxau0")).body)
        self.assertEqual(resolved["extension"], "epub-chapters")
        self.assertIn("/api/reader-bucket-resource?path=ebook-chapters/", resolved["chapter_manifest"])
        self.assertEqual(resolved["chapter_manifest"].count("pdf-pages"), 0)
        bucket_body = b'{"kind":"epub-chapters"}'
        bucket_url = "/api/reader-bucket-resource?path=" + quote(chapter_path, safe="")
        with patch.object(app, "read_reader_bucket_resource", return_value=bucket_body) as read:
            response = await app.api_reader_content(
                SimpleNamespace(method="GET", headers={}), bucket_url, None)
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.body, bucket_body)
        self.assertEqual(read.call_args.args[0], chapter_path)

    async def test_bucket_content_rejects_arbitrary_relative_and_ambiguous_paths(self):
        for url in (
            "https://[broken", "/api/other?path=x", "//evil.test/api/reader-bucket-resource?path=x",
            "/api/reader-bucket-resource?path=x&path=y",
            "/api/reader-bucket-resource?bucket=vomebook%2Fpdf-optimized&path=x",
            "/api/reader-bucket-resource?path=../escape",
            "https://huggingface.co/buckets/vomebook/pdf-optimized/resolve/objects/aa/document.pdf",
        ):
            response = await app.api_reader_content(SimpleNamespace(method="GET", headers={}), url, None)
            self.assertIn(response.status_code, (400, 403))

    async def test_source_id_does_not_hash_originals_and_conversion_precedence_agrees(self):
        asset = {"s": 2, "m": "p", "p": OBJECT + "/page-manifest.json", "c": OBJECT + "/chapters-v1/chapter-manifest.json", "f": OBJECT + "/pdf-v1/document.pdf"}
        self.assets["f"][app.build_reader_asset_key(self.record)] = asset
        resolved = json.loads((await app.api_reader_resolve("0ym7e7wbcxau0")).body)
        enriched = app.add_reader_assets({"results": [self.record]}, self.assets)["results"][0]
        self.assertEqual(resolved["extension"], "pdf-pages")
        self.assertEqual(resolved["extension"], enriched["ReaderExtension"])
        self.assertEqual(resolved["chapter_manifest"], enriched["ReaderChapterManifest"])
        self.assertEqual(resolved["fallback"], enriched["ReaderFallback"])


if __name__ == "__main__":
    unittest.main()