433 lines
18 KiB
Python
433 lines
18 KiB
Python
"""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 <h1> 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 <dc:title> 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"<dc:title[^>]*>.*?</dc:title>",
|
|
f"<dc:title>{title}</dc: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("<I", data, pos + 24, 3 * 1024 ** 3)
|
|
return bytes(data)
|
|
|
|
|
|
def _forge_zip_slip(raw: bytes) -> 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\n\nIMGXMARK body.\n")
|
|
md("io.md", f"# Embedded Image Odt\n\n\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
|
|
# <h1> 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 <h1>
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
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"
|