"""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) == 11  # K4's seven, the two index candidate routes (K6, routers/publish.py), the brand split and the word lists (K9)
    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