"""Pictures at three levels (Stream AW3.2): the cascade, the tier, the one writer, the credit and the store. Pinned: `resolve` shows the product variant's picture over the line's over the brand's, with the level exposed, so a reader can style a brand mark as one and keep it out of a product's structured data; `image_tier` returns every one of the seven values the featured selection reads; `set_image` keeps a supplied picture against a fetched one and lets a supplied one replace a fetched one, and a re-run of the same picture is a no-op; `credit` prints a line for every word in the vocabulary; `store` writes both WebP derivatives from a 4 px RGBA PNG (alpha kept) and from a CMYK JPEG (converted), the original kept beside them and never served; the merge inheritance goes through the writer, so a survivor's supplied picture survives a merge. What the wrong answer costs: a brand logo published as a bottle's image in JSON-LD, or a client's own bottle shot silently replaced by a community photo of the wrong bottle. """ from __future__ import annotations import io from datetime import UTC, datetime from types import SimpleNamespace import pytest from PIL import Image from app.services import imagery def row(url=None, source=None, level=None, gtin=None, **kw): return SimpleNamespace(image_url=url, thumb_url=(url + "-t") if url else None, image_source=source, image_level=level, image_licence=kw.get("licence"), image_attribution=kw.get("attribution"), image_set_at=None, gtin=gtin) ADMIN = "admin:william-grant-via-adam" class TestVocabulary: @pytest.mark.parametrize("source", [ADMIN, "admin:pernod-ricard-via-adam", imagery.PUBLIC_OFF_BARCODE, imagery.PUBLIC_OFF_NAME, imagery.PUBLIC_COMMONS]) def test_every_word_is_valid(self, source): assert imagery.valid_source(source) @pytest.mark.parametrize("source", [None, "", "barcode", "Open Food Facts (barcode)", "admin:", "admin:With Caps", "public:retailer", "admin:a--b"]) def test_anything_else_is_not(self, source): assert not imagery.valid_source(source) def test_the_legacy_strings_move_to_their_successors(self): assert imagery.normalise_source("Open Food Facts (barcode)") == imagery.PUBLIC_OFF_BARCODE assert imagery.normalise_source("Open Food Facts (name)") == imagery.PUBLIC_OFF_NAME assert imagery.normalise_source("Open Food Facts", has_gtin=True) == imagery.PUBLIC_OFF_BARCODE assert imagery.normalise_source("Open Food Facts", has_gtin=False) == imagery.PUBLIC_OFF_NAME assert imagery.normalise_source(ADMIN) == ADMIN assert imagery.normalise_source("something else") == "something else" def test_credit_for_every_value(self): assert imagery.credit(ADMIN) == "William Grant" assert imagery.credit("admin:pernod-ricard-via-adam") == "Pernod Ricard" assert imagery.credit("admin:bruichladdich-via-adam") == "Bruichladdich" assert imagery.credit(imagery.PUBLIC_OFF_BARCODE) == "Open Food Facts" assert imagery.credit(imagery.PUBLIC_OFF_NAME) == "Open Food Facts" assert imagery.credit(imagery.PUBLIC_COMMONS) == "Wikimedia Commons" assert imagery.credit("Open Food Facts (barcode)") == "Open Food Facts" assert imagery.credit(None) == "" assert imagery.credit("") == "" class TestCascade: def test_variant_over_line_over_brand_with_the_level_exposed(self): v = row("https://x/v.webp", imagery.PUBLIC_OFF_BARCODE, "variant") line = row("/uploads/images/line/glenfiddich-12-abcd1234.webp", ADMIN, "line") brand = row("/uploads/images/brand/glenfiddich-abcd1234.webp", "admin:pernod-ricard-via-adam", "brand") assert imagery.resolve(v, line, brand).level == "variant" assert imagery.resolve(row(), line, brand).level == "line" picture = imagery.resolve(row(), row(), brand) assert (picture.level, picture.url, picture.thumb_url, picture.credit) == ( "brand", brand.image_url, brand.image_url + "-t", "Pernod Ricard") assert imagery.resolve(row(), row(), row()) is None assert imagery.resolve(None, None, None) is None def test_a_row_written_before_the_level_existed_is_read_at_its_tables_level(self): # The state between the deploy and `backfill image_sources`: a product variant's Open # Food Facts photo with no level must keep showing. v = row("https://x/v.jpg", "Open Food Facts (barcode)") picture = imagery.resolve(v, None, None) assert (picture.level, picture.credit) == ("variant", "Open Food Facts") def test_the_level_is_the_rows_word_not_the_tables(self): # A product variant row could hold a picture that depicts only its brand; the reader # is told so, whatever table the row sits in. v = row("/uploads/images/brand/x-abcd1234.webp", ADMIN, "brand") assert imagery.resolve(v, None, None).level == "brand" class TestTier: def test_all_seven_values(self): assert imagery.image_tier(None, None, None) == 0 assert imagery.image_tier(row(), row(), row("u", imagery.PUBLIC_COMMONS, "brand")) == 1 assert imagery.image_tier(row(), row("u", imagery.PUBLIC_OFF_BARCODE, "line"), row()) == 2 assert imagery.image_tier(row("u", imagery.PUBLIC_OFF_NAME, "variant"), row(), row()) == 3 assert imagery.image_tier(row(), row(), row("u", ADMIN, "brand")) == 4 assert imagery.image_tier(row(), row("u", ADMIN, "line"), row()) == 5 assert imagery.image_tier(row("u", ADMIN, "variant"), row(), row()) == 6 def test_the_tier_is_the_shown_pictures(self): # The cascade shows the variant's own photo even when the brand holds a supplied # mark; the tier says what is shown, so a card is ranked by what a shopper sees. assert imagery.image_tier(row("u", imagery.PUBLIC_OFF_BARCODE, "variant"), row(), row("u", ADMIN, "brand")) == 3 class TestSetImage: def test_a_fetched_picture_never_replaces_a_supplied_one(self): r = row("/uploads/images/variant/1-abcd1234.webp", ADMIN, "variant") outcome = imagery.set_image(r, url="https://off/x.jpg", thumb_url="https://off/x_t.jpg", source=imagery.PUBLIC_OFF_BARCODE, level="variant") assert outcome == imagery.IMAGE_ADMIN_KEPT assert (r.image_url, r.image_source) == ("/uploads/images/variant/1-abcd1234.webp", ADMIN) def test_a_supplied_picture_replaces_a_fetched_one_with_its_record(self): r = row("https://off/x.jpg", "Open Food Facts (barcode)", "variant") when = datetime(2026, 9, 19, 12, 0, tzinfo=UTC) outcome = imagery.set_image(r, url="/uploads/images/variant/1-abcd1234.webp", thumb_url="/uploads/images/variant/1-abcd1234-thumb.webp", source=ADMIN, level="variant", licence="Brand-supplied", attribution="William Grant & Sons", set_at=when) assert outcome == imagery.APPLIED assert (r.image_url, r.thumb_url, r.image_source, r.image_level, r.image_licence, r.image_attribution, r.image_set_at) == ( "/uploads/images/variant/1-abcd1234.webp", "/uploads/images/variant/1-abcd1234-thumb.webp", ADMIN, "variant", "Brand-supplied", "William Grant & Sons", when) def test_the_same_picture_again_is_a_no_op(self): r = row(None) assert imagery.set_image(r, url="u", thumb_url="t", source=ADMIN, level="line") == imagery.APPLIED first = r.image_set_at assert first is not None and first.tzinfo is not None assert imagery.set_image(r, url="u", thumb_url="t", source=ADMIN, level="line") == imagery.UNCHANGED assert r.image_set_at == first def test_a_public_picture_replaces_a_public_one_and_a_legacy_source_is_normalised(self): r = row("https://off/old.jpg", "Open Food Facts (name)", "variant", gtin="0083664990412") assert imagery.set_image(r, url="https://off/new.jpg", thumb_url=None, source="Open Food Facts (barcode)", level="variant") == imagery.APPLIED assert (r.image_url, r.thumb_url, r.image_source) == ("https://off/new.jpg", "https://off/new.jpg", imagery.PUBLIC_OFF_BARCODE) def test_the_vocabulary_and_the_levels_are_enforced(self): with pytest.raises(ValueError): imagery.set_image(row(), url="u", thumb_url=None, source="retailer", level="variant") with pytest.raises(ValueError): imagery.set_image(row(), url="u", thumb_url=None, source=ADMIN, level="product") with pytest.raises(ValueError): imagery.set_image(row(), url="", thumb_url=None, source=ADMIN, level="brand") def _png_rgba_4px() -> bytes: image = Image.new("RGBA", (4, 4), (255, 0, 0, 0)) image.putpixel((0, 0), (0, 0, 255, 255)) out = io.BytesIO() image.save(out, format="PNG") return out.getvalue() def _jpeg_cmyk(width=1600, height=400) -> bytes: image = Image.new("CMYK", (width, height), (0, 255, 255, 0)) out = io.BytesIO() image.save(out, format="JPEG") return out.getvalue() class TestStore: def test_a_4px_rgba_png_writes_both_derivatives_and_keeps_alpha(self, tmp_path): stored = imagery.store(_png_rgba_4px(), "line", "glenfiddich-12", uploads=tmp_path) sha8 = imagery.sha8_of(_png_rgba_4px()) assert stored.original_path == tmp_path / "originals" / "line" / f"glenfiddich-12-{sha8}.png" assert stored.original_path.read_bytes() == _png_rgba_4px() assert stored.url == f"/uploads/images/line/glenfiddich-12-{sha8}.webp" assert stored.thumb_url == f"/uploads/images/line/glenfiddich-12-{sha8}-thumb.webp" full, thumb = imagery.derivative_paths("line", "glenfiddich-12", sha8, uploads=tmp_path) for path in (full, thumb): with Image.open(path) as derived: assert derived.format == "WEBP" assert derived.size == (4, 4) # never upscaled assert derived.mode == "RGBA" assert derived.getpixel((3, 3))[3] == 0 # the transparent corner stays transparent def test_a_cmyk_jpeg_is_converted_and_bounded_at_1200_and_400(self, tmp_path): stored = imagery.store(_jpeg_cmyk(), "variant", 1234, uploads=tmp_path) assert stored.original_path.suffix == ".jpg" full, thumb = imagery.derivative_paths("variant", 1234, stored.sha8, uploads=tmp_path) with Image.open(full) as derived: assert (derived.mode, derived.size) == ("RGB", (1200, 300)) with Image.open(thumb) as derived: assert (derived.mode, derived.size) == ("RGB", (400, 100)) def test_the_same_bytes_land_at_the_same_address_and_are_not_re_encoded(self, tmp_path): first = imagery.store(_png_rgba_4px(), "brand", "glenfiddich", uploads=tmp_path) full, _ = imagery.derivative_paths("brand", "glenfiddich", first.sha8, uploads=tmp_path) stamp = full.stat().st_mtime_ns second = imagery.store(_png_rgba_4px(), "brand", "glenfiddich", uploads=tmp_path) assert second == first assert full.stat().st_mtime_ns == stamp def test_bytes_that_are_not_an_image_are_refused(self, tmp_path): with pytest.raises(ValueError): imagery.store(b"blocked", "brand", "x", uploads=tmp_path) assert not (tmp_path / "originals").exists() with pytest.raises(ValueError): imagery.derivative_paths("product", "x", "abcd1234", uploads=tmp_path)