"""Wider legitimate coverage (Stream AW3.4): the Open Food Facts name search reads the payload
it is given, a Commons logo is accepted only when every link in the chain agrees, and a product
line's picture is promoted only from its representative variant.
The record that failed: `search_by_name` read `payload.get("product_variants")`. Open Food Facts
answers its search under `products`; the K1 rename swept the key with everything else, and from
that deploy the name fallback found nothing for every product variant without a barcode match,
silently, with no error anywhere, while `images fetch` reported `name=0` as if the search had
simply missed. The test below replays a search answer in the shape Open Food Facts serves (five
rows under `products`, the 18 Year Old listed before the 12) and pins both halves: the 12 is
found now, and the 18 is still refused for it (`numbers_agree`).
The logo rules pinned: a label off by one word ("Glenfiddich Distillery" for "Glenfiddich") is
refused; an entity whose `P31` is outside the allow set (a human, a village) is refused however
well its label matches; a non-free licence (CC BY-NC-SA, fair use) is refused; two entities that
both pass is a stop; `P18` stands in when `P154` is absent. The promotion pinned: the line takes
the representative's photo (most comparison units, then lowest id) and a line whose only photo
sits on another size stays empty. What the wrong answer costs: a village's coat of arms, or a
person's portrait, published as a brand's mark on every card, or a 1 litre bottle's photo
standing for a line whose card names the 70 cl.
"""
from __future__ import annotations
import argparse
import io
import urllib.parse
from datetime import UTC, datetime, timedelta
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 Award, Base, Brand, Listing, Place, PriceObservation, ProductLine, ProductVariant, Redirect, Retailer, Shop, ShopPlace, Source
from app.services import catalog_queries as cq
from app.services import images, images_public as ip, imagery
from app.services.collectors import robots
from app.services.collectors.fetch import SourceBlocked
NOW = datetime(2026, 9, 19, 21, 0, tzinfo=UTC)
TABLES = [Brand.__table__, ProductLine.__table__, ProductVariant.__table__, Retailer.__table__, Shop.__table__,
Listing.__table__, PriceObservation.__table__, Place.__table__, ShopPlace.__table__, Award.__table__, Source.__table__,
Redirect.__table__]
def png(colour=(10, 20, 30, 255)) -> bytes:
out = io.BytesIO()
Image.new("RGBA", (4, 4), colour).save(out, format="PNG")
return out.getvalue()
# --- Open Food Facts: the search payload -------------------------------------------------------------
OFF_IMG = "https://images.openfoodfacts.org/images/products/501/032/700/{tail}/front_en.4.400.jpg"
#: The shape `search.pl?json=1` serves: `count`, `page`, `page_size`, `skip` and the rows under
#: `products`. Five candidates for "glenfiddich 12", the 18 first.
OFF_SEARCH_PAYLOAD = {
"count": 5, "page": 1, "page_count": 1, "page_size": 5, "skip": 0,
"products": [
{"product_name": "Glenfiddich 18 Year Old Small Batch Reserve", "brands": "Glenfiddich",
"image_front_url": OFF_IMG.format(tail="0183"), "image_front_small_url": OFF_IMG.format(tail="0183").replace("400", "200")},
{"product_name": "Glenfiddich 15 Year Old Solera", "brands": "Glenfiddich",
"image_front_url": OFF_IMG.format(tail="0176"), "image_front_small_url": OFF_IMG.format(tail="0176").replace("400", "200")},
{"product_name": "Glenlivet 12 Year Old", "brands": "The Glenlivet",
"image_front_url": OFF_IMG.format(tail="0121"), "image_front_small_url": OFF_IMG.format(tail="0121").replace("400", "200")},
{"product_name": "Glenfiddich 12 Year Old", "brands": "Glenfiddich"}, # no photo: skipped
{"product_name": "Glenfiddich 12 Year Old Single Malt Scotch Whisky", "brands": "Glenfiddich",
"image_front_url": OFF_IMG.format(tail="0129"), "image_front_small_url": OFF_IMG.format(tail="0129").replace("400", "200")},
],
}
class TestOpenFoodFactsSearch:
def test_the_search_reads_products_and_refuses_the_18_for_the_12(self, monkeypatch):
seen: list[str] = []
def fake_fetch(url, timeout=25):
seen.append(url)
return OFF_SEARCH_PAYLOAD
monkeypatch.setattr(images, "_fetch_json", fake_fetch)
result = images.search_by_name("Glenfiddich", "Glenfiddich 12 Year Old 70cl")
assert result.image_url == OFF_IMG.format(tail="0129"), "the 12 with a photo, not the 18 listed first"
assert result.thumb_url == OFF_IMG.format(tail="0129").replace("400", "200")
assert len(seen) == 1 and "search_terms=" in seen[0]
def test_the_old_key_found_nothing_and_that_is_the_bug(self, monkeypatch):
"""Under the swept key the same answer yields nothing: what shipped after K1."""
monkeypatch.setattr(images, "_fetch_json", lambda url, timeout=25: {"product_variants": OFF_SEARCH_PAYLOAD["products"]})
assert images.search_by_name("Glenfiddich", "Glenfiddich 12 Year Old 70cl") == images.ImageResult(None, None)
def test_the_18_alone_is_refused_for_the_12(self, monkeypatch):
monkeypatch.setattr(images, "_fetch_json", lambda url, timeout=25: {"products": OFF_SEARCH_PAYLOAD["products"][:1]})
assert images.search_by_name("Glenfiddich", "Glenfiddich 12 Year Old 70cl") == images.ImageResult(None, None)
# --- Wikidata and Commons: the fake service --------------------------------------------------------
def claim(value, rank="normal"):
return {"rank": rank, "mainsnak": {"datavalue": {"value": value}}}
def entity(qid, label, kinds, logo=None, image=None):
claims = {"P31": [claim({"id": k}) for k in kinds]}
if logo:
claims["P154"] = [claim(logo)]
if image:
claims["P18"] = [claim(image)]
return {"id": qid, "labels": {"en": {"value": label}}, "claims": claims}
def imageinfo(file, licence="CC BY-SA 4.0", artist="Someone", mime="image/svg+xml"):
meta = {"LicenseShortName": {"value": licence}, "LicenseUrl": {"value": "https://creativecommons.org/licenses/by-sa/4.0"},
"Artist": {"value": artist}, "Credit": {"value": "Own work"}}
return {"query": {"pages": {"123": {"title": "File:" + file, "imageinfo": [{
"url": f"https://upload.wikimedia.org/wikipedia/commons/a/ab/{file}",
"thumburl": f"https://upload.wikimedia.org/wikipedia/commons/thumb/a/ab/{file}/1200px-{file}.png",
"descriptionurl": f"https://commons.wikimedia.org/wiki/File:{file}", "mime": mime, "extmetadata": meta}]}}}}
class FakeWiki:
"""Answers `wbsearchentities`, `wbgetentities` and Commons `imageinfo` from what a test seeds,
and records every address asked."""
def __init__(self, hits, entities, files):
self.hits, self.entities, self.files = hits, entities, files
self.calls: list[dict] = []
def __call__(self, url):
params = {k: v[0] for k, v in urllib.parse.parse_qs(urllib.parse.urlsplit(url).query).items()}
self.calls.append(params)
if params["action"] == "wbsearchentities":
return {"search": [{"id": e["id"], "label": e["labels"]["en"]["value"]} for e in self.hits]}
if params["action"] == "wbgetentities":
return {"entities": {q: self.entities[q] for q in params["ids"].split("|") if q in self.entities}}
if params["action"] == "query":
name = params["titles"].removeprefix("File:")
return self.files.get(name, {"query": {"pages": {"-1": {"title": params["titles"], "missing": ""}}}})
raise AssertionError(params)
GLEN = entity("Q1479716", "Glenfiddich", ["Q431289"], logo="Glenfiddich logo.svg")
GLEN_DISTILLERY = entity("Q1479717", "Glenfiddich Distillery", ["Q1207302"], logo="Glenfiddich Distillery.svg")
GLEN_FILES = {"Glenfiddich logo.svg": imageinfo("Glenfiddich logo.svg")}
def wiki(hits=(GLEN,), entities=None, files=None) -> FakeWiki:
entities = entities or {e["id"]: e for e in hits}
return FakeWiki(list(hits), entities, files if files is not None else GLEN_FILES)
#: What `www.wikidata.org/robots.txt` and `commons.wikimedia.org/robots.txt` both publish for
#: everyone (the lines that matter; the file is long): an Allow for one mobile action and a
#: Disallow for the whole of `/w/`, which is where `api.php` lives. Under the project's policy a
#: longer Allow does not out-lawyer a Disallow, so `/w/api.php?action=wbsearchentities` is a no.
WIKIMEDIA_ROBOTS = """\
User-agent: *
Allow: /w/api.php?action=mobileview&
Allow: /w/load.php?
Disallow: /w/
Disallow: /api/
Disallow: /trap/
"""
def robots_saying(text: str):
"""A robots reader that answers every host with `text`, recording the hosts asked."""
asked: list[str] = []
def read(origin: str) -> robots.Robots:
asked.append(origin)
return robots.parse(text, host=origin)
read.asked = asked # type: ignore[attr-defined]
return read
NO_RULES = robots_saying("")
class TestWikidataLogo:
def test_an_exact_label_a_brand_kind_and_a_free_licence_are_accepted(self):
fake = wiki(hits=(GLEN_DISTILLERY, GLEN))
logo = ip.wikidata_brand_logo("Glenfiddich", fetch_json=fake)
assert isinstance(logo, ip.Logo)
assert (logo.entity, logo.file, logo.licence, logo.mime) == ("Q1479716", "File:Glenfiddich logo.svg", "CC BY-SA 4.0", "image/svg+xml")
assert logo.download_url.endswith("/1200px-Glenfiddich logo.svg.png"), "the Commons render, never the SVG itself"
assert logo.attribution == "Wikimedia Commons: File:Glenfiddich logo.svg by Someone, https://commons.wikimedia.org/wiki/File:Glenfiddich logo.svg"
assert [c["action"] for c in fake.calls] == ["wbsearchentities", "wbgetentities", "query"]
assert fake.calls[1]["ids"] == "Q1479716", "only the entity whose label matched is fetched"
def test_a_label_off_by_one_word_is_refused(self):
fake = wiki(hits=(GLEN_DISTILLERY,))
found = ip.wikidata_brand_logo("Glenfiddich", fetch_json=fake)
assert found == ip.Refusal(ip.LABEL_MISMATCH, "Q1479717 'Glenfiddich Distillery'")
assert len(fake.calls) == 1, "no second request for a name that does not match"
def test_a_kind_outside_the_set_is_refused_however_well_the_label_matches(self):
person = entity("Q999", "Glenfiddich", ["Q5"], image="A person.jpg") # a human
found = ip.wikidata_brand_logo("Glenfiddich", fetch_json=wiki(hits=(person,)))
assert found == ip.Refusal(ip.KIND_OUTSIDE_SET, "Q5")
def test_a_non_free_licence_is_refused(self):
for licence in ("CC BY-NC-SA 2.0", "CC BY-ND 4.0", "Fair use", "Copyrighted", ""):
files = {"Glenfiddich logo.svg": imageinfo("Glenfiddich logo.svg", licence=licence)}
found = ip.wikidata_brand_logo("Glenfiddich", fetch_json=wiki(files=files))
assert found == ip.Refusal(ip.LICENCE_NOT_FREE, licence or "no licence"), licence
def test_two_entities_that_both_pass_are_a_stop(self):
twin = entity("Q2", "Glenfiddich", ["Q167270"], logo="Other.svg")
found = ip.wikidata_brand_logo("Glenfiddich", fetch_json=wiki(hits=(GLEN, twin)))
assert found == ip.Refusal(ip.AMBIGUOUS, "Q1479716, Q2")
def test_p18_stands_in_when_p154_is_absent_and_nothing_at_all_is_no_logo(self):
with_image = entity("Q1", "Moët & Chandon", ["Q4830453"], image="Moet.jpg")
files = {"Moet.jpg": imageinfo("Moet.jpg", licence="Public domain", artist=None, mime="image/jpeg")}
logo = ip.wikidata_brand_logo("MOET & CHANDON", fetch_json=wiki(hits=(with_image,), files=files))
assert isinstance(logo, ip.Logo) and logo.file == "File:Moet.jpg" and logo.licence == "Public domain"
assert logo.attribution == "Wikimedia Commons: File:Moet.jpg, https://commons.wikimedia.org/wiki/File:Moet.jpg"
bare = entity("Q1", "Moët & Chandon", ["Q4830453"])
assert ip.wikidata_brand_logo("Moët & Chandon", fetch_json=wiki(hits=(bare,))) == ip.Refusal(ip.NO_LOGO, "Q1")
def test_no_entity_a_missing_file_and_no_answer_are_refusals(self):
assert ip.wikidata_brand_logo("Nobody", fetch_json=wiki(hits=())) == ip.Refusal(ip.NO_ENTITY)
assert ip.wikidata_brand_logo("Glenfiddich", fetch_json=wiki(files={})) == ip.Refusal(ip.NO_FILE_INFO, "Glenfiddich logo.svg")
assert ip.wikidata_brand_logo("Glenfiddich", fetch_json=lambda url: None) == ip.Refusal(ip.NO_ANSWER, "wbsearchentities")
def test_a_deprecated_logo_claim_is_skipped_and_a_preferred_one_wins(self):
claims = {"P31": [claim({"id": "Q431289"})],
"P154": [claim("Old.svg", rank="deprecated"), claim("Normal.svg"), claim("Preferred.svg", rank="preferred")]}
e = {"id": "Q1", "labels": {"en": {"value": "Glenfiddich"}}, "claims": claims}
files = {"Preferred.svg": imageinfo("Preferred.svg")}
logo = ip.wikidata_brand_logo("Glenfiddich", fetch_json=wiki(hits=(e,), files=files))
assert isinstance(logo, ip.Logo) and logo.file == "File:Preferred.svg"
class TestRules:
@pytest.mark.parametrize("a, b, same", [
("Moët & Chandon", "MOET & CHANDON", True), ("Glenfiddich", "Glenfiddich Distillery", False),
("Rémy Martin", "Remy Martin", True), ("Chanel", "Chanel S.A.", False), ("", None, True),
])
def test_fold(self, a, b, same):
assert (ip.fold(a) == ip.fold(b)) is same
@pytest.mark.parametrize("licence, free", [
("CC BY-SA 4.0", True), ("CC BY 2.0", True), ("CC BY-SA 3.0", True), ("CC0", True), ("CC0 1.0", True),
("Public domain", True), ("PD-textlogo", True), ("cc by-sa 4.0", True),
("CC BY-NC-SA 2.0", False), ("CC BY-NC 4.0", False), ("CC BY-ND 3.0", False), ("Fair use", False),
("Copyrighted", False), ("GFDL", False), ("", False), (None, False),
])
def test_free_licence(self, licence, free):
assert ip.free_licence(licence) is free
def test_plain_text_strips_the_html_commons_wraps_an_artist_in(self):
assert ip.plain_text("X&Y") == "X&Y"
assert ip.plain_text("") is None and ip.plain_text("
") is None
def test_the_pace_keeps_one_second_between_requests_across_hosts(self):
clock, waits = [0.0], []
def sleep(s):
waits.append(s)
clock[0] += s
pace = ip.Pace(1.0, sleep=sleep, clock=lambda: clock[0])
calls = []
f = pace.paced(lambda url: calls.append(url))
f("a"); clock[0] += 0.25; f("b"); clock[0] += 1.5; f("c")
assert calls == ["a", "b", "c"] and pace.requests == 3
assert waits == [0.75], "the second request waited the rest of the second; the third came late and waited nothing"
# --- the database: brands, lines, listings ----------------------------------------------------------
@pytest.fixture
def db(monkeypatch):
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
Base.metadata.create_all(engine, tables=TABLES)
monkeypatch.setattr(cq, "blocked_shop_ids", lambda db: []) # verification lives in JSONB tables SQLite cannot build
factory = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False, future=True)
with factory() as s:
s.add_all([Brand(id=1, slug="glenfiddich", name="Glenfiddich"), Brand(id=2, slug="chanel", name="Chanel"),
Brand(id=3, slug="ysl", name="YSL", alias_of_id=2), Retailer(id=1, slug="wdf", name="World Duty Free")])
s.flush()
s.add_all([Shop(id=1, retailer_id=1, code="LHR1", iata="LHR", name="Heathrow T5", city="London", currency="GBP", visible=True),
Shop(id=2, retailer_id=1, code="CDG1", iata="CDG", name="Paris CDG", city="Paris", currency="EUR", visible=True),
Shop(id=3, retailer_id=1, code="HID1", iata="HID", name="Hidden", city="Nowhere", currency="EUR", visible=False)])
s.add_all([ProductLine(id=11, brand_id=1, key="12yo", name="12 Year Old", slug="glenfiddich-12yo"),
ProductLine(id=12, brand_id=1, key="15yo solera", name="15 Year Old Solera", slug="glenfiddich-15yo-solera"),
ProductLine(id=13, brand_id=1, key="18yo", name="18 Year Old", slug="glenfiddich-18yo"),
ProductLine(id=14, brand_id=2, key="no 5", name="No 5", slug="chanel-no-5")])
s.flush()
def variant(vid, line, name, ml, gtin=None, **kw):
s.add(ProductVariant(id=vid, name=name, brand="Glenfiddich", brand_id=1, product_line_id=line, quantity_ml=ml,
gtin=gtin, match_key=f"k{vid}", attributes={}, **kw))
off = dict(image_url=OFF_IMG.format(tail="0129"), thumb_url=OFF_IMG.format(tail="0129").replace("400", "200"),
image_source=imagery.PUBLIC_OFF_BARCODE, image_level="variant", image_licence=images.LICENCE, image_attribution=images.ATTRIBUTION)
# Line 11: the 70cl (id 21) at two airports carries the photo; the 1L (id 22) at one does not.
variant(21, 11, "Glenfiddich 12 Year Old 70cl", 700, gtin="5010327000121", **off)
variant(22, 11, "Glenfiddich 12 Year Old 1L", 1000)
# Line 12: only the 1L (id 24, one airport) has a photo; the 70cl (id 23, two airports) is the representative.
variant(23, 12, "Glenfiddich 15 Year Old Solera 70cl", 700)
variant(24, 12, "Glenfiddich 15 Year Old Solera 1L", 1000, **dict(off, image_url=OFF_IMG.format(tail="0176")))
# Line 13: the representative's picture was supplied, not fetched.
variant(25, 13, "Glenfiddich 18 Year Old 70cl", 700, image_url="/uploads/images/variant/25-abcd1234.webp",
thumb_url="/uploads/images/variant/25-abcd1234-thumb.webp", image_source="admin:william-grant-via-adam", image_level="variant")
# Line 14: a variant priced only at a hidden shop: no representative.
variant(26, 14, "No 5 EDP 100ml", 100, **off)
s.flush()
lid = 0
for vid, shop_id in ((21, 1), (21, 2), (22, 1), (23, 1), (23, 2), (24, 1), (25, 1), (26, 3)):
lid += 1
s.add(Listing(id=lid, variant_id=vid, shop_id=shop_id, source_sku=f"sku{lid}"))
s.flush()
s.add(PriceObservation(listing_id=lid, price=40, currency="GBP", price_usd=50.0, price_type="duty_free",
in_stock=True, observed_at=NOW - timedelta(days=1)))
s.flush()
yield s
class TestPromotion:
def test_the_line_takes_its_representatives_photo_and_other_sizes_stay_empty(self, db):
counts = ip.promote_lines(db, now=NOW)
assert dict(counts) == {"line:applied": 1, f"line:{ip.OTHER_SIZE_ONLY}": 1,
f"line:{ip.REPRESENTATIVE_NOT_OFF}": 1, f"line:{ip.NO_REPRESENTATIVE}": 1}
line = db.get(ProductLine, 11)
assert (line.image_url, line.thumb_url, line.image_source, line.image_level) == (
OFF_IMG.format(tail="0129"), OFF_IMG.format(tail="0129").replace("400", "200"), imagery.PUBLIC_OFF_BARCODE, "line")
assert (line.image_licence, line.image_attribution) == (images.LICENCE, images.ATTRIBUTION)
assert line.image_set_at.replace(tzinfo=UTC) == NOW # SQLite hands a naive datetime back
for empty in (12, 13, 14):
assert db.get(ProductLine, empty).image_url is None
assert dict(ip.promote_lines(db, now=NOW)) == {f"line:{ip.OTHER_SIZE_ONLY}": 1, f"line:{ip.REPRESENTATIVE_NOT_OFF}": 1,
f"line:{ip.NO_REPRESENTATIVE}": 1}, "a second run has nothing to set"
def test_the_representative_is_most_units_then_lowest_id(self, db):
reps = ip.representatives_of(db, [db.get(ProductLine, i) for i in (11, 12, 13, 14)])
assert {k: (v.id if v else None) for k, v in reps.items()} == {11: 21, 12: 23, 13: 25, 14: None}
# The line page agrees on the one it can render.
assert cq.get_product_line(db, "glenfiddich-15yo-solera").representative_variant_id == 23
def test_a_line_with_a_picture_is_never_visited(self, db):
line = db.get(ProductLine, 11)
line.image_url, line.image_source, line.image_level = "/uploads/images/line/x-00000000.webp", "admin:william-grant-via-adam", "line"
db.flush()
assert ip.promote_lines(db, now=NOW)["line:applied"] == 0 and line.image_source == "admin:william-grant-via-adam"
def test_limit(self, db):
assert sum(ip.promote_lines(db, limit=2, now=NOW).values()) == 2
class TestLogosCommand:
def test_the_brands_visited_are_canonical_without_a_picture_biggest_first(self, db):
brands = ip.brands_without_a_picture(db)
assert [b.slug for b in brands] == ["glenfiddich", "chanel"], "the alias is not a brand; the one with more variants first"
assert [b.slug for b in ip.brands_without_a_picture(db, limit=1)] == ["glenfiddich"]
assert [b.slug for b in ip.brands_without_a_picture(db, slug="chanel")] == ["chanel"]
assert ip.brands_without_a_picture(db, slug="ysl") == [] and ip.brands_without_a_picture(db, slug="nobody") == []
db.get(Brand, 1).image_url = "/uploads/images/brand/glenfiddich-00000000.webp"
db.flush()
assert [b.slug for b in ip.brands_without_a_picture(db)] == ["chanel"]
def test_check_looks_up_but_downloads_and_writes_nothing(self, db, tmp_path):
fake, downloads, lines = wiki(), [], []
counts = cli_images.logos_for_brands(db, [db.get(Brand, 1)], uploads=tmp_path, check=True, fetch_json=fake,
fetch_bytes=downloads.append, pace=ip.Pace(0), read_robots=NO_RULES, out=lines.append)
assert dict(counts) == {"brand:would_apply": 1, "requests": 5} and downloads == [], "three lookups and one robots.txt per host"
assert lines == ["glenfiddich: would apply File:Glenfiddich logo.svg (CC BY-SA 4.0; Q1479716; image/svg+xml)"]
assert db.get(Brand, 1).image_url is None and not tmp_path.exists() or not any(tmp_path.iterdir())
def test_an_accepted_logo_is_downloaded_stored_and_written_with_its_provenance(self, db, tmp_path):
fake, asked, lines = wiki(), [], []
def fetch_bytes(url):
asked.append(url)
return png()
counts = cli_images.logos_for_brands(db, [db.get(Brand, 1)], uploads=tmp_path, fetch_json=fake, fetch_bytes=fetch_bytes,
pace=ip.Pace(0), read_robots=NO_RULES, now=NOW, out=lines.append)
assert dict(counts) == {"brand:applied": 1, "requests": 7}, "three lookups, one download, three hosts' robots.txt"
assert asked == ["https://upload.wikimedia.org/wikipedia/commons/thumb/a/ab/Glenfiddich logo.svg/1200px-Glenfiddich logo.svg.png"]
brand = db.get(Brand, 1)
assert brand.image_url.startswith("/uploads/images/brand/glenfiddich-") and brand.thumb_url.endswith("-thumb.webp")
assert (brand.image_source, brand.image_level, brand.image_licence, brand.image_set_at) == (imagery.PUBLIC_COMMONS, "brand", "CC BY-SA 4.0", NOW)
assert brand.image_attribution == "Wikimedia Commons: File:Glenfiddich logo.svg by Someone, https://commons.wikimedia.org/wiki/File:Glenfiddich logo.svg"
assert (tmp_path / "images" / "brand" / Path(brand.image_url).name).exists()
assert lines == ["glenfiddich: applied File:Glenfiddich logo.svg (CC BY-SA 4.0)"]
def test_a_refusal_a_failed_download_and_a_supplied_picture_are_counted_and_leave_the_row(self, db, tmp_path):
glen, chanel = db.get(Brand, 1), db.get(Brand, 2)
glen.image_url, glen.image_source, glen.image_level = "/uploads/images/brand/glenfiddich-00000000.webp", "admin:pernod-ricard-via-adam", "brand"
def failing(url):
raise ValueError("not an image: content type 'text/html'")
counts = cli_images.logos_for_brands(db, [glen, chanel], uploads=tmp_path, fetch_json=wiki(), fetch_bytes=failing,
pace=ip.Pace(0), read_robots=NO_RULES, out=lambda s: None)
assert dict(counts) == {"brand:image_admin_kept": 1, imagery.IMAGE_ADMIN_KEPT: 1, "brand:refused:label_mismatch": 1, "requests": 6}, \
"three lookups for the supplied brand and no download; one for the name that did not match; two robots.txt"
assert glen.image_source == "admin:pernod-ricard-via-adam" and chanel.image_url is None
counts = cli_images.logos_for_brands(db, [chanel], uploads=tmp_path, fetch_json=wiki(hits=(entity("Q3", "Chanel", ["Q4830453"], logo="Glenfiddich logo.svg"),)),
fetch_bytes=failing, pace=ip.Pace(0), read_robots=NO_RULES, out=lambda s: None)
assert counts["brand:download_failed"] == 1 and chanel.image_url is None
def test_the_cli_surface(self):
parser = argparse.ArgumentParser()
cli_images.register(parser.add_subparsers(dest="command"))
args = parser.parse_args(["images", "logos", "--limit", "3", "--check"])
assert cli_images.logos_for_brands.__kwdefaults__["read_robots"] is robots.read, "the command reads the real robots.txt"
assert (args.func, args.limit, args.check, args.delay, args.brand, args.uploads) == (cli_images.cmd_images_logos, 3, True, 1.0, None, None)
args = parser.parse_args(["images", "lines", "--check"])
assert (args.func, args.limit, args.check) == (cli_images.cmd_images_lines, None, True)
class TestRobotsBeforeTheFirstRequest:
"""The record that failed: the `images logos --limit 3 --check` probe of 19 Sep 2026 made
eight requests to `/w/api.php` on `www.wikidata.org` and `commons.wikimedia.org` without
reading either host's robots.txt, and both publish `Disallow: /w/` for everyone
(`WIKIMEDIA_ROBOTS`). The collectors re-read robots.txt on every run and treat any matching
Disallow as a no; the image code read none. What it cost: eight requests a host had said no
to, and a full `images logos` over every brand handed to rian as ready to run. Now the
guard reads each host's rules before its first request and the run stops there."""
def test_the_wikimedia_rule_refuses_the_api_before_any_request_is_made(self, db, tmp_path):
fake, lines, read = wiki(), [], robots_saying(WIKIMEDIA_ROBOTS)
counts = cli_images.logos_for_brands(db, [db.get(Brand, 1), db.get(Brand, 2)], uploads=tmp_path, check=True, fetch_json=fake,
fetch_bytes=lambda url: png(), pace=ip.Pace(0), read_robots=read, out=lines.append)
assert fake.calls == [], "no API request was made"
assert read.asked == ["https://www.wikidata.org"], "one robots.txt read, then the stop; the second brand never looked up"
assert dict(counts) == {"stopped:robots_disallow": 1, "requests": 1}
assert lines == ["glenfiddich: stopped, robots_disallow (www.wikidata.org robots.txt disallows "
"/w/api.php?action=wbsearchentities&search=Glenfiddich&language=en&uselang=en&type=item&limit=7&format=json); no further request"]
assert db.get(Brand, 1).image_url is None and db.get(Brand, 2).image_url is None
def test_a_host_with_no_rules_is_read_once_and_then_left_alone(self, db, tmp_path):
read = robots_saying("User-agent: *\nDisallow: /trap/\n")
counts = cli_images.logos_for_brands(db, [db.get(Brand, 1), db.get(Brand, 2)], uploads=tmp_path, check=True, fetch_json=wiki(),
fetch_bytes=lambda url: png(), pace=ip.Pace(0), read_robots=read, out=lambda s: None)
assert read.asked == ["https://www.wikidata.org", "https://commons.wikimedia.org"], "each host once, however many brands"
assert counts["brand:would_apply"] == 1 and counts["brand:refused:label_mismatch"] == 1 and "stopped:robots_disallow" not in counts
def test_the_download_host_is_checked_too(self, db, tmp_path):
def read(origin):
return robots.parse("User-agent: *\nDisallow: /wikipedia/commons/thumb/\n" if "upload." in origin else "", host=origin)
asked = []
counts = cli_images.logos_for_brands(db, [db.get(Brand, 1)], uploads=tmp_path, fetch_json=wiki(), fetch_bytes=asked.append,
pace=ip.Pace(0), read_robots=read, out=lambda s: None)
assert asked == [] and dict(counts) == {"stopped:robots_disallow": 1, "requests": 6}, "three lookups and three robots.txt, no download"
assert db.get(Brand, 1).image_url is None
def test_a_host_refusing_us_at_robots_and_an_unreadable_robots_both_stop(self, db, tmp_path):
def forbidden(origin):
raise SourceBlocked(f"{origin}/robots.txt answered HTTP 403: the host refuses us")
def down(origin):
raise robots.RobotsUnavailable(f"{origin}/robots.txt answered HTTP 503; run stops, retry next run")
for read, reason in ((forbidden, ip.ROBOTS_DISALLOW), (down, ip.ROBOTS_UNAVAILABLE)):
fake = wiki()
counts = cli_images.logos_for_brands(db, [db.get(Brand, 1)], uploads=tmp_path, check=True, fetch_json=fake,
fetch_bytes=lambda url: png(), pace=ip.Pace(0), read_robots=read, out=lambda s: None)
assert fake.calls == [] and counts[f"stopped:{reason}"] == 1
def test_the_guard_matches_the_collectors_reading_of_the_rule(self):
rules = robots.parse(WIKIMEDIA_ROBOTS, host="https://www.wikidata.org")
assert not rules.allows("https://www.wikidata.org/w/api.php?action=wbgetentities&ids=Q1")
assert not rules.allows("https://www.wikidata.org/w/api.php?action=mobileview&page=Q1"), "a longer Allow does not out-lawyer the Disallow"
assert rules.allows("https://www.wikidata.org/wiki/Q1")
guard = ip.RobotsGuard(read=lambda origin: rules)
with pytest.raises(ip.RobotsRefused) as caught:
guard.check("https://www.wikidata.org/w/api.php?action=wbgetentities&ids=Q1")
assert caught.value.reason == ip.ROBOTS_DISALLOW
guard.check("https://www.wikidata.org/wiki/Q1")