"""The proposals load and withdraw (Stream K4; the escalation-2 spec test 15, and the file's schema).
What it cost before: `propose --file` carried staging's row ids and could not load on production
(plan W13); a re-load could have re-opened what a person had already approved; a check that wrote
would have left half a pass behind; a spot-check sample that moved between loads would have let a
person approve in bulk the rows they had been asked to read. SQLite kit, no network."""
from __future__ import annotations

import copy
import json
from datetime import UTC, datetime

import pytest
from sqlalchemy import create_engine, delete, event, 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, PriceObservation,
                        ProductLine, ProductVariant, Proposal, ProposalPass, Retailer, Shop, Source, Suggestion, AttributeAlias)
from app.models.places import Place, ShopPlace
from app.services import keying, proposals
from app.services.decisions import natural_keys
from app.services.decisions.writer import Refused

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__, PriceObservation.__table__, *LEDGER_TABLES]
NAMES = ["Rouge Allure 3.5 gr / 99 Pirate", "Rouge Allure 3.5 gr / 104 Passion", "Rouge Allure Velvet 3.5 g", "Rouge Allure 3.5g 99 Pirate"]


@pytest.fixture
def db():
    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
    # pysqlite releases a SAVEPOINT as a commit unless the transaction is begun explicitly; the
    # load's --check must roll back here exactly as it does on Postgres (SQLAlchemy's recipe).
    event.listen(engine, "connect", lambda conn, _rec: setattr(conn, "isolation_level", None))
    event.listen(engine, "begin", lambda conn: conn.exec_driver_sql("BEGIN"))
    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=1, username="rian", display_name="rian"))
        s.add_all([Brand(id=1, slug="chanel", name="CHANEL"), Retailer(id=1, slug="attenza", name="Attenza"), Retailer(id=2, slug="avolta", name="Avolta")])
        s.flush()
        s.add_all([ProductLine(id=1, brand_id=1, key="rouge allure", slug="chanel-rouge-allure", name="Rouge Allure"),
                   ProductLine(id=2, brand_id=1, key="rouge allure velvet", slug="chanel-rouge-allure-velvet", name="Rouge Allure Velvet"),
                   ProductLine(id=3, brand_id=1, key="rouge allure 99 pirate", slug="chanel-rouge-allure-99-pirate", name="Rouge Allure 99 Pirate")])
        s.add_all([Shop(id=1, retailer_id=1, code="PTY1", iata="PTY", name="Panama", currency="USD"),
                   Shop(id=2, retailer_id=2, code="MEX1", iata="MEX", name="Mexico", currency="USD")])
        s.flush()
        for n, name in enumerate(NAMES, start=1):
            line = 3 if n == 4 else (2 if n == 3 else 1)
            s.add(ProductVariant(id=n, name=name, brand="CHANEL", brand_id=1, vertical="beauty", category="Makeup", match_key=f"k{n}",
                                 product_line_id=line, quantity_value=3.5, quantity_unit="g", quantity_state="stated", form="single", attributes={}))
            s.flush()
            s.add(Listing(id=n, variant_id=n, shop_id=1 if n != 4 else 2, source_sku=f"sku{n}", listed_name=name,
                          url=f"https://example.test/p/{n}", listed_variant="99 Pirate" if n == 1 else None))
            s.flush()
            s.add(PriceObservation(listing_id=n, price=40 + n, currency="USD", price_usd=40 + n,
                                   observed_at=datetime(2026, 9, 17, tzinfo=UTC), in_stock=True))
        s.commit()
        yield s
    keying.invalidate()


def _key(db, model, id_):
    return natural_keys.build(db.get(model, id_))[0]


def _file(db, name="claude/2026-09-17/chanel-test", **over) -> dict:
    v1, v4 = _key(db, ProductVariant, 1), _key(db, ProductVariant, 4)
    l1, l4 = _key(db, Listing, 1), _key(db, Listing, 4)
    line = "line:new:chanel-rouge-allure"
    data = {"pass": {"name": name, "kind": "session", "process_version": "2", "rules_version": "6", "generator": "claude-session"},
            "brand": "chanel",
            "proposals": [
                {"sheet_line_ref": line, "position": 0,
                 "entity": {"type": "product_line", "key": line, "detail": {"brand_slug": "chanel", "slug": "chanel-rouge-allure", "name": "Rouge Allure"}},
                 "field": "name", "value": "Rouge Allure", "confidence": 0.97, "reason": "the head words every shade shares",
                 "evidence": [{"listing": l1, "source": "listed_name", "span": [0, 12], "text": "Rouge Allure"}]},
                {"sheet_line_ref": line, "position": 1, "entity": {"type": "product_variant", "key": v4, "detail": {"listings": [l4]}},
                 "field": "product_line", "value": line, "confidence": 0.9, "reason": "a shade of Rouge Allure written loose",
                 "evidence": [{"listing": l4, "source": "listed_name", "span": [0, 12], "text": "Rouge Allure"}]},
                {"sheet_line_ref": line, "position": 2, "entity": {"type": "product_variant", "key": v1, "detail": {"listings": [l1]}},
                 "field": "attribute:color", "value": "99 Pirate", "confidence": 0.99, "reason": "the shop's option field",
                 "evidence": [{"listing": l1, "source": "option:shade", "span": [0, 9], "text": "99 Pirate"}]},
                {"sheet_line_ref": line, "position": 3,
                 "entity": {"type": "suggestion", "key": f"pair:product:{min(v1, v4)}||{max(v1, v4)}", "detail": {"level": "product"}},
                 "field": "decision", "value": {"decision": "same", "survivor": v1, "name": None, "note": None}, "confidence": 0.8,
                 "reason": "same shade and weight, two shops", "evidence": [{"listing": l4, "source": "listed_name", "span": [18, 27], "text": "99 Pirate"}]},
            ]}
    data.update(over)
    return data


def _counts(db) -> dict:
    return {m.__tablename__: db.scalar(select(func.count()).select_from(m)) for m in
            (ProposalPass, Proposal, Decision, DecisionBatch, ProductLine, ProductVariant, Suggestion, Listing)}


def test_check_leaves_every_table_unchanged_and_reports_what_a_load_would_do(db):
    before = _counts(db)
    counts = proposals.load(db, _file(db), check=True)
    assert counts["check"] and counts["inserted"] == 4
    assert _counts(db) == before


def test_an_identical_file_is_a_no_op_by_content_and_a_corrected_file_updates_open_rows_only(db):
    data = _file(db)
    first = proposals.load(db, data)
    # The line's slug already exists (adopted at approval, so resolved); the pair no rule offered is creatable.
    assert first["inserted"] == 4 and first["creatable"] == 1 and first["resolved"] == 3
    assert proposals.load(db, copy.deepcopy(data))["unchanged"] is True
    row = db.scalar(select(Proposal).where(Proposal.field == "attribute:color"))
    row.status = "rejected"; db.commit()
    changed = copy.deepcopy(data)
    for p in changed["proposals"]:
        p["reason"] = p["reason"] + " (re-read)"
    counts = proposals.load(db, changed)
    assert counts["kept"] == 1 and counts["updated"] == 3
    assert db.scalar(select(Proposal).where(Proposal.field == "attribute:color")).reason == "the shop's option field", "a resolved row is never changed"


def test_rows_missing_from_a_reload_go_stale_unless_partial(db):
    data = _file(db)
    proposals.load(db, data)
    fewer = copy.deepcopy(data)
    fewer["proposals"] = fewer["proposals"][:2]
    proposals.load(db, fewer, partial=True)
    assert db.scalar(select(func.count()).select_from(Proposal).where(Proposal.status == "stale")) == 0
    fewer["pass"]["note"] = "second statement"
    counts = proposals.load(db, fewer)
    assert counts["stale_missing"] == 2
    assert {r.field for r in db.scalars(select(Proposal).where(Proposal.status == "stale"))} == {"attribute:color", "decision"}


def test_spot_check_is_deterministic_per_pass_name_and_never_cleared(db):
    data = _file(db)
    proposals.load(db, data)
    first = {(r.natural_key, r.field): r.spot_check for r in db.scalars(select(Proposal))}
    assert first[(data["proposals"][3]["entity"]["key"], "decision")] is True, "a merge is always spot-checked"
    # Five rows or fewer: every one is among the five lowest; the rule's sample is at least three.
    again = proposals.spot_checks(data["pass"]["name"], [{"entity_type": p["entity"]["type"], "natural_key": p["entity"]["key"],
                                                          "field": p["field"], "value": p["value"], "confidence": p["confidence"]} for p in data["proposals"]])
    assert again == proposals.spot_checks(data["pass"]["name"], [{"entity_type": p["entity"]["type"], "natural_key": p["entity"]["key"],
                                                                  "field": p["field"], "value": p["value"], "confidence": p["confidence"]} for p in reversed(data["proposals"])])
    many = [{"entity_type": "product_variant", "natural_key": f"variant:{i}", "field": "name", "value": "x", "confidence": 0.95, "position": i}
            for i in range(100)]
    sample_a = proposals.spot_checks("pass-a", many)
    assert sample_a == proposals.spot_checks("pass-a", list(reversed(many))) and len(sample_a) == 5 + 5
    assert sample_a != proposals.spot_checks("pass-b", many), "the sample is seeded by the pass name"
    # A reload that would no longer mark a row keeps its mark while it is open.
    raised = copy.deepcopy(data)
    for p in raised["proposals"]:
        p["confidence"] = 0.99
    raised["proposals"][3]["value"]["decision"] = "separate"
    proposals.load(db, raised)
    assert db.scalar(select(Proposal).where(Proposal.field == "decision")).spot_check is True


def test_the_schema_error_names_the_row_and_the_field(db):
    data = _file(db)
    del data["proposals"][2]["evidence"]
    with pytest.raises(proposals.FileInvalid) as bad:
        proposals.load(db, data)
    assert "proposals[2]" in str(bad.value) and "evidence" in str(bad.value)
    data = _file(db)
    data["proposals"][1]["entity"]["key"] = "17"
    with pytest.raises(proposals.FileInvalid) as bad:
        proposals.load(db, data)
    assert "proposals[1]" in str(bad.value) and "entity.key" in str(bad.value), "a database id is not a natural key"
    data = _file(db)
    data["proposals"][2]["evidence"][0]["text"] = "99 Pirates"
    with pytest.raises(proposals.FileInvalid) as bad:
        proposals.load(db, data)
    assert "proposals[2]" in str(bad.value) and "text" in str(bad.value)


def test_a_span_the_listing_does_not_contain_loads_stale_and_an_unknown_variant_parks(db):
    data = _file(db)
    data["proposals"][1]["evidence"][0].update(span=[1, 13], text="Rouge Allure")
    data["proposals"].append({"entity": {"type": "product_variant", "key": "variant:00000000-0000-0000-0000-000000000000",
                                         "detail": {"listings": ["listing:attenza/PTY1/nope"]}},
                              "field": "name", "value": "Ghost", "confidence": 0.9, "reason": "a row this host never had",
                              "evidence": [{"listing": "listing:attenza/PTY1/nope", "source": "listed_name", "span": [0, 5], "text": "Ghost"}]})
    counts = proposals.load(db, data)
    assert counts["evidence_mismatch"] == 1 and counts["parked"] == 1
    stale = db.scalar(select(Proposal).where(Proposal.field == "product_line"))
    assert stale.status == "stale" and stale.resolution_note.startswith("EVIDENCE_MISMATCH")
    parked = db.scalar(select(Proposal).where(Proposal.field == "name", Proposal.entity_type == "product_variant"))
    assert parked.status == "parked" and parked.entity_id is None


def test_rule_and_current_values_are_captured_and_a_previous_rejection_is_noted(db):
    proposals.load(db, _file(db))
    color = db.scalar(select(Proposal).where(Proposal.field == "attribute:color"))
    assert color.current_value is None and color.entity_id == 1
    member = db.scalar(select(Proposal).where(Proposal.field == "product_line"))
    assert member.current_value == _key(db, ProductLine, 3)
    color.status = "rejected"; db.commit()
    counts = proposals.load(db, _file(db, name="claude/2026-09-18/chanel-2"))
    assert counts["previously_rejected"] == 1
    again = db.scalar(select(Proposal).join(ProposalPass).where(ProposalPass.name == "claude/2026-09-18/chanel-2", Proposal.field == "attribute:color"))
    assert again.detail["previously_rejected"] == str(color.uid)


def test_a_replayed_decision_marks_its_proposal_approved_on_load(db):
    data = _file(db)
    proposals.load(db, data)
    color = db.scalar(select(Proposal).where(Proposal.field == "attribute:color"))
    result = proposals.approve(db, "chanel", data["pass"]["name"], scope={"proposal_uids": [str(color.uid)]}, by="rian")
    assert result["counts"]["approved"] == 1
    # The other host: the pass arrived with the ledger (replay), its proposals did not.
    db.execute(delete(Proposal)); db.commit()
    data["pass"]["note"] = "loaded on the second host"
    counts = proposals.load(db, data)
    assert counts["approved_on_load"] == 1
    assert db.scalar(select(Proposal).where(Proposal.field == "attribute:color")).status == "approved"


def test_a_replayed_keep_separate_marks_its_pair_proposal_rejected_on_load(db):
    """The CHANEL rehearsal (dfp_k4 -> dfp_k4b): the refill pair rejected on the first host loaded
    as approved on the second, because its Keep separate is a decision of the pass like any other."""
    data = _file(db)
    proposals.load(db, data)
    pair = db.scalar(select(Proposal).where(Proposal.field == "decision"))
    proposals.approve(db, "chanel", data["pass"]["name"], scope=None, reject=[str(pair.uid)], note="a refill is not its lipstick", by="rian")
    db.execute(delete(Proposal)); db.commit()
    data["pass"]["note"] = "loaded on the second host"
    counts = proposals.load(db, data)
    assert counts["approved_on_load"] == 1
    again = db.scalar(select(Proposal).where(Proposal.field == "decision"))
    assert again.status == "rejected" and again.resolution_note == "a refill is not its lipstick"


def test_withdraw_closes_the_waiting_rows_lists_the_approved_batches_and_never_touches_the_ledger(db):
    data = _file(db)
    proposals.load(db, data)
    color = db.scalar(select(Proposal).where(Proposal.field == "attribute:color"))
    batch_uid = proposals.approve(db, "chanel", data["pass"]["name"], scope={"proposal_uids": [str(color.uid)]}, by="rian")["batch_uid"]
    ledger = db.scalar(select(func.count()).select_from(Decision))
    with pytest.raises(Refused):
        proposals.withdraw(db, data["pass"]["name"], "", "rian")
    result = proposals.withdraw(db, data["pass"]["name"], "the pass read shades as lines", "rian")
    assert result["withdrawn_rows"] == 3 and [b["uid"] for b in result["approved_batches"]] == [batch_uid]
    assert "decisions undo-batch " + batch_uid in result["approved_batches"][0]["undo"]
    assert db.scalar(select(func.count()).select_from(Decision)) == ledger
    with pytest.raises(Refused) as refused:
        proposals.load(db, _file(db))
    assert refused.value.code == "PASS_WITHDRAWN"


def test_the_cli_load_check_prints_and_writes_nothing(db, tmp_path, monkeypatch, capsys):
    from app import cli_proposals

    path = tmp_path / "chanel.json"
    path.write_text(json.dumps(_file(db)))
    monkeypatch.setattr(cli_proposals, "SessionLocal", lambda: _Borrowed(db))
    before = _counts(db)
    import argparse

    parser = argparse.ArgumentParser()
    cli_proposals.register(parser.add_subparsers(dest="command"))
    args = parser.parse_args(["proposals", "load", "--file", str(path), "--check"])
    assert args.func(args) == 0
    assert "rolled back" in capsys.readouterr().out
    assert _counts(db) == before


class _Borrowed:
    def __init__(self, db):
        self.db = db

    def __enter__(self):
        return self.db

    def __exit__(self, *exc):
        return False