"""Applying an alias makes the catalogue follow (Stream M, task M4). The fourteen 1 Million rows again, on the in-memory SQLite, with a listing each: before rian confirms "Paco Rabanne is Rabanne" they are two brands with two lines and the 100 ml Eau de Toilette is two product variants; applying the alias points the row, moves the lines and the product variants, re-keys them under the brand, folds the two Eau de Toilette rows and the two 200 ml Elixirs, and records who decided. A line alias and a attribute alias do the same at their level. A second application changes nothing. """ from __future__ import annotations import pytest from sqlalchemy import create_engine, select from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool from app import cli from app.models import ( RawRecord, Account, Award, Base, Brand, Listing, Shop, Suggestion, ProductVariant, ProductLine, Merge, Retailer, AttributeAlias, ) from app.models.decisions import LEDGER_TABLES from app.models.places import Place, ShopPlace from app.services import merges, overrides from tests.test_product_lines import ONE_MILLION TABLES = [ Account.__table__, Brand.__table__, ProductLine.__table__, ProductVariant.__table__, AttributeAlias.__table__, Retailer.__table__, Shop.__table__, Listing.__table__, Award.__table__, Merge.__table__, Suggestion.__table__, RawRecord.__table__, *LEDGER_TABLES, Place.__table__, ShopPlace.__table__, ] 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) factory = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False, future=True) overrides._present.clear() with factory() as session: session.add(Account(id=RIAN, username="rian", display_name="rian")) yield session overrides._present.clear() def seed(db): """The fourteen rows, each at its own shop, keyed and lined the way the deploy leaves them.""" 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 = ProductVariant(id=n, name=name, brand=brand, brand_id=paco.id if brand == "Paco Rabanne" else rabanne.id, vertical="beauty", quantity_ml=size, match_key="pending", attributes={}) shop = Shop(retailer_id=retailer.id, code=f"S{n}", iata=f"A{n:02d}", name=f"Shop {n}", currency="EUR") db.add_all([product, shop]) db.flush() db.add(Listing(variant_id=product.id, shop_id=shop.id, source_sku=f"sku{n}")) db.commit() cli.backfill_lines(db) cli.backfill_attributes(db) from app.services import keying maps = keying.load_maps(db) merges.rekey_product_variants(db, list(db.scalars(select(ProductVariant))), maps) merges.merge_duplicates(db) db.commit() return rabanne, paco def alive(db): return {p.id: p for p in db.scalars(select(ProductVariant).where(ProductVariant.merged_into_id.is_(None)))} class TestBrandAlias: def test_before_the_alias_two_houses_keep_the_same_bottle_apart(self, db): seed(db) rows = alive(db) # Inside the Rabanne brand the rules already fold the two 100 ml Eau de Toilette # spellings (the brief's 14747 and 17217) and the two 200 ml Elixirs; nothing of # Paco Rabanne's agrees with anything across the two brands. assert len(rows) == 12 assert db.get(ProductVariant, 10).merged_into_id == 6 and db.get(ProductVariant, 13).merged_into_id == 9 assert rows[6].match_key == "rabanne|1-million|concentration=edt|100ml" and rows[4].match_key == "paco-rabanne|1-million||100ml" def test_the_alias_moves_lines_rekeys_products_and_folds_what_then_agrees(self, db): rabanne, paco = seed(db) counts = merges.apply_brand_alias(db, paco, rabanne, decided_by=RIAN, preferred_name="Rabanne") db.commit() assert (paco.alias_of_id, paco.decided_by) == (rabanne.id, RIAN) and paco.decided_at is not None # The one ledger (the catalogue decisions §2.2): the alias, the preferred name with the # spelling the rules had, and the line that followed the alias, each set by rian. alias_row = overrides.fields_of(db, "brand", str(paco.id))["alias_of_id"] assert (alias_row.value, alias_row.set_by, alias_row.collected_value) == (rabanne.id, RIAN, None) name_row = overrides.fields_of(db, "brand", str(rabanne.id))["name"] assert (name_row.value, name_row.collected_value, name_row.set_by) == ("Rabanne", "Rabanne", RIAN) line_rows = overrides.read(db, "line", fields=("alias_of_id",)) assert len(line_rows) == 1 and next(iter(line_rows.values()))["alias_of_id"].set_by == RIAN assert counts["lines_pointed"] == 1 and counts["product_variants_rekeyed"] == 5 assert counts["groups_merged"] == 0, "the alias's own five rows agreed with nothing yet: their attribute is unknown or Parfum at other sizes" rows = alive(db) assert all(p.match_key.startswith("rabanne|") for p in rows.values()) bottle_line = db.scalar(select(ProductLine.id).where(ProductLine.brand_id == rabanne.id, ProductLine.key == "1 million")) # Every bottle is on the one line; the coffret keeps its own (identity rules v6; a set is # never a member of the bottle's line). assert {p.product_line_id for p in rows.values() if "Set" not in p.name} == {bottle_line} alias_line = db.scalar(select(ProductLine).where(ProductLine.brand_id == paco.id)) assert alias_line.alias_of_id is not None and alias_line.decided_by == RIAN assert rows[4].match_key == "rabanne|1-million||100ml", "the bare 10cl keys apart on an unknown attribute and waits for a person" again = merges.apply_brand_alias(db, paco, rabanne, decided_by=RIAN) assert again["lines_pointed"] == 0 and again["groups_merged"] == 0 def test_the_same_bottle_at_two_shops_folds_once_it_agrees(self, db): """14747 and 17217 in the brief: one EDT 100 ml, two spellings, two shops.""" rabanne, paco = seed(db) db.add(ProductVariant(id=99, name="1 Million EDT 100ml", brand="Paco Rabanne", brand_id=paco.id, vertical="beauty", quantity_ml=100, match_key="paco-rabanne|1-million|concentration=edt|100ml", attributes={"concentration": "edt", "attribute": "edt"}, product_line_id=db.scalar(select(ProductLine.id).where(ProductLine.brand_id == paco.id)))) db.flush() shop = Shop(retailer_id=1, code="S99", iata="A99", name="Shop 99", currency="EUR") db.add(shop) db.flush() db.add(Listing(variant_id=99, shop_id=shop.id, source_sku="sku99")) db.commit() counts = merges.apply_brand_alias(db, paco, rabanne, decided_by=RIAN) db.commit() assert counts["groups_merged"] == 1 and counts["rows_merged"] == 1 survivor = db.get(ProductVariant, 6) if db.get(ProductVariant, 99).merged_into_id else db.get(ProductVariant, 99) assert survivor.merged_into_id is None record = db.scalar(select(Merge).where((Merge.from_id == 99) | (Merge.to_id == 99))) assert record.merged_by == RIAN, "a merge an alias caused is recorded against the person who confirmed the alias" assert len(db.scalars(select(Listing).where(Listing.variant_id == survivor.id)).all()) == 3 # What moved is on the record, so an unmerge can reverse it from the record alone. loser_id = record.from_id assert record.detail["listings"] == [db.scalar(select(Listing.id).where(Listing.source_sku == ("sku99" if loser_id == 99 else "sku6")))] def test_the_alias_lines_own_decisions_follow_its_product_variants(self, db): """A review or a preferred name on the alias brand's line moves to the brand's line the product variants went to (the catalogue decisions §2.2): the decision is about the line, not the row id the rules minted for one spelling.""" rabanne, paco = seed(db) paco_line = db.scalar(select(ProductLine).where(ProductLine.brand_id == paco.id)) overrides.decide(db, "line", str(paco_line.id), "review", "hidden", set_by=RIAN) overrides.decide(db, "line", str(paco_line.id), "name", "One Million", set_by=RIAN, collected_value="1 Million") db.commit() counts = merges.apply_brand_alias(db, paco, rabanne, decided_by=RIAN) db.commit() target = db.scalar(select(ProductLine).where(ProductLine.brand_id == rabanne.id, ProductLine.key == "1 million")) assert counts["line_decisions_moved"] == 2 and counts["line_decisions_kept"] == 0 assert set(overrides.fields_of(db, "line", str(target.id))) >= {"review", "name"} assert target.name == "One Million", "a moved name reaches the column too" # The ledger is append-only (K2): the alias line keeps its rows as history beside its alias. assert set(overrides.fields_of(db, "line", str(paco_line.id))) == {"alias_of_id", "review", "name"} class TestLineAlias: def test_a_line_alias_moves_its_products_and_rekeys_them(self, db): rabanne, _ = seed(db) one = db.scalar(select(ProductLine).where(ProductLine.brand_id == rabanne.id, ProductLine.key == "1 million")) other = ProductLine(brand_id=rabanne.id, key="one million", name="One Million", slug="rabanne-one-million") db.add(other) db.flush() db.add(ProductVariant(id=50, name="One Million Eau de Toilette 200 ml", brand="Rabanne", brand_id=rabanne.id, vertical="beauty", quantity_ml=200, match_key="rabanne|one-million|concentration=edt|200ml", attributes={"attribute": "edt"}, product_line_id=other.id)) db.commit() counts = merges.apply_line_alias(db, other, one, decided_by=RIAN, preferred_name="1 Million") db.commit() assert counts["product_variants_moved"] == 1 and counts["groups_merged"] == 1 assert (other.alias_of_id, other.decided_by) == (one.id, RIAN) and one.name == "1 Million" assert overrides.fields_of(db, "line", str(other.id))["alias_of_id"].value == one.id assert overrides.fields_of(db, "line", str(one.id))["name"].collected_value == "1 Million" moved = db.get(ProductVariant, 50) assert moved.product_line_id == one.id and moved.match_key == "rabanne|1-million|concentration=edt|200ml" assert moved.merged_into_id == 11 or db.get(ProductVariant, 11).merged_into_id == 50 def test_two_houses_lines_never_fold_before_the_brand_alias(self, db): rabanne, paco = seed(db) mine = db.scalar(select(ProductLine).where(ProductLine.brand_id == paco.id)) theirs = db.scalar(select(ProductLine).where(ProductLine.brand_id == rabanne.id)) with pytest.raises(ValueError): merges.apply_line_alias(db, mine, theirs, decided_by=RIAN) class TestVariationAlias: def test_a_wording_mapped_elsewhere_rekeys_every_product_that_carries_it(self, db): seed(db) counts = merges.apply_variation_alias(db, "beauty", "elixir parfum intense", "parfum intense", decided_by=RIAN) db.commit() assert counts["product_variants_rekeyed"] == 3 row = db.scalar(select(AttributeAlias).where(AttributeAlias.raw == "elixir parfum intense")) assert (row.canonical, row.display, row.decided_by) == ("parfum intense", "Parfum Intense", RIAN) ledger = overrides.fields_of(db, "attribute_alias", overrides.attribute_alias_key("beauty", "elixir parfum intense")) assert (ledger["canonical"].value, ledger["canonical"].collected_value, ledger["canonical"].set_by) == ("parfum intense", "elixir", RIAN) assert ledger["display"].value == "Parfum Intense" keys = {p.id: p.match_key for p in db.scalars(select(ProductVariant)) if p.merged_into_id is None} assert keys[7] == "rabanne|1-million|concentration=parfum-intense|50ml" and keys[9] == "rabanne|1-million|concentration=parfum-intense|200ml" # 13 ("Elixir Eau de Parfum Intense", the other wording) was folded into 9 while both # read Elixir; a merge is forward and recorded, so a later alias never undoes it: a # person does, by hand, if the new reading says they were two bottles. assert db.get(ProductVariant, 13).merged_into_id == 9