test: add low-level preview success tests for all formats

Add fixtures and tests covering image EXIF orientations, HDR AVIF, SDR video, 90/270 rotated video, HDR video with and without rotation, PDF, dispatch, and OnlyOffice (skipped unless configured).

Regenerate rotated/HDR video fixtures and sample.pdf with tests/files/generate_fixtures.py.

tests/files/generate_fixtures.py

@@ -0,0 +1,180 @@
+#!/usr/bin/env python3
+"""Generate missing test fixtures for the low-level preview tests.
+
+This script only creates files that are not already present in tests/files/.
+Run it whenever you need to rebuild the rotated/HDR video fixtures or the
+sample PDF.
+"""
+
+from __future__ import annotations
+
+import logging
+import subprocess
+import sys
+from pathlib import Path
+
+try:
+    import pymupdf
+except ImportError as e:  # pragma: no cover - optional pdf extra
+    raise ImportError(
+        "Fixture generation requires pymupdf: pip install mediapreview[pdf]"
+    ) from e
+
+ROOT = Path(__file__).resolve().parent
+FILES = ROOT
+logger = logging.getLogger(__name__)
+
+
+def run(cmd: list[str]) -> None:
+    subprocess.run(cmd, check=True, capture_output=True, text=True)
+
+
+def _find_box(data: bytearray, box_type: bytes, start: int = 0, end: int | None = None) -> int:
+    end = end or len(data)
+    i = start
+    while i + 8 <= end:
+        size = int.from_bytes(data[i : i + 4], "big")
+        btype = data[i + 4 : i + 8]
+        if btype == box_type:
+            return i
+        if size == 0:
+            break
+        if size == 1:
+            size = int.from_bytes(data[i + 8 : i + 16], "big")
+        if size < 8:
+            raise ValueError(f"Invalid box size {size} for {btype.decode('ascii', errors='replace')}")
+        i += size
+    return -1
+
+
+def _matrix_90_cw() -> list[int]:
+    return [0, 1 << 16, 0, -(1 << 16), 0, 0, 0, 0, 1 << 16]
+
+
+def _matrix_270_cw() -> list[int]:
+    return [0, -(1 << 16), 0, 1 << 16, 0, 0, 0, 0, 1 << 16]
+
+
+def _patch_tkhd_rotation(in_path: Path, out_path: Path, degrees: int) -> None:
+    """Write a copy of *in_path* with the video track display matrix rotated."""
+    data = bytearray(in_path.read_bytes())
+
+    moov_idx = _find_box(data, b"moov")
+    if moov_idx < 0:
+        raise ValueError("moov box not found")
+    moov_end = moov_idx + int.from_bytes(data[moov_idx : moov_idx + 4], "big")
+
+    trak_idx = _find_box(data, b"trak", moov_idx + 8, moov_end)
+    while trak_idx >= 0:
+        trak_size = int.from_bytes(data[trak_idx : trak_idx + 4], "big")
+        trak_end = trak_idx + trak_size
+        mdia_idx = _find_box(data, b"mdia", trak_idx + 8, trak_end)
+        if mdia_idx < 0:
+            trak_idx = _find_box(data, b"trak", trak_end, moov_end)
+            continue
+        mdia_end = mdia_idx + int.from_bytes(data[mdia_idx : mdia_idx + 4], "big")
+        hdlr_idx = _find_box(data, b"hdlr", mdia_idx + 8, mdia_end)
+        if hdlr_idx < 0:
+            trak_idx = _find_box(data, b"trak", trak_end, moov_end)
+            continue
+        handler_type = data[hdlr_idx + 16 : hdlr_idx + 20]
+        if handler_type == b"vide":
+            tkhd_idx = _find_box(data, b"tkhd", trak_idx + 8, trak_end)
+            if tkhd_idx < 0:
+                raise ValueError("video track has no tkhd")
+            version = data[tkhd_idx + 8]
+            if version != 0:
+                raise ValueError(f"unsupported tkhd version {version}")
+            matrix_offset = tkhd_idx + 48
+            matrix = _matrix_90_cw() if degrees == 90 else _matrix_270_cw()
+            for i, val in enumerate(matrix):
+                data[matrix_offset + i * 4 : matrix_offset + (i + 1) * 4] = val.to_bytes(
+                    4, "big", signed=True
+                )
+            out_path.write_bytes(data)
+            return
+        trak_idx = _find_box(data, b"trak", trak_end, moov_end)
+    raise ValueError("video track not found")
+
+
+def generate_pdf() -> None:
+    pdf_path = FILES / "sample.pdf"
+    if pdf_path.exists():
+        return
+    doc = pymupdf.open()
+    page = doc.new_page(width=595, height=842)
+    page.insert_text((100, 100), "Hello, PDF preview test!")
+    page.draw_rect((100, 150, 400, 250), color=(1, 0, 0), width=2)
+    doc.save(str(pdf_path))
+    doc.close()
+    logger.info("generated %s", pdf_path)
+
+
+def generate_sdr_rotated() -> None:
+    source = FILES / "sample-1mb.mp4"
+    if not source.exists():
+        raise FileNotFoundError(f"Missing source fixture: {source}")
+    for degrees in (90, 270):
+        out = FILES / f"rotated_{degrees}.mp4"
+        if out.exists():
+            continue
+        _patch_tkhd_rotation(source, out, degrees)
+        logger.info("generated %s", out)
+
+
+def generate_hdr_video() -> None:
+    out = FILES / "hdr_video.mp4"
+    if out.exists():
+        return
+    run(
+        [
+            "ffmpeg",
+            "-y",
+            "-hide_banner",
+            "-loglevel",
+            "error",
+            "-nostdin",
+            "-f",
+            "lavfi",
+            "-i",
+            "testsrc=duration=1:size=320x240:rate=30",
+            "-c:v",
+            "libx265",
+            "-preset",
+            "fast",
+            "-crf",
+            "30",
+            "-pix_fmt",
+            "yuv420p10le",
+            "-x265-params",
+            "colorprim=bt2020:transfer=smpte2084:colormatrix=bt2020nc:repeat-headers=1",
+            "-an",
+            str(out),
+        ]
+    )
+    logger.info("generated %s", out)
+
+
+def generate_hdr_rotated() -> None:
+    source = FILES / "hdr_video.mp4"
+    if not source.exists():
+        raise FileNotFoundError(f"Generate hdr_video.mp4 first: {source}")
+    for degrees in (90, 270):
+        out = FILES / f"hdr_rotated_{degrees}.mp4"
+        if out.exists():
+            continue
+        _patch_tkhd_rotation(source, out, degrees)
+        logger.info("generated %s", out)
+
+
+def main() -> int:
+    logging.basicConfig(level=logging.INFO, format="%(message)s")
+    generate_pdf()
+    generate_sdr_rotated()
+    generate_hdr_video()
+    generate_hdr_rotated()
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_previews.py

@@ -0,0 +1,191 @@
+"""Success tests for low-level preview conversion functions.
+
+These tests exercise the concrete backend converters (image, video, PDF,
+office) against the fixture files in tests/files/.  The only thing they
+assert is that the conversion succeeds and returns non-empty AVIF bytes, plus
+basic metadata sanity checks.
+
+Office tests are skipped unless an OnlyOffice Document Server is reachable.
+Configure them with environment variables before running pytest:
+
+    ONLYOFFICE_URL=http://localhost:8988
+    ONLYOFFICE_JWT_SECRET=<same-secret-you-gave-the-oo-container>
+    ONLYOFFICE_CALLBACK_HOST=<host the OO container can reach>
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import pytest
+
+from mediapreview import dispatch
+from mediapreview.backends.image import (
+    process_image,
+    process_image_buffer,
+    process_image_pyvips,
+)
+from mediapreview.backends.pdf import process_pdf
+from mediapreview.backends.video import process_video
+from mediapreview.office import is_available_async
+from mediapreview.pool import generate_office_preview
+
+FILES = Path(__file__).resolve().parent / "files"
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _assert_ok(data, resp, backend: str | None = None) -> None:
+    assert resp.ok, f"conversion failed: {resp.error}"
+    assert data
+    assert resp.mime == "image/avif"
+    if backend:
+        assert resp.backend == backend
+
+
+# ---------------------------------------------------------------------------
+# Images
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize("path", sorted(FILES.glob("Landscape_*.jpg")))
+def test_process_image_exif_orientations(path: Path) -> None:
+    """Every EXIF orientation fixture must produce a valid preview."""
+    data, resp = process_image(path, maxsize=512, quality=60)
+    _assert_ok(data, resp, backend="pyvips")
+    assert resp.width in (1200, 1800)
+    assert resp.height in (1200, 1800)
+
+
+def test_process_image_hdr_avif() -> None:
+    """HDR AVIF images are routed through ffmpeg to preserve colour metadata."""
+    path = FILES / "hdr_cosmos01650_cicp9-16-9_yuv444_full_qp10.avif"
+    data, resp = process_image(path, maxsize=512, quality=60)
+    _assert_ok(data, resp, backend="ffmpeg")
+    assert resp.width is not None
+    assert resp.height is not None
+
+
+def test_process_image_pyvips() -> None:
+    """The pyvips-only image backend works on a plain JPEG."""
+    path = FILES / "Landscape_1.jpg"
+    data, resp = process_image_pyvips(path, maxsize=512, quality=60)
+    _assert_ok(data, resp, backend="pyvips")
+
+
+def test_process_image_buffer() -> None:
+    """Processing a JPEG from a buffer produces the same valid preview."""
+    path = FILES / "Landscape_1.jpg"
+    data, resp = process_image_buffer(
+        path.read_bytes(), maxsize=512, quality=60, maxzoom=2.0
+    )
+    _assert_ok(data, resp, backend="pyvips")
+
+
+# ---------------------------------------------------------------------------
+# Videos
+# ---------------------------------------------------------------------------
+
+
+VIDEO_FIXTURES = [
+    ("sample-1mb.mp4", 854, 480),
+    ("rotated_90.mp4", 480, 854),
+    ("rotated_270.mp4", 480, 854),
+    ("hdr_video.mp4", 320, 240),
+    ("hdr_rotated_90.mp4", 240, 320),
+    ("hdr_rotated_270.mp4", 240, 320),
+]
+
+
+@pytest.mark.parametrize(
+    ("filename", "expected_width", "expected_height"),
+    VIDEO_FIXTURES,
+    ids=[f[0] for f in VIDEO_FIXTURES],
+)
+def test_process_video(filename: str, expected_width: int, expected_height: int) -> None:
+    """SDR and HDR video clips, with and without rotation, convert successfully."""
+    data, resp = process_video(FILES / filename, maxsize=512, quality=60)
+    _assert_ok(data, resp, backend="video")
+    assert resp.width == expected_width
+    assert resp.height == expected_height
+
+
+# ---------------------------------------------------------------------------
+# PDF
+# ---------------------------------------------------------------------------
+
+
+def test_process_pdf() -> None:
+    """A simple single-page PDF converts to a valid preview."""
+    data, resp = process_pdf(
+        FILES / "sample.pdf",
+        maxsize=512,
+        maxzoom=2.0,
+        quality=60,
+    )
+    _assert_ok(data, resp, backend="pdf+pyvips")
+    assert resp.width == 595
+    assert resp.height == 842
+
+
+# ---------------------------------------------------------------------------
+# Public dispatch entry point
+# ---------------------------------------------------------------------------
+
+
+DISPATCH_FIXTURES = [
+    ("Landscape_1.jpg", "pyvips", 1800, 1200),
+    ("sample-1mb.mp4", "video", 854, 480),
+    ("sample.pdf", "pdf+pyvips", 595, 842),
+]
+
+
+@pytest.mark.parametrize(
+    ("filename", "backend", "expected_width", "expected_height"),
+    DISPATCH_FIXTURES,
+    ids=[f[0] for f in DISPATCH_FIXTURES],
+)
+def test_dispatch(
+    filename: str, backend: str, expected_width: int, expected_height: int
+) -> None:
+    """The public dispatch() wrapper routes to the correct backend and succeeds."""
+    data, resp = dispatch(FILES / filename, quality=60, maxsize=512, maxzoom=2.0)
+    _assert_ok(data, resp, backend=backend)
+    assert resp.width == expected_width
+    assert resp.height == expected_height
+
+
+# ---------------------------------------------------------------------------
+# Office previews via OnlyOffice
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_generate_office_preview() -> None:
+    """DOCX preview through OnlyOffice.
+
+    Skipped unless an OnlyOffice server is reachable and ONLYOFFICE_JWT_SECRET
+    is set to the same secret the OO container is using. The shared secret must
+    be at least 32 bytes long so PyJWT doesn't warn about weak HMAC keys.
+    """
+    secret = os.environ.get("ONLYOFFICE_JWT_SECRET", "")
+    if len(secret.encode()) < 32:
+        pytest.skip("ONLYOFFICE_JWT_SECRET not set or shorter than 32 bytes")
+    if not await is_available_async():
+        pytest.skip("OnlyOffice Document Server not reachable")
+
+    data, resp = await generate_office_preview(
+        FILES / "file-sample_100kB.docx",
+        quality=60,
+        maxsize=512,
+        maxzoom=2.0,
+    )
+    assert data is not None
+    assert resp is not None
+    _assert_ok(data, resp)
+    assert resp.width is not None
+    assert resp.height is not None