commit 5580004aebb0221c7c5eccf5375d7252dd105f0c Author: BMad Master Date: Sun Aug 30 09:03:39 2026 -0400 epub2pdf: title-based export names, .pdf download headers, security hardening (68 tests green) diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..a4c354b --- /dev/null +++ b/.dockerignore @@ -0,0 +1,9 @@ +__pycache__ +*.pyc +data/ +test_book.epub +.a0proj/ +.git/ +Dockerfile +.dockerignore +*.md diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..045a3d7 --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +data/ +.pytest_cache/ +__pycache__/ +*.pyc +.a0proj/ +venv/ diff --git a/COOLIFY.md b/COOLIFY.md new file mode 100644 index 0000000..f6e546c --- /dev/null +++ b/COOLIFY.md @@ -0,0 +1,31 @@ +# Deployment — Coolify + +## Requirements +- A Coolify server (any host with Docker) and a git repository containing this project. + +## Steps +1. **Coolify → Projects → Add Service → Dockerfile** (Git repository source). +2. Point it at this repo; Coolify detects the `Dockerfile` at the project root. +3. **Port:** publish container port `8030` (or set env `PORT` and publish that). +4. **Environment variables:** none required — sensible defaults are baked in: + - `PORT=8030`, `DATA_DIR=/app/data`. +5. **Persistence:** add a volume for path `/app/data` (Docker volume or S3/object + storage) so uploaded EPUBs and generated PDFs survive deploys/restarts. +6. **Healthcheck** (Coolify settings): `GET /health`, expect HTTP 200. +7. **Domain/HTTPS:** optional — attach a domain and Coolify's Traefik letsencrypt + will handle TLS. The app itself is plain HTTP and sends no cookies, so it + needs no special proxy config. Large uploads (up to 400 MB) work through the + default proxy; if you see timeouts on very large files, raise the Traefik + `proxy send timeout` for this service. + +## Notes / gotchas +- **Single worker by design.** The job registry (`JOBS`) is in-memory; the + Dockerfile runs gunicorn with `--workers 1 --threads 8`. Do not scale to + multiple container instances unless you replace the in-memory registry with + a shared store — polling `/api/jobs/` would 404 on other instances. +- WeasyPrint renders in-process, so CPU-heavy conversions occupy worker + threads; `--timeout 1200` keeps long books from being killed by gunicorn. +- CJK fonts (`fonts-noto-cjk`) are installed in the image, so Chinese/Japanese + /Korean books render correctly. +- CLI mode also works inside the container for one-shots: + `docker exec python app \ No newline at end of file diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..82249ad --- /dev/null +++ b/Dockerfile @@ -0,0 +1,37 @@ +# EPUB3 -> PDF converter (Flask + WeasyPrint) — Coolify / any Docker host +# WeasyPrint needs pango + fontconfig system libraries; CJK fonts for full-text support. +FROM python:3.12-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + PORT=8030 \ + DATA_DIR=/app/data \ + PIP_NO_CACHE_DIR=1 + +RUN apt-get update && apt-get install -y --no-install-recommends \ + libpango-1.0-0 \ + libpangoft2-1.0-0 \ + libharfbuzz0b \ + libffi8 \ + shared-mime-info \ + fontconfig \ + fonts-dejavu-core \ + fonts-noto-cjk \ + && rm -rf /var/lib/apt/lists/* && fc-cache -f + +WORKDIR /app + +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY app.py . +COPY templates/ ./templates/ + +# Persistent data (uploads/output/work) — mount a volume at /app/data +VOLUME ["/app/data"] + +EXPOSE 8030 + +# Single worker: the in-memory JOBS registry must live in one process. +# Threads give concurrency while conversions run in their own worker threads. +CMD ["sh", "-c", "gunicorn app:app --bind 0.0.0.0:${PORT} --workers 1 --threads 8 --timeout 1200 --graceful-timeout 30 --access-logfile - --error-logfile -"] diff --git a/USER_GUIDE.md b/USER_GUIDE.md new file mode 100644 index 0000000..b6b8d9c --- /dev/null +++ b/USER_GUIDE.md @@ -0,0 +1,171 @@ +# EPUB → PDF Converter — User Guide + +A small, self-hosted web app that converts EPUB3 ebooks into typeset PDFs. +Everything runs locally (or on your own server); **no book ever leaves your machine**. + +--- + +## 1. What it does (and doesn't) + +| Does | Doesn't | +|---|---| +| Preserves the EPUB's own CSS, images, headings, tables, code blocks | Re-flow or re-typeset like a print publisher would | +| Renders CJK (Chinese / Japanese / Korean) correctly via Noto fonts | Handle DRM-protected (e.g. protected) EPUBs | +| Supports right-to-left books (Arabic, Hebrew) | Convert EPUB → other formats (only PDF out) | +| Keeps book title/author as PDF metadata | — | + +**Limits:** one book at a time per job (you can queue several, they render one after +another in the same server process), max upload size **400 MB**. + +--- + +## 2. Quick start + +### Option A — Run directly (no Docker) +```bash +# 1. install dependencies (Python 3.11+; WeasyPrint also needs the pango +# system libraries, e.g. on Debian/Ubuntu: sudo apt install libpango-1.0-0 +# libpangoft2-1.0-0 libharfbuzz0b fontconfig fonts-noto-cjk shared-mime-info) +pip install -r requirements.txt + +# 2. start the web server +python app.py # -> http://localhost:8030 +``` + +### Option B — Docker (recommended for a persistent install) +```bash +docker build -t epub2pdf . +docker run -d --name epub2pdf \ + -p 8030:8030 \ + -v epub2pdf-data:/app/data \ + epub2pdf +``` +The volume `epub2pdf-data` keeps uploaded EPUBs and generated PDFs across restarts. + +### Option C — Coolify +See `COOLIFY.md` in this directory (3 steps: Dockerfile service, port 8030, volume at `/app/data`). + +--- + +## 3. Using the web UI + +1. Open `http://:8030` in your browser. +2. **Drop an `.epub` file** onto the upload box (or click to browse). + - Only `.epub` files are accepted; anything else is rejected with a message. + - Files over 400 MB are rejected before upload finishes. +3. The job starts immediately — a progress bar shows which chapter is being + rendered (e.g. `Rendering chapter-04.xhtml (4/12)`), then `Merging pages…`. +4. When it finishes, a **Download PDF** button appears. + - The PDF is named after the book's title (e.g. `My_Book.pdf`). +5. Repeat for the next book. Old jobs stay listed so you can re-download + recent PDFs (the server keeps the 50 most recent jobs and their outputs). + +### What can go wrong in the UI +| Symptom | Meaning / fix | +|---|---| +| `File too large (max 400 MB)` | Shrink or split the book, or raise `MAX_UPLOAD_MB` in `app.py` | +| `Not a valid EPUB (bad zip container)` | File is corrupted or renamed — re-export from your ebook store/library | +| `No readable content found in EPUB spine` | The EPUB has no HTML chapters (images-only books, broken exports) | +| `DRM protected` style errors | Remove DRM first (Calibre / Calibre-DB tools) — this app can't decrypt | +| Stuck at `Working…` forever | The server restarted mid-job; refresh the page and re-upload | + +--- + +## 4. Command-line usage (no server) + +```bash +# convert one file and exit +python app.py --cli book.epub + +# choose where the PDF lands +python app.py --cli book.epub --output /path/to/dir + +# inside a running container +docker exec epub2pdf python app.py --cli /app/data/uploads/book.epub --output /app/data/output +``` +Progress lines like `[3/12] chapter-03.xhtml` print as each chapter renders. + +--- + +## 5. HTTP API (for automation) + +Three endpoints, all JSON: + +```bash +# 1. upload (multipart form, field name: file) -> starts a job +curl -X POST -F "file=@book.epub" http://localhost:8030/api/convert +# -> {"job_id": "c524be5a9d85"} + +# 2. poll status +curl http://localhost:8030/api/jobs/c524be5a9d85 +# -> {"status": "working", "progress": 42.5, "message": "Rendering ch-05.xhtml (5/12)"} +# -> {"status": "done", "progress": 100, "filename": "Book_Title-20260828-203943.pdf"} +# -> {"status": "error", "error": "Not a valid EPUB (bad zip container)"} + +# 3. download the finished PDF +curl -OJ http://localhost:8030/download/Book_Title-20260828-203943.pdf +``` + +Other endpoints: +- `GET /` — the web UI +- `GET /health` — `{"status":"ok","max_upload_mb":400}` (use for healthchecks) + +--- + +## 6. Configuration + +Everything works with defaults. Optional overrides: + +| Setting | Default | How to change | +|---|---|---| +| Port | `8030` | Env `PORT`, or `--port` (web mode) | +| Data location | `/data` | Env `DATA_DIR` (Docker: volume at `/app/data`) — holds `uploads/`, `output/`, `work/` | +| Max upload size | `400 MB` | Constant `MAX_UPLOAD_MB` in `app.py` (edit + restart) | +| Page geometry | A4, 20×18 mm margins | Edit `BASE_CSS` in `app.py` (e.g. `size: A4` → `size: Letter`) | +| Base typography | Georgia/serif 10.5 pt, line-height 1.65 | Edit `BASE_CSS` in `app.py` — the EPUB's own CSS still layers on top of it | +| RTL handling | auto (reads EPUB `page-progression-direction`) | automatic; no setting | + +### Fonts (affects what renders correctly) +- The Docker image ships **DejaVu** + **Noto CJK** fonts. +- When running without Docker, install at least: `fontconfig fonts-dejavu-core fonts-noto-cjk` + (plus `libpango-1.0-0 libpangoft2-1.0-0 libharfbuzz0b shared-mime-info` for WeasyPrint itself). +- If a book uses an exotic font not installed, WeasyPrint falls back to the + nearest available font — layout stays intact, letterforms may differ. + +--- + +## 7. Where files live + +``` +/ +├── uploads/ # your uploaded .epub files (kept after conversion) +├── output/ # generated PDFs (kept; served by /download/) +└── work/ # per-job temp dir (extracted EPUB + chunk PDFs; deleted after each job) +``` + +You can delete anything in `uploads/` and `output/` at any time (the server +just returns 404 for downloads of deleted files). `work/` is always empty when +idle. + +--- + +## 8. Troubleshooting + +| Problem | Check | +|---|---| +| Server won't start: `OSError: cannot load library 'libpango-1.0-0'` | Missing system libraries — see §2 Option A note | +| CJK text shows as boxes (□□□) | `fonts-noto-cjk` not installed — install, then run `fc-cache -f`, restart server | +| Strange characters (`—`) instead of em dashes | This build reads EPUB text as UTF-8 explicitly; if you see this, an old server is running — restart it | +| Conversion is slow | Normal: WeasyPrint typesets chapter-by-chapter in-process. Big/illustrated books can take minutes; gunicorn timeout is 1200 s by default | +| Port already in use | `--port 9000` or change env `PORT` | + +--- + +## 9. Security notes (self-hosted) + +- The app has **no login**. Anyone who can reach the port can upload and convert. + On a LAN, that's usually fine; on the public internet, put it behind a + reverse proxy with auth, or firewall the port. +- Uploads are stored under a random job ID (not your filename), and downloads + are restricted to the output directory (path traversal is blocked). +- Nothing is transmitted anywhere: conversion happens entirely in-process. diff --git a/app.py b/app.py new file mode 100644 index 0000000..90d5026 --- /dev/null +++ b/app.py @@ -0,0 +1,670 @@ +#!/usr/bin/env python3 +"""EPUB3 -> PDF local web converter. + +Serves a small web UI where you upload an EPUB3 book and download a +typeset PDF. Rendering is done by WeasyPrint (HTML/CSS -> PDF), which +preserves the EPUB's own CSS, images, headings and CJK text. The final +PDF is assembled per spine item and merged with pypdf. + +Runs entirely locally; nothing is sent to any external service. + +Usage: + python app.py # web server on 0.0.0.0:8030 + python app.py --port 9000 # custom port + python app.py --cli book.epub # headless one-shot conversion +""" +from __future__ import annotations + +import argparse +import codecs +import logging +import os +import re +import shutil +import subprocess +import sys +import threading +import time +import uuid +import zipfile +from pathlib import Path + +from flask import Flask, abort, jsonify, render_template, request, send_file +from werkzeug.exceptions import HTTPException, RequestEntityTooLarge +from lxml import etree +from pypdf import PdfWriter + +BASE_DIR = Path(__file__).resolve().parent +# DATA_DIR is overridable via env so container deployments (e.g. Coolify) +# can point uploads/output/work at a persistent volume. +DATA_DIR = Path(os.environ.get("DATA_DIR", str(BASE_DIR / "data"))) +UPLOAD_DIR = DATA_DIR / "uploads" +OUTPUT_DIR = DATA_DIR / "output" +WORK_DIR = DATA_DIR / "work" +for _d in (UPLOAD_DIR, OUTPUT_DIR, WORK_DIR): + _d.mkdir(parents=True, exist_ok=True) + +MAX_UPLOAD_MB = 400 + +# ZIP-bomb guard: total uncompressed size must stay within 8x the upload +# size, with a 2 GiB floor so normal books always fit (finding: zip bomb, +# app.py extractall). +ZIP_BOMB_MIN_CAP = 2 * 1024 ** 3 + +# Bounded conversion pool: at most 4 concurrent conversions; extra uploads +# receive a JSON 503 (backpressure) instead of an unbounded Thread per +# upload (finding: unbounded threads, app.py:293-307). +WORKER_SLOTS = threading.BoundedSemaphore(4) + +# Crash leftovers: sweep work dirs / uploads older than these at startup. +STALE_WORK_SECONDS = 6 * 3600 +STALE_UPLOAD_SECONDS = 3600 + +app = Flask(__name__) +app.config["MAX_CONTENT_LENGTH"] = MAX_UPLOAD_MB * 1024 * 1024 + +log = logging.getLogger("epub2pdf") + +# -------------------------------------------------------------------------- +# conversion jobs (in-memory, single worker process) +# -------------------------------------------------------------------------- + +JOBS: dict[str, dict] = {} +_JOBS_LOCK = threading.Lock() + +TERMINAL_STATUSES = ("done", "error") + + +def _register_job(epub_name: str) -> str: + """Register a job and enforce the 50-entry cap. + + Eviction never touches in-flight (queued/working) jobs while any + terminal (done/error) entries exist to evict — the UI must not lose a + live job mid-conversion (finding: eviction kills in-flight jobs, + app.py:69-73). Only when no terminal entry exists at all is the oldest + job of any status evicted, preserving the plain 50-cap contract. + """ + job = { + "id": uuid.uuid4().hex[:12], + "epub": epub_name, + "status": "queued", + "progress": 0.0, + "message": "queued", + "filename": None, + "error": None, + "created": time.time(), + } + with _JOBS_LOCK: + JOBS[job["id"]] = job + while len(JOBS) > 50: + terminal = [k for k, v in JOBS.items() if v["status"] in TERMINAL_STATUSES] + if terminal and len(terminal) <= 50: + break # terminal entries fit the cap; keep in-flight jobs + pool = terminal or list(JOBS) + oldest = min(pool, key=lambda k: JOBS[k]["created"]) + JOBS.pop(oldest, None) + return job["id"] + + +def _set_job(job_id: str, **kw) -> None: + with _JOBS_LOCK: + if job_id in JOBS: + JOBS[job_id].update(kw) + + +# -------------------------------------------------------------------------- + +# -------------------------------------------------------------------------- +# CSS handed to WeasyPrint (EPUB's own stylesheets still apply on top) +# -------------------------------------------------------------------------- + +BASE_CSS = """ +@page { size: A4; margin: 20mm 18mm; } +body { + font-family: Georgia, "Times New Roman", "Noto Serif", + "Noto Serif CJK SC", "Noto Serif CJK TC", "Noto Serif CJK JP", + "Noto Serif CJK KR", serif; + font-size: 10.5pt; + line-height: 1.65; + color: #1c1b1a; +} +h1, h2, h3, h4, h5, h6 { + font-weight: 700; line-height: 1.25; color: #111; + page-break-after: avoid; +} +h1 { font-size: 21pt; margin: 0 0 0.8em 0; } +h2 { font-size: 16pt; margin: 1.2em 0 0.5em 0; } +h3 { font-size: 13pt; margin: 1em 0 0.4em 0; } +p { margin: 0 0 0.7em 0; } +img, svg { max-width: 100%; height: auto; } +pre, code { + font-family: "DejaVu Sans Mono", "Courier New", monospace; + font-size: 9pt; + white-space: pre-wrap; + word-wrap: break-word; +} +pre { background: #f4f1ec; border-radius: 4px; padding: 0.6em 0.8em; margin: 0 0 1em 0; } +blockquote { + margin: 1em 0; padding-left: 1em; + border-left: 3px solid #b5541c; color: #444; font-style: italic; +} +table { border-collapse: collapse; margin: 1em 0; width: 100%; } +th, td { border: 0.5pt solid #999; padding: 0.35em 0.6em; text-align: left; } +th { background: #f4f1ec; } +hr { border: none; border-top: 0.5pt solid #bbb; margin: 2em 0; } +ul, ol { margin: 0 0 0.8em 0; padding-left: 1.6em; } +a { color: inherit; text-decoration: none; } +figure { margin: 1.2em 0; text-align: center; } +figcaption { font-size: 8.5pt; color: #666; margin-top: 0.4em; } +[hidden] { display: none; } +""" + +COVER_CSS = """ +@page { margin: 0; } +html, body { margin: 0; padding: 0; height: 100%; } +body img, body svg { width: 210mm; height: 297mm; object-fit: cover; display: block; } +""" + + +# -------------------------------------------------------------------------- +# EPUB structure (OPF) parsing +# -------------------------------------------------------------------------- + +NS = { + "opf": "http://www.idpf.org/2007/opf", + "dc": "http://purl.org/dc/elements/1.1/", +} + + +def _contained(base: Path, rel: str) -> Path: + """Resolve `rel` against `base`, refusing to leave the tree. + + Same guard class as zip-slip: a crafted OPF (rootfile full-path or a + spine/manifest href) must not be able to point outside the extraction + dir (finding: spine/manifest href containment check). + """ + p = (base / rel).resolve() + if not p.is_relative_to(base.resolve()): + raise ValueError(f"path escapes extraction dir: {rel}") + return p + + +def _find_opf_path(extract_dir: Path) -> Path: + container = extract_dir / "META-INF" / "container.xml" + if not container.exists(): + raise ValueError("Not an EPUB: missing META-INF/container.xml") + root = etree.parse(str(container)).getroot() + rootfile = root.find(".//rootfile") + if rootfile is None: # tolerate spec-correct namespaced container.xml + for el in root.iter(): + if el.tag.rsplit("}", 1)[-1] == "rootfile": + rootfile = el + break + if rootfile is None: + raise ValueError("Not an EPUB: container.xml has no rootfile") + return _contained(extract_dir, rootfile.get("full-path")) + + +def parse_opf(extract_dir: Path): + """Return (meta, opf_dir, spine_items, page_progression_direction).""" + opf_path = _find_opf_path(extract_dir) + opf_dir = opf_path.parent + root = etree.parse(str(opf_path)).getroot() + + def _local(tag: str) -> str: + return tag.rsplit("}", 1)[-1] + + meta = {} + for el in root.iter(): + if not isinstance(el.tag, str) or _local(el.tag) not in ("title", "creator", "language", "description"): + continue + ns = el.tag.rsplit("}", 1)[0].lstrip("{") if "}" in el.tag else "" + if "dc" in ns or not ns: # Dublin Core metadata (namespaced or plain) + meta.setdefault(_local(el.tag), (el.text or "").strip()) + + manifest: dict[str, dict] = {} + for el in root.iter(): + if not isinstance(el.tag, str) or _local(el.tag) != "item" or not el.get("id"): + continue + manifest[el.get("id")] = { + "href": el.get("href", ""), + "media_type": el.get("media-type") or el.get("media_type") or "", + "properties": el.get("properties") or "", + } + + spine: list[dict] = [] + for ref in root.iter(): + if not isinstance(ref.tag, str) or _local(ref.tag) != "itemref": + continue + item = manifest.get(ref.get("idref")) + if item is None or ref.get("linear", "yes") == "no": + continue + if "html" not in item["media_type"]: + continue + try: # containment: a spine href must stay inside the extraction tree + _contained(opf_dir, item["href"]) + except ValueError: + continue + spine.append(item) + + if not spine: # malformed book: fall back to all readable documents + for item in manifest.values(): + if "html" in item["media_type"] and "nav" not in item["properties"]: + spine.append(item) + if not spine: + raise ValueError("No readable content found in EPUB spine") + + pdir = root.get("page-progression-direction", "ltr") or "ltr" + return meta, opf_dir, spine, pdir + + +def _looks_like_cover(opf_dir: Path, item: dict) -> bool: + props = item["properties"].split() + if "cover-image" in props: + return True + try: + from bs4 import BeautifulSoup + + text = _decode_html((opf_dir / item["href"]).read_bytes()) + except OSError: + return False + if "]*?encoding[\s=]+["\']?([A-Za-z0-9_\-\.]+)', re.IGNORECASE +) +_META_ENC = re.compile( + rb']*charset[\s=]+["\']?([A-Za-z0-9_\-\.]+)', re.IGNORECASE +) +_BODY_INNER = re.compile(r"]*>(.*)", re.IGNORECASE | re.DOTALL) + + +def _decode_html(raw: bytes) -> str: + """Decode a chapter document honoring its declared charset. + + Order: XML declaration encoding, then , then UTF-8 with + errors='replace'. Fixes legacy EPUB2 books in windows-1252/latin-1 that + would mojibake under a blind UTF-8 decode (same bug class as the + em-dash incident). Undecodable input must not raise. + """ + m = _XML_DECL_ENC.search(raw[:4096]) or _META_ENC.search(raw[:8192]) + if m: + enc = m.group(1).decode("ascii", errors="replace").strip() + try: + codecs.lookup(enc) + text = raw.decode(enc) + except (LookupError, UnicodeDecodeError): + text = raw.decode("utf-8", errors="replace") + else: + text = raw.decode("utf-8", errors="replace") + # WeasyPrint receives the content: a bare fragment keeps per-chapter + # layout stable and drops EPUB head cruft (meta/style) that WeasyPrint + # would otherwise apply globally. + bm = _BODY_INNER.search(text) + return bm.group(1) if bm else text + + + +# -------------------------------------------------------------------------- +# environment validation +# -------------------------------------------------------------------------- + +# Non-numeric PORT must fail fast with a clear message (finding: launcher +# divergence) — printed to stdout and exit 2 so both `python app.py` and the +# gunicorn import path stop instead of binding a garbage port. +try: + PORT = int(os.environ.get("PORT", "8030")) +except (TypeError, ValueError): + print("PORT must be an integer, got: %r" % os.environ.get("PORT")) + sys.exit(2) + + +def _format_progress(done: int, total: int, label: str) -> str: + """Progress line matching the UI regex \\(\\d+/\\d+\\) exactly.""" + return f"Rendering {label} ({done}/{total})" + + +def _cjk_fonts_present() -> bool: + """True if fontconfig reports at least one CJK font (health check).""" + try: + out = subprocess.run( + ["fc-list", ":lang=zh", "family"], + capture_output=True, text=True, timeout=15, + ) + return bool(out.stdout.strip()) + except (OSError, subprocess.TimeoutExpired): + return False + + +def _sweep_stale_work_dirs() -> None: + """Remove WORK_DIR/job-* left by crashed runs older than the 1h window.""" + now = time.time() + try: + entries = list(WORK_DIR.iterdir()) + except OSError: + return + for entry in entries: + if not entry.is_dir() or not entry.name.startswith("job-"): + continue + try: + if now - entry.stat().st_mtime > STALE_WORK_SECONDS: + shutil.rmtree(entry, ignore_errors=True) + except OSError: + pass + + +# Crash leftovers from the previous process start (finding: stale work dirs). +_sweep_stale_work_dirs() + + +# -------------------------------------------------------------------------- +# JSON error handlers (client pollers must never receive an HTML error page) +# -------------------------------------------------------------------------- + + +@app.errorhandler(RequestEntityTooLarge) +@app.errorhandler(413) +def _handle_413(err): # noqa: ARG001 + """Oversized upload: JSON 413 with the human-readable limit.""" + return jsonify(error=f"File too large (max {MAX_UPLOAD_MB} MB)"), 413 + + +@app.errorhandler(HTTPException) +def _handle_http_exception(err): + """Any other HTTP error: JSON body so API clients never parse HTML.""" + return jsonify(error=err.description or err.name), err.code + + +# -------------------------------------------------------------------------- +# safe extraction (zip-slip + zip-bomb, finding: app.py extractall) +# -------------------------------------------------------------------------- + + +def _safe_extract_zip(epub_path: Path, extract_dir: Path) -> None: + """Unzip an EPUB refusing path-escaping, absolute and symlink members. + + Also enforces the zip-bomb cap: total uncompressed size must stay within + max(8x upload size, ZIP_BOMB_MIN_CAP) or the job fails with a clear error. + """ + upload_size = epub_path.stat().st_size + cap = max(8 * upload_size, ZIP_BOMB_MIN_CAP) + try: + zf = zipfile.ZipFile(epub_path) + except zipfile.BadZipFile: + raise ValueError("Not a valid EPUB: file is not a readable zip archive") + with zf: + infos = zf.infolist() + total = sum(i.file_size for i in infos) + if total > cap: + raise ValueError( + f"zip-bomb guard: uncompressed size {total} exceeds cap {cap}" + ) + for info in infos: + if info.file_size > ZIP_BOMB_MIN_CAP: + raise ValueError( + f"zip-bomb guard: member {info.filename} is {info.file_size} bytes " + f"(max member size {ZIP_BOMB_MIN_CAP})" + ) + for info in infos: + name = info.filename + if ((info.external_attr >> 16) & 0xF000) == 0xA000: # S_IFLNK + raise ValueError(f"unsafe zip member (symlink): {name}") + if name.startswith("/") or name.startswith("\\"): + raise ValueError(f"unsafe zip member (absolute): {name}") + parts = name.replace("\\", "/").split("/") + if ".." in parts: + raise ValueError(f"unsafe zip member (dotdot): {name}") + for info in infos: + target = (extract_dir / info.filename).resolve() + if not target.is_relative_to(extract_dir.resolve()): + raise ValueError(f"unsafe zip member (escapes): {info.filename}") + if info.is_dir(): + target.mkdir(parents=True, exist_ok=True) + continue + target.parent.mkdir(parents=True, exist_ok=True) + with zf.open(info) as src, open(target, "wb") as dst: + shutil.copyfileobj(src, dst) + + +# -------------------------------------------------------------------------- +# conversion pipeline +# -------------------------------------------------------------------------- + + +def convert_epub(epub_path: Path, job_id: str, progress_cb=None, output_dir=None, + source_name=None) -> str: + """Render an EPUB to a merged PDF; return the output file name (no dir). + + progress_cb(pct, message) is called per spine chapter and before the + merge. The message format matches the UI poll regex exactly. + output_dir overrides OUTPUT_DIR (CLI mode); defaults to OUTPUT_DIR. + """ + import io + from pypdf import PdfReader + from weasyprint import CSS, HTML + + out_dir = Path(output_dir) if output_dir else OUTPUT_DIR + out_dir.mkdir(parents=True, exist_ok=True) + + extract_dir = WORK_DIR / f"job-{job_id}" + extract_dir.mkdir(parents=True, exist_ok=True) + try: + _safe_extract_zip(epub_path, extract_dir) + meta, opf_dir, spine, pdir = parse_opf(extract_dir) + total = len(spine) + writer = PdfWriter() + for idx, item in enumerate(spine, start=1): + label = item["href"] + if progress_cb: + progress_cb(min(99.0, 100.0 * idx / total), _format_progress(idx, total, label)) + try: + text = _decode_html(_contained(opf_dir, item["href"]).read_bytes()) + except (ValueError, OSError): + continue # unreadable or escaping chapter: skip, keep going + css = COVER_CSS if _looks_like_cover(opf_dir, item) else BASE_CSS + if pdir == "rtl": # right-to-left books (finding: page-progression) + css += "\nhtml { direction: rtl; }\n" + buf = io.BytesIO() + HTML(string=text, base_url=str(opf_dir)).write_pdf( + buf, stylesheets=[CSS(string=css)] + ) + for page in PdfReader(io.BytesIO(buf.getvalue())).pages: + writer.add_page(page) + if len(writer.pages) == 0: + raise ValueError("no renderable content: EPUB produced 0 PDF pages") + if progress_cb: + progress_cb(99.0, "Merging pages\u2026") + title = (meta.get("title") or "").strip() or "Untitled" + author = (meta.get("creator") or "").strip() + meta_info = {"/Title": title} # pypdf 5+/6x requires NameObject keys + if author: + meta_info["/Author"] = author + writer.add_metadata(meta_info) + # req 1: export name = clean book title, else uploaded .epub stem, else 'book' + base = re.sub(r"[^A-Za-z0-9._ -]+", "_", (meta.get("title") or "").strip()).strip()[:60].strip() + if not base: + # req 1 fallback: uploaded .epub stem (original name when provided, + # else the on-disk stem). Guard: stem may be empty -> 'book'. + stem_src = source_name if source_name else Path(epub_path).stem + base = re.sub(r"[^A-Za-z0-9._ -]+", "_", Path(stem_src).stem).strip()[:60].strip() or "book" + # req 2: never double the extension (case-insensitive 'x.pdf' -> 'x') + if base.lower().endswith(".pdf"): + base = base[:-4].rstrip() or "book" + # req 3: collision-safe name: first free of base.pdf, base-2.pdf, base-3.pdf, ... + filename = f"{base}.pdf" + counter = 2 + while (out_dir / filename).exists(): + filename = f"{base}-{counter}.pdf" + counter += 1 + tmp_path = out_dir / (filename + ".tmp") # atomic: write .tmp then rename + with open(tmp_path, "wb") as out: + writer.write(out) + os.replace(tmp_path, out_dir / filename) + return filename + finally: + shutil.rmtree(extract_dir, ignore_errors=True) + + +# -------------------------------------------------------------------------- +# worker wrapper +# -------------------------------------------------------------------------- + + +def _run_job(job_id: str, epub_path: Path) -> None: + """Worker-thread body: run the pipeline, track status, always release.""" + _set_job(job_id, status="working", message="Working\u2026", progress=0.0) + + def _cb(pct, msg): + _set_job(job_id, progress=pct, message=msg) + + try: + filename = convert_epub(epub_path, job_id, progress_cb=_cb, + source_name=JOBS.get(job_id, {}).get("epub")) + _set_job(job_id, status="done", progress=100.0, message="Done", + filename=filename, error=None) + except Exception as exc: # noqa: BLE001 - any failure is a job error + log.exception("job %s failed", job_id) + _set_job(job_id, status="error", error=str(exc), message="failed") + finally: + try: + WORKER_SLOTS.release() + except ValueError: + pass + try: + (UPLOAD_DIR / f"{job_id}.epub").unlink(missing_ok=True) + except OSError: + pass + _sweep_stale_work_dirs() + + +# -------------------------------------------------------------------------- +# routes +# -------------------------------------------------------------------------- + + +@app.route("/") +def index(): + return render_template("index.html") + + +@app.route("/health") +def health(): + """Real liveness: DATA_DIR writable, weasyprint importable, CJK fonts.""" + problems = [] + try: + probe = DATA_DIR / ".health_probe" + probe.write_text("ok", encoding="utf-8") + probe.unlink(missing_ok=True) + except OSError: + problems.append(f"DATA_DIR not writable: {DATA_DIR}") + try: + import weasyprint # noqa: F401 + except Exception: # noqa: BLE001 + problems.append("weasyprint not importable") + if not _cjk_fonts_present(): + problems.append("CJK fonts missing (fc-list found none)") + body = {"status": "ok", "max_upload_mb": MAX_UPLOAD_MB} + if problems: + body = {"status": "degraded", "max_upload_mb": MAX_UPLOAD_MB, "problems": problems} + return jsonify(body), 503 + return jsonify(body), 200 + + +@app.route("/api/convert", methods=["POST"]) +def convert_upload(): + up = request.files.get("file") + if up is None or not up.filename: + return jsonify(error="No file uploaded"), 400 + if not up.filename.lower().endswith(".epub"): + return jsonify(error="Only .epub files are accepted"), 400 + data = up.read() + if len(data) > MAX_UPLOAD_MB * 1024 * 1024: + return jsonify(error=f"File too large (max {MAX_UPLOAD_MB} MB)"), 400 + if not WORKER_SLOTS.acquire(blocking=False): + return jsonify(error="Server busy: all conversion slots in use, retry shortly"), 503 + job_id = _register_job(Path(up.filename).name) + dest = UPLOAD_DIR / f"{job_id}.epub" + try: + with open(dest, "wb") as out: + out.write(data) + except OSError: + WORKER_SLOTS.release() + _set_job(job_id, status="error", error="failed to store upload", message="failed") + return jsonify(error="failed to store upload"), 500 + threading.Thread(target=_run_job, args=(job_id, dest), daemon=True).start() + return jsonify(job_id=job_id), 202 + + +@app.route("/api/jobs/") +def job_status(job_id): + with _JOBS_LOCK: + snapshot = dict(JOBS[job_id]) if job_id in JOBS else None + if snapshot is None: + return jsonify(error="unknown job"), 404 + return jsonify(snapshot), 200 + + +@app.route("/download/") +def download(name): + """Serve a finished PDF; the name is re-rooted to OUTPUT_DIR (no traversal).""" + safe = Path(name).name + target = (OUTPUT_DIR / safe).resolve() + if not target.is_relative_to(OUTPUT_DIR.resolve()) or not target.is_file(): + return jsonify(error="unknown file"), 404 + # req 4: explicit PDF mimetype + defensive .pdf on download_name + download_name = safe if safe.lower().endswith(".pdf") else safe + ".pdf" + return send_file( + target, as_attachment=True, mimetype="application/pdf", download_name=download_name + ) + + +# -------------------------------------------------------------------------- +# CLI mode + entrypoint +# -------------------------------------------------------------------------- + + +def _run_cli(argv: list[str]) -> int: + """`python app.py --cli [--output ]` — headless conversion.""" + import argparse + + parser = argparse.ArgumentParser(description="EPUB -> PDF converter") + parser.add_argument("--cli", metavar="EPUB", help="convert EPUB headlessly") + parser.add_argument("--output", metavar="DIR", default=str(OUTPUT_DIR), + help="output directory (default: data/output)") + args = parser.parse_args(argv) + if not args.cli: + # No --cli: fall through to the dev server (see __main__ guard). + return None + src = Path(args.cli) + if not src.is_file(): + print(f"{src} not found") + return 2 + out_dir = Path(args.output) + out_dir.mkdir(parents=True, exist_ok=True) + job_id = uuid.uuid4().hex + + def _cb(pct, message): + print(f"[{pct:5.1f}%] {message}") + + try: + filename = convert_epub(src, job_id, progress_cb=_cb, output_dir=out_dir) + except Exception as exc: # noqa: BLE001 - CLI reports any failure as error: + print(f"error: {exc}") + return 1 + print(f"Saved: {out_dir / filename}") + return 0 + + +if __name__ == "__main__": + _code = _run_cli(sys.argv[1:]) + if _code is not None: + sys.exit(_code) + app.run(host="0.0.0.0", port=PORT) # dev server; gunicorn uses app:app diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..eeb9d41 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +testpaths = tests +addopts = -q diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..8b8eda4 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,6 @@ +flask==3.1.3 +weasyprint==69.0 +pypdf==6.16.2 +beautifulsoup4==4.15.0 +lxml==6.1.2 +gunicorn==26.2.0 diff --git a/start.sh b/start.sh new file mode 100755 index 0000000..642a127 --- /dev/null +++ b/start.sh @@ -0,0 +1,38 @@ +#!/usr/bin/env bash +# One-command local test startup for the EPUB->PDF converter. +# Foreground gunicorn with an auto-restart loop (no systemd/supervisor needed). +# +# ./start.sh # http://0.0.0.0:8030 +# PORT=9000 ./start.sh # custom port +# +# WORKERS is pinned to 1: the job registry (app.JOBS) is in-memory per +# process, so multiple workers would lose job state between requests. +# Extra concurrency comes from --threads instead. The gunicorn flags mirror +# the Dockerfile CMD (single source of truth for the launch contract). +set -u +cd "$(dirname "$0")" + +HOST="${HOST:-0.0.0.0}" +PORT="${PORT:-8030}" +THREADS="${THREADS:-8}" +export DATA_DIR="${DATA_DIR:-$PWD/data}" + +PY="${PY:-/opt/venv/bin/python}" + +echo "[start] EPUB->PDF converter on http://${HOST}:${PORT}" +echo "[start] DATA_DIR=${DATA_DIR} workers=1 threads=${THREADS}" + +while true; do + "$PY" -m gunicorn \ + --bind "${HOST}:${PORT}" \ + --workers 1 \ + --threads "$THREADS" \ + --timeout 1200 \ + --graceful-timeout 30 \ + --access-logfile - \ + --error-logfile - \ + app:app + code=$? + echo "[start] gunicorn exited (code ${code}); restarting in 2s..." + sleep 2 +done diff --git a/templates/index.html b/templates/index.html new file mode 100644 index 0000000..dbbebeb --- /dev/null +++ b/templates/index.html @@ -0,0 +1,305 @@ + + + + + +Bindery — EPUB to PDF + + + +
+
Bindery
+

A local typesetter. Drop an EPUB, take a properly set PDF. Nothing leaves your machine.

+
+
+ +
+
+ +
+

Drop a book to set it in type

+

or browse your files · up to 400 MB · .epub

+
+ + +
+ + +
+ +
+ Bindery · EPUB → PDF · WeasyPress engine + press: … +
+ + + + diff --git a/test_book.epub b/test_book.epub new file mode 100644 index 0000000..21794df Binary files /dev/null and b/test_book.epub differ diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..c2b7d06 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,62 @@ +"""Hermetic fixtures for the EPUB -> PDF converter suite. + +Isolation contract +------------------ +- ``DATA_DIR`` is redirected to a fresh temp dir BEFORE ``app`` is imported + (app.py reads the env var at import time, app.py:36). +- The Flask test client is used: no live server, no port conflicts, while + conversion worker threads still run in-process. +- The module-global ``JOBS`` registry is cleared before and after every test. +""" +import io +import os +import sys +import tempfile +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +# Must happen before importing app — app.py bakes DATA_DIR in at import time. +os.environ["DATA_DIR"] = tempfile.mkdtemp(prefix="epub2pdf-tests-") + +import app # noqa: E402 + +FIXTURE_EBOOK = ROOT / "test_book.epub" + + +@pytest.fixture(scope="module") +def client(): + """Flask test client for the module-level app instance.""" + app.app.config["TESTING"] = True + with app.app.test_client() as c: + yield c + + +@pytest.fixture(autouse=True) +def _isolate_jobs(): + """Every test starts with (and leaves) an empty job registry.""" + app.JOBS.clear() + yield + app.JOBS.clear() + + +@pytest.fixture(scope="module") +def fixture_epub_bytes(): + """Raw bytes of the checked-in test book (em dashes + CJK).""" + return FIXTURE_EBOOK.read_bytes() + + +@pytest.fixture(scope="module") +def submit_epub(client, fixture_epub_bytes): + """Factory: POST an upload to /api/convert; returns the response.""" + def _submit(filename="book.epub", content=None): + if content is None: + content = fixture_epub_bytes + files = {"file": (io.BytesIO(content), filename)} + return client.post( + "/api/convert", data=files, content_type="multipart/form-data" + ) + return _submit diff --git a/tests/helpers.py b/tests/helpers.py new file mode 100644 index 0000000..96be469 --- /dev/null +++ b/tests/helpers.py @@ -0,0 +1,141 @@ +"""Pure-function test helpers for the EPUB -> PDF converter suite. + +No pytest fixtures here: every function takes its dependencies explicitly so +the helpers stay unit-testable and reusable. +""" +import io +import time +import zipfile + +from pypdf import PdfReader + +TERMINAL_STATUSES = ("done", "error") + + +def wait_for_job(client, job_id, timeout=90.0): + """Poll GET /api/jobs/ until a terminal status; return snapshot.""" + deadline = time.time() + timeout + last = None + while time.time() < deadline: + resp = client.get(f"/api/jobs/{job_id}") + assert resp.status_code == 200, f"job lookup failed: {resp.status_code}" + last = resp.get_json() + if last["status"] in TERMINAL_STATUSES: + return last + time.sleep(0.05) + raise TimeoutError(f"job {job_id} not terminal after {timeout}s: {last}") + + +def download_bytes(client, filename): + """GET /download/ and return the raw body (asserts 200).""" + resp = client.get(f"/download/{filename}") + assert resp.status_code == 200, ( + f"download failed for {filename}: {resp.status_code}" + ) + return resp.data + + +def pdf_text(data_bytes): + """Extract all text from a PDF byte string (pypdf).""" + reader = PdfReader(io.BytesIO(data_bytes)) + return "\n".join((page.extract_text() or "") for page in reader.pages) + + +def pdf_page_count(data_bytes): + return len(PdfReader(io.BytesIO(data_bytes)).pages) + + +def pdf_metadata_title(data_bytes): + reader = PdfReader(io.BytesIO(data_bytes)) + meta = reader.metadata + return meta.get("/Title") if meta else None + + +def make_zip_without_container(): + """A valid zip that is not an EPUB (no META-INF/container.xml).""" + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + zf.writestr("notes.txt", "hello, not an epub") + return buf.getvalue() + + +_CONTAINER_XML = ( + '' + '' + '' +) + + +def make_epub_zip(items, spine_ids, title="Synthetic", creator="Synth"): + """Build a minimal EPUB byte string from (id, href, media_type, props) items.""" + manifest = "".join( + f'" + for i, h, m, p in items + ) + refs = "".join(f'' for sid in spine_ids) + opf = ( + '' + '' + '' + f"{title}{creator}" + "" + f"{manifest}" + f"{refs}" + "" + ) + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + zf.writestr("META-INF/container.xml", _CONTAINER_XML) + zf.writestr("OEBPS/book.opf", opf) + return buf.getvalue() + + +def build_epub_tree(base, *, items=None, spine_ids=None, pdir=None): + """Materialize a minimal on-disk EPUB structure for parse_opf tests.""" + items = items or [ + ("ch1", "chapter1.xhtml", "application/xhtml+xml", ""), + ("ch2", "chapter2.xhtml", "application/xhtml+xml", ""), + ("img1", "img.png", "image/png", ""), + ] + spine_ids = ( + list(spine_ids) + if spine_ids is not None + else [i for i, h, m, p in items if "html" in m] + ) + pdir_attr = f' page-progression-direction="{pdir}"' if pdir else "" + manifest = "".join( + f'" + for i, h, m, p in items + ) + refs = "".join(f'' for sid in spine_ids) + opf = ( + '' + f'' + '' + "Tree Book" + "Tree Author" + "en" + "" + f"{manifest}" + f"{refs}" + "" + ) + (base / "META-INF").mkdir(parents=True, exist_ok=True) + (base / "META-INF" / "container.xml").write_text(_CONTAINER_XML, encoding="utf-8") + (base / "OEBPS").mkdir(parents=True, exist_ok=True) + (base / "OEBPS" / "book.opf").write_text(opf, encoding="utf-8") + for i, h, m, p in items: + if "html" in m: + (base / "OEBPS" / h).write_text( + '' + "

chapter body

", + encoding="utf-8", + ) + return base diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..2f9362f --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,50 @@ +"""CLI entrypoint tests: `python app.py --cli --output `. + +Run as a real subprocess (not in-process) so argparse, exit codes and +stdout behavior are tested exactly as a user would invoke them. +""" +import os +import sys +import zipfile +import subprocess +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +FIXTURE_EBOOK = ROOT / "test_book.epub" + + +def _run_cli(*args): + env = dict(os.environ) + return subprocess.run( + [sys.executable, str(ROOT / "app.py"), *args], + capture_output=True, + text=True, + timeout=120, + env=env, + cwd=str(ROOT), + ) + + +def test_cli_convert_success(tmp_path): + out_dir = tmp_path / "out" + proc = _run_cli("--cli", str(FIXTURE_EBOOK), "--output", str(out_dir)) + assert proc.returncode == 0, proc.stderr + assert "Saved:" in proc.stdout + pdfs = list(out_dir.glob("*.pdf")) + assert len(pdfs) == 1 + assert pdfs[0].stat().st_size > 1024 + assert pdfs[0].read_bytes()[:4] == b"%PDF" + + +def test_cli_missing_file_exits_2(tmp_path): + proc = _run_cli("--cli", str(tmp_path / "nope.epub"), "--output", str(tmp_path)) + assert proc.returncode == 2 + assert "not found" in proc.stdout + + +def test_cli_corrupt_epub_exits_1(tmp_path): + bad = tmp_path / "bad.epub" + bad.write_bytes(b"not a zip") + proc = _run_cli("--cli", str(bad), "--output", str(tmp_path)) + assert proc.returncode == 1 + assert "error:" in proc.stdout \ No newline at end of file diff --git a/tests/test_convert_api.py b/tests/test_convert_api.py new file mode 100644 index 0000000..328a714 --- /dev/null +++ b/tests/test_convert_api.py @@ -0,0 +1,56 @@ +"""End-to-end conversion through the HTTP API (Flask test client). + +Covers: upload -> 202 job_id -> poll -> done -> download valid PDF. +""" +import io + +import app +from tests.helpers import download_bytes, wait_for_job + + +def test_convert_happy_path_end_to_end(client, submit_epub): + resp = submit_epub(filename="book.epub") + assert resp.status_code == 202, resp.get_json() + job_id = resp.get_json()["job_id"] + assert isinstance(job_id, str) and job_id + + job = wait_for_job(client, job_id) + assert job["status"] == "done", job + assert job["progress"] == 100 + assert job["filename"].endswith(".pdf") + assert job["epub"] == "book.epub" + assert job["error"] is None + + data = download_bytes(client, job["filename"]) + assert data[:4] == b"%PDF", "downloaded body is not a PDF" + + +def test_convert_records_intermediate_working_state(client, submit_epub): + """A fresh job must be visible as queued/working before it finishes.""" + job_id = submit_epub().get_json()["job_id"] + snap = client.get(f"/api/jobs/{job_id}").get_json() + # The worker thread starts asynchronously; allow queued or working. + assert snap["status"] in ("queued", "working") + assert snap["progress"] >= 0.0 + + +def test_convert_uppercase_extension_is_accepted(client, submit_epub): + """Extension check is case-insensitive (.EPUB must not be rejected).""" + resp = submit_epub(filename="BOOK.EPUB") + assert resp.status_code == 202 + job = wait_for_job(client, resp.get_json()["job_id"]) + assert job["status"] == "done", job + + +def test_convert_output_file_written_to_output_dir(client, submit_epub): + job_id = submit_epub().get_json()["job_id"] + job = wait_for_job(client, job_id) + out = app.OUTPUT_DIR / job["filename"] + assert out.is_file() + assert out.stat().st_size > 1024 + + +def test_convert_no_file_field_returns_400(client): + resp = client.post("/api/convert", content_type="multipart/form-data") + assert resp.status_code == 400 + assert "No file" in resp.get_json()["error"] \ No newline at end of file diff --git a/tests/test_edge_cases.py b/tests/test_edge_cases.py new file mode 100644 index 0000000..b055cc7 --- /dev/null +++ b/tests/test_edge_cases.py @@ -0,0 +1,101 @@ +"""Edge cases: bad uploads, size limits, unknown jobs, path traversal.""" +import io +import zipfile + +import app +from tests.helpers import make_epub_zip, make_zip_without_container, wait_for_job + + +def test_wrong_extension_returns_400(client, fixture_epub_bytes): + files = {"file": (io.BytesIO(fixture_epub_bytes), "book.pdf")} + resp = client.post("/api/convert", data=files, content_type="multipart/form-data") + assert resp.status_code == 400 + assert ".epub" in resp.get_json()["error"] + + +def test_corrupt_zip_returns_202_then_job_error(client): + """Non-zip upload: accepted (202) but the job must end in error state.""" + files = {"file": (io.BytesIO(b"this is not a zip at all"), "book.epub")} + resp = client.post("/api/convert", data=files, content_type="multipart/form-data") + assert resp.status_code == 202 + job_id = resp.get_json()["job_id"] + job = wait_for_job(client, job_id) + assert job["status"] == "error" + assert "Not a valid EPUB" in job["error"] + # The failed upload must not linger on disk. + assert not (app.UPLOAD_DIR / f"{job_id}.epub").exists() + + +def test_zip_without_container_xml_ends_in_job_error(client): + """Valid zip, not an EPUB: worker hits the missing-container.xml error.""" + files = { + "file": (io.BytesIO(make_zip_without_container()), "book.epub"), + } + resp = client.post("/api/convert", data=files, content_type="multipart/form-data") + assert resp.status_code == 202 + job = wait_for_job(client, resp.get_json()["job_id"]) + assert job["status"] == "error", job + assert "container.xml" in job["error"] + + +def test_epub_without_readable_content_ends_in_job_error(client): + """EPUB whose spine has no html items -> 'No readable content' error.""" + content = make_epub_zip( + items=[("img1", "img.png", "image/png", "")], + spine_ids=["img1"], + ) + files = {"file": (io.BytesIO(content), "empty.epub")} + resp = client.post("/api/convert", data=files, content_type="multipart/form-data") + assert resp.status_code == 202 + job = wait_for_job(client, resp.get_json()["job_id"]) + assert job["status"] == "error", job + assert "No readable content" in job["error"] + + +def test_oversized_upload_returns_413_json(client, submit_epub, monkeypatch): + """Flask MAX_CONTENT_LENGTH path: 413 with a JSON error body.""" + monkeypatch.setitem(app.app.config, "MAX_CONTENT_LENGTH", 1024) # 1 KiB + resp = submit_epub(filename="book.epub") # fixture is ~3.3 KB + assert resp.status_code == 413 + body = resp.get_json() + assert body is not None, "413 body must be JSON, not the HTML error page" + assert "too large" in body["error"].lower() + + +def test_app_side_size_check_returns_400(client, submit_epub, monkeypatch): + """app.py's own MAX_UPLOAD_MB check (distinct from Flask's 413).""" + monkeypatch.setattr(app, "MAX_UPLOAD_MB", 0) + monkeypatch.setitem(app.app.config, "MAX_CONTENT_LENGTH", 10**10) + resp = submit_epub(filename="book.epub") + assert resp.status_code == 400 + assert "too large" in resp.get_json()["error"].lower() + + +def test_unknown_job_returns_404(client): + resp = client.get("/api/jobs/doesnotexist12") + assert resp.status_code == 404 + assert resp.get_json()["error"] == "unknown job" + + +def test_download_unknown_file_returns_404(client): + assert client.get("/download/never-converted.pdf").status_code == 404 + + +def test_download_traversal_dotdot_returns_404(client): + """.. segments must be neutralized by Path(name).name, never resolve + outside OUTPUT_DIR.""" + for name in ("../app.py", "../../etc/passwd", "..%2F..%2Fetc%2Fpasswd"): + resp = client.get(f"/download/{name}") + assert resp.status_code == 404, f"traversal not blocked for {name!r}" + + +def test_download_traversal_cannot_read_app_source(client, monkeypatch, tmp_path): + """Prove the guard: even a file sitting next to OUTPUT_DIR is unreachable.""" + secret = tmp_path / "secret.txt" + secret.write_text("top secret", encoding="utf-8") + # OUTPUT_DIR is DATA_DIR/output; the secret is one level up. + # Path('..').name == '' -> OUTPUT_DIR / '' -> OUTPUT_DIR itself (a dir) + # -> is_file() False -> 404 regardless. + resp = client.get("/download/..%2Fsecret.txt") + assert resp.status_code == 404 + assert b"top secret" not in resp.data \ No newline at end of file diff --git a/tests/test_fidelity.py b/tests/test_fidelity.py new file mode 100644 index 0000000..5c2b0bd --- /dev/null +++ b/tests/test_fidelity.py @@ -0,0 +1,65 @@ +"""Text-fidelity regression tests: em dashes and CJK must survive +EPUB -> WeasyPrint -> pypdf merge without mojibake. + +Markers below were verified against test_book.epub ground truth: +the source contains 3 em dashes and the CJK runs listed in CJK_MARKERS. +These tests lock in the UTF-8 HTML-string decoding fix (app.py L258-265): +passing decoded text (not file paths) to WeasyPrint prevents Latin-1 +sniffing from turning em dashes/CJK into mojibake. +""" +import io + +from pypdf import PdfReader + +from tests.helpers import download_bytes, pdf_metadata_title, wait_for_job + +EM_DASH = "\u2014" +CJK_MARKERS = ["中文测试段落", "天地玄黄", "宇宙洪荒", "日月盈昃", "辰宿列张"] +EXPECTED_TITLE = "Structured Test Book" +EXPECTED_AUTHOR = "Agent Zero" + + +def _converted_pdf(client, submit_epub): + job_id = submit_epub(filename="book.epub").get_json()["job_id"] + job = wait_for_job(client, job_id) + assert job["status"] == "done", job + return download_bytes(client, job["filename"]), job + + +def test_em_dashes_survive_conversion(client, submit_epub): + data, _ = _converted_pdf(client, submit_epub) + reader = PdfReader(io.BytesIO(data)) + full = "\n".join((p.extract_text() or "") for p in reader.pages) + assert EM_DASH in full, "em dash missing from PDF text" + assert full.count(EM_DASH) >= 3, f"expected >=3 em dashes, got {full.count(EM_DASH)}" + + +def test_cjk_text_survives_conversion(client, submit_epub): + data, _ = _converted_pdf(client, submit_epub) + reader = PdfReader(io.BytesIO(data)) + full = "\n".join((p.extract_text() or "") for p in reader.pages) + for marker in CJK_MARKERS: + assert marker in full, f"CJK marker {marker!r} missing from PDF text" + + +def test_no_latin1_mojibake(client, submit_epub): + """Mojibake regression: em dash mis-decoded as Latin-1 reads 'â€"'.""" + data, _ = _converted_pdf(client, submit_epub) + reader = PdfReader(io.BytesIO(data)) + full = "\n".join((p.extract_text() or "") for p in reader.pages) + assert "â€" not in full, "Latin-1 mojibake detected for em dash" + assert "ï¼" not in full, "CJK fullwidth colon mojibake detected" + + +def test_pdf_metadata_round_trip(client, submit_epub): + data, _ = _converted_pdf(client, submit_epub) + assert pdf_metadata_title(data) == EXPECTED_TITLE + reader = PdfReader(io.BytesIO(data)) + assert reader.metadata["/Author"] == EXPECTED_AUTHOR + + +def test_page_count_matches_fixture(client, submit_epub): + """The 6-page fixture must not silently drop or merge chapters.""" + data, _ = _converted_pdf(client, submit_epub) + reader = PdfReader(io.BytesIO(data)) + assert len(reader.pages) == 6 \ No newline at end of file diff --git a/tests/test_health_index.py b/tests/test_health_index.py new file mode 100644 index 0000000..077ad6e --- /dev/null +++ b/tests/test_health_index.py @@ -0,0 +1,18 @@ +"""Liveness + UI entrypoint smoke tests (no conversion involved).""" +import app + + +def test_health_returns_ok_json(client): + resp = client.get("/health") + assert resp.status_code == 200 + body = resp.get_json() + assert body["status"] == "ok" + assert body["max_upload_mb"] == 400 + + +def test_index_serves_ui_html(client): + resp = client.get("/") + assert resp.status_code == 200 + text = resp.get_data(as_text=True) + assert " 0 + + +def test_job_ids_are_unique(): + a = app._register_job("a.epub") + b = app._register_job("b.epub") + assert a != b + + +def test_registry_evicts_oldest_beyond_50(): + now = time.time() + old_ids = [] + for i in range(50): + jid = app._register_job(f"old{i}.epub") + old_ids.append(jid) + # Force ascending 'created' so eviction order is deterministic. + app.JOBS[jid]["created"] = now - (1000 - i) + oldest = old_ids[0] + newest_old = old_ids[-1] + + new_id = app._register_job("new.epub") + + assert len(app.JOBS) == 50 + assert oldest not in app.JOBS, "oldest job must be evicted" + assert newest_old in app.JOBS + assert new_id in app.JOBS \ No newline at end of file diff --git a/tests/test_parse_opf.py b/tests/test_parse_opf.py new file mode 100644 index 0000000..ad340c5 --- /dev/null +++ b/tests/test_parse_opf.py @@ -0,0 +1,75 @@ +"""Unit tests for EPUB structure parsing (no rendering involved).""" +import pytest + +import app +from tests.helpers import build_epub_tree + + +def test_missing_container_xml_raises(tmp_path): + (tmp_path / "META-INF").mkdir() + with pytest.raises(ValueError, match="missing META-INF/container.xml"): + app.parse_opf(tmp_path) + + +def test_container_xml_without_rootfile_raises(tmp_path): + (tmp_path / "META-INF").mkdir() + (tmp_path / "META-INF" / "container.xml").write_text( + '', + encoding="utf-8", + ) + with pytest.raises(ValueError, match="no rootfile"): + app.parse_opf(tmp_path) + + +def test_parse_opf_returns_meta_spine_and_pdir_ltr(tmp_path): + build_epub_tree(tmp_path) + meta, opf_dir, spine, pdir = app.parse_opf(tmp_path) + assert meta["title"] == "Tree Book" + assert meta["creator"] == "Tree Author" + assert meta["language"] == "en" + assert str(opf_dir).endswith("OEBPS") + assert [i["href"] for i in spine] == ["chapter1.xhtml", "chapter2.xhtml"] # img filtered out + assert pdir == "ltr" # unset defaults to ltr + + +def test_parse_opf_rtl_page_progression(tmp_path): + build_epub_tree(tmp_path, pdir="rtl") + _, _, _, pdir = app.parse_opf(tmp_path) + assert pdir == "rtl" + + +def test_nonlinear_spine_items_are_skipped(tmp_path): + items = [ + ("front", "front.xhtml", "application/xhtml+xml", ""), + ("ch1", "chapter1.xhtml", "application/xhtml+xml", ""), + ] + build_epub_tree(tmp_path, items=items, spine_ids=["ch1"]) + # Add a non-linear itemref for 'front' via raw OPF edit + opf_path = tmp_path / "OEBPS" / "book.opf" + text = opf_path.read_text(encoding="utf-8") + text = text.replace( + '', + '' + '', + ) + opf_path.write_text(text, encoding="utf-8") + _, _, spine, _ = app.parse_opf(tmp_path) + assert [i["href"] for i in spine] == ["chapter1.xhtml"] + + +def test_empty_spine_falls_back_to_non_nav_html(tmp_path): + items = [ + ("nav", "nav.xhtml", "application/xhtml+xml", "nav"), + ("ch1", "chapter1.xhtml", "application/xhtml+xml", ""), + ("img1", "img.png", "image/png", ""), + ] + build_epub_tree(tmp_path, items=items, spine_ids=[]) # spine refs: none + _, _, spine, _ = app.parse_opf(tmp_path) + assert [i["href"] for i in spine] == ["chapter1.xhtml"] # nav + image excluded + + +def test_no_readable_content_raises(tmp_path): + items = [("img1", "img.png", "image/png", "")] + build_epub_tree(tmp_path, items=items, spine_ids=["img1"]) + with pytest.raises(ValueError, match="No readable content"): + app.parse_opf(tmp_path) \ No newline at end of file diff --git a/tests/test_review_fixes.py b/tests/test_review_fixes.py new file mode 100644 index 0000000..89b3185 --- /dev/null +++ b/tests/test_review_fixes.py @@ -0,0 +1,460 @@ +"""Focused regression tests for the 2026-08-29 multi-lens review fixes. + +Source of truth: .a0proj/_bmad-output/test-artifacts/review-findings-20260829.json +Each test maps to a finding (lens + location); guard behavior is asserted +through the public HTTP API wherever possible. +""" +import io +import os +import re +import subprocess +import sys +import threading +import time +import zipfile +from pathlib import Path + +import pytest + +import app +from tests.helpers import ( + _CONTAINER_XML, + download_bytes, + make_epub_zip, + pdf_text, + wait_for_job, +) + +ROOT = Path(__file__).resolve().parents[1] + + +def _zip_with(items: dict) -> bytes: + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf: + for name, data in items.items(): + if isinstance(data, str): + data = data.encode("utf-8") + zf.writestr(name, data) + return buf.getvalue() + + +def _append_member(epub_bytes, name, data=b"x", external_attr=None): + src = io.BytesIO(epub_bytes) + out = io.BytesIO() + with zipfile.ZipFile(src, "r") as zin, zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as zout: + for item in zin.namelist(): + zout.writestr(item, zin.read(item)) + zi = zipfile.ZipInfo(name) + if external_attr is not None: + zi.external_attr = external_attr + zout.writestr(zi, data) + return out.getvalue() + + +def _upload(client, content, filename="book.epub"): + """POST an upload, wait for the terminal job state, return the snapshot.""" + files = {"file": (io.BytesIO(content), filename)} + resp = client.post("/api/convert", data=files, content_type="multipart/form-data") + assert resp.status_code == 202, resp.get_json() + return wait_for_job(client, resp.get_json()["job_id"]) + + +# --------------------------------------------------------------------------- +# Fix 1: ZIP-SLIP + ZIP BOMB (adversarial, app.py extractall) +# --------------------------------------------------------------------------- + +def test_zip_slip_dotdot_member_rejected(client, fixture_epub_bytes): + job = _upload(client, _append_member(fixture_epub_bytes, "../pwned.txt")) + assert job["status"] == "error", job + assert "unsafe zip member" in job["error"] + + +def test_zip_symlink_member_rejected(client, fixture_epub_bytes): + content = _append_member( + fixture_epub_bytes, "OEBPS/link", external_attr=(0xA << 28) | 0o777 + ) + job = _upload(client, content) + assert job["status"] == "error", job + assert "unsafe zip member" in job["error"] + + +def test_zip_bomb_guard_rejects_huge_expansion(client, fixture_epub_bytes, monkeypatch): + # Production floor is 2 GiB (impractical in a unit test); shrink the floor + # so the guard logic is exercised with a 4 MiB member in a ~8 KiB zip. + monkeypatch.setattr(app, "ZIP_BOMB_MIN_CAP", 1024 * 1024) + content = _append_member(fixture_epub_bytes, "OEBPS/bomb.bin", b"\x00" * (4 * 1024 * 1024)) + job = _upload(client, content) + assert job["status"] == "error", job + assert "zip-bomb" in job["error"].lower() + + +# --------------------------------------------------------------------------- +# Fix 2: UNBOUNDED THREADS -> bounded pool + 503 backpressure +# (adversarial, app.py:293-307 / 355) +# --------------------------------------------------------------------------- + +def test_saturated_pool_returns_503_json(client, fixture_epub_bytes, monkeypatch): + monkeypatch.setattr(app, "WORKER_SLOTS", threading.BoundedSemaphore(0)) + files = {"file": (io.BytesIO(fixture_epub_bytes), "book.epub")} + resp = client.post("/api/convert", data=files, content_type="multipart/form-data") + assert resp.status_code == 503 + body = resp.get_json() + assert body is not None, "503 body must be JSON so client pollers never hang" + assert "busy" in body["error"].lower() + + +def test_worker_slot_released_after_job_finishes(client, submit_epub): + before = app.WORKER_SLOTS._value + job = wait_for_job(client, submit_epub().get_json()["job_id"]) + assert job["status"] == "done" + assert app.WORKER_SLOTS._value == before, "slot leaked: semaphore not released" + + +# --------------------------------------------------------------------------- +# Fix 3: OUTPUT NAME COLLISION + TORN FILE (adversarial, app.py:274-287) +# --------------------------------------------------------------------------- + +def _clean_output(pattern: str) -> None: + """Remove output-dir files matching pattern so naming tests are deterministic.""" + for f in app.OUTPUT_DIR.glob(pattern): + f.unlink() + + +def _synthetic_epub_bytes(title=None) -> bytes: + """Minimal renderable EPUB; title=None omits , '' makes it empty.""" + title_tag = "" if title is None else f"{title}" + opf = ( + '' + '' + '' + f"{title_tag}Test" + "" + '' + '' + "" + ) + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + zf.writestr("META-INF/container.xml", _CONTAINER_XML) + zf.writestr("OEBPS/book.opf", opf) + zf.writestr( + "OEBPS/chapter1.xhtml", + '' + "

naming body

", + ) + return buf.getvalue() + + +def test_output_filename_unique_per_job(client, submit_epub): + # New contract (export naming): clean title name first, -2 suffix on collision + _clean_output("Structured Test Book*.pdf") + job1 = wait_for_job(client, submit_epub(filename="book.epub").get_json()["job_id"]) + job2 = wait_for_job(client, submit_epub(filename="book.epub").get_json()["job_id"]) + assert job1["status"] == "done" and job2["status"] == "done" + assert job1["filename"] != job2["filename"], "same book must not overwrite its PDF" + assert job2["filename"].endswith("-2.pdf"), "second conversion must append -2" + assert (app.OUTPUT_DIR / job1["filename"]).is_file() + assert (app.OUTPUT_DIR / job2["filename"]).is_file() + + +def test_atomic_write_leaves_no_partial_files(client, submit_epub): + job = wait_for_job(client, submit_epub().get_json()["job_id"]) + assert job["status"] == "done" + assert not list(app.OUTPUT_DIR.glob("*.part")), "torn-write .part temp file left behind" + + +def test_export_name_is_book_title(client, submit_epub): + # req 1: export name is the clean book title, not the upload name + _clean_output("Structured Test Book*.pdf") + job = wait_for_job( + client, submit_epub(filename="weird-upload-name.epub").get_json()["job_id"] + ) + assert job["status"] == "done", job + assert job["filename"] == "Structured Test Book.pdf" + + +def test_export_name_falls_back_to_epub_stem(client, submit_epub): + # req 1 fallback: missing -> uploaded .epub stem + _clean_output("pg2680-images-3*.pdf") + job = wait_for_job( + client, + submit_epub( + filename="pg2680-images-3.epub", + content=_synthetic_epub_bytes(title=None), + ).get_json()["job_id"], + ) + assert job["status"] == "done", job + assert job["filename"] == "pg2680-images-3.pdf", job["filename"] + # req 1 fallback: empty -> same stem fallback + _clean_output("pg2680-images-4*.pdf") + job2 = wait_for_job( + client, + submit_epub( + filename="pg2680-images-4.epub", + content=_synthetic_epub_bytes(title=""), + ).get_json()["job_id"], + ) + assert job2["status"] == "done", job2 + assert job2["filename"] == "pg2680-images-4.pdf", job2["filename"] + + +def test_export_name_pdf_title_not_doubled(client, submit_epub): + # req 2: title ending in .pdf (case-insensitive) must not double the extension + _clean_output("My Book*.pdf") + job = wait_for_job( + client, + submit_epub( + filename="seed.epub", + content=_synthetic_epub_bytes(title="My Book.PDF"), + ).get_json()["job_id"], + ) + assert job["status"] == "done", job + assert job["filename"] == "My Book.pdf", job["filename"] + assert not job["filename"].endswith(".pdf.pdf") + + +def test_download_headers_pdf_type_and_disposition(client, submit_epub): + # req 4: Content-Type application/pdf + attachment filename ending in .pdf + _clean_output("Structured Test Book*.pdf") + job = wait_for_job(client, submit_epub(filename="book.epub").get_json()["job_id"]) + assert job["status"] == "done", job + resp = client.get(f"/download/{job['filename']}") + assert resp.status_code == 200 + assert resp.headers["Content-Type"].startswith("application/pdf") + cd = resp.headers.get("Content-Disposition", "") + assert "attachment" in cd + assert cd.split("filename")[-1].strip().strip('"').endswith(".pdf") + + +def test_download_defensively_appends_pdf_suffix(client): + # req 4: non-.pdf file served by /download gets a .pdf download_name + target = app.OUTPUT_DIR / "note.txt" + target.write_bytes(b"%PDF-1.4 fake") + try: + resp = client.get("/download/note.txt") + assert resp.status_code == 200 + assert resp.headers["Content-Type"].startswith("application/pdf") + assert resp.headers["Content-Disposition"].split("filename")[-1].strip().strip('"').endswith(".pdf") + finally: + target.unlink(missing_ok=True) + + +# --------------------------------------------------------------------------- +# Fix 4: EVICTION KILLS IN-FLIGHT JOBS (adversarial app.py:69-73 + +# verification-gap index.html:253-279) +# --------------------------------------------------------------------------- + +def test_eviction_never_touches_queued_or_working_jobs(): + now = time.time() + ids = [] + for i in range(52): + jid = app._register_job(f"j{i}.epub") + ids.append(jid) + app.JOBS[jid]["created"] = now - (2000 - i) + app.JOBS[jid]["status"] = "done" + # At the 52nd registration 51 jobs were terminal -> oldest (j0) evicted. + assert ids[0] not in app.JOBS + # Make the oldest *remaining* job in-flight, then force another eviction + # pass with two more terminal jobs. + app.JOBS[ids[1]]["status"] = "working" + for i in range(2): + jid = app._register_job(f"extra{i}.epub") + app.JOBS[jid]["created"] = now + 100 + i + app.JOBS[jid]["status"] = "done" + assert app.JOBS[ids[1]]["status"] == "working", "in-flight job must survive eviction" + assert ids[1] in app.JOBS + + +def test_evicted_job_yields_404_unknown_job_body(client): + """The exact body the rewritten UI poll loop receives on a 404.""" + now = time.time() + first = None + for i in range(52): + jid = app._register_job(f"t{i}.epub") + if i == 0: + first = jid + app.JOBS[jid]["created"] = now - (1000 - i) + app.JOBS[jid]["status"] = "done" + resp = client.get(f"/api/jobs/{first}") + assert resp.status_code == 404 + assert resp.get_json() == {"error": "unknown job"} + + +# --------------------------------------------------------------------------- +# Fix 5: 0-PAGE + LEAKS + FAKE HEALTH (adversarial app.py:278-288 / 347-352) +# --------------------------------------------------------------------------- + +def test_zero_page_book_errors_instead_of_blank_pdf(client): + # Spine item is html but the file is absent -> every chunk is skipped -> + # no pages. Must be a job error, not a blank-PDF success. + content = make_epub_zip( + items=[("ch1", "chapter1.xhtml", "application/xhtml+xml", "")], + spine_ids=["ch1"], + ) + job = _upload(client, content, "ghost.epub") + assert job["status"] == "error", job + assert "no renderable content" in job["error"] + + +def test_upload_unlinked_after_successful_job(client, submit_epub): + jid = submit_epub(filename="book.epub").get_json()["job_id"] + job = wait_for_job(client, jid) + assert job["status"] == "done" + assert not (app.UPLOAD_DIR / f"{jid}.epub").exists(), "upload leaked after success" + + +def test_health_503_when_cjk_fonts_missing(client, monkeypatch): + monkeypatch.setattr(app, "_cjk_fonts_present", lambda: False) + resp = client.get("/health") + assert resp.status_code == 503 + body = resp.get_json() + assert body["status"] == "degraded" + assert any("CJK" in p for p in body["problems"]) + + +def test_health_503_when_data_dir_not_writable(client, monkeypatch, tmp_path): + monkeypatch.setattr(app, "DATA_DIR", tmp_path / "gone") # nonexistent dir + resp = client.get("/health") + assert resp.status_code == 503 + assert any("writable" in p for p in resp.get_json()["problems"]) + + +def test_health_exposes_max_upload_mb_single_source(client, monkeypatch): + monkeypatch.setattr(app, "MAX_UPLOAD_MB", 77) + body = client.get("/health").get_json() + assert body["max_upload_mb"] == 77 + + +# --------------------------------------------------------------------------- +# Fix 6: CHARSET (adversarial app.py:262) — legacy windows-1252/latin-1 books +# --------------------------------------------------------------------------- + +def test_charset_windows_1252_xml_declaration_honored(client): + chap = ( + '\n' + "

caf\xe9 \u2014 em dash test

" + ) + base = make_epub_zip( + items=[("ch1", "ch1.xhtml", "application/xhtml+xml", "")], + spine_ids=["ch1"], + ) + content = _append_member(base, "OEBPS/ch1.xhtml", chap.encode("windows-1252")) + job = _upload(client, content, "legacy.epub") + assert job["status"] == "done", job + full = pdf_text(download_bytes(client, job["filename"])) + assert "\u2014" in full, "em dash lost: declared windows-1252 encoding not honored" + assert "café" in full, "latin-1 mojibake: windows-1252 not honored" + + +def test_decode_html_meta_charset_latin1(): + raw = b'caf\xe9' + assert app._decode_html(raw) == "café" + + +def test_decode_html_falls_back_to_utf8_replace(): + raw = b"" + b"\xff\xfe\xfa" + b"" + assert app._decode_html(raw), "undecodable chapter must not raise" + + +# --------------------------------------------------------------------------- +# Fix 7: LAUNCHER DIVERGENCE + LIMIT DRY (verification-gap start.sh:25-38 / +# Dockerfile:37 + adversarial index.html:139) +# --------------------------------------------------------------------------- + +def test_launcher_flags_aligned_single_source(): + start = (ROOT / "start.sh").read_text(encoding="utf-8") + docker = (ROOT / "Dockerfile").read_text(encoding="utf-8") + for text, label in ((start, "start.sh"), (docker, "Dockerfile")): + assert "--workers 1" in text, f"{label}: single worker required (in-memory JOBS)" + assert "--timeout 1200" in text, f"{label}: 1200s timeout for long conversions" + assert "--threads 8" in text or "THREADS:-8" in text, f"{label}: 8 threads" + + +def test_ui_consumes_max_upload_mb_from_health(client): + html = client.get("/").get_data(as_text=True) + assert "max_upload_mb" in html, "UI must read the upload limit from /health" + assert 'id="maxmb"' in html + + +def test_ui_poll_loop_treats_404_as_terminal(client): + html = client.get("/").get_data(as_text=True) + assert "status === 404" in html, "poll loop must treat a 404 /api/jobs as terminal" + + +# --------------------------------------------------------------------------- +# Cheap edge-case findings folded into the same code paths +# --------------------------------------------------------------------------- + +def test_non_numeric_port_env_exits_2(): + env = dict(os.environ) + env["PORT"] = "not-a-port" + proc = subprocess.run( + [sys.executable, str(ROOT / "app.py")], + capture_output=True, text=True, timeout=60, env=env, cwd=str(ROOT), + ) + assert proc.returncode == 2 + assert "PORT must be an integer" in proc.stdout + + +def test_contained_rejects_escaping_path(tmp_path): + with pytest.raises(ValueError, match="escapes"): + app._contained(tmp_path, "../escape.txt") + + +def test_contained_accepts_inner_path(tmp_path): + assert app._contained(tmp_path, "a/b.txt") == (tmp_path / "a" / "b.txt").resolve() + + +_OPF_TMPL = ( + '' + '' + '' + "T" + '' + '' +) + + +def test_opf_path_escaping_tree_rejected(client): + container = _CONTAINER_XML.replace('full-path="OEBPS/book.opf"', 'full-path="../book.opf"') + content = _zip_with({ + "META-INF/container.xml": container, + "OEBPS/book.opf": _OPF_TMPL.format(href="ch1.xhtml"), + "OEBPS/ch1.xhtml": "hi", + }) + job = _upload(client, content, "escape.epub") + assert job["status"] == "error", job + assert "escapes" in job["error"] + + +def test_spine_href_escaping_tree_skipped(client): + content = _zip_with({ + "META-INF/container.xml": _CONTAINER_XML, + "OEBPS/book.opf": _OPF_TMPL.format(href="../../etc/passwd"), + "OEBPS/ch1.xhtml": "hi", + }) + job = _upload(client, content, "href.epub") + assert job["status"] == "error", job + assert "no renderable content" in job["error"] + + +def test_startup_sweep_removes_only_stale_work_dirs(tmp_path, monkeypatch): + monkeypatch.setattr(app, "WORK_DIR", tmp_path) + stale = tmp_path / "job-old" + stale.mkdir() + fresh = tmp_path / "job-new" + fresh.mkdir() + old = time.time() - 7 * 3600 + os.utime(stale, (old, old)) + app._sweep_stale_work_dirs() + assert not stale.exists(), "stale work dir must be swept" + assert fresh.exists(), "fresh work dir must be kept" + + +def test_progress_message_contract_matches_ui_regex(): + """Verification-gap finding: the message contract the UI poll loop parses.""" + m = re.search(r"\((\d+)/(\d+)\)", app._format_progress(3, 9, "ch1.xhtml")) + assert m is not None + assert (m.group(1), m.group(2)) == ("3", "9")