"""The proposals load and withdraw (Stream K4; the escalation-2 spec test 15, and the file's schema). What it cost before: `propose --file` carried staging's row ids and could not load on production (plan W13); a re-load could have re-opened what a person had already approved; a check that wrote would have left half a pass behind; a spot-check sample that moved between loads would have let a person approve in bulk the rows they had been asked to read. SQLite kit, no network.""" from __future__ import annotations import copy import json from datetime import UTC, datetime import pytest from sqlalchemy import create_engine, delete, event, func, select from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool from app.models import (Account, Award, Base, Brand, CollectionRun, Decision, DecisionBatch, LEDGER_TABLES, Listing, Merge, PriceObservation, ProductLine, ProductVariant, Proposal, ProposalPass, Retailer, Shop, Source, Suggestion, AttributeAlias) from app.models.places import Place, ShopPlace from app.services import keying, proposals from app.services.decisions import natural_keys from app.services.decisions.writer import Refused TABLES = [Account.__table__, Source.__table__, CollectionRun.__table__, Brand.__table__, ProductLine.__table__, ProductVariant.__table__, AttributeAlias.__table__, Award.__table__, Retailer.__table__, Shop.__table__, Listing.__table__, Merge.__table__, Suggestion.__table__, Place.__table__, ShopPlace.__table__, PriceObservation.__table__, *LEDGER_TABLES] NAMES = ["Rouge Allure 3.5 gr / 99 Pirate", "Rouge Allure 3.5 gr / 104 Passion", "Rouge Allure Velvet 3.5 g", "Rouge Allure 3.5g 99 Pirate"] @pytest.fixture def db(): engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) # pysqlite releases a SAVEPOINT as a commit unless the transaction is begun explicitly; the # load's --check must roll back here exactly as it does on Postgres (SQLAlchemy's recipe). event.listen(engine, "connect", lambda conn, _rec: setattr(conn, "isolation_level", None)) event.listen(engine, "begin", lambda conn: conn.exec_driver_sql("BEGIN")) Base.metadata.create_all(engine, tables=TABLES) keying.invalidate() with sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)() as s: s.add(Account(id=1, username="rian", display_name="rian")) s.add_all([Brand(id=1, slug="chanel", name="CHANEL"), Retailer(id=1, slug="attenza", name="Attenza"), Retailer(id=2, slug="avolta", name="Avolta")]) s.flush() s.add_all([ProductLine(id=1, brand_id=1, key="rouge allure", slug="chanel-rouge-allure", name="Rouge Allure"), ProductLine(id=2, brand_id=1, key="rouge allure velvet", slug="chanel-rouge-allure-velvet", name="Rouge Allure Velvet"), ProductLine(id=3, brand_id=1, key="rouge allure 99 pirate", slug="chanel-rouge-allure-99-pirate", name="Rouge Allure 99 Pirate")]) s.add_all([Shop(id=1, retailer_id=1, code="PTY1", iata="PTY", name="Panama", currency="USD"), Shop(id=2, retailer_id=2, code="MEX1", iata="MEX", name="Mexico", currency="USD")]) s.flush() for n, name in enumerate(NAMES, start=1): line = 3 if n == 4 else (2 if n == 3 else 1) s.add(ProductVariant(id=n, name=name, brand="CHANEL", brand_id=1, vertical="beauty", category="Makeup", match_key=f"k{n}", product_line_id=line, quantity_value=3.5, quantity_unit="g", quantity_state="stated", form="single", attributes={})) s.flush() s.add(Listing(id=n, variant_id=n, shop_id=1 if n != 4 else 2, source_sku=f"sku{n}", listed_name=name, url=f"https://example.test/p/{n}", listed_variant="99 Pirate" if n == 1 else None)) s.flush() s.add(PriceObservation(listing_id=n, price=40 + n, currency="USD", price_usd=40 + n, observed_at=datetime(2026, 9, 17, tzinfo=UTC), in_stock=True)) s.commit() yield s keying.invalidate() def _key(db, model, id_): return natural_keys.build(db.get(model, id_))[0] def _file(db, name="claude/2026-09-17/chanel-test", **over) -> dict: v1, v4 = _key(db, ProductVariant, 1), _key(db, ProductVariant, 4) l1, l4 = _key(db, Listing, 1), _key(db, Listing, 4) line = "line:new:chanel-rouge-allure" data = {"pass": {"name": name, "kind": "session", "process_version": "2", "rules_version": "6", "generator": "claude-session"}, "brand": "chanel", "proposals": [ {"sheet_line_ref": line, "position": 0, "entity": {"type": "product_line", "key": line, "detail": {"brand_slug": "chanel", "slug": "chanel-rouge-allure", "name": "Rouge Allure"}}, "field": "name", "value": "Rouge Allure", "confidence": 0.97, "reason": "the head words every shade shares", "evidence": [{"listing": l1, "source": "listed_name", "span": [0, 12], "text": "Rouge Allure"}]}, {"sheet_line_ref": line, "position": 1, "entity": {"type": "product_variant", "key": v4, "detail": {"listings": [l4]}}, "field": "product_line", "value": line, "confidence": 0.9, "reason": "a shade of Rouge Allure written loose", "evidence": [{"listing": l4, "source": "listed_name", "span": [0, 12], "text": "Rouge Allure"}]}, {"sheet_line_ref": line, "position": 2, "entity": {"type": "product_variant", "key": v1, "detail": {"listings": [l1]}}, "field": "attribute:color", "value": "99 Pirate", "confidence": 0.99, "reason": "the shop's option field", "evidence": [{"listing": l1, "source": "option:shade", "span": [0, 9], "text": "99 Pirate"}]}, {"sheet_line_ref": line, "position": 3, "entity": {"type": "suggestion", "key": f"pair:product:{min(v1, v4)}||{max(v1, v4)}", "detail": {"level": "product"}}, "field": "decision", "value": {"decision": "same", "survivor": v1, "name": None, "note": None}, "confidence": 0.8, "reason": "same shade and weight, two shops", "evidence": [{"listing": l4, "source": "listed_name", "span": [18, 27], "text": "99 Pirate"}]}, ]} data.update(over) return data def _counts(db) -> dict: return {m.__tablename__: db.scalar(select(func.count()).select_from(m)) for m in (ProposalPass, Proposal, Decision, DecisionBatch, ProductLine, ProductVariant, Suggestion, Listing)} def test_check_leaves_every_table_unchanged_and_reports_what_a_load_would_do(db): before = _counts(db) counts = proposals.load(db, _file(db), check=True) assert counts["check"] and counts["inserted"] == 4 assert _counts(db) == before def test_an_identical_file_is_a_no_op_by_content_and_a_corrected_file_updates_open_rows_only(db): data = _file(db) first = proposals.load(db, data) # The line's slug already exists (adopted at approval, so resolved); the pair no rule offered is creatable. assert first["inserted"] == 4 and first["creatable"] == 1 and first["resolved"] == 3 assert proposals.load(db, copy.deepcopy(data))["unchanged"] is True row = db.scalar(select(Proposal).where(Proposal.field == "attribute:color")) row.status = "rejected"; db.commit() changed = copy.deepcopy(data) for p in changed["proposals"]: p["reason"] = p["reason"] + " (re-read)" counts = proposals.load(db, changed) assert counts["kept"] == 1 and counts["updated"] == 3 assert db.scalar(select(Proposal).where(Proposal.field == "attribute:color")).reason == "the shop's option field", "a resolved row is never changed" def test_rows_missing_from_a_reload_go_stale_unless_partial(db): data = _file(db) proposals.load(db, data) fewer = copy.deepcopy(data) fewer["proposals"] = fewer["proposals"][:2] proposals.load(db, fewer, partial=True) assert db.scalar(select(func.count()).select_from(Proposal).where(Proposal.status == "stale")) == 0 fewer["pass"]["note"] = "second statement" counts = proposals.load(db, fewer) assert counts["stale_missing"] == 2 assert {r.field for r in db.scalars(select(Proposal).where(Proposal.status == "stale"))} == {"attribute:color", "decision"} def test_spot_check_is_deterministic_per_pass_name_and_never_cleared(db): data = _file(db) proposals.load(db, data) first = {(r.natural_key, r.field): r.spot_check for r in db.scalars(select(Proposal))} assert first[(data["proposals"][3]["entity"]["key"], "decision")] is True, "a merge is always spot-checked" # Five rows or fewer: every one is among the five lowest; the rule's sample is at least three. again = proposals.spot_checks(data["pass"]["name"], [{"entity_type": p["entity"]["type"], "natural_key": p["entity"]["key"], "field": p["field"], "value": p["value"], "confidence": p["confidence"]} for p in data["proposals"]]) assert again == proposals.spot_checks(data["pass"]["name"], [{"entity_type": p["entity"]["type"], "natural_key": p["entity"]["key"], "field": p["field"], "value": p["value"], "confidence": p["confidence"]} for p in reversed(data["proposals"])]) many = [{"entity_type": "product_variant", "natural_key": f"variant:{i}", "field": "name", "value": "x", "confidence": 0.95, "position": i} for i in range(100)] sample_a = proposals.spot_checks("pass-a", many) assert sample_a == proposals.spot_checks("pass-a", list(reversed(many))) and len(sample_a) == 5 + 5 assert sample_a != proposals.spot_checks("pass-b", many), "the sample is seeded by the pass name" # A reload that would no longer mark a row keeps its mark while it is open. raised = copy.deepcopy(data) for p in raised["proposals"]: p["confidence"] = 0.99 raised["proposals"][3]["value"]["decision"] = "separate" proposals.load(db, raised) assert db.scalar(select(Proposal).where(Proposal.field == "decision")).spot_check is True def test_the_schema_error_names_the_row_and_the_field(db): data = _file(db) del data["proposals"][2]["evidence"] with pytest.raises(proposals.FileInvalid) as bad: proposals.load(db, data) assert "proposals[2]" in str(bad.value) and "evidence" in str(bad.value) data = _file(db) data["proposals"][1]["entity"]["key"] = "17" with pytest.raises(proposals.FileInvalid) as bad: proposals.load(db, data) assert "proposals[1]" in str(bad.value) and "entity.key" in str(bad.value), "a database id is not a natural key" data = _file(db) data["proposals"][2]["evidence"][0]["text"] = "99 Pirates" with pytest.raises(proposals.FileInvalid) as bad: proposals.load(db, data) assert "proposals[2]" in str(bad.value) and "text" in str(bad.value) def test_a_span_the_listing_does_not_contain_loads_stale_and_an_unknown_variant_parks(db): data = _file(db) data["proposals"][1]["evidence"][0].update(span=[1, 13], text="Rouge Allure") data["proposals"].append({"entity": {"type": "product_variant", "key": "variant:00000000-0000-0000-0000-000000000000", "detail": {"listings": ["listing:attenza/PTY1/nope"]}}, "field": "name", "value": "Ghost", "confidence": 0.9, "reason": "a row this host never had", "evidence": [{"listing": "listing:attenza/PTY1/nope", "source": "listed_name", "span": [0, 5], "text": "Ghost"}]}) counts = proposals.load(db, data) assert counts["evidence_mismatch"] == 1 and counts["parked"] == 1 stale = db.scalar(select(Proposal).where(Proposal.field == "product_line")) assert stale.status == "stale" and stale.resolution_note.startswith("EVIDENCE_MISMATCH") parked = db.scalar(select(Proposal).where(Proposal.field == "name", Proposal.entity_type == "product_variant")) assert parked.status == "parked" and parked.entity_id is None def test_rule_and_current_values_are_captured_and_a_previous_rejection_is_noted(db): proposals.load(db, _file(db)) color = db.scalar(select(Proposal).where(Proposal.field == "attribute:color")) assert color.current_value is None and color.entity_id == 1 member = db.scalar(select(Proposal).where(Proposal.field == "product_line")) assert member.current_value == _key(db, ProductLine, 3) color.status = "rejected"; db.commit() counts = proposals.load(db, _file(db, name="claude/2026-09-18/chanel-2")) assert counts["previously_rejected"] == 1 again = db.scalar(select(Proposal).join(ProposalPass).where(ProposalPass.name == "claude/2026-09-18/chanel-2", Proposal.field == "attribute:color")) assert again.detail["previously_rejected"] == str(color.uid) def test_a_replayed_decision_marks_its_proposal_approved_on_load(db): data = _file(db) proposals.load(db, data) color = db.scalar(select(Proposal).where(Proposal.field == "attribute:color")) result = proposals.approve(db, "chanel", data["pass"]["name"], scope={"proposal_uids": [str(color.uid)]}, by="rian") assert result["counts"]["approved"] == 1 # The other host: the pass arrived with the ledger (replay), its proposals did not. db.execute(delete(Proposal)); db.commit() data["pass"]["note"] = "loaded on the second host" counts = proposals.load(db, data) assert counts["approved_on_load"] == 1 assert db.scalar(select(Proposal).where(Proposal.field == "attribute:color")).status == "approved" def test_a_replayed_keep_separate_marks_its_pair_proposal_rejected_on_load(db): """The CHANEL rehearsal (dfp_k4 -> dfp_k4b): the refill pair rejected on the first host loaded as approved on the second, because its Keep separate is a decision of the pass like any other.""" data = _file(db) proposals.load(db, data) pair = db.scalar(select(Proposal).where(Proposal.field == "decision")) proposals.approve(db, "chanel", data["pass"]["name"], scope=None, reject=[str(pair.uid)], note="a refill is not its lipstick", by="rian") db.execute(delete(Proposal)); db.commit() data["pass"]["note"] = "loaded on the second host" counts = proposals.load(db, data) assert counts["approved_on_load"] == 1 again = db.scalar(select(Proposal).where(Proposal.field == "decision")) assert again.status == "rejected" and again.resolution_note == "a refill is not its lipstick" def test_withdraw_closes_the_waiting_rows_lists_the_approved_batches_and_never_touches_the_ledger(db): data = _file(db) proposals.load(db, data) color = db.scalar(select(Proposal).where(Proposal.field == "attribute:color")) batch_uid = proposals.approve(db, "chanel", data["pass"]["name"], scope={"proposal_uids": [str(color.uid)]}, by="rian")["batch_uid"] ledger = db.scalar(select(func.count()).select_from(Decision)) with pytest.raises(Refused): proposals.withdraw(db, data["pass"]["name"], "", "rian") result = proposals.withdraw(db, data["pass"]["name"], "the pass read shades as lines", "rian") assert result["withdrawn_rows"] == 3 and [b["uid"] for b in result["approved_batches"]] == [batch_uid] assert "decisions undo-batch " + batch_uid in result["approved_batches"][0]["undo"] assert db.scalar(select(func.count()).select_from(Decision)) == ledger with pytest.raises(Refused) as refused: proposals.load(db, _file(db)) assert refused.value.code == "PASS_WITHDRAWN" def test_the_cli_load_check_prints_and_writes_nothing(db, tmp_path, monkeypatch, capsys): from app import cli_proposals path = tmp_path / "chanel.json" path.write_text(json.dumps(_file(db))) monkeypatch.setattr(cli_proposals, "SessionLocal", lambda: _Borrowed(db)) before = _counts(db) import argparse parser = argparse.ArgumentParser() cli_proposals.register(parser.add_subparsers(dest="command")) args = parser.parse_args(["proposals", "load", "--file", str(path), "--check"]) assert args.func(args) == 0 assert "rolled back" in capsys.readouterr().out assert _counts(db) == before class _Borrowed: def __init__(self, db): self.db = db def __enter__(self): return self.db def __exit__(self, *exc): return False def test_collected_name_cites_a_fragment_less_listing_and_is_refused_on_one_with_a_fragment(db): """The five refused sources (four Heinemann-family hosts and Dubai) were last read before fragments were kept and are never contacted again, so their 5,737 listings have no listed column and never will; the only text a pass can cite is the name the collector built from the shop's fields (K12.2, the plan's A1). Cost: without it section 3's rule made 3,847 product variants uncitable and the 18 Sep pass wrote no file for glenfiddich (28 of 59 variants had nothing to cite). """ from app.models import Listing as L bare = db.get(L, 4) bare.listed_name = None bare.listed_record_id = None db.commit() v4, l4 = _key(db, ProductVariant, 4), _key(db, Listing, 4) name = db.get(ProductVariant, 4).name data = _file(db) data["proposals"] = [{"entity": {"type": "product_variant", "key": v4, "detail": {"listings": [l4]}}, "field": "name", "value": name[:12], "confidence": 0.9, "reason": "the name as the collector read it", "evidence": [{"listing": l4, "source": "collected_name", "span": [0, 12], "text": name[:12]}]}] counts = proposals.load(db, data) row = db.scalar(select(Proposal).where(Proposal.field == "name", Proposal.entity_type == "product_variant")) assert row.status == "open" and counts.get("evidence_mismatch", 0) == 0 ok, why, spans = proposals.evidence_state(proposals.ListingCache(db), row.evidence) assert ok and spans[0]["listed_text"] == name # A listing whose fragment WAS kept is cited by its listed column, never by the collected name. bare.listed_record_id = 1 db.commit() data["pass"]["name"] = "claude/2026-09-18/chanel-2" # an identical file is a no-op by content with pytest.raises(proposals.Refused) as refused: proposals.load(db, data, check=True) assert refused.value.code == "EVIDENCE_HAS_FRAGMENT"