"""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 = ( "My HTML Doc" "" "

Heading One

HTML MARKER ALPHA

" ) 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", "

FRAG MARK BRAVO

") 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 = ( '

' ).encode("utf-8") + body.encode("cp1252") + b"

" 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 = '\nGAMMA_XMLtwo' job, _, text = _e2e(client, "feed.xml", doc) assert job["filename"] == "feed.pdf" assert "GAMMA_XML" in text def test_svg_e2e(client): svg = ( '' "" "SVG MARK DELTA" ) 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", "

UPPER MARK ECHO

") 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", "") assert "XML" in job["error"] def test_invalid_svg_is_job_error(client): job = _error_job(client, "bad.svg", "") 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 = ( '' '' '' '' + marker + '' '' ) buf = io.BytesIO() with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as z: z.writestr('[Content_Types].xml', '' '' '' '' '' '') z.writestr('_rels/.rels', '' '' '' '') 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- 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- 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