"""Undo restores the predecessor, batch undo is one transaction, consequences carry the cause (Stream K2; spec ยง7, tests 4, 5, 6). What it cost: a confirm had no undo at all, and the 15 Sep decisions refused any 200-row batch until one existed. SQLite kit, no network.""" from __future__ import annotations import uuid import pytest from sqlalchemy import create_engine, 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, ProductLine, ProductVariant, Retailer, Shop, Source, Suggestion, AttributeAlias) from app.models.places import Place, ShopPlace from app.services import keying, merges from app.services.decisions import effective, undo as undo_service, writer from tests.conftest import forbid_ledger_updates 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__, *LEDGER_TABLES] 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) forbid_ledger_updates(engine) # the real database refuses an UPDATE on decisions; so does this one now keying.invalidate() with sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)() as s: s.add(Account(id=RIAN, username="rian", display_name="rian")) rabanne, paco = Brand(id=1, slug="rabanne", name="Rabanne"), Brand(id=2, slug="paco-rabanne", name="Paco Rabanne") retailer = Retailer(slug="r", name="R"); s.add_all([rabanne, paco, retailer]); s.flush() one_r = ProductLine(id=1, brand_id=1, key="1 million", slug="rabanne-1-million", name="1 Million") one_p = ProductLine(id=2, brand_id=2, key="1 million", slug="paco-rabanne-1-million", name="1 Million") s.add_all([one_r, one_p]); s.flush() for n, (brand_id, line_id) in enumerate(((1, 1), (2, 2), (2, 2)), start=1): v = ProductVariant(id=n, name=f"1 Million {n} 100ml", brand="Rabanne" if brand_id == 1 else "Paco Rabanne", brand_id=brand_id, vertical="beauty", match_key=f"k{n}", product_line_id=line_id, quantity_value=100, quantity_unit="ml", quantity_state="stated", quantity_ml=100, form="single", attributes={}) shop = Shop(id=n, retailer_id=retailer.id, code=f"S{n}", iata=f"A0{n}", name=f"Shop {n}", currency="EUR") s.add_all([v, shop]); s.flush() s.add(Listing(variant_id=n, shop_id=n, source_sku=f"sku{n}")) s.commit() yield s keying.invalidate() def name_of(db, vid): return db.get(ProductVariant, vid).name def test_undo_restores_the_predecessor_then_releases_and_an_undo_of_an_undo_reinstates(db): ids = [] for value in ("One", "Two", "Three"): with writer.batch(db, "route", "individual", RIAN, commit=True, tail=False) as b: ids.append(writer.record(b, "product_variant", 1, "name", value).id) assert name_of(db, 1) == "Three" with writer.batch(db, "route", "individual", RIAN, commit=True, tail=False) as b: u3 = undo_service.undo(b, ids[2], "typo") assert name_of(db, 1) == "Two" and u3.restores_id == ids[1] and u3.reverses_id == ids[2] and u3.effect == "set" assert effective(db, "product_variant", [1], ["name"])[(1, "name")].id == u3.id with writer.batch(db, "route", "individual", RIAN, commit=True, tail=False) as b: with pytest.raises(writer.Refused) as refused: undo_service.undo(b, ids[2], "again") # Three is not in force any more assert refused.value.code == "DECISION_SUPERSEDED" # Undo of the undo reinstates Three. with writer.batch(db, "route", "individual", RIAN, commit=True, tail=False) as b: again = undo_service.undo(b, u3.id, "Three was right") assert name_of(db, 1) == "Three" and again.reverses_id == u3.id and again.restores_id == ids[2] # Down the chain: undoing what is in force restores the predecessor each time, then releases to the rules. with writer.batch(db, "route", "individual", RIAN, commit=True, tail=False) as b: undo_service.undo(b, ids[2], "no") assert name_of(db, 1) == "Two" with writer.batch(db, "route", "individual", RIAN, commit=True, tail=False) as b: undo_service.undo(b, ids[1], "no") assert name_of(db, 1) == "One" with writer.batch(db, "route", "individual", RIAN, commit=True, tail=False) as b: last = undo_service.undo(b, ids[0], "to the rules") assert last.effect == "release" and name_of(db, 1) == "1 Million 1 100ml" assert effective(db, "product_variant", [1], ["name"]) == {} def test_a_brand_alias_is_a_row_with_consequences_that_an_undo_reverses_together(db): with writer.batch(db, "route", "individual", RIAN, commit=True) as b: row = merges.apply_brand_alias(db, db.get(Brand, 2), db.get(Brand, 1), decided_by=RIAN, preferred_name="Rabanne", batch=b) alias_row = db.scalar(select(Decision).where(Decision.entity_type == "brand", Decision.field == "alias_of")) consequences = list(db.scalars(select(Decision).where(Decision.caused_by_id == alias_row.id))) fields = sorted(d.field for d in consequences) assert fields == ["alias_of", "name"], fields line_row = next(d for d in consequences if d.field == "alias_of") assert line_row.uid == uuid.uuid5(alias_row.uid, f"product_line|{line_row.natural_key}|alias_of") assert db.get(Brand, 2).alias_of_id == 1 and db.get(ProductLine, 2).alias_of_id == 1 assert {v.product_line_id for v in db.scalars(select(ProductVariant).where(ProductVariant.id.in_([2, 3])))} == {1} assert line_row.detail["products_moved"] == [[2, 2], [3, 2]] with writer.batch(db, "route", "individual", RIAN, commit=True) as b: skipped = [] undo_service.undo(b, alias_row.id, "not the same brand after all", skipped=skipped) assert skipped == [] assert db.get(Brand, 2).alias_of_id is None and db.get(ProductLine, 2).alias_of_id is None assert {v.product_line_id for v in db.scalars(select(ProductVariant).where(ProductVariant.id.in_([2, 3])))} == {2} assert effective(db, "brand", [2], ["alias_of"]) == {} and effective(db, "product_line", [2], ["alias_of"]) == {} def test_batch_undo_skips_a_row_a_later_batch_superseded_and_reverses_its_tail_fold_first(db): # One bottle at Rabanne (3) and at Paco Rabanne (2): different keys until the alias; the # alias's rekey lands them on one key and the batch's tail folds them. v3 = db.get(ProductVariant, 3) v3.brand_id, v3.brand, v3.product_line_id = 1, "Rabanne", 1 for v in db.scalars(select(ProductVariant).where(ProductVariant.id.in_([2, 3]))): v.name = "1 Million 100ml" db.get(ProductVariant, 1).name = "1 Million Elixir 100ml" db.commit() keying.invalidate() merges.rekey_product_variants(db, list(db.scalars(select(ProductVariant))), keying.load_maps(db)) # real keys, no fold yet db.commit() assert len({v.match_key for v in db.scalars(select(ProductVariant))}) == 3 with writer.batch(db, "desk", "bulk", RIAN, commit=True) as a: writer.record(a, "brand", 1, "hidden", True) merges.apply_brand_alias(db, db.get(Brand, 2), db.get(Brand, 1), decided_by=RIAN, batch=a) fold = db.scalar(select(Merge).where(Merge.batch_id == a.row.id)) assert fold is not None and fold.decision_id is None, "the tail fold carries the batch id" loser = db.get(ProductVariant, fold.from_id) assert loser.merged_into_id == fold.to_id and db.get(Brand, 2).alias_of_id == 1 # Batch B supersedes the hidden flag. with writer.batch(db, "route", "individual", RIAN, commit=True, tail=False) as bb: writer.record(bb, "brand", 1, "hidden", False) undo_batch = undo_service.undo_batch(db, a.row.uid, RIAN, "wrong sheet") assert undo_batch.kind == "undo" and undo_batch.reverses_batch_id == a.row.id skipped = undo_batch.summary.get("skipped", []) assert len(skipped) == 1 and skipped[0]["field"] == "hidden" and skipped[0]["superseded_by_batch"] == str(bb.row.uid) assert db.get(Brand, 1).hidden is False, "a later person's decision is never reversed under him" db.refresh(loser) assert loser.merged_into_id is None and db.get(Merge, fold.id).reversed_by_id is not None assert db.get(Brand, 2).alias_of_id is None and effective(db, "brand", [2], ["alias_of"]) == {} with pytest.raises(writer.Refused) as refused: undo_service.undo_batch(db, a.row.uid, RIAN, "twice") assert refused.value.code == "DECISION_SUPERSEDED" def test_a_brand_fold_carries_the_variants_it_moved_when_the_decision_row_is_first_written(db): """The variants on the folded brand move across, and what they moved from is in the decision row from the moment it is written. Cost: a confirm of the very first brand fold raised "decisions is append-only: UPDATE refused" against the real database, because the applier moved the variants and then went back to write their old brand into the row's detail. Every test passed, since SQLite has no such trigger; the fixture now refuses that UPDATE too (`forbid_ledger_updates`). Undo needs the detail, so the fix is to compute it before the row exists, not to drop it. """ with writer.batch(db, "route", "individual", RIAN, commit=True) as b: merges.apply_brand_alias(db, db.get(Brand, 2), db.get(Brand, 1), decided_by=RIAN, preferred_name="Rabanne", batch=b) alias_row = db.scalar(select(Decision).where(Decision.entity_type == "brand", Decision.field == "alias_of")) moved = (alias_row.detail or {}).get("brands_moved") assert moved, "the decision must carry what it moved, or undo cannot put it back" assert sorted(v for v, _ in moved) == [2, 3] assert all(was == 2 for _, was in moved) assert [db.get(ProductVariant, i).brand_id for i in (1, 2, 3)] == [1, 1, 1] with writer.batch(db, "route", "individual", RIAN, commit=True) as b: undo_service.undo(b, alias_row.id, "not the same house after all") assert [db.get(ProductVariant, i).brand_id for i in (1, 2, 3)] == [1, 2, 2]