tts-voice-creator-clone-and-design-2/tests/test_magazine_sources.py

"""Magazine sources: parsing, host restriction, size/type checks. No network."""
import os
import tempfile
import unittest
from pathlib import Path

os.environ.setdefault("CONFIG_DIR", tempfile.mkdtemp(prefix="ttvc-test-config-"))

from core import magazine_sources as ms


class FakeResp:
    def __init__(self, status=200, body=b"", headers=None, json_data=None):
        self.status_code = status
        self._body = body
        self.headers = headers or {}
        self._json = json_data
        self.is_redirect = status in (301, 302, 303, 307, 308)

    def json(self):
        if self._json is None:
            raise ValueError("no json")
        return self._json

    def iter_content(self, n):
        for i in range(0, len(self._body), n):
            yield self._body[i:i + n]

    def close(self):
        pass


class FakeSession:
    def __init__(self, routes):
        self.routes = routes
        self.calls = []

    def get(self, url, **kw):
        self.calls.append(url)
        for prefix, resp in self.routes.items():
            if url.startswith(prefix):
                return resp() if callable(resp) else resp
        return FakeResp(404)


OPDS = b"""<?xml version="1.0"?>
<feed xmlns="http://www.w3.org/2005/Atom">
 <entry><title>c't 12/2026</title><author><name>Heise</name></author><summary>Issue</summary>
  <link rel="http://opds-spec.org/acquisition" type="application/pdf" href="/dl/1.pdf"/>
  <link rel="http://opds-spec.org/image/thumbnail" href="/thumb/1.jpg"/></entry>
 <entry><title>Elsewhere</title>
  <link rel="http://opds-spec.org/acquisition" type="application/pdf" href="http://evil.example/x.pdf"/></entry>
 <entry><title>EPUB only</title>
  <link rel="http://opds-spec.org/acquisition" type="application/epub+zip" href="/dl/2.epub"/></entry>
</feed>"""


class OpdsTests(unittest.TestCase):
    def provider(self, routes):
        return ms.OpdsProvider("opds-1", "NAS", "http://nas.local:8083/opds/", FakeSession(routes))

    def test_search_keeps_only_same_host_pdf_entries(self):
        # Most specific route first: the fake session matches by prefix.
        p = self.provider({"http://nas.local:8083/opds/search": FakeResp(body=OPDS),
                           "http://nas.local:8083/opds/": FakeResp(body=b'<feed xmlns="http://www.w3.org/2005/Atom"/>')})
        items = p.search("ct")
        self.assertEqual([i.title for i in items], ["c't 12/2026"])
        self.assertEqual(items[0].id, "http://nas.local:8083/dl/1.pdf")
        self.assertEqual(items[0].thumb_url, "http://nas.local:8083/thumb/1.jpg")

    def test_resolve_rejects_other_hosts(self):
        p = self.provider({})
        with self.assertRaises(ms.SourceError):
            p.resolve_pdf("http://169.254.169.254/latest/meta-data")

    def test_doctype_and_entities_are_refused(self):
        evil = b'<?xml version="1.0"?><!DOCTYPE lol [<!ENTITY a "aaaa">]><feed xmlns="http://www.w3.org/2005/Atom"/>'
        with self.assertRaises(ms.SourceError):
            ms.parse_xml_safely(evil)

    def test_malformed_feed_is_a_source_error(self):
        with self.assertRaises(ms.SourceError):
            ms.parse_xml_safely(b"<feed")

    def test_empty_query_rejected(self):
        with self.assertRaises(ms.SourceError):
            self.provider({}).search("   ")


class ArchiveTests(unittest.TestCase):
    def test_search_builds_items_and_skips_bad_identifiers(self):
        docs = [{"identifier": "ct-2026-01", "title": "c't", "creator": ["Heise"], "date": "2026-01-01T00:00:00Z"},
                {"identifier": "../../etc/passwd", "title": "x"}]
        s = FakeSession({"https://archive.org/advancedsearch.php": FakeResp(json_data={"response": {"docs": docs}})})
        items = ms.InternetArchiveProvider(s).search("ct")
        self.assertEqual([i.id for i in items], ["ct-2026-01"])
        self.assertEqual(items[0].date, "2026-01-01")
        self.assertEqual(items[0].author, "Heise")

    def test_resolve_prefers_original_pdf(self):
        meta = {"metadata": {}, "files": [
            {"name": "a_text.pdf", "source": "derivative", "size": "9999"},
            {"name": "issue.pdf", "source": "original", "size": "100"}]}
        s = FakeSession({"https://archive.org/metadata/": FakeResp(json_data=meta)})
        url = ms.InternetArchiveProvider(s).resolve_pdf("ct-2026-01")
        self.assertEqual(url, "https://archive.org/download/ct-2026-01/issue.pdf")

    def test_lending_only_is_refused(self):
        s = FakeSession({"https://archive.org/metadata/": FakeResp(
            json_data={"metadata": {"access-restricted-item": "true"}, "files": []})})
        with self.assertRaises(ms.SourceError):
            ms.InternetArchiveProvider(s).resolve_pdf("loan")

    def test_rate_limit_message(self):
        s = FakeSession({"https://archive.org/advancedsearch.php": FakeResp(status=429)})
        with self.assertRaisesRegex(ms.SourceError, "rate-limiting"):
            ms.InternetArchiveProvider(s).search("ct")

    def test_bad_identifier_refused(self):
        with self.assertRaises(ms.SourceError):
            ms.InternetArchiveProvider(FakeSession({})).resolve_pdf("a/b")


class DownloadTests(unittest.TestCase):
    def setUp(self):
        self.dir = Path(tempfile.mkdtemp(prefix="ttvc-dl-"))
        self.p = ms.OpdsProvider("opds-1", "NAS", "http://nas.local/opds/", None)

    def test_downloads_pdf(self):
        s = FakeSession({"http://nas.local/dl": FakeResp(body=b"%PDF-1.7 hello")})
        n = ms.download_pdf(self.p, "http://nas.local/dl/1.pdf", self.dir / "source.pdf", session=s)
        self.assertEqual(n, 14)
        self.assertEqual((self.dir / "source.pdf").read_bytes()[:5], b"%PDF-")
        self.assertFalse((self.dir / "source.part").exists())

    def test_rejects_non_pdf_and_cleans_up(self):
        s = FakeSession({"http://nas.local/dl": FakeResp(body=b"<html>login</html>")})
        with self.assertRaisesRegex(ms.SourceError, "PDF"):
            ms.download_pdf(self.p, "http://nas.local/dl/1.pdf", self.dir / "source.pdf", session=s)
        self.assertEqual(list(self.dir.iterdir()), [])

    def test_redirect_to_other_host_is_refused(self):
        s = FakeSession({"http://nas.local/dl": FakeResp(status=302, headers={"location": "http://evil.example/x"})})
        with self.assertRaisesRegex(ms.SourceError, "different host"):
            ms.download_pdf(self.p, "http://nas.local/dl/1.pdf", self.dir / "source.pdf", session=s)

    def test_oversize_declared_is_refused(self):
        s = FakeSession({"http://nas.local/dl": FakeResp(body=b"%PDF-", headers={"content-length": str(ms.MAX_DOWNLOAD_BYTES + 1)})})
        with self.assertRaisesRegex(ms.SourceError, "larger"):
            ms.download_pdf(self.p, "http://nas.local/dl/1.pdf", self.dir / "source.pdf", session=s)

    def test_http_error(self):
        s = FakeSession({"http://nas.local/dl": FakeResp(status=403)})
        with self.assertRaisesRegex(ms.SourceError, "403"):
            ms.download_pdf(self.p, "http://nas.local/dl/1.pdf", self.dir / "source.pdf", session=s)


class BuildProvidersTests(unittest.TestCase):
    def test_skips_bad_and_credentialed_catalogues(self):
        provs = ms.build_providers([
            {"name": "NAS", "url": "http://nas.local/opds"},
            {"name": "bad", "url": "ftp://x"},
            {"name": "creds", "url": "http://user:pw@nas.local/opds"},
            {"url": ""}])
        self.assertEqual(sorted(provs), ["internet-archive", "opds-1"])


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


class RouteTests(unittest.TestCase):
    """The HTTP layer over the providers, on a minimal app (no auth middleware, no network)."""

    def setUp(self):
        from fastapi import FastAPI
        from fastapi.testclient import TestClient
        from unittest import mock
        from routes import magazine_sources as routes
        self.routes = routes
        self.app = FastAPI()
        self.app.include_router(routes.router)
        self.client = TestClient(self.app)
        self.mock = mock

    def test_lists_archive_and_configured_catalogues(self):
        with self.mock.patch.object(self.routes, "_load_settings",
                                    return_value={"magazine_opds_catalogs": [{"name": "NAS", "url": "http://nas.local/opds"}]}):
            data = self.client.get("/api/magazine-sources").json()
        self.assertEqual([s["key"] for s in data["sources"]], ["internet-archive", "opds-1"])

    def test_unknown_source_is_404(self):
        with self.mock.patch.object(self.routes, "_load_settings", return_value={}):
            self.assertEqual(self.client.get("/api/magazine-sources/nope/search?q=x").status_code, 404)
            self.assertEqual(self.client.get("/api/magazine-sources/..%2f/search?q=x").status_code, 404)

    def test_source_errors_become_502_with_a_message(self):
        class Boom:
            key = "internet-archive"
            label = "x"
            def search(self, q, page=1):
                raise ms.SourceError("The source is rate-limiting requests; try again in a minute")
        with self.mock.patch.object(self.routes, "_providers", return_value={"internet-archive": Boom()}):
            r = self.client.get("/api/magazine-sources/internet-archive/search?q=x")
        self.assertEqual(r.status_code, 502)
        self.assertIn("rate-limiting", r.json()["detail"])

    def test_import_requires_an_item(self):
        with self.mock.patch.object(self.routes, "_load_settings", return_value={}):
            r = self.client.post("/api/magazine-sources/import", json={"source": "internet-archive"})
        self.assertEqual(r.status_code, 400)