"""Splitting a brand row the trailer fold joined wrongly (Stream K9).

The rows are the real ones: `appleton` on a copy of the 17 Sep post-chain dump holds four
listed spellings on one row -- "Appleton", "Appleton Estate", "Appleton Estate(R)" and
"Appleton Rum" -- because `normalize.brand_key` drops "estate" and "rum" from the end. That
fold is right, and rian kept it for launch; what was missing is any way back when one is
wrong, which is what these tests pin. The second brand here is the case the fold will meet
the day a category beyond drinks and beauty is collected: two unrelated companies whose
spellings differ only by a listed word.

What it cost before: a wrong fold had no way back at all. The fold is a rule, not a decision,
so nothing recorded it, and `undo.unmerge` works on variants only.

SQLite kit, no network.
"""
from __future__ import annotations

import json
from datetime import UTC, datetime

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, Redirect, Retailer, Shop, Source, Suggestion,
                        AttributeAlias)
from app.models.places import Place, ShopPlace
from app.services import brands as brands_service, ingest, keying, merges, proposal_rules, proposals
from app.services.decisions import replay as replay_service, 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

#: The four spellings the catalogue really holds on `appleton`, with the product line each
#: variant sits on (`tests/fixtures/brand_folds.json`).
APPLETON = [
    ("Appleton", "Signature Blend"),
    ("Appleton Estate", "Signature Blend"),
    ("Appleton Estate®", "12 Year Old Rare Casks"),
    ("Appleton Rum", "White Overproof"),
]


@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"))
        retailer = Retailer(slug="r", name="R")
        s.add_all([Brand(id=1, slug="appleton", name="Appleton Estate"), retailer])
        s.flush()
        s.add(Shop(id=1, retailer_id=retailer.id, code="S1", iata="KIN", name="Kingston", currency="USD"))
        s.flush()
        for n, (listed, line_name) in enumerate(APPLETON, start=1):
            line = s.scalar(select(ProductLine).where(ProductLine.brand_id == 1, ProductLine.key == line_name.lower()))
            if line is None:
                line = ProductLine(brand_id=1, key=line_name.lower(), slug=f"appleton-{line_name.lower().replace(' ', '-')}", name=line_name)
                s.add(line)
                s.flush()
            s.add(ProductVariant(id=n, name=f"{listed} {line_name} 750ml", brand=listed, brand_id=1, vertical="liquor",
                                 match_key=f"appleton|{line_name.lower()}||750ml", product_line_id=line.id, quantity_value=750,
                                 quantity_unit="ml", quantity_state="stated", quantity_ml=750, form="single", attributes={}))
            s.flush()
            s.add(Listing(variant_id=n, shop_id=1, source_sku=f"sku{n}"))
        s.flush()
        # The keys the rules actually compute, so "the state before" is the rules' own state and
        # a round trip is compared against something the catalogue could really hold.
        merges.rekey_product_variants(s, list(s.scalars(select(ProductVariant))), keying.load_maps(s))
        s.commit()
        yield s
    keying.invalidate()


def _appleton(db) -> Brand:
    return db.scalar(select(Brand).where(Brand.slug == "appleton"))


def _split(db, spellings=("Appleton Rum",), name="Appleton Rum"):
    return brands_service.split_brand(db, _appleton(db), list(spellings), new_name=name, decided_by=RIAN)


class TestTheSplit:
    def test_the_named_spelling_ends_on_a_row_of_its_own_with_its_own_address(self, db):
        report = _split(db)
        assert report["product_variants_moved"] == 1 and report["minted"] is True
        new = db.scalar(select(Brand).where(Brand.slug == "appleton-rum"))
        assert new is not None and new.name == "Appleton Rum" and new.uid is not None
        moved = db.scalar(select(ProductVariant).where(ProductVariant.brand == "Appleton Rum"))
        assert moved.brand_id == new.id
        assert {v.brand for v in db.scalars(select(ProductVariant).where(ProductVariant.brand_id == _appleton(db).id))} == {
            "Appleton", "Appleton Estate", "Appleton Estate®"}

    def test_the_moved_variant_keys_under_its_new_brand(self, db):
        before = db.get(ProductVariant, 4).match_key
        _split(db)
        after = db.get(ProductVariant, 4).match_key
        assert before.startswith("appleton|") and after.startswith("appleton-rum|")

    def test_it_sits_on_a_product_line_of_the_new_brand_and_the_emptied_one_is_pruned(self, db):
        _split(db)
        new = db.scalar(select(Brand).where(Brand.slug == "appleton-rum"))
        moved = db.get(ProductVariant, 4)
        line = db.get(ProductLine, moved.product_line_id)
        assert line.brand_id == new.id
        assert db.scalar(select(ProductLine).where(ProductLine.brand_id == _appleton(db).id,
                                                   ProductLine.key == "white overproof")) is None

    def test_one_decision_row_carries_everything_the_undo_needs(self, db):
        report = _split(db)
        row = db.get(Decision, report["decision_id"])
        assert row.entity_type == "brand" and row.field == "split" and row.entity_id == _appleton(db).id
        assert row.value == {"brand_slug": "appleton-rum", "name": "Appleton Rum", "spellings": ["Appleton Rum"]}
        assert row.detail["moved"] == [[4, 1, 3]]
        assert row.detail["new_brand"]["slug"] == "appleton-rum" and row.detail["new_brand"]["minted"] is True
        assert row.detail["new_brand"]["uid"] == str(db.scalar(select(Brand).where(Brand.slug == "appleton-rum")).uid)

    def test_the_ledger_agrees_with_the_catalogue_afterwards(self, db):
        _split(db)
        assert undo_service.verify(db)["drift"] == 0

    def test_a_second_split_moves_the_second_spelling_without_touching_the_first(self, db):
        _split(db)
        brands_service.split_brand(db, _appleton(db), ["Appleton Estate®"], new_name="Appleton Estate Rare", decided_by=RIAN)
        assert db.get(ProductVariant, 4).brand_id == db.scalar(select(Brand.id).where(Brand.slug == "appleton-rum"))
        assert db.get(ProductVariant, 3).brand_id == db.scalar(select(Brand.id).where(Brand.slug == "appleton-estate-rare"))
        assert undo_service.verify(db)["drift"] == 0


class TestTheClaim:
    """The split must survive the next collection: the fold key of a split-off spelling IS the
    row it was split away from, so the resolver has to read the claim before the fold."""

    def test_the_next_sighting_of_the_spelling_lands_on_the_new_row_not_the_old_one(self, db):
        _split(db)
        keying.invalidate()
        assert ingest.brand_slug("Appleton Rum") == "appleton"  # the fold itself is unchanged
        assert ingest.resolve_brand(db, "Appleton Rum").slug == "appleton-rum"
        assert ingest.resolve_brand(db, "APPLETON RUM").slug == "appleton-rum"  # case and accents folded
        assert ingest.resolve_brand(db, "Appleton Estate").slug == "appleton"

    def test_an_unclaimed_spelling_still_folds_as_before(self, db):
        _split(db)
        keying.invalidate()
        assert ingest.resolve_brand(db, "Appleton Distillery").slug == "appleton"

    def test_the_claim_goes_when_the_split_is_undone(self, db):
        report = _split(db)
        keying.invalidate()
        batch_uid = db.get(DecisionBatch, db.get(Decision, report["decision_id"]).batch_id).uid
        undo_service.undo_batch(db, batch_uid, RIAN, "wrong call")
        keying.invalidate()
        assert ingest.resolve_brand(db, "Appleton Rum").slug == "appleton"


class TestWhatItRefuses:
    def test_a_spelling_no_variant_of_the_row_carries(self, db):
        with pytest.raises(writer.Refused) as refused:
            _split(db, spellings=("Appleton Gin",))
        assert refused.value.code == "SPELLING_NOT_ON_ROW"

    def test_a_split_that_would_empty_the_source(self, db):
        with pytest.raises(writer.Refused) as refused:
            _split(db, spellings=[listed for listed, _ in APPLETON], name="Appleton Everything")
        assert refused.value.code == "SPLIT_EMPTIES_SOURCE"

    def test_an_aliased_brand_row(self, db):
        db.add(Brand(id=2, slug="appleton-jamaica", name="Appleton Jamaica", alias_of_id=1))
        db.commit()
        with pytest.raises(writer.Refused) as refused:
            brands_service.split_brand(db, db.get(Brand, 2), ["Appleton Rum"], new_name="X", decided_by=RIAN)
        assert refused.value.code == "ENTITY_ALIASED"

    def test_no_spelling_and_no_name(self, db):
        for spellings, name in (([], "X"), (["Appleton Rum"], "  ")):
            with pytest.raises(writer.Refused) as refused:
                brands_service.split_brand(db, _appleton(db), spellings, new_name=name, decided_by=RIAN)
            assert refused.value.code == "VALUE_INVALID"

    def test_a_running_collection(self, db):
        db.add(Source(id=1, slug="s", name="S", enabled=True))
        db.flush()
        db.add(CollectionRun(source_id=1, status="running", started_at=datetime.now(UTC)))
        db.commit()
        with pytest.raises(writer.Refused) as refused:
            _split(db)
        assert refused.value.code == "COLLECTION_RUNNING"


class TestTheUndo:
    """A split returns the database to its exact prior state: row counts, keys, brand ids."""

    @staticmethod
    def _state(db) -> dict:
        """Row counts, keys and brand ids, which is what the brief asks the round trip to
        restore. A product line is compared by its brand and key, not by its row id: the split
        prunes the rows its move empties, and the undo re-derives them under the same brand with
        the same key, slug and name. Nothing named those rows, which is why the prune was allowed
        to take them."""
        brands = {b.id: b.slug for b in db.scalars(select(Brand))}
        lines = {l.id: (brands.get(l.brand_id), l.key, l.slug, l.name) for l in db.scalars(select(ProductLine))}
        return {
            "brands": sorted((b.slug, b.name, b.alias_of_id) for b in db.scalars(select(Brand))),
            "lines": sorted(lines.values()),
            "variants": sorted((v.id, v.brand, brands.get(v.brand_id), lines.get(v.product_line_id), v.match_key)
                               for v in db.scalars(select(ProductVariant))),
        }

    @staticmethod
    def _undo(db, report):
        uid = db.get(DecisionBatch, db.get(Decision, report["decision_id"]).batch_id).uid
        return undo_service.undo_batch(db, uid, RIAN, "the fold was right after all")

    def test_split_then_undo_leaves_the_database_as_it_was(self, db):
        before = self._state(db)
        report = _split(db)
        assert self._state(db) != before
        self._undo(db, report)
        assert self._state(db) == before

    def test_the_minted_brand_row_is_deleted_because_nothing_references_it(self, db):
        report = _split(db)
        assert db.scalar(select(Brand).where(Brand.slug == "appleton-rum")) is not None
        self._undo(db, report)
        assert db.scalar(select(Brand).where(Brand.slug == "appleton-rum")) is None

    def test_a_minted_row_something_else_points_at_is_kept(self, db):
        """The same rule `prune_lines` follows for lines: a row a decision, an alias, a
        suggestion, a redirect or a variant names is a record, not a cache."""
        report = _split(db)
        new = db.scalar(select(Brand).where(Brand.slug == "appleton-rum"))
        with writer.batch(db, "route", "individual", RIAN, commit=True, tail=False) as b:
            writer.record(b, "brand", new, "name", "Appleton Rum Co", reason="rian typed it")
        self._undo(db, report)
        kept = db.scalar(select(Brand).where(Brand.slug == "appleton-rum"))
        assert kept is not None and kept.name == "Appleton Rum Co"
        assert db.get(ProductVariant, 4).brand_id == _appleton(db).id  # the variants still came back

    def test_nothing_is_deleted_by_the_undo_it_is_a_row_like_any_other(self, db):
        report = _split(db)
        self._undo(db, report)
        rows = list(db.scalars(select(Decision).where(Decision.field == "split").order_by(Decision.id)))
        assert [r.effect for r in rows] == ["set", "release"]
        assert rows[1].reverses_id == rows[0].id
        assert undo_service.verify(db)["drift"] == 0

    def test_after_an_undo_the_split_can_simply_be_taken_again(self, db):
        """There is no re-do row for a released field, here or anywhere else in the ledger: an
        undo whose target had no predecessor writes `release`, and a release is never the
        effective row, so `undo` of it refuses `DECISION_SUPERSEDED` (the same for an alias).
        Taking the split again is the way back, and it is clean: a new decision, a new row."""
        report = _split(db)
        self._undo(db, report)
        release = db.scalar(select(Decision).where(Decision.field == "split", Decision.effect == "release"))
        with writer.batch(db, "route", "individual", RIAN, commit=True) as b:
            with pytest.raises(writer.Refused) as refused:
                undo_service.undo(b, release.id, "no, it was wrong")
        assert refused.value.code == "DECISION_SUPERSEDED"
        _split(db)
        moved = db.get(ProductVariant, 4)
        assert db.get(Brand, moved.brand_id).slug == "appleton-rum"
        assert moved.match_key.startswith("appleton-rum|")
        assert undo_service.verify(db)["drift"] == 0

    def test_a_variant_whose_product_line_a_person_decided_carries_and_comes_back(self, db):
        """A rule never overwrites a decision: the moved variant's decided product line is
        carried to the matching row under the new brand as a consequence, and the undo reverses
        the consequence before the split, so the variant ends on the line it was decided onto."""
        line = ProductLine(brand_id=1, key="decided:rare blend", slug="appleton-rare-blend", name="Rare Blend")
        db.add(line)
        db.flush()
        with writer.batch(db, "route", "individual", RIAN, commit=True, tail=False) as b:
            writer.record(b, "product_variant", 4, "product_line", line, reason="rian put it there")
        report = _split(db)
        assert report["line_decisions_carried"] == 1
        carried = db.get(ProductLine, db.get(ProductVariant, 4).product_line_id)
        assert carried.id != line.id and carried.key == line.key
        assert db.get(Brand, carried.brand_id).slug == "appleton-rum"
        assert undo_service.verify(db)["drift"] == 0
        self._undo(db, report)
        assert db.get(ProductVariant, 4).product_line_id == line.id
        assert undo_service.verify(db)["drift"] == 0


class TestARejectedFoldOffersTheSplit:
    """`rule:brand_trailers` asks a person to confirm the spellings on a brand row. Rejecting
    one used to record a rejection and move nothing, which left the wrong fold exactly where it
    was: the fold is a rule, not a decision, so there was nothing to undo. The rejection now
    carries the split, and a person runs it in one click; the rejection alone still moves
    nothing, which is the whole point of a proposal."""

    @staticmethod
    def _reject(db):
        counts = proposal_rules.generate(db, ("brand_trailers",))
        assert counts["brand_trailers"] == 1
        db.commit()
        pass_name = db.scalar(select(ProposalPass.name).where(ProposalPass.generator == "rule"))
        row = db.scalar(select(Proposal).where(Proposal.entity_type == "brand"))
        proposals.approve(db, "appleton", pass_name, scope=None, by="rian", reject=[str(row.uid)],
                          note="Appleton Rum is a different company")
        return pass_name

    def test_an_open_fold_proposal_offers_nothing_until_it_is_rejected(self, db):
        proposal_rules.generate(db, ("brand_trailers",))
        db.commit()
        row = next(r for g in proposals.sheet(db, "appleton")["groups"] for r in g["rows"])
        assert row["status"] == "open" and row["split_offer"] is None

    def test_the_rejection_carries_the_split_and_marks_what_the_list_folded(self, db):
        self._reject(db)
        row = next(r for g in proposals.sheet(db, "appleton")["groups"] for r in g["rows"])
        assert row["status"] == "rejected"
        offer = row["split_offer"]
        assert offer["brand_slug"] == "appleton"
        assert [s["spelling"] for s in offer["spellings"]] == sorted(listed for listed, _ in APPLETON)
        assert {s["spelling"] for s in offer["spellings"] if s["folded"]} == {
            "Appleton Estate", "Appleton Estate®", "Appleton Rum"}
        assert next(s for s in offer["spellings"] if s["spelling"] == "Appleton Rum")["words"] == ["rum"]

    def test_the_rejection_by_itself_splits_nothing(self, db):
        before = TestTheUndo._state(db)
        self._reject(db)
        assert TestTheUndo._state(db) == before
        assert db.scalar(select(func.count(Decision.id)).where(Decision.field == "split")) == 0

    def test_the_person_runs_it_and_it_is_one_undoable_batch(self, db):
        self._reject(db)
        out = proposals.run_split(db, "appleton", ["Appleton Rum"], name="Appleton Rum", by="rian")
        assert out["product_variants_moved"] == 1 and out["from_brand"] == "appleton" and out["brand_slug"] == "appleton-rum"
        assert "undo-batch" in out["undo"]
        undo_service.undo_batch(db, out["batch_uid"], RIAN, "second thoughts")
        assert db.get(ProductVariant, 4).brand_id == _appleton(db).id

    def test_the_offer_refuses_what_the_split_refuses(self, db):
        self._reject(db)
        with pytest.raises(writer.Refused) as refused:
            proposals.run_split(db, "appleton", [listed for listed, _ in APPLETON], name="Appleton Everything", by="rian")
        assert refused.value.code == "SPLIT_EMPTIES_SOURCE"


class TestAFoldInAnUnlistedVerticalIsAProposal:
    """The other half of scoping the list (K9.5): where the fold no longer acts, the rule still
    says what it WOULD have joined, as a brand pair a person confirms or keeps separate. Two
    fashion companies differing only by "London" are two rows, and the sheet shows the question
    instead of the answer."""

    @staticmethod
    def _two_fashion_brands(db):
        for n, (slug, name, listed) in enumerate([("acme", "Acme", "Acme"), ("acme-london", "Acme London", "Acme London")], start=10):
            db.add(Brand(id=n, slug=slug, name=name))
            db.flush()
            line = ProductLine(brand_id=n, key="coat", slug=f"{slug}-coat", name="Coat")
            db.add(line)
            db.flush()
            db.add(ProductVariant(id=n, name=f"{listed} Coat", brand=listed, brand_id=n, vertical="fashion",
                                  match_key=f"{slug}|coat||n/a", product_line_id=line.id, quantity_state="none",
                                  form="single", attributes={}))
            db.flush()
            db.add(Listing(variant_id=n, shop_id=1, source_sku=f"fsku{n}"))
        db.commit()

    def test_the_two_rows_stay_two_rows(self, db):
        self._two_fashion_brands(db)
        keying.invalidate()
        assert ingest.resolve_brand(db, "Acme London", "fashion").slug == "acme-london"
        assert ingest.resolve_brand(db, "Acme", "fashion").slug == "acme"

    def test_the_rule_proposes_the_pair_instead_of_folding_it(self, db):
        self._two_fashion_brands(db)
        proposal_rules.generate(db, ("brand_trailers",))
        db.commit()
        pair = db.scalar(select(Proposal).where(Proposal.entity_type == "suggestion"))
        assert pair is not None and pair.value == {"decision": "same"}
        assert "the fashion vertical has no trailer list, so they were kept apart" in pair.reason
        assert "'london'" in pair.reason
        row = db.scalar(select(Suggestion).where(Suggestion.level == "brand"))
        assert row is not None and row.decision is None, "an open question, not an act"
        assert db.get(ProductVariant, 11).brand_id == 11, "and nothing moved"

    def test_a_liquor_pair_is_not_proposed_because_the_list_already_folded_it(self, db):
        """The list still acts where it was written for, so those rows are already one and there
        is no pair to ask about; what they get is the confirm-the-spellings row instead."""
        proposal_rules.generate(db, ("brand_trailers",))
        db.commit()
        assert db.scalar(select(func.count(Proposal.id)).where(Proposal.entity_type == "suggestion")) == 0
        assert db.scalar(select(func.count(Proposal.id)).where(Proposal.entity_type == "brand")) == 1


class TestAPlacedSpellingIsNeverReHomedByARuleChange:
    """Scoping the list decides where an UNSEEN spelling goes, never where a placed one moves.

    Measured on the 17 Sep catalogue: three listed spellings would have resolved to a different
    row at the next collection under the scoped list -- 'Souvenir de Paris' (62 live variants on
    `souvenir-de`), 'Distillerie De Paris' (3) and 'Huda Beauty' (13), all classified liquor, so
    the beauty list's "paris" and "beauty" stopped applying to them. Three new brand rows would
    have appeared beside 78 existing variants, with no decision behind any of it. The same rule
    REVIEW-PROCESS.md section 1 already states for a listing and its variant: what is placed
    stays placed while the shop's words are unchanged.
    """

    def test_the_spelling_stays_on_the_row_its_variants_are_on(self, db):
        db.add(Brand(id=20, slug="souvenir-de", name="Souvenir de Paris"))
        db.flush()
        db.add(ProductVariant(id=20, name="Souvenir de Paris Rum 700ml", brand="Souvenir de Paris", brand_id=20,
                              vertical="liquor", match_key="souvenir-de|rum||700ml", quantity_state="none", attributes={}))
        db.commit()
        keying.invalidate()
        assert ingest.brand_slug("Souvenir de Paris", "liquor") == "souvenir-de-paris"  # the scoped fold says otherwise
        assert ingest.resolve_brand(db, "Souvenir de Paris", "liquor").slug == "souvenir-de"  # and is not asked

    def test_a_spelling_nobody_has_seen_takes_the_scoped_fold(self, db):
        keying.invalidate()
        assert ingest.resolve_brand(db, "Acme London", "fashion").slug == "acme-london"
        assert ingest.resolve_brand(db, "Acme London", "beauty").slug == "acme"

    def test_a_split_beats_where_the_spelling_sits(self, db):
        """The one thing that DOES re-home a placed spelling is a person's decision."""
        _split(db)
        keying.invalidate()
        assert ingest.resolve_brand(db, "Appleton Rum", "liquor").slug == "appleton-rum"


class TestTheListsAreVisible:
    """Rian, 17 Sep: *"do we have a list of those stop words somewhere maybe in the review
    dashboard so we know when we're doing that"*. Until now the only way to read one of the word
    lists that decide what two spellings have in common was to open the source."""

    def test_every_list_is_listed_with_its_words_and_its_vertical(self, db):
        rows = {r["key"]: r for r in proposal_rules.word_lists(db)}
        assert "estate" in rows["brand_trailers"]["words"] and rows["brand_trailers"]["verticals"] == ["liquor"]
        assert "london" in rows["brand_trailers_beauty"]["words"] and rows["brand_trailers_beauty"]["verticals"] == ["beauty"]
        assert "ltd" in rows["brand_trailers_every"]["words"] and rows["brand_trailers_every"]["verticals"] == ["every vertical"]
        assert rows["noise"]["words"] == [] and rows["noise"]["pattern"], "a pattern says so instead of pretending to be a list"

    def test_the_one_list_that_acts_is_marked_and_the_rest_are_not(self, db):
        rows = {r["key"]: r for r in proposal_rules.word_lists(db)}
        assert [k for k, r in rows.items() if r["acts"]] == ["brand_trailers", "brand_trailers_beauty", "brand_trailers_every"]
        assert rows["drink_words"]["acts"] is False and rows["stopwords"]["acts"] is False

    def test_it_counts_the_brand_rows_the_acting_list_folded(self, db):
        rows = {r["key"]: r for r in proposal_rules.word_lists(db)}
        assert rows["brand_trailers"]["brand_rows_folded"] == 1, "appleton, on estate and rum"
        assert rows["drink_words"]["brand_rows_folded"] is None, "a proposing list's count is its proposals"

    def test_it_links_a_list_to_the_sheets_its_proposals_are_waiting_on(self, db):
        proposal_rules.generate(db, ("brand_trailers",))
        db.commit()
        row = next(r for r in proposal_rules.word_lists(db) if r["key"] == "brand_trailers")
        assert row["pass_name"] == "rule:brand_trailers:6" and row["proposals"]["open"] == 1
        assert row["brands"] == ["appleton"]

    def test_reading_the_lists_writes_nothing(self, db):
        proposal_rules.generate(db, ("brand_trailers",))
        db.commit()
        before = (db.scalar(select(func.count(Decision.id))), db.scalar(select(func.count(Proposal.id))),
                  db.scalar(select(func.count(Brand.id))), db.scalar(select(func.count(ProductVariant.id))))
        proposal_rules.word_lists(db)
        assert not db.new and not db.dirty and not db.deleted
        assert (db.scalar(select(func.count(Decision.id))), db.scalar(select(func.count(Proposal.id))),
                db.scalar(select(func.count(Brand.id))), db.scalar(select(func.count(ProductVariant.id)))) == before


class TestItSurvivesTheReplayToAnotherHost:
    """A split taken on staging has to reach production, and production is a different host with
    different row ids. The decision therefore carries the WHOLE claim in its value -- the new
    brand's slug, its name and the spellings -- and nothing local: the target host resolves
    `brand:<slug>`, mints the new row itself and recomputes which variants move. The detail's
    variant ids are this host's working note, never the instruction."""

    @staticmethod
    def _export(db, report) -> list[dict]:
        import io

        buf = io.StringIO()
        batch = db.get(DecisionBatch, db.get(Decision, report["decision_id"]).batch_id)
        replay_service.export(db, batch_uid=str(batch.uid), out=buf)
        return [json.loads(line) for line in buf.getvalue().splitlines() if line.strip()]

    def test_the_export_carries_the_claim_and_no_local_id_in_its_value(self, db):
        lines = self._export(db, _split(db))
        row = next(line for line in lines if line["kind"] == "decision")
        assert row["natural_key"] == "brand:appleton" and row["field"] == "split"
        assert row["value"] == {"brand_slug": "appleton-rum", "name": "Appleton Rum", "spellings": ["Appleton Rum"]}
        assert "id" not in row

    def test_the_target_host_mints_the_row_and_works_out_the_move_for_itself(self, db, tmp_path):
        lines = self._export(db, _split(db))
        # The source's working note, made deliberately wrong: the target must ignore it.
        for line in lines:
            if line["kind"] == "decision":
                line["detail"] = {**(line["detail"] or {}), "moved": [[9999, 9999, 9999]],
                                  "new_brand": {"slug": "appleton-rum", "uid": "0" * 32, "name": "Appleton Rum", "minted": True}}
        undo_service.undo_batch(db, db.get(DecisionBatch, db.get(Decision, db.scalar(
            select(Decision.id).where(Decision.field == "split"))).batch_id).uid, RIAN, "back to the start")
        db.execute(Decision.__table__.delete())
        db.execute(DecisionBatch.__table__.delete())
        db.execute(Brand.__table__.delete().where(Brand.slug == "appleton-rum"))
        db.commit()
        keying.invalidate()

        was_on = db.get(ProductVariant, 4).product_line_id  # where the target host has it NOW
        path = tmp_path / "split.jsonl"
        path.write_text("\n".join(json.dumps(line) for line in lines))
        counts = replay_service.replay(db, path, label="staging")
        assert counts["applied"] == 1 and counts["parked"] == 0 and counts["held"] == 0

        new = db.scalar(select(Brand).where(Brand.slug == "appleton-rum"))
        assert new is not None, "the target minted the row the claim named"
        moved = db.get(ProductVariant, 4)
        assert moved.brand_id == new.id and moved.match_key.startswith("appleton-rum|")
        row = db.scalar(select(Decision).where(Decision.field == "split"))
        (vid, old_brand_id, old_line_id), = row.detail["moved"]
        assert (vid, old_brand_id) == (4, _appleton(db).id), "recomputed here, not taken from the file"
        assert old_line_id == was_on, "the product line it came from HERE, which the split then pruned"
        assert row.detail["new_brand"]["uid"] == str(new.uid) != "0" * 32
        assert undo_service.verify(db)["drift"] == 0