"""The merge desk (Stream L, LT7): the tables count what the database holds, a batch applies each pair as the signed-in person's decision and regenerates once, a proposal survives the rules, a refused pair is reported and the rest still apply. What it cost: the keyboard queue regenerated every suggestion inside each brand or line confirm (one full pass per row of a bulk session), and a rule that stopped producing a pair withdrew it, so a session's own proposals had no way to wait for a person. The SQLite kit with the ONE_MILLION fixture; no network. """ from __future__ import annotations import pytest from sqlalchemy import select from app.models import Brand, MergeCandidate, Product, ProductLine from app.services import merge_desk, suggest from tests import _accounts as T from tests.kit import _env from tests.test_merge_session import CATALOGUE, rian, world # noqa: F401 (the seeded world and the owner client) def _db(): return _env.TestSessionLocal() class TestTheTables: def test_the_brands_table_counts_what_the_database_holds(self, world): with _db() as db: houses = db.scalar(select(__import__("sqlalchemy").func.count(Brand.id)).where(Brand.canonical_id.is_(None))) page = merge_desk.brands(db, per_page=50) assert page["total"] == houses == 2 rabanne = next(r for r in page["rows"] if r["slug"] == "rabanne") assert rabanne["other_name"] == "Paco Rabanne" and rabanne["reason"] == "known_rebrand" and rabanne["score"] == 0.95 assert rabanne["products"] > 0 and rabanne["listings"] > 0 and rabanne["lines"] >= 1 assert page["counts"]["remaining"] == 1 and page["counts"]["decided_total"] == 0 assert merge_desk.brands(db, has_suggestion=False)["total"] == 0 assert merge_desk.brands(db, min_score=0.96)["total"] == 0 assert [r["slug"] for r in merge_desk.brands(db, sort="name", direction="asc")["rows"]] == ["paco-rabanne", "rabanne"] def test_the_lines_table_has_the_same_shape_with_the_product_pairs_under_it(self, world): with _db() as db: page = merge_desk.lines(db, per_page=50) assert page["total"] == db.scalar(select(__import__("sqlalchemy").func.count(ProductLine.id)).where(ProductLine.canonical_id.is_(None))) row = next(r for r in page["rows"] if r["house"] == "Paco Rabanne") assert row["products"] > 0 and row["product_pairs"] >= 1 and row["vertical"] == "beauty" assert merge_desk.lines(db, house="paco-rabanne")["total"] == 1 def test_the_reads_refuse_anyone_without_the_permission(self, world): c = T.client() assert c.get("/api/collectors/desk/brands").status_code == 401 assert c.post("/api/collectors/merge/batch", json={"confirm": [], "reject": []}).status_code == 401 T.as_user(c, "adam") assert c.get("/api/collectors/desk/lines").status_code == 403 assert c.post("/api/collectors/merge/batch", json={"confirm": [], "reject": []}).status_code == 403 assert c.post("/api/collectors/merge/propose", json={"entries": []}).status_code == 403 c = rian() assert c.get("/api/collectors/desk/brands").status_code == 200 assert c.get("/api/collectors/desk/lines?house=rabanne").status_code == 200 class TestTheBatch: def test_three_confirms_and_two_rejects_regenerate_once_and_record_rian(self, world, monkeypatch): calls = [] real = suggest.generate monkeypatch.setattr(suggest, "generate", lambda db: (calls.append(1), real(db))[1]) c = rian() with _db() as db: # Two proposed product pairs beside the rules' two unknown-variation pairs. edt = {p.name: p.id for p in db.scalars(select(Product).where(Product.merged_into_id.is_(None)))} merge_desk.propose(db, [ {"level": "product", "left_id": edt["1 Million Eau de Toilette 50 ml"], "right_id": edt["1 Million Eau de Toilette 200 ml"], "score": 0.6, "why": "test"}, {"level": "product", "left_id": edt["Rabanne 1 Million Elixir Parfum Intense 50ml"], "right_id": edt["1 Million Elixir Parfum Intense 200 ml"], "score": 0.6, "why": "test"}, ], proposed_by="test") db.commit() products = c.get("/api/collectors/merge?level=product&per_page=50").json()["pairs"] brand = c.get("/api/collectors/merge?level=brand").json()["pairs"][0] ids = [p["id"] for p in products] assert len(ids) >= 4 confirm = [{"candidate_id": brand["id"], "preferred_name": "Rabanne"}, {"candidate_id": ids[0]}, {"candidate_id": ids[1]}] reject = [{"candidate_id": ids[2], "note": "two bottles"}, {"candidate_id": ids[3]}] r = c.post("/api/collectors/merge/batch", json={"confirm": confirm, "reject": reject}) assert r.status_code == 200, r.text body = r.json() assert len(calls) == 1, "the batch regenerates once, at the end" assert len(body["applied"]) + len(body["refused"]) == 5 with _db() as db: rian_id = db.scalar(select(__import__("app.models", fromlist=["Account"]).Account.id).where( __import__("app.models", fromlist=["Account"]).Account.username == "rian")) for item in body["applied"]: row = db.get(MergeCandidate, item["candidate_id"]) assert row.decided_by == rian_id and row.decision in ("merged", "kept_apart") assert db.scalar(select(Brand).where(Brand.slug == "paco-rabanne")).canonical_id is not None assert body["counts"]["brand"]["decided_total"] >= 1 assert T.audit_actions().count("merge.confirm") + T.audit_actions().count("merge.reject") >= len(body["applied"]) def test_a_barcode_veto_without_a_note_is_reported_and_the_others_still_apply(self, world): with _db() as db: a = db.scalar(select(Product).where(Product.name == "1 Million Eau de Toilette 200 ml")) b = db.scalar(select(Product).where(Product.name == "1 Million Elixir Parfum Intense 200 ml")) a.gtin, b.gtin = "3349668508587", "5010327703053" veto = MergeCandidate(level="product", left_id=min(a.id, b.id), right_id=max(a.id, b.id), reason="gtin_differs", score=0.5, detail={"why": "two barcodes"}) db.add(veto) db.commit() veto_id = veto.id other = db.scalar(select(MergeCandidate).where(MergeCandidate.level == "product", MergeCandidate.id != veto_id, MergeCandidate.decision.is_(None))) other_id = other.id c = rian() r = c.post("/api/collectors/merge/batch", json={"confirm": [{"candidate_id": veto_id}], "reject": [{"candidate_id": other_id}]}) assert r.status_code == 200 body = r.json() assert [x["error_code"] for x in body["refused"]] == ["NOTE_REQUIRED"] assert [x["candidate_id"] for x in body["applied"]] == [other_id] r = c.post("/api/collectors/merge/batch", json={"confirm": [], "reject": [{"candidate_id": other_id}]}) assert r.json()["refused"][0]["error_code"] == "PAIR_DECIDED" def test_a_batch_over_the_cap_is_refused_whole(self, world): c = rian() r = c.post("/api/collectors/merge/batch", json={"confirm": [{"candidate_id": i} for i in range(1, 202)], "reject": []}) assert r.status_code == 409 and r.json()["detail"]["error_code"] == "BATCH_TOO_LARGE" class TestProposals: def test_a_proposed_pair_survives_generate_and_is_never_merged_by_filing(self, world): with _db() as db: lines = list(db.scalars(select(ProductLine).where(ProductLine.canonical_id.is_(None)).order_by(ProductLine.id))) left, right = lines[0], lines[1] result = merge_desk.propose(db, [ {"level": "line", "left_id": right.id, "right_id": left.id, "score": 0.7, "why": "one line spelled two ways"}, {"level": "line", "left_id": left.id, "right_id": right.id, "score": 0.7, "why": "again"}, {"level": "line", "left_id": left.id, "right_id": 999999, "score": 0.7, "why": "gone"}, {"level": "brand", "left_id": 1, "right_id": 1, "score": 0.7, "why": "self"}, ], proposed_by="Stream L", check=True) assert (result["inserted"], result["skipped_present"], result["skipped_missing"], result["invalid"]) == (1, 1, 1, 1) # A rule's own undecided pair is endorsed, not duplicated: the desk's "tick all proposed" finds it. rule_pair = db.scalar(select(MergeCandidate).where(MergeCandidate.level == "brand", MergeCandidate.decision.is_(None))) endorsed = merge_desk.propose(db, [{"level": "brand", "left_id": rule_pair.left_id, "right_id": rule_pair.right_id, "score": 0.9, "why": "the same house"}], proposed_by="Stream L") assert endorsed["endorsed"] == 1 and endorsed["inserted"] == 0 db.refresh(rule_pair) assert rule_pair.reason == "known_rebrand" and rule_pair.detail["proposed_by"] == "Stream L" assert merge_desk.brands(db, per_page=5)["counts"]["proposed"] == 1 assert any(r["proposed"] for r in merge_desk.brands(db, per_page=5)["rows"]) assert db.scalar(select(MergeCandidate).where(MergeCandidate.reason == "proposed")) is None, "check writes nothing" result = merge_desk.propose(db, [{"level": "line", "left_id": right.id, "right_id": left.id, "score": 0.7, "why": "one line spelled two ways"}], proposed_by="Stream L") db.commit() assert result["inserted"] == 1 row = db.scalar(select(MergeCandidate).where(MergeCandidate.reason == "proposed")) assert (row.left_id, row.right_id) == (left.id, right.id) and row.detail["proposed_by"] == "Stream L" assert left.canonical_id is None and right.canonical_id is None suggest.generate(db) db.commit() db.refresh(row) assert row.decision is None and row.reason == "proposed" and float(row.score) == 0.7 assert merge_desk.propose(db, [{"level": "line", "left_id": left.id, "right_id": right.id, "score": 0.7, "why": "x"}], proposed_by="again")["skipped_present"] == 1 def test_the_route_and_the_files_shape(self, world, tmp_path): c = rian() with _db() as db: lines = list(db.scalars(select(ProductLine).where(ProductLine.canonical_id.is_(None)).order_by(ProductLine.id))) r = c.post("/api/collectors/merge/propose", json={"proposed_by": "Stream L", "entries": [ {"level": "line", "left_id": lines[0].id, "right_id": lines[1].id, "score": 0.6, "why": "test"}]}) assert r.status_code == 200 and r.json()["inserted"] == 1 path = tmp_path / "p.json" path.write_text('{"proposed_by": "Stream L", "entries": [{"level": "line", "left_id": 1, "right_id": 2, "score": 0.6, "why": "t"}]}') entries, by = merge_desk.load_proposals(path) assert len(entries) == 1 and by == "Stream L" def test_the_committed_proposal_files_validate(self): import pathlib from app.routers.collectors import ProposalEntry root = pathlib.Path(__file__).resolve().parents[2] / "import" / "proposals" files = sorted(root.glob("2026-09-15-*.json")) if not files: pytest.skip("no proposal files committed") for path in files: entries, by = merge_desk.load_proposals(path) assert by and entries seen = set() for entry in entries: ProposalEntry(**entry) key = (entry["level"], min(entry["left_id"], entry["right_id"]), max(entry["left_id"], entry["right_id"])) assert key not in seen, key seen.add(key) assert "—" not in entry["why"] and "free" not in entry["why"].lower() and "cheap" not in entry["why"].lower() if path.name.endswith("brands.json"): assert len(entries) <= 150 and all(e["level"] == "brand" for e in entries) if path.name.endswith("lines.json"): assert len(entries) <= 300 and all(e["level"] == "line" and e["score"] <= 0.8 for e in entries)