"""Decided beats rule (Stream L, LT3): an override survives every rekey and backfill, a human product merge holds across the next collection, a pin holds, an ignore hides. What it cost: `overrides` had zero rows, zero writers and no reader in the keying path; a survivor renamed at confirm to "1 Million" re-keyed `rabanne|1-million||100ml` on the next `rederive` and the next collection created a new product beside it; `_resolve_product` looked up live rows only, so a confirmed merge of the bare "1 Million 10cl" into the Eau de Toilette came undone on the next sighting (576 `variation_unknown` pairs on staging were that class). The SQLite kit with the ONE_MILLION fixture; no network. """ 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, Award, Base, Brand, CollectionRun, Listing, Location, MergeCandidate, PriceObservation, Product, ProductLine, ProductMerge, RawRecord, Retailer, Source, VariationAlias, ) from app.models.accounts import Override from app.models.catalog import IDENTITY_RULES_VERSION from app.services import catalog_queries as cq from app.services import collector_view, ingest, keying, merges, overrides from app.services.collectors.base import RawListing from tests import _accounts as T from tests.kit import _env from tests.test_lines import ONE_MILLION TABLES = [ Account.__table__, Source.__table__, CollectionRun.__table__, Brand.__table__, ProductLine.__table__, Product.__table__, VariationAlias.__table__, Retailer.__table__, Location.__table__, RawRecord.__table__, Listing.__table__, Award.__table__, ProductMerge.__table__, MergeCandidate.__table__, PriceObservation.__table__, Override.__table__, ] RIAN = 1 @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) keying.invalidate() overrides._present.clear() with factory() as session: session.add(Account(id=RIAN, username="rian", display_name="rian")) yield session keying.invalidate() def seed(db): """The fourteen rows at their shops, keyed and lined the way the deploy leaves them.""" rabanne, paco = Brand(slug="rabanne", name="Rabanne"), Brand(slug="paco-rabanne", name="Paco Rabanne") retailer = Retailer(slug="shop", name="Shop") db.add_all([rabanne, paco, retailer]) db.flush() for n, (brand, name, size) in enumerate(ONE_MILLION, start=1): product = Product(id=n, name=name, brand=brand, brand_id=paco.id if brand == "Paco Rabanne" else rabanne.id, vertical="beauty", size_ml=size, match_key="pending", attributes={}) location = Location(retailer_id=retailer.id, code=f"S{n}", iata=f"A{n:02d}", name=f"Shop {n}", currency="EUR") db.add_all([product, location]) db.flush() listing = Listing(product_id=product.id, location_id=location.id, source_sku=f"sku{n}") db.add(listing) db.flush() db.add(PriceObservation(listing_id=listing.id, price=100 + n, currency="EUR", price_usd=110 + n, observed_at=datetime.now(UTC))) db.commit() cli.backfill_lines(db) cli.backfill_variations(db) merges.rekey_products(db, list(db.scalars(select(Product))), keying.load_maps(db)) merges.merge_duplicates(db) db.commit() keying.invalidate() return rabanne, paco def rekey_all(db): keying.invalidate() maps = keying.load_maps(db) merges.rekey_products(db, list(db.scalars(select(Product))), maps) db.commit() keying.invalidate() def edt_100(db) -> Product: return db.scalar(select(Product).where(Product.name == "Rabanne 1 Million EDT 100ml")) class TestOverridesSurvive: def test_a_variation_and_a_name_override_survive_rekey_and_backfill_lines(self, db): seed(db) survivor = edt_100(db) assert survivor.merged_into_id is None and survivor.match_key == "rabanne|1-million|edt|100ml" # The typed name drops the concentration; the collected spelling keeps the key. overrides.decide(db, "product", str(survivor.id), "name", "1 Million", set_by=RIAN, reason="typed", collected_value=survivor.name) survivor.name = "1 Million" parfum = db.scalar(select(Product).where(Product.name == "1 Million Parfum 10cl")) overrides.decide(db, "product", str(parfum.id), "variation", "edp", set_by=RIAN, reason="a person read the box") db.commit() for _ in range(2): rekey_all(db) cli.backfill_lines(db) db.refresh(survivor) db.refresh(parfum) assert survivor.name == "1 Million" assert survivor.match_key == "rabanne|1-million|edt|100ml" assert survivor.line_id is not None assert parfum.match_key == "paco-rabanne|1-million|edp|100ml" assert (parfum.attributes or {}).get("variation") == "edp" assert parfum.identity_rules_version == IDENTITY_RULES_VERSION def test_a_line_override_is_never_pointed_elsewhere_by_backfill_lines(self, db): seed(db) survivor = edt_100(db) other = ProductLine(brand_id=survivor.brand_id, key="somewhere else", slug="rabanne-somewhere-else", name="Elsewhere") db.add(other) db.flush() overrides.decide(db, "product", str(survivor.id), "line_id", other.id, set_by=RIAN, collected_value=survivor.line_id) survivor.line_id = other.id db.commit() cli.backfill_lines(db) rekey_all(db) db.refresh(survivor) assert survivor.line_id == other.id assert survivor.match_key == "rabanne|somewhere-else|edt|100ml" def test_a_quantity_override_names_the_slot(self, db): seed(db) survivor = edt_100(db) overrides.decide(db, "product", str(survivor.id), "quantity", {"value": 75, "unit": "g"}, set_by=RIAN) db.commit() rekey_all(db) db.refresh(survivor) assert survivor.match_key == "rabanne|1-million|edt|75g" def test_a_merge_moves_the_losers_override_and_keeps_the_survivors(self, db): seed(db) survivor = edt_100(db) loser = db.scalar(select(Product).where(Product.name == "1 Million Eau de Toilette 200 ml")) overrides.decide(db, "product", str(loser.id), "variation", "edp", set_by=RIAN) overrides.decide(db, "product", str(loser.id), "name", "Loser", set_by=RIAN, collected_value=loser.name) overrides.decide(db, "product", str(survivor.id), "name", "Winner", set_by=RIAN, collected_value=survivor.name) db.flush() merges.merge_products(db, survivor, [loser], reason="confirmed", merged_by=RIAN) db.commit() rows = overrides.fields_of(db, "product", str(survivor.id)) assert set(rows) == {"name", "variation"} and rows["name"].value == "Winner" record = db.scalar(select(ProductMerge).where(ProductMerge.from_id == loser.id)) assert record.detail["overrides"] == {"moved": ["variation"], "kept": ["name"]} def test_enrich_never_writes_over_a_guarded_column(self, db): seed(db) survivor = edt_100(db) loser = db.scalar(select(Product).where(Product.name == "1 Million Eau de Toilette 200 ml")) overrides.decide(db, "product", str(survivor.id), "quantity", {"value": 75, "unit": "g"}, set_by=RIAN) survivor.size_ml = None loser.size_ml = 200 db.flush() merges.merge_products(db, survivor, [loser], reason="confirmed", merged_by=RIAN) assert survivor.size_ml is None, "a decided quantity is never overwritten by a merged row's size" def _tombstone(db, *, key: str, into: Product, version: str = IDENTITY_RULES_VERSION, reason: str = "confirmed", name: str = "1 Million 10cl") -> Product: row = Product(name=name, brand="Rabanne", brand_id=into.brand_id, vertical="beauty", size_ml=100, match_key=key, attributes={}, identity_rules_version=version, merged_into_id=into.id, line_id=into.line_id) db.add(row) db.flush() db.add(ProductMerge(from_id=row.id, to_id=into.id, merged_by=RIAN, reason=reason, detail={})) db.flush() return row def bare_tile(size_ml: int | None = 100) -> RawListing: return RawListing(source_sku="bare", name="1 Million 10cl", price=80.0, currency="EUR", location_code="S1", brand="Rabanne", size_ml=size_ml, vertical="beauty") class TestTombstoneFollow: def test_the_bare_10cl_resolves_to_the_confirmed_survivor_not_a_new_product(self, db): seed(db) survivor = edt_100(db) bare = db.scalar(select(Product).where(Product.name == "1 Million 10cl", Product.brand == "Paco Rabanne")) # A person confirmed the bare 10cl (keyed apart on an unknown variation) into the EDT. bare.brand, bare.brand_id = "Rabanne", survivor.brand_id merges.merge_products(db, survivor, [bare], reason="confirmed", merged_by=RIAN) rekey_all(db) db.refresh(bare) assert bare.match_key == "rabanne|1-million||100ml" and bare.merged_into_id == survivor.id before = db.scalar(select(Product.id).order_by(Product.id.desc())) resolved = ingest._resolve_product(db, bare_tile()) assert resolved.id == survivor.id assert db.scalar(select(Product.id).order_by(Product.id.desc())) == before, "no new product" def test_a_tombstone_at_a_stale_rules_version_is_ignored_until_rederive_moves_it(self, db): seed(db) survivor = edt_100(db) tomb = _tombstone(db, key="rabanne|1-million||100", into=survivor, version="3") db.commit() assert ingest._follow_tombstone(db, "rabanne|1-million||100ml", None) is None rekey_all(db) db.refresh(tomb) assert tomb.match_key == "rabanne|1-million||100ml" and tomb.identity_rules_version == IDENTITY_RULES_VERSION assert tomb.merged_into_id == survivor.id, "rederive keeps the tombstone a tombstone" assert ingest._follow_tombstone(db, "rabanne|1-million||100ml", None).id == survivor.id def test_a_rules_fold_is_never_followed_only_a_human_merge(self, db): seed(db) survivor = edt_100(db) _tombstone(db, key="rabanne|1-million||100ml", into=survivor, reason="duplicate_match_key") db.commit() assert ingest._follow_tombstone(db, "rabanne|1-million||100ml", None) is None def test_a_tombstone_keyed_unknown_is_never_followed(self, db): seed(db) survivor = edt_100(db) _tombstone(db, key="rabanne|1-million||unknown", into=survivor) db.commit() assert ingest._follow_tombstone(db, "rabanne|1-million||unknown", None) is None silent = RawListing(source_sku="silent", name="1 Million", price=80.0, currency="EUR", location_code="S1", brand="Rabanne", vertical="beauty") resolved = ingest._resolve_product(db, silent) assert resolved.id != survivor.id and resolved.match_key == "rabanne|1-million||unknown" def test_two_tombstones_on_one_key_follow_the_newest_merge(self, db): seed(db) first = edt_100(db) second = db.scalar(select(Product).where(Product.name == "1 Million Parfum 10cl")) _tombstone(db, key="rabanne|1-million||100ml", into=first) _tombstone(db, key="rabanne|1-million||100ml", into=second) db.commit() assert ingest._follow_tombstone(db, "rabanne|1-million||100ml", None).id == second.id def test_a_survivor_with_another_barcode_is_not_followed(self, db): seed(db) survivor = edt_100(db) survivor.gtin = "3349668508587" _tombstone(db, key="rabanne|1-million||100ml", into=survivor) db.commit() assert ingest._follow_tombstone(db, "rabanne|1-million||100ml", "5010327703053") is None assert ingest._follow_tombstone(db, "rabanne|1-million||100ml", "3349668508587").id == survivor.id class TestPinAndIgnore: def test_a_pinned_listing_stays_pinned_when_resolved_with_another_product(self, db): seed(db) survivor = edt_100(db) listing = db.scalar(select(Listing).where(Listing.source_sku == "sku4")) other = db.get(Product, listing.product_id) location = db.get(Location, listing.location_id) overrides.pin_listing(db, listing.id, survivor.id, set_by=RIAN, reason="same bottle") assert listing.product_id == survivor.id and listing.pinned_by == RIAN raw = RawListing(source_sku="sku4", name=other.name, price=1.0, currency="EUR", location_code=location.code) again = ingest._resolve_listing(db, other, location, raw) assert again.id == listing.id and again.product_id == survivor.id overrides.unpin_listing(db, listing.id, set_by=RIAN) assert ingest._resolve_listing(db, other, location, raw).product_id == other.id def test_a_pin_follows_the_pinned_products_survivor(self, db): seed(db) survivor = edt_100(db) loser = db.scalar(select(Product).where(Product.name == "1 Million Eau de Toilette 200 ml")) listing = db.scalar(select(Listing).where(Listing.source_sku == "sku4")) overrides.pin_listing(db, listing.id, loser.id, set_by=RIAN) merges.merge_products(db, survivor, [loser], reason="confirmed", merged_by=RIAN) db.commit() location = db.get(Location, listing.location_id) raw = RawListing(source_sku="sku4", name="x", price=1.0, currency="EUR", location_code=location.code) assert ingest._resolve_listing(db, db.get(Product, 1), location, raw).product_id == survivor.id def test_an_ignored_listing_leaves_the_collectors_page_and_returns_with_the_filter(self, db): seed(db) listing = db.scalar(select(Listing).where(Listing.source_sku == "sku6")) overrides.ignore_listing(db, listing.id, set_by=RIAN, reason="a display tester, not for sale") assert listing.ignored_by == RIAN and listing.ignore_reason.startswith("a display") page = collector_view.listings(db, per_page=100) assert listing.id not in {r["listing_id"] for r in page["rows"]} only = collector_view.listings(db, ignored="only", per_page=100) assert {r["listing_id"] for r in only["rows"]} == {listing.id} assert only["rows"][0]["ignore_reason"].startswith("a display") both = collector_view.listings(db, ignored="any", per_page=100) assert listing.id in {r["listing_id"] for r in both["rows"]} table = collector_view.products(db, q="1 Million EDT 100ml") assert all(s["iata"] != "A06" for row in table["rows"] for s in row["shops"]) overrides.unignore_listing(db, listing.id, set_by=RIAN) assert listing.id in {r["listing_id"] for r in collector_view.listings(db, per_page=100)["rows"]} def test_live_listings_is_the_condition_and_the_site_readers_carry_it(self): assert "ignored_at IS NULL" in str(cq.live_listings()) wired = ("_latest_observation_subquery", "_summary_base", "_priced_locations", "suggest_products", "get_product", "featured_savings", "_airport_shops", "_airport_summary", "_airport_rows", "list_products", "airport_locations") for name in wired: fn = getattr(cq, name, None) assert fn is not None, name assert "live_listings()" in inspect.getsource(fn), name # --------------------------------------------------------------------------- the routes, on the kit CATALOGUE = [ Source.__table__, CollectionRun.__table__, Retailer.__table__, Location.__table__, ProductLine.__table__, Product.__table__, RawRecord.__table__, Listing.__table__, PriceObservation.__table__, Award.__table__, ProductMerge.__table__, MergeCandidate.__table__, VariationAlias.__table__, Override.__table__, ] @pytest.fixture def world(monkeypatch): T.fresh(monkeypatch) Base.metadata.drop_all(_env.engine, tables=CATALOGUE) Base.metadata.create_all(_env.engine, tables=CATALOGUE) overrides._present.clear() keying.invalidate() T.person("rian") T.person("adam", level="admin") with _env.TestSessionLocal() as db: seed(db) yield Base.metadata.drop_all(_env.engine, tables=CATALOGUE) overrides._present.clear() def _ids(): with _env.TestSessionLocal() as db: survivor = edt_100(db) listing = db.scalar(select(Listing).where(Listing.source_sku == "sku4")) return survivor.id, listing.id, listing.product_id class TestRoutes: def test_refused_without_a_session_or_the_permission(self, world): c = T.client() assert c.post("/api/collectors/products/1/override", json={"field": "name", "value": "x"}).status_code == 401 assert c.post("/api/collectors/listings/1/ignore", json={}).status_code == 401 T.as_user(c, "adam") assert c.post("/api/collectors/products/1/override", json={"field": "name", "value": "x"}).status_code == 403 assert c.post("/api/collectors/listings/1/pin", json={"product_id": 1}).status_code == 403 assert c.post("/api/collectors/listings/1/unpin", json={}).status_code == 403 assert c.post("/api/collectors/listings/1/ignore", json={}).status_code == 403 assert c.post("/api/collectors/listings/1/unignore", json={}).status_code == 403 def test_the_owner_decides_and_the_who_columns_come_from_the_session(self, world): survivor_id, listing_id, other_id = _ids() c = T.client() T.as_user(c, "rian") r = c.post(f"/api/collectors/products/{survivor_id}/override", json={"field": "name", "value": "1 Million", "reason": "the box says so"}) assert r.status_code == 200, r.text assert r.json()["match_key"] == "rabanne|1-million|edt|100ml" r = c.post(f"/api/collectors/listings/{listing_id}/pin", json={"product_id": survivor_id, "reason": "same"}) assert r.status_code == 200 and r.json()["product_id"] == survivor_id r = c.post(f"/api/collectors/listings/{listing_id}/ignore", json={"reason": "tester"}) assert r.status_code == 200 and r.json()["ignored"] is True with _env.TestSessionLocal() as db: rian_id = db.scalar(select(Account.id).where(Account.username == "rian")) row = overrides.fields_of(db, "product", str(survivor_id))["name"] assert row.set_by == rian_id and row.collected_value == "Rabanne 1 Million EDT 100ml" listing = db.get(Listing, listing_id) assert listing.pinned_by == rian_id and listing.ignored_by == rian_id assert c.post(f"/api/collectors/listings/{listing_id}/unpin", json={}).status_code == 200 assert c.post(f"/api/collectors/listings/{listing_id}/unignore", json={}).status_code == 200 r = c.post(f"/api/collectors/products/{survivor_id}/override", json={"field": "quantity", "value": {"value": 1, "unit": "cups"}}) assert r.status_code == 422 and r.json()["detail"]["error_code"] == "VALUE_INVALID" assert c.post("/api/collectors/products/999999/override", json={"field": "name", "value": "x"}).status_code == 404 def test_a_typed_name_at_confirm_leaves_an_override_set_by_rian(self, world): from app.services import merge_session with _env.TestSessionLocal() as db: rian_id = db.scalar(select(Account.id).where(Account.username == "rian")) survivor = edt_100(db) other = db.scalar(select(Product).where(Product.name == "1 Million Eau de Toilette 200 ml")) candidate = MergeCandidate(level="product", left_id=min(survivor.id, other.id), right_id=max(survivor.id, other.id), reason="variation_unknown", score=0.9, detail={"why": "test"}) db.add(candidate) db.commit() result = merge_session.confirm(db, candidate.id, decided_by=rian_id, preferred_name="1 Million", keep="left") kept = db.get(Product, result["applied"]["survivor_id"]) row = overrides.fields_of(db, "product", str(kept.id))["name"] assert kept.name == "1 Million" and row.set_by == rian_id and row.reason == "chosen at merge" assert row.collected_value in ("Rabanne 1 Million EDT 100ml", "1 Million Eau de Toilette 200 ml") rekey_all(db) db.refresh(kept) assert kept.match_key.startswith("rabanne|1-million|edt|"), "the key keeps the collected spelling"