"""The listed layer (Stream L, LT5): the shop's words on `listings`, written by one reader from the fragment at ingest and by `backfill listed`, NULL where no fragment exists. What it cost: the listings table read every fragment back into Python per row and could not sort 24k rows; a backfill that copied ProductVariant fields into the listed columns would have labelled our text as the shop's (12,673 listings have no fragment at all: CDG's 7,032-row crawl predates `raw_records`). The fixtures are the eight real fragments `test_listings_view.py` pins; the SQLite kit runs the backfill twice. """ from __future__ import annotations import inspect from datetime import UTC, datetime import pytest from sqlalchemy import create_engine, select from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool from app import cli from app.models import ( Account, Base, Brand, CollectionRun, Listing, Shop, PriceObservation, ProductVariant, ProductLine, RawRecord, Retailer, Source, ) from app.models.decisions import LEDGER_TABLES from app.models.places import Place, ShopPlace from app.services import ingest from app.services.collected import LISTED_FIELDS, _raw, listed_fields, write_listed from tests.test_listings_view import ARI, AVOLTA, CHANGI, DUBAI, EXTIME, HEINEMANN, SHILLA, SHOPIFY TABLES = [ Account.__table__, Source.__table__, CollectionRun.__table__, Brand.__table__, ProductLine.__table__, ProductVariant.__table__, Retailer.__table__, Shop.__table__, RawRecord.__table__, Listing.__table__, PriceObservation.__table__, *LEDGER_TABLES, Place.__table__, ShopPlace.__table__, ] @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: yield session class TestTheReaderPerPlatform: def test_avolta(self): f = listed_fields("avolta/2026-09-05", AVOLTA) assert (f["listed_name"], f["listed_brand"]) == ("Genifique Serum Ultimate 10cl", "Lancôme") assert f["listed_brand_key"] == "lancome" and f["listed_name_key"] == "genifique serum ultimate 10cl" assert f["listed_category"] == "beauty > skincare" assert f["listed_quantity_text"] == "10cl" and (f["listed_quantity_value"], f["listed_quantity_unit"]) == (100, "ml") assert f["listed_gtin"] is None and f["listed_variant"] is None def test_shopify(self): f = listed_fields("shopify/2026-09-05", SHOPIFY) assert (f["listed_name"], f["listed_brand"], f["listed_variant"]) == ("N°5", "CHANEL", "100ml") assert f["listed_category"] == "Fragancias" and f["listed_gtin"] is None assert (f["listed_quantity_value"], f["listed_quantity_unit"]) == (100, "ml") default = listed_fields("shopify/2026-09-05", {**SHOPIFY, "variant": {**SHOPIFY["variant"], "title": "Default Title", "option1": "Default Title"}}) assert default["listed_variant"] is None def test_ari(self): f = listed_fields("ari/2026-09-05", ARI) assert (f["listed_name"], f["listed_brand"]) == ("Perfect Absolute Eau de Parfum 50ml", "Marc Jacobs") assert f["listed_brand_key"] == "marc-jacobs", "two spaces in the shop's spelling fold to the brand slug" assert f["listed_category"] == "New In Beauty" assert f["listed_gtin"] is None, "a six-digit supplier code is never a barcode" assert (f["listed_quantity_value"], f["listed_quantity_unit"]) == (50, "ml") def test_extime(self): payload = {**EXTIME, "main_offer": {"brand_name": "DIOR", "gtin": "3348901682534", "categories_name": ["Beauty", "Perfume", "Men"]}} f = listed_fields("extime/2026-09-05", payload) assert (f["listed_name"], f["listed_brand"], f["listed_variant"]) == ("Dior Homme Parfum", "DIOR", "50 ml") assert f["listed_quantity_text"] == "50 ml" and f["listed_category"] == "Beauty > Perfume > Men" assert f["listed_gtin"] == "3348901682534" def test_shilla(self): f = listed_fields("shilla/2026-09-05", {**SHILLA, "categories": ["Skin Care", "Basic Care", "Creams"]}) assert f["listed_name"] == "Moisture Surge™ Intense 72H Lipid Replenishing Hydrator" assert f["listed_brand"] == "CLINIQUE" and f["listed_quantity_text"] == "75ML/2.5OZ" assert (f["listed_quantity_value"], f["listed_quantity_unit"]) == (75, "ml") assert f["listed_category"] == "Skin Care > Basic Care > Creams" and f["listed_gtin"] is None def test_changi(self): payload = {**CHANGI, "breadcrumb": [{"code": "body", "name": "Body Lotion"}, {"code": "bath", "name": "Bath & Body"}, {"code": "beauty", "name": "Beauty"}]} f = listed_fields("changi/2026-09-05", payload) assert f["listed_brand"] == "ELIZABETH ARDEN" and f["listed_quantity_text"] == "400ml" assert f["listed_category"] == "Beauty > Bath & Body > Body Lotion" def test_dubai(self): f = listed_fields("dubai/2026-09-05", DUBAI) assert (f["listed_name"], f["listed_brand"], f["listed_gtin"]) == ("Johnnie Walker Blue Label 1L", "Johnnie Walker", "5000267014203") assert f["listed_category"] is None and f["listed_quantity_text"] == "1L" def test_heinemann(self): f = listed_fields("heinemann/2026-09-04", {**HEINEMANN, "typeOfGoods": "Whisky", "gtin": "5010327000176"}) assert (f["listed_name"], f["listed_brand"], f["listed_category"]) == ("Glenfiddich 12 Years", "Glenfiddich", "Whisky") assert f["listed_quantity_text"] == "0.7 L" and (f["listed_quantity_value"], f["listed_quantity_unit"]) == (700, "ml") assert f["listed_gtin"] == "5010327000176" def test_an_unknown_shape_yields_every_column_none(self): f = listed_fields("mystery/1", {"foo": "bar"}) assert set(f) == set(LISTED_FIELDS) and all(v is None for v in f.values()) def test_a_listed_text_is_unescaped_and_never_trimmed_or_cased(self): assert _raw(" Bacardí ") == " Bacardí " assert _raw("") is None and _raw(" ") is None and _raw(None) is None def seed(db, *, with_fragment=True, brand="Marc Jacobs"): retailer = Retailer(slug="shop", name="Shop") db.add(retailer) db.flush() shop = Shop(retailer_id=retailer.id, code="DUB", iata="DUB", name="Dublin", currency="EUR") product = ProductVariant(name="Perfect Absolute Eau de Parfum 50ml", brand=brand, vertical="beauty", match_key="k", attributes={}) db.add_all([shop, product]) db.flush() listing = Listing(variant_id=product.id, shop_id=shop.id, source_sku="397595") db.add(listing) db.flush() db.add(PriceObservation(listing_id=listing.id, price=98.5, currency="EUR", price_usd=110, observed_at=datetime(2026, 9, 1, tzinfo=UTC))) if with_fragment: db.add(RawRecord(listing_id=listing.id, payload=ARI, parser_version="ari/2026-09-05", created_at=datetime(2026, 9, 5, tzinfo=UTC))) db.commit() return listing class TestTheBackfill: def test_the_newest_fragment_fills_the_columns_and_a_second_run_writes_zero(self, db): listing = seed(db) older = db.scalar(select(RawRecord)) newer = RawRecord(listing_id=listing.id, parser_version="ari/2026-09-05", payload={"jsonld": {**ARI["jsonld"], "name": "Perfect Absolute EDP 50ml", "category": None}}, created_at=datetime(2026, 9, 6, tzinfo=UTC)) db.add(newer) db.commit() first = cli.backfill_listed(db) assert first.startswith("listed: 1 listing(s) written from their newest fragment, 1 last_seen_at filled, 0 listing(s) with no fragment") db.refresh(listing) assert listing.listed_record_id == newer.id and listing.listed_record_id != older.id assert listing.listed_name == "Perfect Absolute EDP 50ml" assert listing.listed_category is None, "a newer fragment lacking a field nulls it" assert listing.listed_brand_key == "marc-jacobs" and listing.listed_name_key == "perfect absolute edp 50ml" assert listing.last_seen_at is not None and listing.last_seen_at.replace(tzinfo=UTC) >= datetime(2026, 9, 6, tzinfo=UTC) assert cli.backfill_listed(db).startswith("listed: 0 listing(s) written from their newest fragment, 0 last_seen_at filled") def test_no_fragment_means_null_never_our_text(self, db): listing = seed(db, with_fragment=False) message = cli.backfill_listed(db) assert message.endswith("1 listing(s) with no fragment (listed columns NULL)") db.refresh(listing) assert listing.listed_name is None and listing.listed_record_id is None assert listing.last_seen_at is not None, "the newest price observation dates it" def test_a_two_word_house_folds_to_its_brand_slug(self, db): listing = seed(db) db.add(Brand(slug=ingest.brand_slug("Marc Jacobs"), name="Marc Jacobs")) db.commit() cli.backfill_listed(db) db.refresh(listing) slug = db.scalar(select(Brand.slug)) assert listing.listed_brand_key == slug == "marc-jacobs" def test_write_listed_reports_change_and_writes_as_one_unit(self, db): listing = seed(db, with_fragment=False) assert write_listed(listing, 7, "ari/2026-09-05", ARI) is True assert listing.listed_record_id == 7 and listing.listed_name == ARI["jsonld"]["name"] assert write_listed(listing, 7, "ari/2026-09-05", ARI) is False assert write_listed(listing, 8, "mystery/1", {}) is True assert listing.listed_record_id == 8 and listing.listed_name is None and listing.listed_brand_key is None class TestTheIngestHook: def test_run_collector_writes_the_listed_columns_from_the_record_it_just_kept(self): """The hook sits right after `_resolve_listing`, where the raw record is written, and before the price quarantine, so a quarantined sighting still refreshes the words.""" source = inspect.getsource(ingest.run_collector) resolve = source.index("_resolve_listing(db, product, shop, raw)") seen = source.index("listing.last_seen_at = datetime.now(UTC)") hook = source.index("write_listed(listing, record.id, parser_version, raw.raw)") quarantine = source.index("_implausibly_cheap(db, product, shop, price_usd)") assert resolve < seen < hook < quarantine