"""The sheet read (Stream K4; the escalation-2 spec ยง6.3 and test 16). What it cost before: a listing endpoint that created rows on read deadlocked against a running collection and the page hung (agents.md, "A GET must never write"); a sheet computed from its load snapshot would show a decision undone an hour ago as still standing; a staleness rule that followed a rederive would stale a whole sheet between its load and its approval. SQLite kit, no network.""" from __future__ import annotations import pytest from sqlalchemy import select from app.main import app import uuid as _uuid from uuid import uuid4 as _uuid4 from app.models import Decision, Listing, ProductVariant, Proposal from app.services import access, proposals from app.services.decisions.writer import Refused from tests.test_proposals_load import _counts, _file, db # noqa: F401 (the fixture) def _sheet(db, **kw): return proposals.sheet(db, "chanel", **kw) def test_the_sheet_read_writes_nothing(db): proposals.load(db, _file(db)) before = _counts(db) stamps = {r.uid: (r.updated_at, r.status, r.detail) for r in db.scalars(select(Proposal))} db.get(Listing, 4).listed_name = "Rouge Allure 3.5g 104 Passion" # a stale row is shown stale, never written stale db.commit() data = _sheet(db) assert data["counts"]["stale_now"] >= 1 assert not db.new and not db.dirty and not db.deleted assert _counts(db) == before assert {r.uid: (r.updated_at, r.status, r.detail) for r in db.scalars(select(Proposal))} == stamps proposals.sheets(db) assert not db.new and not db.dirty def test_rows_group_under_their_product_line_in_the_pass_order_with_spot_checks_listed_first(db): data = _file(db) data["proposals"].insert(0, {"entity": {"type": "brand", "key": "brand:chanel"}, "field": "name", "value": "Chanel", "confidence": 0.95, "reason": "the brand as a shopper writes it", "position": 9, "evidence": [{"listing": data["proposals"][0]["evidence"][0]["listing"], "source": "listed_name", "span": [0, 5], "text": "Rouge"}]}) proposals.load(db, data) sheet = _sheet(db) assert sheet["groups"][0]["ref"] is None and sheet["groups"][0]["rows"][0]["entity_type"] == "brand", "brand rows first" line = sheet["groups"][1] assert line["ref"] == "line:new:chanel-rouge-allure" and line["name"] == "Rouge Allure" and line["header_uid"] assert [r["position"] for r in line["rows"]] == sorted(r["position"] for r in line["rows"]) merge = next(r for r in line["rows"] if r["field"] == "decision") assert merge["spot_check"] and merge["uid"] in sheet["spot_check_uids"] assert sheet["counts"]["open"] == 5 def test_the_effective_decision_is_computed_now_with_who_mode_and_pass(db): data = _file(db) proposals.load(db, data) color = db.scalar(select(Proposal).where(Proposal.field == "attribute:color")) proposals.approve(db, "chanel", data["pass"]["name"], scope={"proposal_uids": [str(color.uid)]}, by="rian") second = _file(db, name="claude/2026-09-18/chanel-2") second["proposals"][2]["value"] = "99 Pirate Red" second["proposals"][2]["evidence"][0].update(span=[0, 9], text="99 Pirate") proposals.load(db, second) row = next(r for g in _sheet(db, pass_name="claude/2026-09-18/chanel-2")["groups"] for r in g["rows"] if r["field"] == "attribute:color") eff = row["effective"] assert eff["by"] == "rian" and eff["mode"] == "individual" and eff["origin"] == "proposal" assert eff["pass_name"] == data["pass"]["name"] and eff["process_version"] == "2" assert row["held"] is True, "a bulk approval would leave a person's individual decision" assert row["disagrees_with_bulk"] is False, "the flag is for a bulk decision only" def test_drift_is_flagged_when_the_column_changed_since_the_load(db): proposals.load(db, _file(db)) db.get(ProductVariant, 4).product_line_id = 2 db.commit() row = next(r for g in _sheet(db)["groups"] for r in g["rows"] if r["field"] == "product_line") assert row["drift"] is True and row["current_value"] != row["loaded_value"] def test_staleness_is_a_change_in_what_was_read_never_a_rederive(db): """Spec test 16: moving the variant's line and rule (what a rederive does) stales nothing; the listed text at the cited span changing stales the row.""" proposals.load(db, _file(db)) db.get(ProductVariant, 4).product_line_id = 1 db.get(ProductVariant, 4).match_key = "rederived" db.commit() rows = [r for g in _sheet(db)["groups"] for r in g["rows"]] assert not any(r["stale_now"] for r in rows) db.get(Listing, 1).listed_variant = "104 Passion" db.commit() stale = [r for g in _sheet(db)["groups"] for r in g["rows"] if r["stale_now"]] assert [r["field"] for r in stale] == ["attribute:color"] assert "now reads" in stale[0]["stale_why"] db.get(Listing, 4).ignored_at = db.get(Listing, 4).ignored_at or __import__("datetime").datetime.now() db.commit() assert {r["field"] for g in _sheet(db)["groups"] for r in g["rows"] if r["stale_now"]} == {"attribute:color", "product_line", "decision"} def test_the_history_lists_the_brands_batches_undone_or_not_and_the_ledger_reads_newest_first(db): data = _file(db) proposals.load(db, data) color = db.scalar(select(Proposal).where(Proposal.field == "attribute:color")) uid = proposals.approve(db, "chanel", data["pass"]["name"], scope={"proposal_uids": [str(color.uid)]}, by="rian")["batch_uid"] batches = _sheet(db)["batches"] assert [b["uid"] for b in batches] == [uid] and not batches[0]["undone"] and batches[0]["decisions"] == 1 proposals.undo_batch(db, uid, "rian", "the shade was read from the wrong field") sheet = _sheet(db) assert sheet["batches"][0]["undone"] is True row = next(r for g in sheet["groups"] for r in g["rows"] if r["field"] == "attribute:color") assert row["status"] == "open" and row["reopened"] is True, "an undone approval waits again" ledger = proposals.decisions_for(db, "product_variant", 1) assert [d["reverses_id"] is not None for d in ledger] == [True, False] detail = proposals.batch_detail(db, uid) assert detail["undone"] and detail["decisions"][0]["field"] == "attribute:color" def test_every_review_route_is_owner_only(): from tests.test_route_inventory import walk keys = [f"{method} {route.path}" for route in walk(app.routes) if getattr(route, "path", "").startswith("/api/review") for method in sorted(getattr(route, "methods", None) or [])] # K4's seven, the two index candidate routes (K6, routers/publish.py), the brand split and the # word lists (K9), the process doc and the two folding reads (K10). # K4's seven, K6's two index candidates, K9's brand split and word lists, K10's process doc # and two folding reads, K11's hints, notes and line-table reads. Each stream bumped this # number on its own branch; the merged figure is asserted once, here. assert len(keys) == 26 # plus the row detail and its listings; K12's register read and overturn for key in keys: assert access.classify(key) == ("permission", "owner"), key assert access.spa_class("/review") == ("permission", "owner") def test_no_review_route_answers_an_anonymous_caller_or_a_member_who_is_not_the_owner_even_when_the_site_is_open(): """The review writes the catalogue's decided layer: the policy refuses it before its handler runs for anyone but the owner, whatever SITE_ACCESS says (checked live on dfp_k4: 401 anonymous).""" from tests.test_route_inventory import walk keys = [f"{method} {route.path}" for route in walk(app.routes) if getattr(route, "path", "").startswith("/api/review") for method in sorted(getattr(route, "methods", None) or [])] base = dict(page=None, must_change=False, read_only=False, origin_ok=True, wants_html=False) for key in keys: method = key.split()[0] for site_open in (False, True): anonymous = access.decide(method=method, key=key, site_open=site_open, signed_in=False, active=False, holds=lambda p: False, **base) assert anonymous is not None and anonymous.status in (401, 403), key member = access.decide(method=method, key=key, site_open=site_open, signed_in=True, active=True, holds=lambda p: p != "owner", **base) assert member is not None and member.status == 403, key assert access.decide(method=method, key=key, site_open=False, signed_in=True, active=True, holds=lambda p: p == "owner", **base) is None def test_a_pair_row_carries_both_sides_by_name_and_every_row_names_its_subject(db): """What it cost: the sheet sent a pair row's natural key and nothing else, so the question "are these two the same product line?" reached the page with neither side on it. Rian opened a brand's sheet on staging on 17 Sep, found sixteen such rows, and could not answer one of them: "product line 1196afb2 is not useful", "I dont even know what feedback to give". The names were never in the response to begin with, so no amount of wording on the page could have fixed it.""" data = _file(db) proposals.load(db, data) rows = [r for g in _sheet(db)["groups"] for r in g["rows"]] pair = next(r for r in rows if r["field"] == "decision") assert len(pair["sides"]) == 2, "a pair is asked about two things; both must reach the page" assert all(s["name"] for s in pair["sides"]), f"a side with no name is a uuid to compare: {pair['sides']}" assert [s["kind"] for s in pair["sides"]] == ["product_variant", "product_variant"] assert pair["sides"][0]["name"] != pair["sides"][1]["name"], "two sides that read alike explain nothing" # Every other row names what it is about, so no row is identified by its key alone. for row in rows: named = row["subject_name"] or row["sides"] or row["listings"] assert named, f"{row['field']} on {row['natural_key']} reaches the page unidentified" def test_the_sheet_shows_every_live_pass_so_a_brand_is_one_queue(db): """What it cost: each word list loads as its own pass, and the sheet drew the newest one alone. One brand's nineteen waiting rows were three passes (fifteen, three and one), so the page drew three of them while the brand index, which counts every pass, said nineteen. Rian went looking for the nineteen and found three: "where are the 19 rows you're talking about?" The rest were reachable only from a selector labelled with the generator's name, which there was no reason to touch.""" proposals.load(db, _file(db)) second = _file(db, name="rule:format_words:6") second["proposals"] = second["proposals"][:2] proposals.load(db, second) whole = _sheet(db) assert whole["pass_name"] is None, "no pass named means the brand, not one list" shown = {r["pass_name"] for g in whole["groups"] for r in g["rows"]} assert shown == {"claude/2026-09-17/chanel-test", "rule:format_words:6"}, f"a pass is missing: {shown}" # The number on the brand index and the number the sheet draws are the same number. They # disagreed by sixteen rows on the brand that exposed this. index = next(row for row in proposals.sheets(db) if row["brand_slug"] == "chanel") assert index["open"] == whole["counts"]["open"], "the index counts every pass; so must the sheet" # And strictly more than the newest pass alone, which is all the sheet used to draw. newest = _sheet(db, pass_name="rule:format_words:6")["counts"]["open"] assert whole["counts"]["open"] > newest, "the whole brand is more than its newest pass" # Each group names the pass that asked it, because an approval is one batch under one pass. assert all(g["pass_name"] for g in whole["groups"]) # Naming a pass still narrows to it. narrowed = _sheet(db, pass_name="rule:format_words:6") assert {r["pass_name"] for g in narrowed["groups"] for r in g["rows"]} == {"rule:format_words:6"} def test_a_brand_pair_shows_both_brands_by_name(db): """What it cost: the first session pass proposed the rename Paco Rabanne -> Rabanne, which is the flagship case of REVIEW-PROCESS section 2.2 (no key, fold or string distance finds it, only a pass does). It reached the sheet as `pair:brand:brand:paco-rabanne||brand:rabanne`, because the side resolver knew uid-keyed rows and not brand slugs. The one row in the pass that most needed reading was the one row that could not be read.""" from app.models import Brand db.add(Brand(slug="paco-rabanne", name="Paco Rabanne")) db.flush() data = _file(db) other = db.scalar(select(Brand).where(Brand.slug == "chanel")) data["proposals"] = [{ "entity": {"type": "suggestion", "key": f"pair:brand:brand:paco-rabanne||brand:{other.slug}"}, "field": "decision", "value": {"decision": "same", "survivor": f"brand:{other.slug}"}, "reason": "a rename only a pass can find", "confidence": 0.95, "evidence": data["proposals"][0]["evidence"], }] proposals.load(db, data) row = next(r for g in _sheet(db)["groups"] for r in g["rows"] if r["field"] == "decision") assert [s["name"] for s in row["sides"]] == ["Paco Rabanne", other.name], row["sides"] class TestTheBrandFoldIsAnsweredFirst: """Rian, 18 Sep: *"if there's a brand fold question, I want that to be confirmed or rejected first. The same brand fold suggestion should be on all relevant brands. The other suggestions under it should be viewable but not decidable until the brand decision is made."* The reason is not tidiness: every other row on the sheet was written against the brands as they stand, so answering them before the fold decides what they are about.""" @staticmethod def _fold(db, survivor="chanel"): from app.models import Brand db.add(Brand(slug="chanel-paris", name="Chanel Paris")) db.flush() data = _file(db) data["proposals"].append({ "entity": {"type": "suggestion", "key": f"pair:brand:brand:chanel-paris||brand:{survivor}"}, "field": "decision", "value": {"decision": "same", "survivor": f"brand:{survivor}"}, "reason": "one company, two rows", "confidence": 0.95, "evidence": data["proposals"][0]["evidence"], "position": 0, }) proposals.load(db, data) return data["pass"]["name"] def test_the_fold_shows_on_both_brands_and_blocks_everything_else(self, db): self._fold(db) sheet = _sheet(db) assert sheet["brand_gate"] and sheet["brand_gate"]["blocked"] >= 1 rows = [r for g in sheet["groups"] for r in g["rows"]] assert sum(1 for r in rows if r["is_brand_fold"]) == 1 assert all(r["blocked_by_brand_fold"] for r in rows if not r["is_brand_fold"] and r["status"] == "open") # and the same question is on the OTHER brand of the pair, whose file never carried it other = proposals.sheet(db, "chanel-paris") assert other["brand_gate"], "the other half of the pair could not see the question" def test_nothing_else_can_be_approved_while_it_is_open(self, db): name = self._fold(db) with pytest.raises(Refused) as refused: proposals.approve(db, "chanel", name, scope="all", by="rian") assert refused.value.code == "BRAND_FOLD_FIRST" def test_keep_separate_leaves_every_suggestion_where_it_was(self, db): name = self._fold(db) uid = next(r["uid"] for g in _sheet(db)["groups"] for r in g["rows"] if r["is_brand_fold"]) out = proposals.approve(db, "chanel", name, scope=None, by="rian", reject=[uid], note="two companies") assert out.get("redirect_to") is None, "keeping them apart moves nothing" assert _sheet(db)["brand_gate"] is None # and the rest are decidable again: the call that was refused now runs. (It approves # nothing here only because this fixture's rows are all spot-checks, which `all` skips.) assert proposals.approve(db, "chanel", name, scope="all", by="rian")["counts"]["refused"] == 0 assert all(not r["blocked_by_brand_fold"] for g in _sheet(db)["groups"] for r in g["rows"]) def test_confirming_moves_the_suggestions_and_says_where_to_go(self, db): from app.models import Brand name = self._fold(db) # a suggestion filed under the brand that is about to be folded away loser = db.scalar(select(Brand).where(Brand.slug == "chanel-paris")) row = db.scalar(select(Proposal).where(Proposal.field == "attribute:color")) db.add(Proposal(uid=_uuid4(), pass_id=row.pass_id, brand_slug=loser.slug, entity_type="brand", natural_key=f"brand:{loser.slug}", field="name", value="Chanel Paris", reason="its own name", evidence=[], status="open", resolution="resolved", generator="claude-session")) db.flush() uid = next(r["uid"] for g in _sheet(db)["groups"] for r in g["rows"] if r["is_brand_fold"]) out = proposals.approve(db, "chanel", name, scope={"proposal_uids": [uid]}, by="rian") assert out["brand_folded_into"] == "chanel" assert out["redirect_to"] == "/review?tab=cleaning&brand=chanel" assert out["proposals_moved"] == 1, "the loser's open suggestion follows the brand" assert db.scalar(select(Proposal).where(Proposal.brand_slug == "chanel-paris", Proposal.status == "open")) is None class TestTheBrandIndex: """Every brand, with the counts a column could show and the two filters (rian, 18 Sep).""" def test_a_brand_with_nothing_waiting_is_still_listed(self, db): from app.models import Brand db.add(Brand(slug="quiet", name="Quiet")) db.commit() rows = {r["brand_slug"]: r for r in proposals.brand_index(db)["rows"]} assert "quiet" in rows and rows["quiet"]["pending"] == 0, "an index that hides these cannot show a fold's other half" def test_the_shape_columns_count_lines_and_variants(self, db): row = next(r for r in proposals.brand_index(db)["rows"] if r["brand_slug"] == "chanel") assert row["product_variants"] >= 1 and row["product_lines"] >= 1 def test_have_suggestions_keeps_only_the_brands_with_work(self, db): proposals.load(db, _file(db)) only = proposals.brand_index(db, has_suggestions=True)["rows"] assert only and all(r["pending"] > 0 for r in only) def test_have_brand_decisions_finds_both_halves_of_a_fold(self, db): from app.models import Brand db.add(Brand(slug="chanel-paris", name="Chanel Paris")) db.flush() data = _file(db) data["proposals"] = [{ "entity": {"type": "suggestion", "key": "pair:brand:brand:chanel-paris||brand:chanel"}, "field": "decision", "value": {"decision": "same", "survivor": "brand:chanel"}, "reason": "one company", "confidence": 0.9, "evidence": data["proposals"][0]["evidence"], }] proposals.load(db, data) slugs = {r["brand_slug"] for r in proposals.brand_index(db, has_brand_decision=True)["rows"]} assert slugs == {"chanel", "chanel-paris"}, "the half whose file never carried the question must still match" class TestTheRowDetail: """One suggestion opened up (rian, 18 Sep): what variants, what attributes, what conflicts and how we propose to handle them, with a way back to the shop's own page.""" def test_a_pair_names_both_sides_with_their_shape(self, db): from app.services import review_detail proposals.load(db, _file(db)) row = db.scalar(select(Proposal).where(Proposal.field == "decision")) d = review_detail.read(db, str(row.uid)) assert d["question"] == "Same product variant?" assert len(d["subjects"]) == 2 and all(s["name"] for s in d["subjects"]) assert d["listings_total"] >= 1 and d["listings"][0]["key"].startswith("listing:") def test_conflicts_are_only_computed_where_variants_actually_become_one(self, db): """A brand or product line fold MOVES variants and keeps them distinct, so their different barcodes are the catalogue working. Listing them there buried the real ones in noise.""" from app.services import review_detail proposals.load(db, _file(db)) header = db.scalar(select(Proposal).where(Proposal.entity_type == "product_line", Proposal.field == "name")) assert review_detail.read(db, str(header.uid))["merges_variants"] is False assert review_detail.read(db, str(header.uid))["conflicts"] == [] pair = db.scalar(select(Proposal).where(Proposal.field == "decision")) assert review_detail.read(db, str(pair.uid))["merges_variants"] is True def test_two_barcodes_are_reported_and_nothing_is_chosen(self, db): from app.models import ProductVariant as PV from app.services import review_detail db.get(PV, 1).gtin, db.get(PV, 4).gtin = "5010327304212", "5010327324081" db.commit() proposals.load(db, _file(db)) pair = db.scalar(select(Proposal).where(Proposal.field == "decision")) conflicts = review_detail.read(db, str(pair.uid))["conflicts"] barcode = next(c for c in conflicts if c["field"] == "gtin") assert len(barcode["values"]) == 2 assert barcode["proposed"] is None, "a conflict is reported for a person, never resolved here" def test_an_unknown_uid_is_nothing_rather_than_an_error(self, db): from app.services import review_detail assert review_detail.read(db, "not-a-uid") == {} class TestAttention: """Attention is what a row needs of a person, not how sure the pass was (rian, 18 Sep). The two are different axes, and sorting by the wrong one buries the rows that matter.""" def test_the_first_question_of_a_kind_is_critical_and_the_next_is_not(self, db): from app.services import attention proposals.load(db, _file(db)) rows = list(db.scalars(select(Proposal).where(Proposal.field == "attribute:color"))) first = attention.for_rows(db, rows)[rows[0].id] assert first == "critical", "nothing of this kind has been answered yet" # answer it, and the precedent exists proposals.approve(db, "chanel", _file(db)["pass"]["name"], scope={"proposal_uids": [str(rows[0].uid)]}, by="rian") again = list(db.scalars(select(Proposal).where(Proposal.field == "attribute:color"))) assert attention.for_rows(db, again)[again[0].id] != "critical" def test_a_pass_may_set_it_and_critical_still_decays_once_answered(self, db): from app.services import attention data = _file(db) for row in data["proposals"]: row["attention"] = "critical" proposals.load(db, data) rows = list(db.scalars(select(Proposal))) assert set(attention.for_rows(db, rows).values()) == {"critical"} assert all(r.attention == "critical" for r in rows), "the pass's own rating is stored" def test_the_kinds_that_set_their_own_precedent_are_distinct(self, db): """Rian drew a line between a refill and a coffret, so each sets its own precedent and answering one must not silently answer the other.""" from app.services import attention make = lambda reason: Proposal(uid=_uuid4(), pass_id=1, brand_slug="x", entity_type="product_variant", natural_key="variant:x", field="product_line", value="line:y", reason=reason, evidence=[], generator="g", status="open") keys = {attention.precedent_key(make(r)) for r in ("a gift set of two", "a refill for the range", "plain move")} assert len(keys) == 3, keys def test_the_sheet_counts_them_by_level(self, db): proposals.load(db, _file(db)) counts = _sheet(db)["attention_counts"] assert set(counts) == {"critical", "high", "medium", "low", "none"} assert sum(counts.values()) == _sheet(db)["counts"]["open"] class TestTheCardReadsAsOneThing: """The headline says what KIND of suggestion it is and what happens to each side (rian, 18 Sep, reading a card that listed 'Million Gold 4 variants / One Million Gold 2 variants' and could not tell which survived, with one of them listed twice).""" def test_a_line_is_never_listed_twice_under_itself(self, db): """The loader DERIVES the lines a header absorbs, so a file that also named them produced the same line in `absorbs` and `absorbs_derived`, and the card printed it twice.""" from app.services import review_detail data = _file(db) header = data["proposals"][0] header["entity"]["detail"]["absorbs"] = [] proposals.load(db, data) row = db.scalar(select(Proposal).where(Proposal.entity_type == "product_line", Proposal.field == "name")) row.natural_key_detail = {**(row.natural_key_detail or {}), "absorbs": ["line:new:chanel-rouge-allure"]} row.detail = {**(row.detail or {}), "absorbs_derived": ["line:new:chanel-rouge-allure"]} db.commit() keys = [s["key"] for s in review_detail.read(db, str(row.uid))["subjects"]] assert len(keys) == len(set(keys)), f"a side is listed twice: {keys}" def test_every_side_says_what_happens_to_it(self, db): from app.services import review_detail proposals.load(db, _file(db)) pair = db.scalar(select(Proposal).where(Proposal.field == "decision")) sides = review_detail.read(db, str(pair.uid))["subjects"] assert all(s["becomes"] for s in sides), "a name and a count do not say which one survives" assert sum(1 for s in sides if s["becomes"] == "stays as it is") == 1 def test_the_category_names_the_kind_of_suggestion(self, db): from app.services import review_detail proposals.load(db, _file(db)) kinds = {review_detail.category_of(r) for r in db.scalars(select(Proposal))} assert "combine-product-variants" in kinds and "move-a-product-variant" in kinds assert all(k in review_detail.CATEGORIES for k in kinds), kinds def test_suggesting_something_else_needs_the_something_else(self, db): """The note IS the answer here, so confirming without one would say nothing at all.""" data = _file(db) proposals.load(db, data) uid = str(db.scalar(select(Proposal).where(Proposal.field == "decision")).uid) with pytest.raises(Refused) as refused: proposals.approve(db, "chanel", data["pass"]["name"], scope=None, by="rian", counter=[uid], note=" ") assert refused.value.code == "VALUE_INVALID" out = proposals.approve(db, "chanel", data["pass"]["name"], scope=None, by="rian", counter=[uid], note="Paco is its own brand; combine only the other two") assert out["counts"]["deferred"] == 1 row = db.scalar(select(Proposal).where(Proposal.uid == _uuid.UUID(uid))) assert row.status == "deferred" and (row.detail or {}).get("wants") == "a different suggestion" def test_the_next_pass_can_tell_something_else_from_ask_me_later(self, db): """`wants` reaches the pass that reads the notes, or the mechanism does nothing. Cost: a rehearsal of the loop end to end. Answering "suggest something else" wrote the marker on the proposal, the pass is told to treat such a note as what to propose INSTEAD, and the notes it reads dropped the field -- so a deferral and a counter-proposal arrived indistinguishable and the next pass would have re-proposed the same thing. """ data = _file(db) proposals.load(db, data) rows = sorted(db.scalars(select(Proposal)), key=lambda r: (r.field != "decision", r.id)) counter, later = str(rows[0].uid), str(rows[1].uid) proposals.approve(db, "chanel", data["pass"]["name"], scope=None, by="rian", counter=[counter], note="Combine only the other two") proposals.approve(db, "chanel", data["pass"]["name"], scope=None, by="rian", defer=[later], note="Ask me again once the sets are settled") by_uid = {n["uid"]: n for n in proposals.deferred_notes(db)["notes"]} assert by_uid[counter]["wants"] == "a different suggestion" assert by_uid[later]["wants"] is None class TestWhatComesNext: """Rian, 18 Sep: *"I'm nervous to click confirm because I dont want to create a mess, but I do want to see what comes next."* Showing the queued questions would bias the answer he has not made; showing how many there will be and what kind does not.""" def test_a_brand_fold_says_what_moves_and_roughly_what_follows(self, db): from app.models import Brand from app.services import review_detail db.add(Brand(slug="chanel-paris", name="Chanel Paris")) db.flush() data = _file(db) data["proposals"] = [{ "entity": {"type": "suggestion", "key": "pair:brand:brand:chanel-paris||brand:chanel"}, "field": "decision", "value": {"decision": "same", "survivor": "brand:chanel"}, "reason": "one company", "confidence": 0.9, "evidence": data["proposals"][0]["evidence"], }] proposals.load(db, data) row = db.scalar(select(Proposal).where(Proposal.field == "decision")) c = review_detail.read(db, str(row.uid))["consequences"] assert c["if_confirmed"] and c["if_separate"] assert any("move" in x["what"] for x in c["if_confirmed"]) assert any("nothing moves" in x["what"] for x in c["if_separate"]) def test_same_named_product_lines_are_named_as_happening_on_confirm_not_as_questions(self, db): """What a brand fold does to the lines under it, said correctly. Cost: it said "about 15 product line questions follow ... they do not exist yet: the next pass writes them." Both halves were wrong. The applier points every line whose key matches one on the survivor at that line as a consequence of the confirm, so they happen on the click and are not questions at all. Rian caught it by asking whether they were queued. """ from app.models import Brand, ProductLine from app.services import review_detail other = Brand(slug="chanel-paris", name="Chanel Paris") db.add(other) db.flush() mine = db.scalar(select(ProductLine).where(ProductLine.key.isnot(None))) db.add(ProductLine(brand_id=other.id, key=mine.key, name=mine.name, slug="chanel-paris-same")) db.add(ProductLine(brand_id=other.id, key="no twin here", name="No Twin", slug="chanel-paris-solo")) db.flush() data = _file(db) data["proposals"] = [{ "entity": {"type": "suggestion", "key": "pair:brand:brand:chanel-paris||brand:chanel"}, "field": "decision", "value": {"decision": "same", "survivor": "brand:chanel"}, "reason": "one company", "confidence": 0.9, "evidence": data["proposals"][0]["evidence"], }] proposals.load(db, data) row = db.scalar(select(Proposal).where(Proposal.field == "decision")) said = " ".join(x["what"] for x in review_detail.read(db, str(row.uid))["consequences"]["if_confirmed"]) assert "1 of those product lines fold straight into a product line of the same name" in said assert "1 arrive as product lines of their own" in said assert "on confirm, not later" in said assert "question" not in said, said assert "do not exist yet" not in said, said def test_keeping_two_brands_apart_says_a_pass_is_still_needed(self, db): """Rian, 18 Sep: keeping them apart does not mean there is nothing left to look at.""" from app.models import Brand from app.services import review_detail db.add(Brand(slug="chanel-paris", name="Chanel Paris")) db.flush() data = _file(db) data["proposals"] = [{ "entity": {"type": "suggestion", "key": "pair:brand:brand:chanel-paris||brand:chanel"}, "field": "decision", "value": {"decision": "same", "survivor": "brand:chanel"}, "reason": "one company", "confidence": 0.9, "evidence": data["proposals"][0]["evidence"], }] proposals.load(db, data) row = db.scalar(select(Proposal).where(Proposal.field == "decision")) said = " ".join(x["what"] for x in review_detail.read(db, str(row.uid))["consequences"]["if_separate"]) assert "another pass is still needed" in said and "each brand separately" in said def test_it_counts_and_characterises_but_names_no_question(self, db): """The line between informing and biasing: a count and a kind, never a proposal.""" from app.services import review_detail proposals.load(db, _file(db)) row = db.scalar(select(Proposal).where(Proposal.field == "decision")) d = review_detail.read(db, str(row.uid)) text = " ".join(x["what"] for x in d["consequences"]["if_confirmed"] + d["consequences"]["if_separate"]) for name in (s["name"] for s in d["subjects"] if s.get("name")): assert name not in text or "move" in text, "a consequence describes shape, not the questions" def test_the_lines_tab_groups_variants_under_their_product_line(self, db): from app.services import review_detail proposals.load(db, _file(db)) row = db.scalar(select(Proposal).where(Proposal.field == "decision")) d = review_detail.read(db, str(row.uid)) assert d["lines"], "a fold is really a question about product lines" assert all(l["variant_uids"] for l in d["lines"]) # every listing names the variant it hangs under, so a line opens to its variants to its listings assert all(x["variant_uid"] for x in d["listings"]) class TestOpeningARowOntoItsListings: """Rian, 18 Sep: *"what I'm confused by even with this layout is why some of them have no listings? How do we end up with a product line or variant that has no related listings?"* Mostly we did not. The card fetched one capped array of listings for the whole question and the nested rows filtered it, so once a brand fold's 60-listing budget ran out every remaining variant read as having none while the count beside its name said otherwise. Reading per row cannot disagree with itself, and it is also what he asked for: open a row, see where the data came from, follow the link to the shop. """ def test_a_variant_past_the_cards_listing_cap_still_shows_its_own_listings(self, db): from app.services import review_detail proposals.load(db, _file(db)) listed = db.scalar(select(Listing)) v = db.get(ProductVariant, listed.variant_id) # Whatever the card's array holds, asking for this variant answers from the database. out = review_detail.listings_under(db, variant_uid=str(v.uid), limit=5) assert out["total"] >= 1 and out["shown"] >= 1 assert all(l["variant_uid"] == str(v.uid) for l in out["listings"]) assert {"listed_name", "listed_brand", "url", "gtin"} <= set(out["listings"][0]) def test_a_line_answers_with_every_listing_under_its_variants_paged(self, db): from app.models import ProductLine from app.services import review_detail proposals.load(db, _file(db)) v = db.get(ProductVariant, db.scalar(select(Listing)).variant_id) line = db.get(ProductLine, v.product_line_id) first = review_detail.listings_under(db, line_uid=str(line.uid), limit=1) assert first["total"] >= 1 and first["shown"] == 1 rest = review_detail.listings_under(db, line_uid=str(line.uid), limit=50) assert rest["shown"] == rest["total"] == first["total"] def test_an_unknown_row_answers_empty_rather_than_raising(self, db): from app.services import review_detail assert review_detail.listings_under(db, variant_uid=str(_uuid4()))["total"] == 0 assert review_detail.listings_under(db, line_uid=str(_uuid4()))["total"] == 0 assert review_detail.listings_under(db)["total"] == 0 def test_every_variant_carries_its_true_listing_count(self, db): """The count on the row and the listings behind it are read the same way, so a row saying three cannot open onto nothing.""" from app.services import review_detail proposals.load(db, _file(db)) row = db.scalar(select(Proposal).where(Proposal.field == "decision")) for v in review_detail.read(db, str(row.uid))["variants"]: assert v["listings"] == review_detail.listings_under(db, variant_uid=v["uid"], limit=200)["total"]