"""The review area's three tabs, over one staging-shaped database (Stream K10.6). Rian, 17 Sep: *"I'd like to see some tabs. One where a human can see how decisions are being made... I'd like to see the folding/splitting review in another tab... And I'd like to see the publish candidates in a 3rd tab."* Each tab has its own suite for its own rules (`test_process_doc.py`, `test_folding_work.py`, `test_review_sheet.py`); this one pins that the THREE READ TOGETHER off one database that has what staging has -- pairs at every level, a rule's proposals, values on variants, and index candidates -- and that reading the whole area writes nothing. A tab that only works when the other two have no data is a tab that works on a fixture and not on staging. SQLite kit, no network. """ from __future__ import annotations import pytest from sqlalchemy import create_engine, 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, ProductLine, ProductVariant, Proposal, ProposalPass, Retailer, Shop, Source, Suggestion, AttributeAlias) from app.models.places import Place, ShopPlace from app.services import folding, process_doc, proposals_store, publish 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 staging(monkeypatch): """One brand, two product lines, two variants, a pair at each level, a rule's reading of the line pair, a shade on a variant, and three index candidates: the shape of a real host.""" engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) Base.metadata.create_all(engine, tables=TABLES) monkeypatch.setattr(publish, "blocked_shop_ids", lambda db: [], raising=False) 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, line in ((1, 1), (2, 2)): s.add(ProductVariant(id=n, name=f"1 Million {n} 100ml", 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() 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"}), Suggestion(id=10, level="index:brand", left_id=1, reason="index_rule", score=0.5, detail={"why": "3 variants priced across 2 or more places"}), Suggestion(id=11, level="index:line", left_id=1, reason="index_rule", score=0.5, detail={"why": "2 variants compared across 2 places"}), Suggestion(id=12, level="index:line", left_id=2, reason="index_rule", score=0.5, detail={"why": "2 variants compared across 2 places"})]) s.commit() proposals_store.write(s, "rule:drink_words:6", [ proposals_store.ProposalRow( entity_type="suggestion", natural_key=f"pair:line:line:{s.get(ProductLine, 1).uid}||line:{s.get(ProductLine, 2).uid}", natural_key_detail={"level": "line"}, field="decision", value={"decision": "same"}, reason="rule:drink_words: removed 'royal'; the two product lines then read as one", evidence=[{"field": "name", "text": "1 Million Royal", "matched": "royal"}], confidence=None, brand_slug="rabanne")], kind="rule", generator="rule", rules_version="6", process_version="4", note="what the list would have grouped") proposals_store.write(s, "rule:shade_shapes:6", [ proposals_store.ProposalRow( entity_type="product_variant", natural_key=f"variant:{s.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 1 100ml", "matched": "99 pirate"}], confidence=None, brand_slug="rabanne")], kind="rule", generator="rule", rules_version="6", process_version="4", note="unmarked shades") s.commit() yield s class TestTheThreeTabsReadTogether: def test_tab_one_hands_over_the_document_and_every_pass_that_wrote_a_proposal(self, staging): doc = {**process_doc.read(), "passes": process_doc.passes(staging)} assert doc["missing"] is False and doc["version"], "the authority is on the host" assert [s["number"] for s in doc["sections"] if s["number"]][:2] == ["1", "2"] by_name = {p["name"]: p for p in doc["passes"]} assert by_name["rule:drink_words:6"]["open"] == 1 and by_name["rule:drink_words:6"]["generator"] == "rule" assert by_name["rule:shade_shapes:6"]["brands"] == ["rabanne"], "each block links to the rows it produced" def test_tab_two_counts_all_four_levels_and_lists_each(self, staging): counts = {c["level"]: c for c in folding.counts(staging)} assert counts["brand"]["pairs"] == 1 and counts["product_line"]["pairs"] == 1 assert counts["product_line"]["pairs_with_a_reason"] == 1 assert counts["attribute"]["values"] == 1 and counts["product_variant"]["pairs"] == 1 for level in folding.LEVELS: page = folding.rows(staging, level) assert page["rows"], f"{level} has rows" assert all(r["changes"] and (r["reason"] or r["field"]) for r in page["rows"]) def test_tab_three_lists_the_candidates_with_their_reason_and_pages(self, staging): desk = publish.desk(staging, limit=2) assert desk["totals"] == {"brand": 1, "product_line": 2, "place": 0} assert desk["shown"] == 3 and len(desk["rows"]) == 2 assert all(r["why"] for r in desk["rows"]), "what made each a candidate" assert publish.desk(staging, limit=2, offset=2)["rows"], "and page two is reachable" def test_the_rule_s_matched_words_reach_the_folding_tab_and_the_row_is_not_stale(self, staging): """0.50.1 put the words a rule matched in their own field, because a character span broke the sheet and a rule cites a name rather than one listing. The folding tab reads the same evidence, so it is pinned here too.""" from app.services import proposals row = next(r for r in folding.rows(staging, "product_line")["rows"] if r["kind"] == "pair") assert row["evidence"][0]["matched"] == "royal" and "span" not in row["evidence"][0] ok, why, _ = proposals.evidence_state(proposals.ListingCache(staging), db_evidence(staging)) assert ok and why is None def test_reading_the_whole_review_area_writes_nothing(self, staging): before = (staging.query(Suggestion).count(), staging.query(Proposal).count(), staging.query(Decision).count(), staging.query(DecisionBatch).count(), staging.scalar(select(func.count(ProductVariant.id)))) process_doc.read() process_doc.passes(staging) folding.counts(staging) for level in folding.LEVELS: folding.rows(staging, level) publish.desk(staging) assert not staging.new and not staging.dirty and not staging.deleted assert (staging.query(Suggestion).count(), staging.query(Proposal).count(), staging.query(Decision).count(), staging.query(DecisionBatch).count(), staging.scalar(select(func.count(ProductVariant.id)))) == before def db_evidence(db): return db.scalar(select(Proposal).where(Proposal.entity_type == "suggestion")).evidence