"""PHASE 2 (Chunk B): office / LaTeX / structured-text formats via Pandoc. Pins the behavior of converters/pandoc_converter.py end-to-end through the HTTP API (Flask test client + worker threads, hermetic DATA_DIR via conftest): * all 8 formats e2e -> job done, correct export name, real %PDF, content * title chain: docx from docProps/core.xml dc:title, odt from meta.xml dc:title, tex/org/adoc/rst from the

pandoc promotes from the document's own title (\title / #+TITLE / '= Title' / overline) * embedded image (PNG) survives the pandoc -> html5 -> WeasyPrint pipeline * naming collision (-2 suffix) and the .tex not-latex fall-through to the plain-text converter * zip-slip / zip-bomb pre-pass + a non-zip .docx all surface as clean job errors (202 -> error), never an HTTP 500 * /health advertises the 8 new formats; is_supported() accepts them Pandoc-generated fixtures (docx/odt/rtf) are built ONCE per module run from a small markdown source. Pandoc is a hard prerequisite for this module (3.x is a project system dep); if the binary is absent the whole module skips. """ import io import re import shutil import struct import subprocess import tempfile import urllib.parse import zipfile from pathlib import Path import pytest from PIL import Image import converters from tests.helpers import (download_bytes, pdf_page_count, pdf_text, wait_for_job) # -------------------------------------------------------------------------- # skip guard: pandoc is a system dependency of this converter family # -------------------------------------------------------------------------- _pandoc_bin = shutil.which("pandoc") pytestmark = pytest.mark.skipif(_pandoc_bin is None, reason="pandoc not installed") # Extension matrix for the pandoc converter family. PANDOC_EXTS = (".docx", ".odt", ".rtf", ".tex", ".latex", ".org", ".adoc", ".rst") PANDOC_BARE = ("docx", "odt", "rtf", "tex", "latex", "org", "adoc", "rst") # -------------------------------------------------------------------------- # pure builders (zip mutation + pandoc invocation) — no fixtures, testable # -------------------------------------------------------------------------- def _run_pandoc(src: Path, out: Path) -> None: proc = subprocess.run(["pandoc", str(src), "-o", str(out)], capture_output=True, text=True) assert proc.returncode == 0, f"pandoc failed for {src.name}: {proc.stderr}" def _png_bytes(size=(32, 32), color=(255, 0, 0)) -> bytes: buf = io.BytesIO() Image.new("RGB", size, color).save(buf, format="PNG") return buf.getvalue() def _rewrite_zip(raw: bytes, edits: dict) -> bytes: """Rebuild a zip from `raw`, replacing the bytes of each `edits` member.""" with zipfile.ZipFile(io.BytesIO(raw)) as src: items = {i.filename: src.read(i.filename) for i in src.infolist()} for name, new in edits.items(): items[name] = new out = io.BytesIO() with zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as dst: for name, data in items.items(): dst.writestr(name, data) return out.getvalue() def _stamp_dc_title(raw: bytes, member: str, title: str) -> bytes: """Set inside a zip member (core.xml / meta.xml) to `title`.""" with zipfile.ZipFile(io.BytesIO(raw)) as z: cur = z.read(member).decode("utf-8", errors="replace") new = re.sub(r"]*>.*?", f"{title}", cur, flags=re.S) return _rewrite_zip(raw, {member: new.encode("utf-8")}) def _forge_zip_bomb(raw: bytes) -> bytes: """Inflate the first central-directory member's declared uncompressed size to 3 GiB so the zip-bomb guard total exceeds its 2 GiB cap. zipfile.writestr recomputes sizes on write, so the central-directory field is patched in place (offset 24 of the CD file header). """ data = bytearray(raw) pos = data.find(b"PK\x01\x02") assert pos >= 0, "no central directory found in fixture zip" struct.pack_into(" bytes: """Copy a valid office zip and add a dotdot member escaping the root.""" out = io.BytesIO() with zipfile.ZipFile(io.BytesIO(raw)) as src, \ zipfile.ZipFile(out, "w") as dst: for i in src.infolist(): dst.writestr(i.filename, src.read(i.filename)) dst.writestr("../../evil.txt", "pwn") return out.getvalue() def _build_all(root: Path) -> dict: """Build every pandoc fixture once; return {key: bytes}. The PNG is written BEFORE pandoc runs: pandoc embeds by path, and a missing file is silently dropped (PandocResourceNotFound), which would make the embedded-image tests prove nothing. """ (root / "tiny.png").write_bytes(_png_bytes()) png = str(root / "tiny.png") # absolute path — required for embedding def md(name: str, text: str) -> None: (root / name).write_text(text, encoding="utf-8") # Office trio: distinct H1 titles + a per-format ALL-CAPS marker. md("x.md", "# Pandoc DocX Alpha\n\nBody X with **BOLDX** and a code span.\n") md("b.md", "# Pandoc Odt Bravo\n\nBody B with BOLD_B text.\n") md("c.md", "# Pandoc Rtf Charlie\n\nBody C with BOLD_C text.\n") _run_pandoc(root / "x.md", root / "x.docx") _run_pandoc(root / "b.md", root / "b.odt") _run_pandoc(root / "c.md", root / "c.rtf") # Embedded-image pair (PNG embedded by pandoc at build time). md("ix.md", f"# Embedded Image Doc\n\n![]({png})\n\nIMGXMARK body.\n") md("io.md", f"# Embedded Image Odt\n\n![]({png})\n\nIMGOMARK body.\n") _run_pandoc(root / "ix.md", root / "ix.docx") _run_pandoc(root / "io.md", root / "io.odt") # Structured text formats (read directly by pandoc, no zip container). (root / "sample.tex").write_text( "\\documentclass{article}\n\\title{Pandoc TeX Delta}\n" "\\begin{document}\nTTEXMARK body.\n\\end{document}\n", encoding="utf-8") (root / "sample.latex").write_text( "\\documentclass{article}\n\\title{Pandoc LaTeX Echo}\n" "\\begin{document}\nLATEXMARK body.\n\\end{document}\n", encoding="utf-8") (root / "sample.org").write_text( "#+TITLE: Pandoc Org Foxtrot\n\n* A Heading\n\n" "Org body ORGMARK here.\n", encoding="utf-8") (root / "sample.adoc").write_text( "= Pandoc Adoc Golf\n\nAdoc body ADOCMARK here.\n", encoding="utf-8") (root / "sample.rst").write_text( "Pandoc Rst Hotel\n================\n\nRst body RSTMARK here.\n", encoding="utf-8") # Fall-through: a .tex with no \\documentclass and no leading %. (root / "note.tex").write_text( "just a note\nnothing special here\n", encoding="utf-8") # Collision: an org whose #+TITLE is 'Pandoc Collie'. (root / "collide.org").write_text( "#+TITLE: Pandoc Collie\n\nCollision body.\n", encoding="utf-8") docx = (root / "x.docx").read_bytes() odt = (root / "b.odt").read_bytes() # Title-chain fixtures: stamp a dc:title that DIFFERS from the body H1 so # the export name proves which source the converter actually read # (pandoc leaves core.xml / meta.xml dc:title empty; the stamp is non-empty). docx_core = _stamp_dc_title(docx, "docProps/core.xml", "CoreX Wins") odt_meta = _stamp_dc_title(odt, "meta.xml", "MetaX Wins") # Security fixtures forged from a real, small docx. return { "docx": docx, "odt": odt, "rtf": (root / "c.rtf").read_bytes(), "tex": (root / "sample.tex").read_bytes(), "latex": (root / "sample.latex").read_bytes(), "org": (root / "sample.org").read_bytes(), "adoc": (root / "sample.adoc").read_bytes(), "rst": (root / "sample.rst").read_bytes(), "note_tex": (root / "note.tex").read_bytes(), "collide_org": (root / "collide.org").read_bytes(), "docx_core": docx_core, "odt_meta": odt_meta, "img_docx": (root / "ix.docx").read_bytes(), "img_odt": (root / "io.odt").read_bytes(), "bomb_docx": _forge_zip_bomb(docx), "slip_docx": _forge_zip_slip(docx), "fake_docx": b"\x89\x50\x4e\x47\x0d\x0a\x1a\x0a definitely not a docx", } # -------------------------------------------------------------------------- # fixtures # -------------------------------------------------------------------------- @pytest.fixture(scope="module") def pandoc_fixtures(): """Build the pandoc fixture set once; return {key: bytes}.""" root = Path(tempfile.mkdtemp(prefix="pandoc-fix-")) fx = _build_all(root) # Self-check: the image fixtures must really carry a PNG, else the # embedded-image tests would prove nothing (pandoc drops missing files). for key in ("img_docx", "img_odt"): with zipfile.ZipFile(io.BytesIO(fx[key])) as z: assert any(i.filename.lower().endswith(".png") for i in z.infolist()), f"{key}: PNG not embedded" return fx # -------------------------------------------------------------------------- # e2e helpers (mirror tests/test_multi_format.py) # -------------------------------------------------------------------------- def _submit(client, filename, content, render_mode="standard"): # AC: docx/odt/rtf are gated by render_mode; standard keeps them on the # EXISTING pandoc path (no LibreOffice) — this file pins pandoc-family # behavior, so every call here requests standard explicitly. if isinstance(content, str): content = content.encode("utf-8") data = {"file": (io.BytesIO(content), filename), "render_mode": render_mode} return client.post("/api/convert", data=data, content_type="multipart/form-data") def _e2e(client, filename, content, render_mode="standard"): # AC: docx/odt/rtf are gated by render_mode; standard keeps them on the # EXISTING pandoc path (no LibreOffice) — this file pins pandoc-family # behavior, so every call here requests standard explicitly. """Submit, wait for terminal, assert done, return (job, pdf_bytes, pdf_text).""" resp = _submit(client, filename, content, render_mode=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 _error_job(client, filename, content, render_mode="standard"): # AC: docx/odt/rtf are gated by render_mode; standard keeps them on the # EXISTING pandoc path (no LibreOffice) — this file pins pandoc-family # behavior, so every call here requests standard explicitly. """Submit, expect 202 then a terminal error state; return the snapshot.""" resp = _submit(client, filename, content, render_mode=render_mode) 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 # -------------------------------------------------------------------------- # is_supported + /health advertisement # -------------------------------------------------------------------------- def test_is_supported_pandoc_formats(): for ext in (".docx", ".odt", ".rtf", ".tex", ".latex", ".org", ".adoc", ".rst"): assert converters.is_supported(f"book{ext}"), ext def test_health_lists_pandoc_formats(client): resp = client.get("/health") assert resp.status_code == 200, resp.get_json() formats = resp.get_json()["supported_formats"] for bare in ("docx", "odt", "rtf", "tex", "latex", "org", "adoc", "rst"): assert bare in formats, f"{bare} missing from /health supported_formats" # -------------------------------------------------------------------------- # per-format e2e happy paths (all 8) — filename + content marker # -------------------------------------------------------------------------- def test_docx_e2e_title_and_content(client, pandoc_fixtures): # Pandoc leaves core.xml dc:title empty, so the name comes from the first #

pandoc promotes from the markdown source. job, data, text = _e2e(client, "alpha.docx", pandoc_fixtures["docx"]) assert job["filename"] == "Pandoc DocX Alpha.pdf" assert "BOLDX" in text def test_odt_e2e_title_and_content(client, pandoc_fixtures): job, data, text = _e2e(client, "bravo.odt", pandoc_fixtures["odt"]) assert job["filename"] == "Pandoc Odt Bravo.pdf" assert "BOLD_B" in text def test_rtf_e2e_title_and_content(client, pandoc_fixtures): job, data, text = _e2e(client, "charlie.rtf", pandoc_fixtures["rtf"]) assert job["filename"] == "Pandoc Rtf Charlie.pdf" assert "BOLD_C" in text def test_tex_e2e_title_and_content(client, pandoc_fixtures): job, data, text = _e2e(client, "delta.tex", pandoc_fixtures["tex"]) assert job["filename"] == "Pandoc TeX Delta.pdf" assert "TTEXMARK" in text def test_latex_e2e_title_and_content(client, pandoc_fixtures): # .latex is the same pandoc reader as .tex — extension-parity check. job, data, text = _e2e(client, "echo.latex", pandoc_fixtures["latex"]) assert job["filename"] == "Pandoc LaTeX Echo.pdf" assert "LATEXMARK" in text def test_org_e2e_title_and_content(client, pandoc_fixtures): job, data, text = _e2e(client, "foxtrot.org", pandoc_fixtures["org"]) assert job["filename"] == "Pandoc Org Foxtrot.pdf" assert "ORGMARK" in text def test_adoc_e2e_title_and_content(client, pandoc_fixtures): job, data, text = _e2e(client, "golf.adoc", pandoc_fixtures["adoc"]) assert job["filename"] == "Pandoc Adoc Golf.pdf" assert "ADOCMARK" in text def test_rst_e2e_title_and_content(client, pandoc_fixtures): job, data, text = _e2e(client, "hotel.rst", pandoc_fixtures["rst"]) assert job["filename"] == "Pandoc Rst Hotel.pdf" assert "RSTMARK" in text # -------------------------------------------------------------------------- # title chain: zip dc:title beats the body

# -------------------------------------------------------------------------- def test_docx_title_from_core_xml(client, pandoc_fixtures): # A dc:title stamped into docProps/core.xml wins over the body H1 # ('Pandoc DocX Alpha'), proving the core.xml step of the chain runs. job, data, _ = _e2e(client, "corex.docx", pandoc_fixtures["docx_core"]) assert job["filename"] == "CoreX Wins.pdf" def test_odt_title_from_meta_xml(client, pandoc_fixtures): # A dc:title stamped into meta.xml wins over the body H1 ('Pandoc Odt Bravo'). job, data, _ = _e2e(client, "metax.odt", pandoc_fixtures["odt_meta"]) assert job["filename"] == "MetaX Wins.pdf" # -------------------------------------------------------------------------- # embedded image: PNG survives pandoc -> html5 -> WeasyPrint into the PDF # -------------------------------------------------------------------------- def test_docx_embedded_image_reaches_pdf(client, pandoc_fixtures): job, data, text = _e2e(client, "imaged.docx", pandoc_fixtures["img_docx"]) assert job["filename"] == "Embedded Image Doc.pdf" assert pdf_page_count(data) >= 1 # The PNG is an actual XObject on a page (probe-verified: 1 image/page 0). from pypdf import PdfReader pages = PdfReader(io.BytesIO(data)).pages assert any(len(pg.images) >= 1 for pg in pages), "no image in PDF pages" assert "IMGXMARK" in text def test_odt_embedded_image_reaches_pdf(client, pandoc_fixtures): job, data, text = _e2e(client, "imageo.odt", pandoc_fixtures["img_odt"]) assert job["filename"] == "Embedded Image Odt.pdf" assert pdf_page_count(data) >= 1 from pypdf import PdfReader pages = PdfReader(io.BytesIO(data)).pages assert any(len(pg.images) >= 1 for pg in pages), "no image in PDF pages" assert "IMGOMARK" in text # -------------------------------------------------------------------------- # naming: collision suffix + .tex not-latex fall-through # -------------------------------------------------------------------------- def test_collision_second_upload_gets_suffix(client, pandoc_fixtures): job1, _, _ = _e2e(client, "collide.org", pandoc_fixtures["collide_org"]) job2, _, _ = _e2e(client, "collide.org", pandoc_fixtures["collide_org"]) assert job1["filename"] == "Pandoc Collie.pdf" assert job2["filename"] == "Pandoc Collie-2.pdf" def test_tex_not_latex_falls_through_to_text(client, pandoc_fixtures): # No \\documentclass and no leading % -> plain-text converter: named by # stem, raw source text rendered, no error. job, data, text = _e2e(client, "note.tex", pandoc_fixtures["note_tex"]) assert job["filename"] == "note.pdf" assert "just a note" in text assert "nothing special here" in text # -------------------------------------------------------------------------- # download headers (family representative) # -------------------------------------------------------------------------- def test_download_headers_for_pandoc_conversion(client, pandoc_fixtures): job, _, _ = _e2e(client, "alpha.docx", pandoc_fixtures["docx"]) # URL-encode the filename (it contains spaces) for the direct GET. resp = client.get("/download/" + urllib.parse.quote(job["filename"])) assert resp.status_code == 200 assert resp.content_type == "application/pdf" cd = resp.headers["Content-Disposition"] assert "attachment" in cd assert f'filename="{job["filename"]}"' in cd # -------------------------------------------------------------------------- # security: zip-bomb / zip-slip / non-zip .docx -> clean job error, not 500 # -------------------------------------------------------------------------- def test_zip_bomb_docx_rejected(client, pandoc_fixtures): # Declared uncompressed size inflated past the 2 GiB cap -> job error. job = _error_job(client, "bomb.docx", pandoc_fixtures["bomb_docx"]) assert "zip-bomb" in job["error"].lower() def test_zip_slip_docx_rejected(client, pandoc_fixtures): job = _error_job(client, "slip.docx", pandoc_fixtures["slip_docx"]) assert "unsafe zip member" in job["error"] def test_non_zip_docx_rejected_not_500(client, pandoc_fixtures): # Random bytes with a .docx name: clean job error, never an HTTP 500. job = _error_job(client, "fake.docx", pandoc_fixtures["fake_docx"]) assert job["error"], "expected a non-empty error message"