"""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 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) 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"