"""Everything waiting to be joined or kept apart, at every level (Stream K10.3). Rian, 17 Sep, on the review area: *"It seems like it focuses on brand folding, but what about product line decisions? I hope we haven't missed the product line decisions because that's where I think a lot of the folding will take place... eg the folding shades or misspellings, etc. And then there is option folding/splitting too I think?"* Nothing was missed; the sheet is arranged by brand, so a brand's sheet was all he could see. What these pin is the reader behind the page that shows all four levels at once -- and the trap it was written to avoid: a rule's reading of a pair is a REASON on that pair, not a second row beside it. Adding the proposals table to the suggestions table prints 3,255 waiting at line level where 2,572 pairs wait, and a person would go looking for a thousand rows that do not exist. Measured on `dfp_k10`, a copy of the 17 Sep post-chain dump: brands 74 pairs and 91 values, product lines 2,572 pairs of which 683 carry a rule's reason, attributes 164 values, product variants 1,012 pairs. SQLite kit here, no network. """ from __future__ import annotations 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, LEDGER_TABLES, Listing, Merge, ProductLine, ProductVariant, Proposal, ProposalPass, Retailer, Shop, Source, Suggestion, AttributeAlias) from app.models.places import Place, ShopPlace from app.services import folding, proposals_store 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] @pytest.fixture def db(): engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) Base.metadata.create_all(engine, tables=TABLES) 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="rabanne", name="Rabanne"), Brand(id=2, slug="paco-rabanne", name="Paco Rabanne")]) s.flush() s.add_all([ProductLine(id=1, brand_id=1, key="1 million", slug="rabanne-1-million", name="1 Million"), ProductLine(id=2, brand_id=1, key="1 million royal", slug="rabanne-1-million-royal", name="1 Million Royal")]) s.flush() for n, (name, line) in enumerate([("1 Million 100ml", 1), ("1 Million Eau de Toilette 100 ml", 1)], start=1): s.add(ProductVariant(id=n, name=name, brand="Rabanne", brand_id=1, vertical="beauty", match_key=f"k{n}", product_line_id=line, quantity_state="stated", quantity_value=100, quantity_unit="ml", attributes={})) s.flush() # One pair at each level, as `suggest` writes them. s.add_all([Suggestion(id=1, level="brand", left_id=1, right_id=2, reason="brand_within", score=0.95, detail={"why": '"Rabanne" is within "Paco Rabanne"'}), Suggestion(id=2, level="line", left_id=1, right_id=2, reason="line_within", score=0.8, detail={"why": 'Every word of "1 Million" is in "1 Million Royal"'}), Suggestion(id=3, level="product", left_id=1, right_id=2, reason="attribute_silent", score=0.6, detail={"why": "Same line and size; one names no attribute"})]) s.commit() yield s def _rule_rows(db, rows, name="rule:drink_words:6"): proposals_store.write(db, name, rows, kind="rule", generator="rule", rules_version="6", process_version="4", note="what the list would have grouped") db.commit() def _pair_proposal(natural_key): return proposals_store.ProposalRow( entity_type="suggestion", natural_key=natural_key, natural_key_detail={"level": "line"}, field="decision", value={"decision": "same"}, reason="rule:drink_words: removed 'whisky'; the two product lines then read as one", evidence=[{"field": "name", "text": "1 Million Royal", "matched": "royal"}], confidence=None, brand_slug="rabanne") class TestWhereTheWorkIs: def test_all_four_levels_are_counted_not_brands_alone(self, db): by_level = {c["level"]: c for c in folding.counts(db)} assert list(by_level) == ["brand", "product_line", "attribute", "product_variant"] assert by_level["brand"]["pairs"] == 1 assert by_level["product_line"]["pairs"] == 1 assert by_level["product_variant"]["pairs"] == 1 assert sum(c["total"] for c in folding.counts(db)) == 3 def test_each_level_says_what_it_asks_and_what_approving_it_changes(self, db): for c in folding.counts(db): assert c["asks"] and c["changes"], c["level"] by_level = {c["level"]: c for c in folding.counts(db)} assert "one company" in by_level["brand"]["asks"] assert "misspelling" in by_level["product_line"]["asks"] assert "same bottle at two shops" in by_level["product_variant"]["asks"] def test_a_value_proposal_is_counted_at_the_level_its_field_belongs_to(self, db): _rule_rows(db, [proposals_store.ProposalRow( entity_type="product_variant", natural_key=f"variant:{db.get(ProductVariant, 1).uid}", natural_key_detail=None, field="attribute:color", value="99 pirate", reason="rule:shade_shapes: the name ends in a shade", evidence=[{"field": "name", "text": "1 Million 100ml", "matched": "99 pirate"}], confidence=None, brand_slug="rabanne")], name="rule:shade_shapes:6") by_level = {c["level"]: c for c in folding.counts(db)} assert by_level["attribute"]["values"] == 1, "an attribute reading is attribute work, not variant work" assert by_level["product_variant"]["values"] == 0 def test_a_brand_name_proposal_is_brand_work(self, db): _rule_rows(db, [proposals_store.ProposalRow( entity_type="brand", natural_key="brand:rabanne", natural_key_detail={"name": "Rabanne"}, field="name", value="Rabanne", reason="rule:brand_trailers: these spellings sit on this brand", evidence=[], confidence=None, brand_slug="rabanne")], name="rule:brand_trailers:6") by_level = {c["level"]: c for c in folding.counts(db)} assert by_level["brand"]["values"] == 1 and by_level["brand"]["total"] == 2 class TestAPairIsCountedOnce: """The trap this reader exists to avoid. On the real catalogue the line level holds 2,572 pairs, 683 of which a rule has read; adding the two tables prints 3,255 and sends a person looking for 683 rows that are the same 683 pairs.""" def test_a_rules_reading_of_a_pair_is_a_reason_on_it_never_a_second_row(self, db): key = f"pair:line:line:{db.get(ProductLine, 1).uid}||line:{db.get(ProductLine, 2).uid}" _rule_rows(db, [_pair_proposal(key)]) by_level = {c["level"]: c for c in folding.counts(db)} assert by_level["product_line"]["pairs"] == 1 assert by_level["product_line"]["values"] == 0, "a pair proposal is not a value" assert by_level["product_line"]["pairs_with_a_reason"] == 1 assert by_level["product_line"]["total"] == 1, "one thing waits, not two" def test_the_reason_and_its_evidence_ride_on_the_pair_s_own_row(self, db): key = f"pair:line:line:{db.get(ProductLine, 1).uid}||line:{db.get(ProductLine, 2).uid}" _rule_rows(db, [_pair_proposal(key)]) rows = folding.rows(db, "product_line")["rows"] assert len(rows) == 1 row = rows[0] assert row["kind"] == "pair" and row["rule"] == "rule:drink_words:6" assert "the two product lines then read as one" in row["reason"] assert row["evidence"][0]["matched"] == "royal" assert row["acted_on"] == "sheet" and row["brand_slug"] == "rabanne" def test_a_pair_no_rule_has_read_keeps_its_own_why_and_goes_to_the_desk(self, db): row = folding.rows(db, "product_variant")["rows"][0] assert row["rule"] is None and "Same line and size" in row["reason"] and row["acted_on"] == "desk" class TestTheRows: def test_a_pair_names_both_sides_as_a_person_reads_them(self, db): row = folding.rows(db, "brand")["rows"][0] assert {row["left"]["label"], row["right"]["label"]} == {"Rabanne", "Paco Rabanne"} assert row["left"]["path"].startswith("/brands/") and row["what"] == "join" def test_a_product_line_side_says_which_brand_it_sits_under(self, db): row = folding.rows(db, "product_line")["rows"][0] assert row["left"]["detail"].startswith("Rabanne ยท ") def test_a_value_row_names_the_field_and_the_value(self, db): _rule_rows(db, [proposals_store.ProposalRow( entity_type="product_variant", natural_key=f"variant:{db.get(ProductVariant, 1).uid}", natural_key_detail=None, field="attribute:color", value="99 pirate", reason="rule:shade_shapes: the name ends in a shade", evidence=[], confidence=None, brand_slug="rabanne")], name="rule:shade_shapes:6") row = folding.rows(db, "attribute")["rows"][0] assert row["kind"] == "value" and row["field"] == "attribute:color" and row["value"] == "99 pirate" assert row["what"] == "set" and row["left"]["label"] == "1 Million 100ml" def test_the_page_counts_pairs_and_values_and_pages_through_both(self, db): _rule_rows(db, [proposals_store.ProposalRow( entity_type="brand", natural_key="brand:rabanne", natural_key_detail={"name": "Rabanne"}, field="name", value="Rabanne", reason="rule:brand_trailers: the spellings on this row", evidence=[], confidence=None, brand_slug="rabanne")], name="rule:brand_trailers:6") page = folding.rows(db, "brand") assert page["pairs"] == 1 and page["values"] == 1 and page["total"] == 2 assert [r["kind"] for r in page["rows"]] == ["pair", "value"], "pairs first, the older question" first = folding.rows(db, "brand", limit=1, offset=0) second = folding.rows(db, "brand", limit=1, offset=1) assert [r["kind"] for r in first["rows"]] == ["pair"] and [r["kind"] for r in second["rows"]] == ["value"] def test_a_decided_pair_is_gone_from_the_page(self, db): db.get(Suggestion, 1).decision = "separate" db.commit() assert folding.rows(db, "brand")["pairs"] == 0 assert {c["level"]: c["pairs"] for c in folding.counts(db)}["brand"] == 0 def test_an_unknown_level_is_refused_rather_than_answered_empty(self, db): with pytest.raises(ValueError): folding.rows(db, "shops") def test_reading_the_page_writes_nothing(self, db): before = (db.scalar(select(Suggestion.id).order_by(Suggestion.id)), db.query(Proposal).count()) for level in folding.LEVELS: folding.rows(db, level) folding.counts(db) assert not db.new and not db.dirty and not db.deleted assert (db.scalar(select(Suggestion.id).order_by(Suggestion.id)), db.query(Proposal).count()) == before class TestThePublishCandidatesPage: """K10.4: the index candidates become their own tab, which means a person has to be able to REACH page two. The panel asked for the first hundred and said "the rest follow once these are decided"; on the real catalogue that is the first hundred of 1,701 product lines, so finding one page meant deciding a hundred others first.""" @staticmethod def _candidates(db): from app.models import Suggestion for n, (level, left) in enumerate([("index:brand", 1), ("index:brand", 2), ("index:line", 1), ("index:line", 2)], start=10): db.add(Suggestion(id=n, level=level, left_id=left, reason="index_rule", score=0.5, detail={"why": "three variants across two places"})) db.commit() def test_the_filter_leaves_a_count_to_page_through(self, db): from app.services import publish self._candidates(db) page = publish.desk(db, limit=2, offset=0) assert page["shown"] == 4 and page["offset"] == 0 and len(page["rows"]) == 2 assert publish.desk(db, limit=2, offset=2)["rows"], "page two is reachable" assert {r["suggestion_id"] for r in page["rows"]} != {r["suggestion_id"] for r in publish.desk(db, limit=2, offset=2)["rows"]} def test_a_name_or_address_finds_one_page_without_deciding_the_others(self, db): from app.services import publish self._candidates(db) found = publish.desk(db, q="royal") assert found["shown"] == 1 and found["rows"][0]["name"] == "1 Million Royal" assert publish.desk(db, q="rabanne-1-million-royal")["shown"] == 1, "the address matches too" assert publish.desk(db, q="nothing here")["shown"] == 0 def test_the_kind_totals_do_not_move_when_the_filter_does(self, db): from app.services import publish self._candidates(db) everything = publish.desk(db)["totals"] assert publish.desk(db, q="royal")["totals"] == everything, "the counts say what is waiting, not what is shown"