epub2pdf: title-based export names, .pdf download headers, security hardening (68 tests green)
This commit is contained in:
BMad Master
@@ -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
|
||||
@@ -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/<job_id> 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/<filename> 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 = (
|
||||
'<?xml version="1.0" encoding="utf-8"?>'
|
||||
'<container version="1.0" xmlns="urn:oasis:names:tc:opendocument:xmlns:container">'
|
||||
'<rootfiles><rootfile full-path="OEBPS/book.opf" '
|
||||
'media-type="application/oebps-package+xml"/></rootfiles></container>'
|
||||
)
|
||||
|
||||
|
||||
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'<item id="{i}" href="{h}" media-type="{m}"'
|
||||
+ (f' properties="{p}"' if p else "")
|
||||
+ "/>"
|
||||
for i, h, m, p in items
|
||||
)
|
||||
refs = "".join(f'<itemref idref="{sid}"/>' for sid in spine_ids)
|
||||
opf = (
|
||||
'<?xml version="1.0" encoding="utf-8"?>'
|
||||
'<package xmlns="http://www.idpf.org/2007/opf" version="3.0" '
|
||||
'unique-identifier="uid">'
|
||||
'<metadata xmlns:dc="http://purl.org/dc/elements/1.1/">'
|
||||
f"<dc:title>{title}</dc:title><dc:creator>{creator}</dc:creator>"
|
||||
"</metadata>"
|
||||
f"<manifest>{manifest}</manifest>"
|
||||
f"<spine>{refs}</spine>"
|
||||
"</package>"
|
||||
)
|
||||
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'<item id="{i}" href="{h}" media-type="{m}"'
|
||||
+ (f' properties="{p}"' if p else "")
|
||||
+ "/>"
|
||||
for i, h, m, p in items
|
||||
)
|
||||
refs = "".join(f'<itemref idref="{sid}"/>' for sid in spine_ids)
|
||||
opf = (
|
||||
'<?xml version="1.0" encoding="utf-8"?>'
|
||||
f'<package xmlns="http://www.idpf.org/2007/opf" version="3.0" '
|
||||
f'unique-identifier="uid"{pdir_attr}>'
|
||||
'<metadata xmlns:dc="http://purl.org/dc/elements/1.1/">'
|
||||
"<dc:title>Tree Book</dc:title>"
|
||||
"<dc:creator>Tree Author</dc:creator>"
|
||||
"<dc:language>en</dc:language>"
|
||||
"</metadata>"
|
||||
f"<manifest>{manifest}</manifest>"
|
||||
f"<spine>{refs}</spine>"
|
||||
"</package>"
|
||||
)
|
||||
(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(
|
||||
'<?xml version="1.0" encoding="utf-8"?>'
|
||||
"<html><body><p>chapter body</p></body></html>",
|
||||
encoding="utf-8",
|
||||
)
|
||||
return base
|
||||
@@ -0,0 +1,50 @@
|
||||
"""CLI entrypoint tests: `python app.py --cli <file> --output <dir>`.
|
||||
|
||||
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
|
||||
@@ -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"]
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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 "<html" in text.lower()
|
||||
assert "/api/convert" in text # UI knows its upload endpoint
|
||||
@@ -0,0 +1,47 @@
|
||||
"""Job registry behavior: registration shape and the 50-entry eviction cap.
|
||||
|
||||
Unit-level tests against app._register_job — the registry is module-global
|
||||
state that every API test isolates via the autouse fixture.
|
||||
"""
|
||||
import time
|
||||
|
||||
import app
|
||||
|
||||
|
||||
def test_register_job_shape():
|
||||
job_id = app._register_job("x.epub")
|
||||
assert job_id in app.JOBS
|
||||
job = app.JOBS[job_id]
|
||||
assert job["id"] == job_id
|
||||
assert job["epub"] == "x.epub"
|
||||
assert job["status"] == "queued"
|
||||
assert job["progress"] == 0.0
|
||||
assert job["message"] == "queued"
|
||||
assert job["filename"] is None
|
||||
assert job["error"] is None
|
||||
assert job["created"] > 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
|
||||
@@ -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(
|
||||
'<?xml version="1.0"?><container version="1.0"><rootfiles/></container>',
|
||||
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(
|
||||
'<spine><itemref idref="ch1"/></spine>',
|
||||
'<spine><itemref idref="front" linear="no"/>'
|
||||
'<itemref idref="ch1"/></spine>',
|
||||
)
|
||||
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)
|
||||
@@ -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 <dc:title>, '' makes it empty."""
|
||||
title_tag = "" if title is None else f"<dc:title>{title}</dc:title>"
|
||||
opf = (
|
||||
'<?xml version="1.0" encoding="utf-8"?>'
|
||||
'<package xmlns="http://www.idpf.org/2007/opf" version="3.0" '
|
||||
'unique-identifier="uid">'
|
||||
'<metadata xmlns:dc="http://purl.org/dc/elements/1.1">'
|
||||
f"{title_tag}<dc:creator>Test</dc:creator>"
|
||||
"</metadata>"
|
||||
'<manifest><item id="ch1" href="chapter1.xhtml" '
|
||||
'media-type="application/xhtml+xml"/></manifest>'
|
||||
'<spine><itemref idref="ch1"/></spine>'
|
||||
"</package>"
|
||||
)
|
||||
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",
|
||||
'<?xml version="1.0" encoding="utf-8"?>'
|
||||
"<html><body><p>naming body</p></body></html>",
|
||||
)
|
||||
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 <dc:title> -> 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 <dc:title></dc:title> -> 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 = (
|
||||
'<?xml version="1.0" encoding="windows-1252"?>\n'
|
||||
"<html><body><p>caf\xe9 \u2014 em dash test</p></body></html>"
|
||||
)
|
||||
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'<html><head><meta charset="latin-1"></head><body>caf\xe9</body></html>'
|
||||
assert app._decode_html(raw) == "café"
|
||||
|
||||
|
||||
def test_decode_html_falls_back_to_utf8_replace():
|
||||
raw = b"<html><body>" + b"\xff\xfe\xfa" + b"</body></html>"
|
||||
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 = (
|
||||
'<?xml version="1.0" encoding="utf-8"?>'
|
||||
'<package xmlns="http://www.idpf.org/2007/opf" version="3.0" unique-identifier="uid">'
|
||||
'<metadata xmlns:dc="http://purl.org/dc/elements/1.1/">'
|
||||
"<dc:title>T</dc:title></metadata>"
|
||||
'<manifest><item id="ch1" href="{href}" media-type="application/xhtml+xml"/></manifest>'
|
||||
'<spine><itemref idref="ch1"/></spine></package>'
|
||||
)
|
||||
|
||||
|
||||
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": "<html><body>hi</body></html>",
|
||||
})
|
||||
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": "<html><body>hi</body></html>",
|
||||
})
|
||||
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")
|
||||
Reference in New Issue
Block a user