bindery/tests/fake_cwa.py

"""A small stand-in for CWA's OPDS feed, book JSON and downloads."""

import base64
import json
import threading
from dataclasses import dataclass, field
from html import escape
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse


@dataclass
class FakeBook:
    id: int
    uuid: str
    title: str
    authors: list[str] = field(default_factory=lambda: ["Writer"])
    publisher: str = ""
    epub: bytes | None = None


class FakeCwa:
    def __init__(self, username: str = "bindery", password: str = "secret", page_size: int = 50) -> None:
        self.username, self.password, self.page_size = username, password, page_size
        self.books: list[FakeBook] = []
        self.requests: list[str] = []
        # CWA's "Show Publisher Section" setting. Without it, the publisher feed is 404.
        self.publisher_view = True
        # CWA's anonymous browsing. With it, the fake answers every request, with or without a login.
        self.anonymous = False
        fake = self

        class Handler(BaseHTTPRequestHandler):
            def log_message(self, *args):  # quiet
                pass

            def do_GET(self):
                fake.requests.append(self.path)
                fake.handle(self)

        self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        self.url = f"http://127.0.0.1:{self.server.server_address[1]}"
        threading.Thread(target=self.server.serve_forever, daemon=True).start()

    def close(self) -> None:
        if self.server.socket.fileno() == -1:
            return
        self.server.shutdown()
        self.server.server_close()

    def _authorized(self, h) -> bool:
        header = h.headers.get("Authorization", "")
        if not header.startswith("Basic "):
            return False
        user, _, password = base64.b64decode(header[6:]).decode().partition(":")
        return user == self.username and password == self.password

    def _send(self, h, status: int, body: bytes, ctype: str) -> None:
        h.send_response(status)
        h.send_header("Content-Type", ctype)
        h.send_header("Content-Length", str(len(body)))
        h.end_headers()
        h.wfile.write(body)

    def _feed(self, entries: list[str], next_href: str | None = None) -> bytes:
        link = f'<link rel="next" href="{escape(next_href)}"/>' if next_href else ""
        return (
            f'<?xml version="1.0" encoding="UTF-8"?><feed xmlns="http://www.w3.org/2005/Atom">'
            f"<title>Fake CWA</title>{link}{''.join(entries)}</feed>"
        ).encode()

    def _book_entry(self, b: FakeBook) -> str:
        authors = "".join(f"<author><name>{escape(a)}</name></author>" for a in b.authors)
        publisher = f"<publisher><name>{escape(b.publisher)}</name></publisher>" if b.publisher else ""
        link = (
            (
                f'<link rel="http://opds-spec.org/acquisition" href="/opds/download/{b.id}/epub/" '
                f'type="application/epub+zip"/>'
            )
            if b.epub is not None
            else ""
        )
        return f"<entry><title>{escape(b.title)}</title><id>urn:uuid:{b.uuid}</id>{authors}{publisher}{link}</entry>"

    def _paged(self, h, books: list[FakeBook], path: str, query: dict) -> None:
        offset = int(query.get("offset", ["0"])[0])
        page = books[offset : offset + self.page_size]
        next_href = f"{path}?offset={offset + self.page_size}" if offset + self.page_size < len(books) else None
        self._send(h, 200, self._feed([self._book_entry(b) for b in page], next_href), "application/atom+xml")

    def handle(self, h) -> None:
        if not self.anonymous and not self._authorized(h):
            self._send(h, 401, b"Unauthorized", "text/plain")
            return
        url = urlparse(h.path)
        query = parse_qs(url.query)
        path = url.path
        if path == "/opds":
            self._send(h, 200, self._feed([]), "application/atom+xml")
        elif path == "/opds/search":
            term = query.get("query", [""])[0].lower()
            hits = [b for b in self.books if term in b.title.lower() or any(term in a.lower() for a in b.authors)]
            # CWA returns every search result on one page.
            self._send(h, 200, self._feed([self._book_entry(b) for b in hits]), "application/atom+xml")
        elif path.startswith("/opds/publisher") and not self.publisher_view:
            self._send(h, 404, b"Not Found", "text/plain")
        elif path == "/opds/publisher":
            names = sorted({b.publisher for b in self.books if b.publisher})
            entries = [
                f"<entry><title>{escape(n)}</title><id>/opds/publisher/{i}</id>"
                f'<link rel="subsection" href="/opds/publisher/{i}"/></entry>'
                for i, n in enumerate(names)
            ]
            self._send(h, 200, self._feed(entries), "application/atom+xml")
        elif path.startswith("/opds/publisher/"):
            names = sorted({b.publisher for b in self.books if b.publisher})
            name = names[int(path.rsplit("/", 1)[1])]
            self._paged(h, [b for b in self.books if b.publisher == name], path, query)
        elif path.startswith("/ajax/book/"):
            uuid = path.rsplit("/", 1)[1]
            book = next((b for b in self.books if b.uuid == uuid), None)
            if not book:
                self._send(h, 200, b"", "text/html")
                return
            data = {
                "uuid": book.uuid,
                "application_id": book.id,
                "title": book.title,
                "authors": book.authors,
                "formats": ["EPUB"] if book.epub is not None else [],
            }
            self._send(h, 200, json.dumps(data).encode(), "application/json")
        elif path.startswith("/opds/download/"):
            book_id = int(path.split("/")[3])
            book = next((b for b in self.books if b.id == book_id and b.epub is not None), None)
            if not book:
                self._send(h, 404, b"Not found", "text/plain")
                return
            self._send(h, 200, book.epub, "application/epub+zip")
        else:
            self._send(h, 404, b"Not found", "text/plain")