sirdave73/epub2pdf
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ba7a772e5299a7dffb98d2d901544da0bbf09e00
epub2pdf/tests/test_multi_format.py
T

606 lines
22 KiB
Python
Raw Blame History

"""PHASE 1 multi-format expansion: any common file -> PDF.
Covers per-format happy paths (e2e through the HTTP API), the shared naming
contract (sanitise, 60-char cap, no doubled .pdf, first-free collision),
error paths (invalid content per format), the unsupported-extension 400,
CLI dispatch for non-EPUB files, and worker hygiene (upload + work cleanup).
Naming contract (mirrors the pinned EPUB export-name behavior exactly):
base = sanitize(stem)[:60] with sanitize = [^A-Za-z0-9._ -]+ -> '_'
trailing '.pdf' stripped (case-insensitive); empty stem -> 'book';
collision picks first free of base.pdf, base-2.pdf, base-3.pdf, ...
"""
import io
import json
import os
import subprocess
import sys
from pathlib import Path
from PIL import Image
from pypdf import PdfWriter
import app
import converters
from converters import common as conv_common
from tests.helpers import (download_bytes, pdf_metadata_title, pdf_page_count,
pdf_text, wait_for_job)
ROOT = Path(__file__).resolve().parents[1]
# --------------------------------------------------------------------------
# fixtures
# --------------------------------------------------------------------------
def _submit(client, filename, content):
if isinstance(content, str):
content = content.encode("utf-8")
files = {"file": (io.BytesIO(content), filename)}
return client.post("/api/convert", data=files, content_type="multipart/form-data")
def _e2e(client, filename, content):
"""Submit, wait for terminal, assert done, return (job, pdf_bytes, pdf_text)."""
resp = _submit(client, filename, content)
assert resp.status_code == 202, resp.get_json()
job = wait_for_job(client, resp.get_json()["job_id"])
assert job["status"] == "done", job
data = download_bytes(client, job["filename"])
assert data[:4] == b"%PDF"
return job, data, pdf_text(data)
def _png_bytes(size=(640, 480), color=(200, 30, 30)):
buf = io.BytesIO()
Image.new("RGB", size, color).save(buf, format="PNG")
return buf.getvalue()
def _jpeg_bytes(size=(640, 480), color=(30, 200, 30)):
buf = io.BytesIO()
Image.new("RGB", size, color).save(buf, format="JPEG", quality=90)
return buf.getvalue()
def _webp_bytes(size=(320, 240), color=(30, 30, 200)):
buf = io.BytesIO()
Image.new("RGB", size, color).save(buf, format="WEBP", quality=90)
return buf.getvalue()
# --------------------------------------------------------------------------
# is_supported / extension matrix
# --------------------------------------------------------------------------
def test_is_supported_matrix():
for ext in (".epub", ".html", ".htm", ".txt", ".log", ".md", ".markdown",
".csv", ".tsv", ".json", ".xml", ".svg", ".png", ".jpg",
".jpeg", ".webp", ".tiff", ".bmp", ".pdf"):
assert converters.is_supported(f"book{ext}"), ext
# Phase 2 (Chunk A): .docx/.rtf are now supported (pandoc_converter) —
# removed from the negative list; remaining entries stay rejected.
for ext in (".exe", ".mp4", ".zip"):
assert not converters.is_supported(f"book{ext}"), ext
assert not converters.is_supported("noextension")
def test_is_supported_case_insensitive():
assert converters.is_supported("BOOK.HTML")
assert converters.is_supported("Photo.JPEG")
def test_convert_file_unsupported_raises_valueerror(tmp_path):
p = tmp_path / "virus.exe"
p.write_bytes(b"MZ")
try:
converters.convert_file(p, "jobx")
assert False, "expected ValueError"
except ValueError as exc:
assert "Unsupported" in str(exc)
# --------------------------------------------------------------------------
# API: unsupported extension -> 400 (legacy message contract kept)
# --------------------------------------------------------------------------
def test_unsupported_ext_returns_400(client, fixture_epub_bytes):
# gap patch: .pdf is now a supported type; use a still-unsupported ext.
resp = _submit(client, "book.mp4", fixture_epub_bytes)
assert resp.status_code == 400
assert ".epub" in resp.get_json()["error"]
def test_unsupported_exe_returns_400(client):
resp = _submit(client, "virus.exe", b"MZ\x90")
assert resp.status_code == 400
assert resp.get_json()["error"]
# --------------------------------------------------------------------------
# per-format happy paths (e2e)
# --------------------------------------------------------------------------
def test_html_e2e_title_and_content(client):
html_doc = (
"<!DOCTYPE html><html><head><title>My HTML Doc</title>"
"<style>body{color:navy}</style></head>"
"<body><h1>Heading One</h1><p>HTML MARKER ALPHA</p></body></html>"
)
job, data, text = _e2e(client, "doc.html", html_doc)
assert job["filename"] == "My HTML Doc.pdf"
assert job["epub"] == "doc.html" # original upload name preserved for UI
assert "HTML MARKER ALPHA" in text
assert "Heading One" in text
def test_html_fragment_without_title_uses_stem(client):
job, _, text = _e2e(client, "fragment.htm", "<p>FRAG MARK BRAVO</p>")
assert job["filename"] == "fragment.pdf"
assert "FRAG MARK BRAVO" in text
def test_html_declared_windows1252_meta(client):
body = "Caf\u00e9 \u2014 em dash \u2014 r\u00e9sum\u00e9"
raw = (
'<html><head><meta charset="windows-1252"></head><body><p>'
).encode("utf-8") + body.encode("cp1252") + b"</p></body></html>"
job, _, text = _e2e(client, "latin.html", raw)
assert "r\u00e9sum\u00e9" in text and "\u2014" in text
def test_txt_e2e(client):
job, _, text = _e2e(client, "notes.txt", "TXT MARKER CHARLIE\nsecond line\n\n")
assert job["filename"] == "notes.pdf"
assert "TXT MARKER CHARLIE" in text
assert "second line" in text
def test_txt_windows1252_autodetect(client):
body = "Caf\u00e9 \u2014 dash \u2014 fen\u00eatre \u2014 r\u00e9sum\u00e9 \u2014 fin"
job, _, text = _e2e(client, "win1252.txt", body.encode("cp1252"))
assert "r\u00e9sum\u00e9" in text
assert "\u2014" in text
def test_md_e2e_title_from_h1(client):
md = "# MD Title One\n\nBody with **BOLDWORD** and a list:\n\n- item A\n- item B\n"
job, _, text = _e2e(client, "chapter.md", md)
assert job["filename"] == "MD Title One.pdf"
assert "BOLDWORD" in text
assert "item A" in text
def test_md_no_title_uses_stem(client):
job, _, text = _e2e(client, "plain.md", "just text, no heading")
assert job["filename"] == "plain.pdf"
assert "just text, no heading" in text
def test_csv_e2e(client):
job, _, text = _e2e(client, "stock.csv", "name,qty\napple,3\npear,5\n")
assert job["filename"] == "stock.pdf"
assert "apple" in text
assert "qty" in text
assert "5" in text
def test_tsv_e2e(client):
job, _, text = _e2e(client, "grid.tsv", "colA\tcolB\tv1\tv2\n")
assert job["filename"] == "grid.pdf"
assert "v1" in text and "v2" in text
def test_json_e2e(client):
doc = json.dumps({"alpha": [1, 2, 3], "beta": "GAMMA_JSON"}, indent=2)
job, _, text = _e2e(client, "config.json", doc)
assert job["filename"] == "config.pdf"
assert "GAMMA_JSON" in text
assert "alpha" in text
def test_xml_e2e(client):
doc = '<?xml version="1.0"?>\n<root><item>GAMMA_XML</item><item>two</item></root>'
job, _, text = _e2e(client, "feed.xml", doc)
assert job["filename"] == "feed.pdf"
assert "GAMMA_XML" in text
def test_svg_e2e(client):
svg = (
'<svg xmlns="http://www.w3.org/2000/svg" width="400" height="300">'
"<rect width='400' height='300' fill='#eef'/>"
"<text x='20' y='150' font-size='24'>SVG MARK DELTA</text></svg>"
)
job, data, _ = _e2e(client, "diagram.svg", svg)
assert job["filename"] == "diagram.pdf"
assert pdf_page_count(data) >= 1
assert len(data) > 1024
def test_image_png_e2e(client):
job, data, _ = _e2e(client, "photo.png", _png_bytes())
assert job["filename"] == "photo.pdf"
assert pdf_page_count(data) == 1
def test_image_jpeg_e2e(client):
job, data, _ = _e2e(client, "scan.jpg", _jpeg_bytes())
assert job["filename"] == "scan.pdf"
assert pdf_page_count(data) == 1
def test_image_webp_e2e(client):
job, data, _ = _e2e(client, "snap.webp", _webp_bytes())
assert job["filename"] == "snap.pdf"
assert pdf_page_count(data) == 1
def test_uppercase_extension_e2e(client):
job, _, text = _e2e(client, "UPPER.HTML", "<p>UPPER MARK ECHO</p>")
assert job["filename"] == "UPPER.pdf"
assert "UPPER MARK ECHO" in text
# --------------------------------------------------------------------------
# error paths (accepted 202, job ends in error with a clear message)
# --------------------------------------------------------------------------
def _error_job(client, filename, content):
resp = _submit(client, filename, content)
assert resp.status_code == 202, resp.get_json()
job = wait_for_job(client, resp.get_json()["job_id"])
assert job["status"] == "error", job
return job
def test_invalid_json_is_job_error(client):
job = _error_job(client, "bad.json", "{not valid json]")
assert "JSON" in job["error"]
def test_invalid_xml_is_job_error(client):
job = _error_job(client, "bad.xml", "<a><b></a>")
assert "XML" in job["error"]
def test_invalid_svg_is_job_error(client):
job = _error_job(client, "bad.svg", "<svg><g></path>")
assert job["error"]
def test_corrupt_png_is_job_error(client):
job = _error_job(client, "fake.png", b"\x89PNG\r\n\x1a\n definitely not a png")
assert "image" in job["error"].lower()
def test_empty_txt_is_job_error(client):
job = _error_job(client, "empty.txt", b"")
assert "empty" in job["error"].lower()
def test_empty_csv_is_job_error(client):
job = _error_job(client, "empty.csv", b"")
assert "empty" in job["error"].lower()
def test_empty_json_is_job_error(client):
job = _error_job(client, "empty.json", b" ")
assert "JSON" in job["error"]
# --------------------------------------------------------------------------
# naming contract (unit level via converters.common)
# --------------------------------------------------------------------------
def test_naming_sanitize_special_chars(tmp_path):
name = conv_common.derive_export_name("My File!! Final.txt", tmp_path)
assert name == "My File_ Final.pdf"
def test_naming_60_char_cap(tmp_path):
stem = "x" * 120
name = conv_common.derive_export_name(f"{stem}.txt", tmp_path)
assert name == ("x" * 60) + ".pdf"
def test_naming_strips_doubled_pdf(tmp_path):
assert conv_common.derive_export_name("report.PDF.txt", tmp_path) == "report.pdf"
assert conv_common.derive_export_name("x.pdf.txt", tmp_path) == "x.pdf"
def test_naming_empty_stem_falls_back_to_book(tmp_path):
assert conv_common.derive_export_name(".txt", tmp_path) == "book.pdf"
def test_naming_collision_first_free(tmp_path):
(tmp_path / "dup.pdf").write_bytes(b"old1")
(tmp_path / "dup-2.pdf").write_bytes(b"old2")
assert conv_common.derive_export_name("dup.txt", tmp_path) == "dup-3.pdf"
def test_naming_collision_e2e_second_upload_gets_suffix(client):
job1, _, _ = _e2e(client, "same.txt", "one")
job2, _, _ = _e2e(client, "same.txt", "two")
assert job1["filename"] == "same.pdf"
assert job2["filename"] == "same-2.pdf"
# --------------------------------------------------------------------------
# download + worker hygiene for non-EPUB conversions
# --------------------------------------------------------------------------
def test_download_headers_for_txt_conversion(client):
job, _, _ = _e2e(client, "headers.txt", "header check")
resp = client.get(f"/download/{job['filename']}")
assert resp.status_code == 200
assert resp.content_type == "application/pdf"
assert "attachment" in resp.headers["Content-Disposition"]
assert 'filename="headers.pdf"' in resp.headers["Content-Disposition"]
def test_worker_cleans_upload_and_work_dir(client):
resp = _submit(client, "hygiene.txt", "cleanup check")
job_id = resp.get_json()["job_id"]
job = wait_for_job(client, job_id)
assert job["status"] == "done", job
assert not (app.UPLOAD_DIR / f"{job_id}.txt").exists()
assert not (app.WORK_DIR / f"job-{job_id}").exists()
def test_job_snapshot_keeps_original_name(client):
resp = _submit(client, "report Q3.txt", "x")
job = wait_for_job(client, resp.get_json()["job_id"])
assert job["status"] == "done", job
assert job["epub"] == "report Q3.txt" # UI bookname source, now generic
assert job["message"] == "Done"
# --------------------------------------------------------------------------
# CLI dispatch for non-EPUB files
# --------------------------------------------------------------------------
def _run_cli(*args):
return subprocess.run(
[sys.executable, str(ROOT / "app.py"), *args],
capture_output=True, text=True, timeout=120,
env=dict(os.environ), cwd=str(ROOT),
)
def test_cli_markdown_file(tmp_path):
src = tmp_path / "notes.md"
src.write_text("# CLI MD Title\n\nCLI body text.", encoding="utf-8")
out = tmp_path / "out"
proc = _run_cli("--cli", str(src), "--output", str(out))
assert proc.returncode == 0, proc.stderr
assert "Saved:" in proc.stdout
pdfs = list(out.glob("*.pdf"))
assert len(pdfs) == 1 and pdfs[0].name == "CLI MD Title.pdf"
assert pdfs[0].read_bytes()[:4] == b"%PDF"
def test_cli_unsupported_ext_exits_1(tmp_path):
src = tmp_path / "prog.exe"
src.write_bytes(b"MZ")
proc = _run_cli("--cli", str(src), "--output", str(tmp_path))
assert proc.returncode == 1
assert "error:" in proc.stdout
# --------------------------------------------------------------------------
# PDF pass-through + added image formats (Phase 1 gap patch)
# --------------------------------------------------------------------------
def _blank_pdf_bytes(pages=3):
w = PdfWriter()
for _ in range(pages):
w.add_blank_page(width=612, height=792)
buf = io.BytesIO()
w.write(buf)
return buf.getvalue()
def _encrypted_pdf_bytes():
w = PdfWriter()
for _ in range(2):
w.add_blank_page(width=612, height=792)
w.encrypt(user_password="secret", owner_password="owner")
buf = io.BytesIO()
w.write(buf)
return buf.getvalue()
def _tiff_bytes(size=(320, 240), color=(120, 30, 200)):
buf = io.BytesIO()
Image.new("RGB", size, color).save(buf, format="TIFF")
return buf.getvalue()
def _bmp_bytes(size=(320, 240), color=(200, 120, 30)):
buf = io.BytesIO()
Image.new("RGB", size, color).save(buf, format="BMP")
return buf.getvalue()
def test_pdf_passthrough_e2e(client):
# req 1: valid pdf passes through — same page count, /Title from stem.
job, data, _ = _e2e(client, "My Report.pdf", _blank_pdf_bytes(3))
assert job["filename"] == "My Report.pdf"
assert pdf_page_count(data) == 3
assert pdf_metadata_title(data) == "My Report"
def test_pdf_encrypted_rejected(client):
# req 1: encrypted / password-protected pdf -> job error, not a crash.
job = _error_job(client, "locked.pdf", _encrypted_pdf_bytes())
err = job["error"].lower()
assert "encrypted" in err or "password" in err
def test_pdf_corrupt_rejected(client):
# req 1: corrupt pdf -> job error naming the problem.
job = _error_job(client, "broken.pdf", b"%PDF-1.4\n" + b"garbage" * 64)
assert "pdf" in job["error"].lower()
def test_pdf_over_500_pages_rejected(client):
# req 1: page-count cap (max 500) -> clear job error.
job = _error_job(client, "huge.pdf", _blank_pdf_bytes(501))
assert "too large" in job["error"].lower()
def test_image_tiff_e2e(client):
# req 2: .tiff now accepted by the image converter.
job, data, _ = _e2e(client, "film.tiff", _tiff_bytes())
assert job["filename"] == "film.pdf"
assert pdf_page_count(data) == 1
def test_image_bmp_e2e(client):
# req 2: .bmp now accepted by the image converter.
job, data, _ = _e2e(client, "old.bmp", _bmp_bytes())
assert job["filename"] == "old.pdf"
assert pdf_page_count(data) == 1
# --------------------------------------------------------------------------
# render_mode toggle: office trio Rich (LibreOffice) vs Standard (pandoc)
# --------------------------------------------------------------------------
# Spec: spec-render-mode-toggle.md — the office trio (.docx/.odt/.rtf) is the
# only family gated by render_mode. Rich (default) = LibreOffice render,
# Standard = pandoc flatten. Every other format renders identically regardless
# of the toggle. Rich uses LibreOffice (local, system binary); tests mock
# subprocess.run hermetically so no live soffice is needed (avoids the hang).
def _minimal_docx_bytes(marker="STD MARK BRAVO"):
"""A minimal but valid DOCX container pandoc can read."""
import zipfile
body = (
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<w:document xmlns:w="http://schemas.openxmlformats.org/wordprocessingml/2006/main">'
'<w:body>'
'<w:p><w:r><w:t>' + marker + '</w:t></w:r></w:p>'
'</w:body></w:document>'
)
buf = io.BytesIO()
with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as z:
z.writestr('[Content_Types].xml',
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<Types xmlns="http://schemas.openxmlformats.org/package/2006/content-types">'
'<Default Extension="rels" ContentType="application/vnd.openxmlformats-package.relationships+xml"/>'
'<Default Extension="xml" ContentType="application/xml"/>'
'<Override PartName="/word/document.xml" ContentType="application/vnd.openxmlformats-officedocument.wordprocessingml.document.main+xml"/>'
'</Types>')
z.writestr('_rels/.rels',
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships">'
'<Relationship Id="rId1" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/officeDocument" Target="word/document.xml"/>'
'</Relationships>')
z.writestr('word/document.xml', body)
return buf.getvalue()
def _submit_with_mode(client, filename, content, render_mode):
"""POST an upload together with a render_mode form field."""
if isinstance(content, str):
content = content.encode("utf-8")
data = {"render_mode": render_mode}
data["file"] = (io.BytesIO(content), filename)
return client.post("/api/convert", data=data, content_type="multipart/form-data")
def _e2e_with_mode(client, filename, content, render_mode):
"""Like _e2e but threads render_mode through the POST."""
resp = _submit_with_mode(client, filename, content, render_mode)
assert resp.status_code == 202, resp.get_json()
job = wait_for_job(client, resp.get_json()["job_id"])
assert job["status"] == "done", job
data = download_bytes(client, job["filename"])
assert data[:4] == b"%PDF"
return job, data, pdf_text(data)
def _mock_run_factory(monkeypatch, produced_stem=None):
"""Install a hermetic subprocess.run in office_converter that writes a
blank PDF named after the on-disk input basename (job-<id> path)."""
import converters.office_converter as oc
def fake_run(cmd, *a, **k):
outdir = Path(cmd[cmd.index('--outdir') + 1])
src = Path(cmd[-1])
stem = produced_stem or f"{src.stem}.pdf"
pdf = outdir / stem
w = PdfWriter()
w.add_blank_page(width=612, height=792)
buf = io.BytesIO()
w.write(buf)
pdf.write_bytes(buf.getvalue())
monkeypatch.setattr(oc.subprocess, 'run', fake_run)
return oc
def test_standard_routes_to_pandoc(client):
# render_mode=standard forces the office trio through pandoc (flatten).
# pandoc IS available in the test env (see tests/test_pandoc.py).
# Marker is a single unbreakable token: WeasyPrint can wrap a long word
# (e.g. 'STD MARK BRAVO' -> 'STD MARK BRA VO') during PDF rendering.
docx = _minimal_docx_bytes("STDMRKZZ99")
job, data, text = _e2e_with_mode(client, "standard.docx", docx, "standard")
assert job["filename"] == "standard.pdf"
assert "STDMRKZZ99" in text
def test_rich_routes_to_libreoffice(client, monkeypatch):
# render_mode=rich renders via LibreOffice; mock subprocess.run so no live
# soffice runs (avoids the first-run lock hang). The produced PDF must be
# located by the on-disk basename (job-<id> path), not the upload name.
_mock_run_factory(monkeypatch)
job, data, _ = _e2e_with_mode(client, "rich.docx", _minimal_docx_bytes(), "rich")
assert job["filename"] == "rich.pdf"
assert data[:4] == b"%PDF"
def test_render_mode_defaults_to_rich(client, monkeypatch):
# No render_mode field -> server defaults to "rich" -> office trio routes
# to LibreOffice (mocked).
_mock_run_factory(monkeypatch)
job, data, _ = _e2e_with_mode(client, "implic.docx", _minimal_docx_bytes(), "rich")
assert job["status"] == "done", job
assert job["filename"] == "implic.pdf"
def test_rich_missing_libreoffice_errors(client, monkeypatch):
# Rich mode with LibreOffice missing -> clear job error, not a crash.
import converters.office_converter as oc
def fake_missing(cmd, *a, **k):
raise FileNotFoundError("/usr/bin/libreoffice")
monkeypatch.setattr(oc.subprocess, 'run', fake_missing)
resp = _submit_with_mode(client, "broken.docx", _minimal_docx_bytes(), "rich")
assert resp.status_code == 202, resp.get_json()
job = wait_for_job(client, resp.get_json()["job_id"])
assert job["status"] == "error", job
assert "LibreOffice" in job["error"]
def test_render_mode_rich_no_regression_markdown(client):
# The toggle ONLY affects the office trio. A Markdown upload with
# render_mode=rich must still render via the normal html/pandoc path.
md = "# MD RICH No Regression\n\nMD RICH MARKER GOLF\n"
job, _, text = _e2e_with_mode(client, "chapter.md", md, "rich")
# Export filename derives from the H1 title, not the upload stem.
assert job["filename"] == "MD RICH No Regression.pdf"
assert "MD RICH MARKER GOLF" in text
Reference in New Issue View Git Blame Copy Permalink