"""`images stage` and `images import` (Stream AW3.3): the manifest's chosen entries into the uploads home and onto their rows. Pinned: `stage` reads a file from the folder and downloads an address once each (the same source chosen at two levels is fetched once), writes the content-hashed derivatives and the served addresses back onto the entries, refuses an address that answers with something other than an image and records why on the entry, and skips every entry whose derivative already exists (a re-run makes no request); `--check` reads and requests nothing. `import` verifies the derivative is on disk, resolves the target by slug, id or barcode (an alias or a merged-away id followed), writes through the one writer so a re-run is a no-op and a supplied picture is never replaced by a fetched one, counts an entry no row answers to and skips it, and `--check` rolls back. What the wrong answer costs: a login page staged as a bottle, a picture set on a row the review merged away, or a second run that downloads the client's export again. """ from __future__ import annotations import argparse import io import json from datetime import UTC, datetime from pathlib import Path import pytest from PIL import Image from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool from app import cli_images from app.models import Base, Brand, ProductLine, ProductVariant from app.services import imagery TABLES = [Brand.__table__, ProductLine.__table__, ProductVariant.__table__] ADMIN = "admin:william-grant-via-adam" LICENCE = "Brand-supplied; permission held by the client" NOW = datetime(2026, 9, 19, 20, 0, tzinfo=UTC) def png(colour=(10, 20, 30, 255)) -> bytes: out = io.BytesIO() Image.new("RGBA", (4, 4), colour).save(out, format="PNG") return out.getvalue() @pytest.fixture def db(): engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) Base.metadata.create_all(engine, tables=TABLES) factory = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False, future=True) with factory() as session: brand = Brand(id=1, slug="glenfiddich", name="Glenfiddich") old = Brand(id=2, slug="glenfiddich-old-spelling", name="GLENFIDDICH", alias_of_id=1) line = ProductLine(id=11, brand_id=1, key="15yo solera", name="15 Year Old Solera", slug="glenfiddich-15yo-solera") alias = ProductLine(id=12, brand_id=1, key="15 year old solera", name="15yo Solera", slug="glenfiddich-15-solera", alias_of_id=11) session.add_all([brand, old, line, alias]) session.flush() session.add_all([ ProductVariant(id=5, name="Glenfiddich 15 Year Old Solera 70cl", brand="Glenfiddich", brand_id=1, product_line_id=11, quantity_ml=700, gtin="5010327000176", match_key="k5", attributes={}), ProductVariant(id=6, name="Glenfiddich 15 Year Old Solera 1L", brand="Glenfiddich", brand_id=1, product_line_id=11, quantity_ml=1000, match_key="k6", attributes={}, merged_into_id=5), ]) session.flush() yield session def entry(level, target, chosen=True, **kw) -> dict: out = dict(level=level, target=target, chosen=chosen, reason="test", image_source=ADMIN, licence=LICENCE, attribution="William Grant & Sons, supplied by the client") out.update(kw) return out @pytest.fixture def manifest(tmp_path: Path) -> tuple[dict, Path, Path]: folder = tmp_path / "folder" (folder / "Glenfiddich").mkdir(parents=True) (folder / "Glenfiddich" / "GLENFIDDICH_750ML_15YO_TRANSPARENT.png").write_bytes(png()) document = { "schema": "dfp-image-manifest/1", "entries": [ entry("line", "line:glenfiddich-15yo-solera", source_path="Glenfiddich/GLENFIDDICH_750ML_15YO_TRANSPARENT.png"), entry("variant", "variant:5", source_path="Glenfiddich/GLENFIDDICH_750ML_15YO_TRANSPARENT.png", gtin="5010327000176"), entry("brand", "brand:glenfiddich", source_url="https://images.example/logo.png", image_source="admin:pernod-ricard-via-adam"), entry("variant", "variant:9", source_url="https://images.example/9.jpg", chosen=False), ], } return document, folder, tmp_path / "uploads" class FakeFetch: def __init__(self, answers: dict[str, bytes | Exception]): self.answers = answers self.calls: list[str] = [] def __call__(self, url: str) -> bytes: self.calls.append(url) answer = self.answers[url] if isinstance(answer, Exception): raise answer return answer class TestStage: def test_files_and_addresses_become_derivatives_and_the_entries_learn_their_addresses(self, manifest): document, folder, uploads = manifest fetch = FakeFetch({"https://images.example/logo.png": png((1, 2, 3, 0))}) counts = cli_images.stage_manifest(document, folder=folder, uploads=uploads, fetch=fetch, sleep=lambda s: None) assert dict(counts) == {"line:stored": 1, "variant:stored": 1, "brand:stored": 1, "downloads": 1} line, variant, brand, unchosen = document["entries"] assert line["url"].startswith("/uploads/images/line/glenfiddich-15yo-solera-") and line["url"].endswith(".webp") assert line["thumb_url"] == line["url"].replace(".webp", "-thumb.webp") assert variant["url"].startswith("/uploads/images/variant/5-") assert brand["url"].startswith("/uploads/images/brand/glenfiddich-") assert "url" not in unchosen for e in (line, variant, brand): assert (uploads / "images" / e["url"].removeprefix("/uploads/images/")).exists() assert (uploads / "images" / e["thumb_url"].removeprefix("/uploads/images/")).exists() # The same file chosen at two levels carries the same content hash at two addresses. assert line["url"].rsplit("-", 1)[1] == variant["url"].rsplit("-", 1)[1] assert document["staged_at"] def test_a_second_run_makes_no_request_and_reads_nothing(self, manifest): document, folder, uploads = manifest fetch = FakeFetch({"https://images.example/logo.png": png()}) cli_images.stage_manifest(document, folder=folder, uploads=uploads, fetch=fetch, sleep=lambda s: None) again = FakeFetch({}) counts = cli_images.stage_manifest(document, folder=Path("/nowhere"), uploads=uploads, fetch=again, sleep=lambda s: None) assert dict(counts) == {"line:existing": 1, "variant:existing": 1, "brand:existing": 1} assert again.calls == [] def test_an_address_that_is_not_an_image_is_refused_and_the_entry_says_why(self, manifest): document, folder, uploads = manifest fetch = FakeFetch({"https://images.example/logo.png": ValueError("not an image: content type 'text/html'")}) counts = cli_images.stage_manifest(document, folder=folder, uploads=uploads, fetch=fetch, sleep=lambda s: None) brand = document["entries"][2] assert counts["brand:failed"] == 1 and "url" not in brand assert brand["stage_error"] == "not an image: content type 'text/html'" assert not (uploads / "images" / "brand").exists() def test_bytes_no_picture_library_can_read_are_refused_the_same_way(self, manifest): document, folder, uploads = manifest fetch = FakeFetch({"https://images.example/logo.png": b"please log in"}) counts = cli_images.stage_manifest(document, folder=folder, uploads=uploads, fetch=fetch, sleep=lambda s: None) assert counts["brand:failed"] == 1 assert document["entries"][2]["stage_error"].startswith("not an image Pillow can read") def test_check_reads_and_requests_nothing(self, manifest): document, folder, uploads = manifest fetch = FakeFetch({}) counts = cli_images.stage_manifest(document, folder=folder, uploads=uploads, check=True, fetch=fetch) assert dict(counts) == {"line:would_read": 1, "variant:would_read": 1, "brand:would_download": 1} assert fetch.calls == [] and not uploads.exists() and "staged_at" not in document def test_the_downloads_keep_one_second_between_requests(self, manifest): document, folder, uploads = manifest document["entries"].append(entry("brand", "brand:glenfiddich-old-spelling", source_url="https://images.example/two.png")) waits: list[float] = [] fetch = FakeFetch({"https://images.example/logo.png": png(), "https://images.example/two.png": png((9, 9, 9, 255))}) cli_images.stage_manifest(document, folder=folder, uploads=uploads, fetch=fetch, delay=1.0, sleep=waits.append) assert len(fetch.calls) == 2 and len(waits) == 1 and 0.9 < waits[0] <= 1.0 def test_the_command_check_prints_and_writes_nothing(self, manifest, tmp_path, capsys): document, folder, uploads = manifest path = tmp_path / "m.json" path.write_text(json.dumps(document)) before = path.read_text() args = argparse.Namespace(manifest=str(path), folder=str(folder), uploads=str(uploads), delay=0, check=True) assert cli_images.cmd_images_stage(args) == 0 out = capsys.readouterr().out assert "would_read=1" in out and "would_download=1" in out and "[check, nothing written]" in out assert path.read_text() == before def _staged(document, folder, uploads) -> dict: cli_images.stage_manifest(document, folder=folder, uploads=uploads, fetch=FakeFetch({"https://images.example/logo.png": png()}), sleep=lambda s: None) return document class TestImport: def test_each_staged_entry_lands_on_its_row_with_its_provenance(self, db, manifest): document, folder, uploads = manifest _staged(document, folder, uploads) counts = cli_images.import_manifest(db, document, uploads=uploads, now=NOW) assert dict(counts) == {"line:applied": 1, "variant:applied": 1, "brand:applied": 1} line, variant, brand = db.get(ProductLine, 11), db.get(ProductVariant, 5), db.get(Brand, 1) assert (line.image_url, line.image_level, line.image_source) == (document["entries"][0]["url"], "line", ADMIN) assert (variant.image_url, variant.thumb_url, variant.image_level) == (document["entries"][1]["url"], document["entries"][1]["thumb_url"], "variant") assert (brand.image_level, brand.image_source, brand.image_licence) == ("brand", "admin:pernod-ricard-via-adam", LICENCE) assert brand.image_attribution == "William Grant & Sons, supplied by the client" and brand.image_set_at == NOW def test_a_second_import_is_a_no_op(self, db, manifest): document, folder, uploads = manifest _staged(document, folder, uploads) cli_images.import_manifest(db, document, uploads=uploads, now=NOW) counts = cli_images.import_manifest(db, document, uploads=uploads) assert dict(counts) == {"line:unchanged": 1, "variant:unchanged": 1, "brand:unchanged": 1} def test_an_alias_slug_a_merged_id_and_a_barcode_each_find_the_live_row(self, db, manifest): document, folder, uploads = manifest _staged(document, folder, uploads) document["entries"][0]["target"] = "line:glenfiddich-15-solera" # the alias line document["entries"][1]["target"] = "variant:6" # merged into 5 document["entries"][2]["target"] = "brand:glenfiddich-old-spelling" # an alias brand counts = cli_images.import_manifest(db, document, uploads=uploads, now=NOW) assert dict(counts) == {"line:applied": 1, "variant:applied": 1, "brand:applied": 1} assert db.get(ProductLine, 11).image_url and db.get(ProductLine, 12).image_url is None assert db.get(ProductVariant, 5).image_url and db.get(ProductVariant, 6).image_url is None assert db.get(Brand, 1).image_url and db.get(Brand, 2).image_url is None document["entries"][1]["target"] = "variant:999" # an id that is gone; the barcode stands db.get(ProductVariant, 5).image_url = None counts = cli_images.import_manifest(db, document, uploads=uploads, now=NOW) assert counts["variant:applied"] == 1 and db.get(ProductVariant, 5).image_url def test_no_row_no_derivative_and_not_staged_are_counted_and_skipped(self, db, manifest): document, folder, uploads = manifest _staged(document, folder, uploads) document["entries"][0]["target"] = "line:nobody-has-this-line" document["entries"][1]["url"] = "/uploads/images/variant/5-00000000.webp" del document["entries"][2]["url"] counts = cli_images.import_manifest(db, document, uploads=uploads) assert dict(counts) == {"line:unresolved": 1, "variant:missing_derivative": 1, "brand:not_staged": 1} assert db.get(ProductLine, 11).image_url is None and db.get(ProductVariant, 5).image_url is None def test_a_supplied_picture_is_kept_against_a_fetched_one_and_counted(self, db, manifest): document, folder, uploads = manifest _staged(document, folder, uploads) cli_images.import_manifest(db, document, uploads=uploads, now=NOW) fetched = dict(document, entries=[dict(document["entries"][1], image_source=imagery.PUBLIC_OFF_BARCODE, url="/uploads/images/variant/5-deadbeef.webp")]) (uploads / "images" / "variant" / "5-deadbeef.webp").write_bytes(b"x") counts = cli_images.import_manifest(db, fetched, uploads=uploads) assert counts["variant:image_admin_kept"] == 1 and counts[imagery.IMAGE_ADMIN_KEPT] == 1 assert db.get(ProductVariant, 5).image_source == ADMIN def test_the_command_check_rolls_back(self, db, manifest, tmp_path, monkeypatch, capsys): document, folder, uploads = manifest _staged(document, folder, uploads) db.commit() # the seed is committed; only the command's own writes are rolled back path = tmp_path / "m.json" path.write_text(json.dumps(document)) class Factory: def __enter__(self): return db def __exit__(self, *exc): return False monkeypatch.setattr(cli_images, "SessionLocal", Factory) args = argparse.Namespace(manifest=str(path), uploads=str(uploads), check=True) assert cli_images.cmd_images_import(args) == 0 out = capsys.readouterr().out assert "applied=1" in out and "image_admin_kept=0" in out and "[check, nothing written]" in out db.expire_all() assert db.get(ProductLine, 11).image_url is None and db.get(Brand, 1).image_url is None class TestTargets: @pytest.mark.parametrize("entry_, want", [ ({"target": "brand:glenfiddich", "level": "brand"}, ("brand", "glenfiddich")), ({"target": "line:x-y", "level": "variant"}, ("line", "x-y")), ({"target": "variant:5", "level": "variant"}, ("variant", "5")), ({"target": "5", "level": "variant"}, ("variant", "5")), # a person typed the id ({"target": "hendricks", "level": "brand"}, ("brand", "hendricks")), # a person typed the slug ({"target": None, "level": "line"}, None), ({"target": "", "level": "line"}, None), ]) def test_a_target_names_its_table_and_token(self, entry_, want): assert cli_images.target_of(entry_) == want def test_the_cli_surface(self): parser = argparse.ArgumentParser() cli_images.register(parser.add_subparsers(dest="command")) args = parser.parse_args(["images", "stage", "--manifest", "m.json", "--folder", "f", "--uploads", "u", "--check"]) assert (args.func, args.check, args.delay) == (cli_images.cmd_images_stage, True, 1.0) args = parser.parse_args(["images", "import", "--manifest", "m.json", "--check"]) assert (args.func, args.uploads) == (cli_images.cmd_images_import, None)