"""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

from sqlalchemy import select

from app.main import app
from app.models import Decision, Listing, ProductVariant, Proposal
from app.services import access, proposals
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 [])]
    assert len(keys) == 12  # K4's seven, K6's two index candidates, and K11's hints, notes and line-table reads
    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"}