"""The merge session end to end: the queue, a confirm, a reject, who decided (Stream M, M6). Runs the real routes in-process, signed in as the owner, against the in-memory SQLite the account suites use, with the catalogue tables added for the fourteen 1 Million rows. What is pinned is the outcome rian asked for: the Paco Rabanne to Rabanne alias is the first pair waiting, confirming it records his account and puts the bare "1 Million 10cl" beside the Eau de Toilette in the product queue, a reject never comes back, and every write refuses anyone without the permission. """ from __future__ import annotations import pytest from sqlalchemy import select from app import cli from app.models import ( Award, Base, Brand, CollectionRun, Listing, Location, MergeCandidate, PriceObservation, Product, ProductLine, ProductMerge, Retailer, Source, VariationAlias, ) from app.services import keying, merges, suggest from tests import _accounts as T from tests.kit import _env from tests.test_lines import ONE_MILLION CATALOGUE = [ Retailer.__table__, Location.__table__, ProductLine.__table__, Product.__table__, Listing.__table__, Source.__table__, CollectionRun.__table__, PriceObservation.__table__, Award.__table__, ProductMerge.__table__, MergeCandidate.__table__, VariationAlias.__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) T.person("rian") T.person("adam", level="admin") with _env.TestSessionLocal() as db: 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() from datetime import UTC, datetime 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) suggest.generate(db) db.commit() keying.invalidate() yield Base.metadata.drop_all(_env.engine, tables=CATALOGUE) def rian(): c = T.client() T.as_user(c, "rian") return c class TestQueue: def test_the_alias_is_the_first_pair_waiting_side_by_side(self, world): c = rian() r = c.get("/api/collectors/merge?level=brand") assert r.status_code == 200 body = r.json() assert body["total"] == 1 and body["progress"]["remaining"]["brand"] == 1 pair = body["pairs"][0] assert pair["why"] == "Paco Rabanne renamed itself Rabanne in 2023; the two rows are one house" assert {pair["left"]["name"], pair["right"]["name"]} == {"Paco Rabanne", "Rabanne"} assert pair["left"]["products"] + pair["right"]["products"] == 12 assert "1 Million" in pair["left"]["lines"] and pair["default_name"] in ("Rabanne", "Paco Rabanne") assert pair["left"]["airport_codes"] and pair["right"]["listings"] > 0 def test_the_product_queue_shows_shops_prices_and_barcodes(self, world): c = rian() body = c.get("/api/collectors/merge?level=product").json() assert body["total"] >= 1 pair = body["pairs"][0] for side in (pair["left"], pair["right"]): assert side["shops"] and side["shops"][0]["price"] is not None and "gtin" in side and "image_url" in side class TestDecisions: def test_confirming_the_alias_records_rian_and_brings_the_bare_row_within_reach(self, world): c = rian() pair = c.get("/api/collectors/merge?level=brand").json()["pairs"][0] before = c.get("/api/collectors/merge?level=product").json()["total"] r = c.post(f"/api/collectors/merge/{pair['id']}/confirm", json={"preferred_name": "Rabanne"}) assert r.status_code == 200, r.text body = r.json() assert body["applied"]["products_rekeyed"] == 5 and body["suggested"]["product"] >= 1 assert body["progress"]["remaining"]["brand"] == 0 and body["progress"]["decided_today"] == 1 with _env.TestSessionLocal() as db: rian_id = db.scalar(select(T.Account.id).where(T.Account.username == "rian")) paco = db.scalar(select(Brand).where(Brand.slug == "paco-rabanne")) house = db.scalar(select(Brand).where(Brand.slug == "rabanne")) assert paco.canonical_id == house.id and paco.decided_by == rian_id and paco.decided_at is not None assert house.name == "Rabanne" and house.decided_by == rian_id candidate = db.get(MergeCandidate, pair["id"]) assert (candidate.decision, candidate.decided_by) == ("merged", rian_id) bare = db.get(Product, 4) assert bare.match_key == "rabanne|1-million||100" after = c.get("/api/collectors/merge?level=product").json() assert after["total"] > before whys = [p["why"] for p in after["pairs"]] assert any('"1 Million 10cl" names no variation and the other is Eau de Toilette' in w for w in whys) assert "merge.confirm" in T.audit_actions() def test_a_reject_is_recorded_and_never_resurfaces(self, world): c = rian() pair = c.get("/api/collectors/merge?level=product").json()["pairs"][0] r = c.post(f"/api/collectors/merge/{pair['id']}/reject", json={"note": "two bottles"}) assert r.status_code == 200 and r.json()["decision"] == "kept_apart" assert pair["id"] not in [p["id"] for p in c.get("/api/collectors/merge?level=product").json()["pairs"]] r = c.post("/api/collectors/merge/suggest", json={}) assert r.status_code == 200 assert pair["id"] not in [p["id"] for p in c.get("/api/collectors/merge?level=product").json()["pairs"]] with _env.TestSessionLocal() as db: candidate = db.get(MergeCandidate, pair["id"]) assert candidate.decision == "kept_apart" and candidate.decided_by is not None assert candidate.detail["decision"]["note"] == "two bottles" r = c.post(f"/api/collectors/merge/{pair['id']}/confirm", json={}) assert r.status_code == 409 and r.json()["detail"]["error_code"] == "PAIR_DECIDED" def test_confirming_a_product_pair_merges_under_the_chosen_name(self, world): c = rian() brand_pair = c.get("/api/collectors/merge?level=brand").json()["pairs"][0] c.post(f"/api/collectors/merge/{brand_pair['id']}/confirm", json={"preferred_name": "Rabanne"}) pairs = c.get("/api/collectors/merge?level=product").json()["pairs"] pair = next(p for p in pairs if "Eau de Toilette" in (p["why"] or "") and "1 Million 10cl" in (p["why"] or "")) edt = pair["left"] if pair["left"]["variation"] == "edt" else pair["right"] r = c.post(f"/api/collectors/merge/{pair['id']}/confirm", json={"preferred_name": edt["name"]}) assert r.status_code == 200, r.text assert r.json()["applied"] == {"rows_merged": 1, "survivor_id": edt["id"], "alias_id": 4} with _env.TestSessionLocal() as db: bare = db.get(Product, 4) assert bare.merged_into_id == edt["id"] record = db.scalar(select(ProductMerge).where(ProductMerge.from_id == 4)) assert record.reason == "confirmed" and record.merged_by is not None survivor = db.get(Product, edt["id"]) assert survivor.name == edt["name"] assert len(db.scalars(select(Listing).where(Listing.product_id == survivor.id)).all()) >= 3 class TestAccess: def test_the_writes_refuse_anyone_without_the_permission(self, world): c = T.client() assert c.get("/api/collectors/merge").status_code == 401 assert c.post("/api/collectors/merge/1/confirm", json={}).status_code == 401 T.as_user(c, "adam") assert c.get("/api/collectors/merge").status_code == 403 assert c.post("/api/collectors/merge/1/reject", json={}).status_code == 403 assert c.post("/api/collectors/merge/suggest", json={}).status_code == 403