"""US07-03: the golden media corpus, driven through the real decode path. Every case in ``tests/fixtures/media_corpus.py`` is exercised here. The claims: * a supported format/orientation/profile renders, with the orientation applied and the decode bounded to roughly the size that was asked for; * a damaged, empty, lying, or gigapixel file becomes one precise item error — the scan still finishes, the other assets still render, and a job that meets one keeps running; * a failure leaves no temporary file behind, and startup removes only the temporaries this service recognises; * cache invalidation follows the pixels, not the metadata. """ from __future__ import annotations import hashlib import shutil import time import tracemalloc from types import SimpleNamespace import pytest from PIL import Image from photo_pipeline import imaging from photo_pipeline.config import Config from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations from photo_pipeline.models import Asset, Thumbnail from photo_pipeline.services.inventory import InventoryService from photo_pipeline.services.thumbnails import ( TEMP_SUFFIX, ThumbnailError, ThumbnailService, ) from tests.fixtures.media_corpus import CASES, CASES_BY_ID, build_corpus RENDERABLE = [case for case in CASES if case.expect_error is None] DAMAGED = [case for case in CASES if case.expect_error is not None] needs_exiftool = pytest.mark.skipif( shutil.which("exiftool") is None, reason="exiftool not installed" ) @pytest.fixture(scope="module") def corpus(tmp_path_factory): root = tmp_path_factory.mktemp("corpus") return SimpleNamespace(root=root, files=build_corpus(root)) @pytest.fixture def env(tmp_path, corpus): """A library holding the whole corpus, scanned into a fresh database.""" data = tmp_path / "data" data.mkdir() lib = tmp_path / "lib" shutil.copytree(corpus.root, lib) config = Config.from_env( {"PHOTO_PIPELINE_DATA_DIR": str(data), "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib)} ) run_migrations(config.database_url) engine = create_db_engine(config.database_url) sf = create_session_factory(engine) scan = InventoryService(sf).scan(lib) by_id = { case.id: next( (aid for path, aid in scan.asset_ids.items() if path.endswith(case.filename)), None ) for case in CASES } yield SimpleNamespace( config=config, lib=lib, sf=sf, scan=scan, assets=by_id, thumbs=ThumbnailService(sf, config), ) engine.dispose() # ── the manifest itself ─────────────────────────────────────────────────────── def test_the_manifest_is_internally_consistent(): assert len({case.id for case in CASES}) == len(CASES), "logical ids must be unique" assert len({case.filename for case in CASES}) == len(CASES), "paths must be unique" for case in CASES: assert case.kind in {"format", "orientation", "profile", "damaged", "metadata"} assert case.expect_error in (None, "unsupported_image", "image_too_large") def test_the_corpus_regenerates_byte_for_byte(tmp_path): """A golden corpus that drifts between runs cannot be a golden corpus.""" first = build_corpus(tmp_path / "one") second = build_corpus(tmp_path / "two") digests = { case_id: ( hashlib.sha256(first[case_id].read_bytes()).hexdigest(), hashlib.sha256(second[case_id].read_bytes()).hexdigest(), ) for case_id in first } drifted = [case_id for case_id, (a, b) in digests.items() if a != b] assert not drifted, f"non-deterministic fixtures: {drifted}" # ── bounded decode ──────────────────────────────────────────────────────────── @pytest.mark.parametrize("case", RENDERABLE, ids=lambda case: case.id) def test_every_supported_case_renders(case, env): asset_id = env.assets[case.id] assert asset_id, f"{case.id} was not discovered by the scan" path = env.thumbs.generate(asset_id, 256) with Image.open(path) as thumb: assert thumb.format == "WEBP" assert max(thumb.size) <= 256 assert min(thumb.size) >= 1 @pytest.mark.parametrize("orientation", range(1, 9)) def test_exif_orientation_is_applied_before_resizing(orientation, env): """All eight tags: the 90° ones must come out portrait from a landscape source.""" asset_id = env.assets[f"orientation_{orientation}"] with Image.open(env.thumbs.generate(asset_id, 256)) as thumb: rotated = orientation in (5, 6, 7, 8) assert (thumb.height > thumb.width) is rotated def test_transparency_and_grayscale_survive_the_pipeline(env): with Image.open(env.thumbs.generate(env.assets["png_alpha"], 256)) as thumb: assert "A" in thumb.getbands() with Image.open(env.thumbs.generate(env.assets["grayscale"], 256)) as thumb: assert thumb.size == (100, 100) # smaller than the request: never upscaled def test_a_broken_colour_profile_still_produces_a_picture(env): """An unusable ICC profile is a metadata problem, not a reason to lose the preview.""" for case_id in ("icc_tagged", "broken_icc"): with Image.open(env.thumbs.generate(env.assets[case_id], 256)) as thumb: assert thumb.size[0] > 0 def test_a_large_jpeg_is_not_decoded_at_full_resolution(env): """12 megapixels would be ~36 MB of pixels; the draft decode keeps it far below.""" tracemalloc.start() try: env.thumbs.generate(env.assets["large_jpeg"], 256) _, peak = tracemalloc.get_traced_memory() finally: tracemalloc.stop() assert peak < 12_000_000, f"decode peaked at {peak} bytes" # ── damaged inputs ──────────────────────────────────────────────────────────── @pytest.mark.parametrize("case", DAMAGED, ids=lambda case: case.id) def test_every_damaged_case_is_a_precise_item_error(case, env): asset_id = env.assets[case.id] assert asset_id, f"{case.id} was not discovered by the scan" with pytest.raises(ThumbnailError) as raised: env.thumbs.generate(asset_id, 256) assert raised.value.code == case.expect_error # Persisted, so a broken original is not re-decoded on every request... with env.sf() as session: rows = [ row for row in session.query(Thumbnail).all() if row.asset_id == asset_id and row.state == "error" ] assert rows and rows[0].error_code == case.expect_error # ...and the cached failure is the same precise error, not a generic one. with pytest.raises(ThumbnailError) as again: env.thumbs.generate(asset_id, 256) assert again.value.code == case.expect_error def test_a_gigapixel_header_is_refused_quickly_and_cheaply(env): """The refusal must come from the declared size, not from decoding it.""" tracemalloc.start() started = time.monotonic() try: with pytest.raises(ThumbnailError): env.thumbs.generate(env.assets["bomb_header"], 1280) _, peak = tracemalloc.get_traced_memory() finally: tracemalloc.stop() assert time.monotonic() - started < 5 assert peak < 5_000_000, f"a 3.6 gigapixel header allocated {peak} bytes" def test_one_broken_file_stops_neither_the_scan_nor_its_neighbours(env): """The whole corpus is in the library, damaged files included.""" assert len(env.scan.asset_ids) >= len(CASES) - 1 # the empty file has no pixels to hash for case in RENDERABLE[:5]: assert env.thumbs.generate(env.assets[case.id], 256).exists() def test_a_job_that_meets_a_broken_file_gets_evidence_not_an_exception(env): """``ensure_protected`` is what the archive lane calls; a plan must not die on one unreadable original.""" evidence = env.thumbs.ensure_protected(env.assets["corrupt_png"]) assert evidence["state"] == "unsupported" assert evidence["error_code"] == "unsupported_image" assert env.thumbs.ensure_protected(env.assets["jpeg"])["state"] == "ready" def test_undecodable_files_do_not_break_hashing(env): """Discovery records what it can: byte identity always, pixel identity when the file has pixels.""" with env.sf() as session: assets = {a.current_path: a for a in session.query(Asset).all()} for case in DAMAGED: asset = next((a for path, a in assets.items() if path.endswith(case.filename)), None) if asset is None: continue assert asset.current_sha256, "byte identity is always available" assert asset.pixel_sha256 is None, "undecodable files must not invent pixel identity" # ── cache lifecycle ─────────────────────────────────────────────────────────── def test_a_failed_render_leaves_no_temporary_behind(env): with pytest.raises(ThumbnailError): env.thumbs.generate(env.assets["truncated_jpeg"], 256) leftovers = list(env.config.thumbnail_cache_dir.rglob(f"*{TEMP_SUFFIX}")) assert leftovers == [] def test_startup_cleanup_removes_only_recognised_temporaries(env): cache = env.config.thumbnail_cache_dir env.thumbs.generate(env.assets["jpeg"], 256) # creates the cache directory stale = cache / f".abc123{TEMP_SUFFIX}" stale.write_bytes(b"half a thumbnail") innocent = cache / "keep-me.webp" innocent.write_bytes(b"not a temporary") assert env.thumbs.cleanup_temp_files() == 1 assert not stale.exists() assert innocent.exists() assert list(cache.rglob("*.webp")), "real cache entries survive" def test_metadata_only_change_reuses_the_thumbnail_and_a_pixel_change_does_not(env): """The cache key follows the pixels: an EXIF edit must not cost a re-render.""" asset_id = env.assets["jpeg"] first = env.thumbs.generate(asset_id, 256) with env.sf() as session: # what a rescan records after an EXIF-only edit asset = session.get(Asset, asset_id) asset.current_sha256 = "different-bytes" session.commit() assert env.thumbs.generate(asset_id, 256) == first with env.sf() as session: # a genuine pixel change asset = session.get(Asset, asset_id) asset.pixel_sha256 = "different-pixels" session.commit() assert env.thumbs.generate(asset_id, 256) != first def test_a_deleted_cache_file_is_regenerated(env): asset_id = env.assets["png"] path = env.thumbs.generate(asset_id, 256) path.unlink() regenerated = env.thumbs.generate(asset_id, 256) assert regenerated == path and regenerated.exists() # ── the imaging door itself ─────────────────────────────────────────────────── def test_open_image_translates_every_decoder_failure(corpus): """Callers must be able to catch two typed errors, never bare ``Exception``.""" for case in DAMAGED: expected = imaging.ImageTooLarge if case.expect_error == "image_too_large" else imaging.UndecodableImage with pytest.raises(expected): with imaging.open_image(corpus.files[case.id]) as image: image.load() def test_open_image_refuses_more_pixels_than_the_caller_allowed(corpus): with pytest.raises(imaging.ImageTooLarge): with imaging.open_image(corpus.files["jpeg"], max_pixels=100): pass with imaging.open_image(corpus.files["jpeg"], max_pixels=100_000) as image: assert image.size == (320, 240) @needs_exiftool def test_malformed_metadata_does_not_stop_the_picture(env): """A broken EXIF block is a metadata fact, not a decode failure.""" assert env.thumbs.generate(env.assets["malformed_metadata"], 256).exists() with env.sf() as session: asset = session.get(Asset, env.assets["malformed_metadata"]) assert asset.pixel_sha256, "pixels are still identifiable" def test_scoring_never_relaxes_truncated_image_handling(corpus): """The donor's process-global ``LOAD_TRUNCATED_IMAGES`` is gone for good. It is global state: switching it on for the safety model would also switch it on for hashing and preview rendering in the same process, and half a file would silently become a valid picture (donor_ledger: nt-score-model). """ import inspect from PIL import ImageFile from photo_pipeline.integrations import nsfw_model code = [ line for line in inspect.getsource(nsfw_model).splitlines() if not line.strip().startswith("#") ] assert not any("LOAD_TRUNCATED_IMAGES" in line for line in code) # And the door holds even when something else in the process turned it on — # the frozen donor does exactly that when the characterization suite imports it. previous = ImageFile.LOAD_TRUNCATED_IMAGES ImageFile.LOAD_TRUNCATED_IMAGES = True try: with pytest.raises(imaging.UndecodableImage): with imaging.open_image(corpus.files["truncated_jpeg"]) as image: image.load() assert ImageFile.LOAD_TRUNCATED_IMAGES is True, "the caller's setting is restored" finally: ImageFile.LOAD_TRUNCATED_IMAGES = previous def test_every_manifest_case_is_exercised(): """The corpus lint: no fixture may sit in the manifest untested.""" covered = {case.id for case in RENDERABLE} | {case.id for case in DAMAGED} assert covered == set(CASES_BY_ID)