"""The image sourcing list: ranked by airports, in the unit the client can source, for client.view. Pins what the client will act on: a product variant listed at one airport never appears (nothing to compare), the product variant carried by the most airports leads its group, the perfume category splits from the rest of beauty, a brand image is credited with every comparable product variant of the brand, and the reads admit the client's level while a stranger is refused. Since the picture cascade (Stream AW3): a product variant *has an image* when a bottle shows for it, its own or its product line's; a brand mark in the bottle's place does not count, but rides on the row as `image_level` with the credit, so the pill can say what shows. The per-level coverage counts every canonical row, and the wanted list per level names only rows with nothing to show anywhere in their cascade. """ from __future__ import annotations import pytest from app.models import RawRecord, AccountLevel, Base, Brand, Listing, Shop, ProductVariant, ProductLine, Retailer from app.services import accounts, image_ask, imagery from tests import _accounts as T from tests.kit import _env CATALOGUE = [Retailer.__table__, Shop.__table__, Brand.__table__, ProductLine.__table__, ProductVariant.__table__, Listing.__table__, RawRecord.__table__] ADMIN = "admin:test-supplier-via-adam" # (id, slug, name, image_url, image_source) BRANDS = [ (1, "johnnie-walker", "Johnnie Walker", "/uploads/images/brand/johnnie-walker-aaaaaaaa.webp", ADMIN), (2, "patron", "Patrón", None, None), (3, "lagavulin", "Lagavulin", "/uploads/images/brand/lagavulin-bbbbbbbb.webp", imagery.PUBLIC_COMMONS), (4, "chanel", "Chanel", None, None), (5, "lancome", "Lancôme", None, None), ] # (id, brand_id, slug, name, image_url, image_source) LINES = [ (1, 1, "johnnie-walker-blue-label", "Blue Label", "https://img/jw-blue-line.jpg", imagery.PUBLIC_OFF_BARCODE), (2, 4, "chanel-no-5", "No 5", None, None), ] # (id, brand_id, line_id, brand, name, vertical, category, image, source, airports) PRODUCTS = [ (1, 1, 1, "Johnnie Walker", "Johnnie Walker Blue Label 1L", "liquor", "Whisky", None, None, 4), (2, 1, None, "Johnnie Walker", "Johnnie Walker Black Label 1L", "liquor", "Whisky", "https://img/jw-black.jpg", imagery.PUBLIC_OFF_BARCODE, 3), (3, 2, None, "Patrón", "Patrón Silver 1L", "liquor", "Tequila & Mezcal", None, None, 2), (4, 3, None, "Lagavulin", "Lagavulin 16 700ml", "liquor", "Whisky", None, None, 1), (5, 4, 2, "Chanel", "Chanel No 5 EDP 100ml", "beauty", "Perfume", None, None, 3), (6, 4, None, "Chanel", "Chanel Hydra Beauty Cream 50ml", "beauty", "Skincare", None, None, 2), (7, 5, None, "Lancôme", "Lancôme Lash Idôle 8.5ml", "beauty", "Makeup", None, None, 2), ] @pytest.fixture def world(monkeypatch): T.fresh(monkeypatch) Base.metadata.drop_all(_env.engine, tables=CATALOGUE) Base.metadata.create_all(_env.engine, tables=CATALOGUE) T.person("rian") with _env.TestSessionLocal() as db: db.add(AccountLevel(name="admin", permissions=list(accounts.SEED_LEVELS["admin"]["permissions"]), assignable=[])) db.commit() T.person("adam", level="admin") with _env.TestSessionLocal() as db: retailer = Retailer(slug="shop", name="Shop") db.add(retailer) db.flush() shops = [Shop(retailer_id=retailer.id, code=f"S{n}", iata=f"A{n:02d}", name=f"Shop {n}", currency="EUR", visible=(n <= 2)) for n in range(1, 5)] db.add_all(shops) db.flush() for bid, slug, name, image, source in BRANDS: db.add(Brand(id=bid, slug=slug, name=name, image_url=image, thumb_url=image, image_source=source, image_level="brand" if image else None)) db.flush() for lid, bid, slug, name, image, source in LINES: db.add(ProductLine(id=lid, brand_id=bid, key=slug, slug=slug, name=name, image_url=image, thumb_url=image, image_source=source, image_level="line" if image else None)) db.flush() for pid, bid, lid, brand, name, vertical, category, image, source, airports in PRODUCTS: db.add(ProductVariant(id=pid, name=name, brand=brand, brand_id=bid, product_line_id=lid, vertical=vertical, category=category, image_url=image, image_source=source, image_level="variant" if image else None, match_key=f"k{pid}", attributes={})) db.flush() for loc in shops[:airports]: db.add(Listing(variant_id=pid, shop_id=loc.id, source_sku=f"sku{pid}-{loc.id}")) db.commit() yield Base.metadata.drop_all(_env.engine, tables=CATALOGUE) class TestRanking: def test_the_product_variant_at_the_most_airports_leads_and_a_single_airport_never_appears(self, world): with _env.TestSessionLocal() as db: out = image_ask.recommended(db, mode="product") alcohol = next(g for g in out["groups"] if g["key"] == "alcohol") assert [r["id"] for r in alcohol["rows"]] == [1, 2, 3] assert alcohol["comparable"] == 3 and alcohol["without_image"] == 1 assert alcohol["rows"][0]["codes"] == ["A01", "A02", "A03", "A04"] assert alcohol["rows"][0]["visible_airports"] == 2 assert alcohol["rows"][1]["has_image"] is True def test_a_row_says_what_shows_and_at_which_level(self, world): """A bottle from the product line counts as the variant's image; a brand mark does not, but the row still says it shows, so the pill reads the level and the credit.""" with _env.TestSessionLocal() as db: out = image_ask.recommended(db, mode="product") rows = {r["id"]: r for g in out["groups"] for r in g["rows"]} blue, black, patron = rows[1], rows[2], rows[3] assert blue["has_image"] is True and blue["image_level"] == "line" assert blue["image_source"] == imagery.PUBLIC_OFF_BARCODE and blue["image_credit"] == "Open Food Facts" assert black["image_level"] == "variant" assert patron["has_image"] is False and patron["image_level"] is None and patron["image_credit"] is None def test_a_brand_mark_alone_is_not_the_variants_image(self, world): """Lagavulin's Commons logo shows on its card, at brand level; the sourcing list still counts the variant as without an image, the rule structured data follows.""" with _env.TestSessionLocal() as db: rows = image_ask.product_variants(db, group=None, brand="Lagavulin", q=None, missing=False, page=1, per_page=10) everything = image_ask._ranked_product_variants(db, None, limit=None)[0] assert rows["rows"] == [] # one airport: never comparable assert all(r["id"] != 4 for r in everything) def test_perfume_splits_from_the_rest_of_beauty(self, world): with _env.TestSessionLocal() as db: out = image_ask.recommended(db, mode="product") by = {g["key"]: [r["id"] for r in g["rows"]] for g in out["groups"]} assert by["perfume"] == [5] assert by["cosmetics"] == [6, 7] assert out["summary"]["alcohol"] == {"comparable": 3, "with_image": 2} def test_a_brand_image_is_credited_with_every_comparable_product_variant_of_the_brand(self, world): with _env.TestSessionLocal() as db: out = image_ask.recommended(db, mode="brand") alcohol = next(g for g in out["groups"] if g["key"] == "alcohol") top = alcohol["rows"][0] assert top["brand"] == "Johnnie Walker" and top["product_variants"] == 2 and top["without_image"] == 0 assert top["airports"] == 4 and top["examples"][0] == "Johnnie Walker Blue Label 1L" assert top["has_image"] is True and top["image_source"] == ADMIN patron = next(r for r in alcohol["rows"] if r["brand"] == "Patrón") assert patron["has_image"] is False and patron["image_source"] is None assert alcohol["covers_product_variants"] == 3 def test_the_full_list_filters_and_pages(self, world): with _env.TestSessionLocal() as db: missing = image_ask.product_variants(db, group="alcohol", brand=None, q=None, missing=True, page=1, per_page=10) searched = image_ask.product_variants(db, group=None, brand=None, q="chanel", missing=False, page=1, per_page=10) paged = image_ask.product_variants(db, group=None, brand=None, q=None, missing=False, page=2, per_page=2) assert [r["id"] for r in missing["rows"]] == [3] assert [r["id"] for r in searched["rows"]] == [5, 6] and searched["total"] == 2 assert paged["total"] == 6 and [r["rank"] for r in paged["rows"]] == [3, 4] class TestCoverage: def test_every_canonical_row_is_counted_per_level_with_what_the_cascade_shows(self, world): with _env.TestSessionLocal() as db: levels = image_ask.coverage(db) summary = image_ask.summary(db) assert levels["brand"] == {"total": 5, "admin": 1, "public": 1, "without": 3, "none": 3} assert levels["line"] == {"total": 2, "admin": 0, "public": 1, "without": 1, "from_brand": 0, "none": 1} assert levels["variant"] == {"total": 7, "admin": 0, "public": 1, "without": 6, "from_line": 1, "from_brand": 1, "none": 4} assert summary["levels"] == levels def test_a_merged_variant_and_an_alias_are_not_counted(self, world): with _env.TestSessionLocal() as db: db.add(ProductVariant(id=8, name="Gone", brand="Patrón", brand_id=2, vertical="liquor", category="Tequila & Mezcal", merged_into_id=3, match_key="k8", attributes={})) db.add(Brand(id=6, slug="ysl", name="YSL", alias_of_id=4)) db.commit() levels = image_ask.coverage(db) assert levels["variant"]["total"] == 7 and levels["brand"]["total"] == 5 class TestWanted: def test_only_rows_with_nothing_to_show_anywhere_are_wanted_by_airports_first(self, world): with _env.TestSessionLocal() as db: out = image_ask.wanted(db, limit=10) # Variant 4 shows its brand's mark and variant 1 its line's bottle: neither is an ask. assert [r["id"] for r in out["variant"]] == [5, 6, 7, 3] assert out["variant"][0]["airports"] == 3 and out["variant"][0]["line"] == "No 5" # The Blue Label line has a picture; No 5 has none on it, its brand or its variant. assert [(r["slug"], r["brand"], r["airports"]) for r in out["line"]] == [("chanel-no-5", "Chanel", 3)] # Johnnie Walker and Lagavulin hold marks; the rest are asks, most airports first. assert [r["slug"] for r in out["brand"]] == ["chanel", "lancome", "patron"] assert out["brand"][0] == {"slug": "chanel", "name": "Chanel", "airports": 3, "product_variants": 2} def test_a_line_with_a_pictured_variant_is_not_wanted_and_a_hidden_row_is_left_out(self, world): with _env.TestSessionLocal() as db: line = db.get(ProductLine, 2) db.add(ProductVariant(id=9, name="Chanel No 5 EDP 50ml", brand="Chanel", brand_id=4, product_line_id=2, vertical="beauty", category="Perfume", image_url="https://img/no5.jpg", image_source=imagery.PUBLIC_OFF_NAME, image_level="variant", match_key="k9", attributes={})) db.get(Brand, 2).hidden = True db.commit() out = image_ask.wanted(db, limit=10) line.hidden = True db.commit() again = image_ask.wanted(db, limit=10) assert out["line"] == [] and "patron" not in [r["slug"] for r in out["brand"]] assert out["limit"] == 10 and again["line"] == [] def test_the_limit_is_bounded(self, world): with _env.TestSessionLocal() as db: assert image_ask.wanted(db, limit=10_000)["limit"] == image_ask.WANTED_MAX assert image_ask.wanted(db, limit=0)["limit"] == 1 class TestAccess: def test_the_client_reads_it_and_a_stranger_is_refused(self, world): client = T.client() assert client.get("/api/images/recommended").status_code in (401, 403) assert client.get("/api/images/wanted").status_code in (401, 403) T.login(client, "adam") r = client.get("/api/images/recommended?mode=brand") assert r.status_code == 200 and r.json()["mode"] == "brand" assert set(r.json()["summary"]["levels"]) == set(imagery.LEVELS) assert client.get("/api/images/products?group=perfume").status_code == 200 assert client.get("/api/images/brands").status_code == 200 w = client.get("/api/images/wanted?limit=5") assert w.status_code == 200 and set(w.json()) == {"limit", "brand", "line", "variant"} assert client.get("/api/images/wanted?limit=0").status_code == 422 assert client.get("/api/images/recommended?mode=logo").status_code == 422