"""The approval, the reject and the undo (Stream K4; the escalation-2 spec §6.4, §7 and tests 14, 17
and 22). What it cost before: a bulk approval could have taken the rows a person had been asked
to read; a line approved by name could have minted a `-2` sibling beside the derived row that
already held its address; a variant merged between the load and the approval would have taken its
decision to the grave; a rejected pair with no record would have been proposed again next pass; an
approval under a running collection deadlocks it. SQLite kit, no network."""
from __future__ import annotations

from datetime import UTC, datetime
from types import SimpleNamespace

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import func, select

from app.db import get_db
from app.models import (Brand, CollectionRun, Decision, DecisionBatch, Listing, ProductLine, ProductVariant, Proposal, Redirect, Source,
                        Suggestion)
from app.routers import review as review_router
from app.services import audit_log, identity, proposals
from app.services.decisions import effective, natural_keys
from app.services.decisions.writer import Refused
from tests.test_proposals_load import _file, _key, db  # noqa: F401 (the fixture)

PASS = "claude/2026-09-17/chanel-test"


def _uid(db, field, entity_type=None):
    stmt = select(Proposal).where(Proposal.field == field)
    if entity_type:
        stmt = stmt.where(Proposal.entity_type == entity_type)
    return str(db.scalar(stmt).uid)


def _with_names(db, n: int) -> dict:
    """The fixture's file plus `n` more variants on line 1, each with a name proposal at 0.95, so a
    bulk approval has rows the spot-check rule does not take."""
    for i in range(10, 10 + n):
        db.add(ProductVariant(id=i, name=f"Rouge Allure 3.5 gr / {i} Shade", brand="CHANEL", brand_id=1, vertical="beauty", category="Makeup",
                              match_key=f"k{i}", product_line_id=1, quantity_value=3.5, quantity_unit="g", quantity_state="stated",
                              form="single", attributes={}))
        db.flush()
        db.add(Listing(id=i, variant_id=i, shop_id=1, source_sku=f"sku{i}", listed_name=f"Rouge Allure 3.5 gr / {i} Shade"))
    db.commit()
    data = _file(db)
    for i in range(10, 10 + n):
        data["proposals"].append({"sheet_line_ref": "line:new:chanel-rouge-allure", "position": i,
                                  "entity": {"type": "product_variant", "key": _key(db, ProductVariant, i), "detail": {"listings": [_key(db, Listing, i)]}},
                                  "field": "name", "value": f"Rouge Allure {i} Shade", "confidence": 0.95, "reason": "the shade read off the tail",
                                  "evidence": [{"listing": _key(db, Listing, i), "source": "listed_name", "span": [0, 12], "text": "Rouge Allure"}]})
    return data


def test_all_leaves_out_spot_checks_and_held_rows_and_the_counts_are_said_loudly(db):
    data = _with_names(db, 20)
    proposals.load(db, data)
    spot = set(db.scalars(select(Proposal.id).where(Proposal.spot_check.is_(True))))
    assert spot and len(spot) < 24
    # A person decided one of the names individually before the pass is approved in bulk.
    from app.services.decisions import writer

    held_target = db.scalar(select(Proposal).where(Proposal.field == "name", Proposal.spot_check.is_(False), Proposal.entity_type == "product_variant"))
    with writer.batch(db, "route", "individual", "rian") as b:
        writer.record(b, "product_variant", held_target.entity_id, "name", "Rian's own spelling")
    result = proposals.approve(db, "chanel", PASS, scope="all", by="rian")
    waiting = 24 - len(spot)
    assert result["mode"] == "bulk" and result["counts"]["held"] == 1
    assert result["counts"]["approved"] == waiting - 1 and result["message"].startswith(f"{waiting - 1} approved, 1 held, 0 refused")
    assert db.get(Proposal, held_target.id).status == "open"
    assert all(db.get(Proposal, i).status == "open" for i in spot), "a spot-check is approved only by a person's own look"
    batch = db.scalar(select(DecisionBatch).where(DecisionBatch.kind == "sheet"))
    assert str(batch.uid) == result["batch_uid"] and batch.scope["brand_slug"] == "chanel" and batch.summary["held"] == 1
    assert db.scalar(select(func.count()).select_from(Decision).where(Decision.batch_id == batch.id, Decision.caused_by_id.is_(None))) == waiting - 1
    # "Approve all" on the page is `all` plus the spot-check uids, which a person has now seen.
    rest = [str(db.get(Proposal, i).uid) for i in spot]
    again = proposals.approve(db, "chanel", PASS, scope={"proposal_uids": rest}, by="rian")
    assert again["mode"] == "individual" and again["counts"]["approved"] + again["counts"]["refused"] == len(rest)


def test_a_sheet_line_adopts_the_row_holding_its_slug_and_keys_it_decided(db):
    """Spec test 14 (first half): an approved "Rouge Allure" adopts the derived row holding the
    slug and rewrites its key into the `decided:` namespace; no `-2` sibling is minted."""
    proposals.load(db, _file(db))
    result = proposals.approve(db, "chanel", PASS, scope={"proposal_uids": [_uid(db, "name", "product_line"), _uid(db, "product_line")]}, by="rian")
    assert result["counts"]["approved"] == 2, result
    line = db.get(ProductLine, 1)
    assert line.key == "decided:chanel-rouge-allure" and line.name == "Rouge Allure"
    assert db.scalar(select(func.count()).select_from(ProductLine)) == 3
    assert db.get(ProductVariant, 4).product_line_id == 1, "the member landed on the adopted line"
    decision = db.scalar(select(Decision).where(Decision.field == "name", Decision.entity_type == "product_line"))
    assert decision.detail["adopted_from_key"] == "rouge allure" and decision.detail["proposal_key"] == "line:new:chanel-rouge-allure"


def test_a_sheet_line_adopts_by_rule_key_moves_the_address_and_otherwise_mints(db):
    data = _file(db)
    data["proposals"][0]["entity"]["key"] = data["proposals"][0]["sheet_line_ref"] = "line:new:chanel-rouge-allure-velvet-matte"
    data["proposals"][0]["value"] = "Rouge Allure Velvet"
    for p in data["proposals"][1:]:
        p["sheet_line_ref"] = "line:new:chanel-rouge-allure-velvet-matte"
    data["proposals"][1]["value"] = "line:new:chanel-rouge-allure-velvet-matte"
    proposals.load(db, data)
    proposals.approve(db, "chanel", PASS, scope={"proposal_uids": [_uid(db, "name", "product_line")]}, by="rian")
    velvet = db.get(ProductLine, 2)
    assert velvet.slug == "chanel-rouge-allure-velvet-matte" and velvet.key == "decided:chanel-rouge-allure-velvet-matte"
    assert db.get(Redirect, "chanel-rouge-allure-velvet").to_slug == "chanel-rouge-allure-velvet-matte"
    fresh = _file(db, name="claude/2026-09-18/chanel-mint")
    fresh["proposals"] = fresh["proposals"][:1]
    fresh["proposals"][0]["entity"]["key"] = fresh["proposals"][0]["sheet_line_ref"] = "line:new:chanel-rouge-coco"
    fresh["proposals"][0]["value"] = "Rouge Coco"
    proposals.load(db, fresh)
    header = db.scalar(select(Proposal).where(Proposal.natural_key == "line:new:chanel-rouge-coco"))
    proposals.approve(db, "chanel", "claude/2026-09-18/chanel-mint", scope={"proposal_uids": [str(header.uid)]}, by="rian")
    minted = db.scalar(select(ProductLine).where(ProductLine.slug == "chanel-rouge-coco"))
    assert minted is not None and minted.key == "decided:chanel-rouge-coco" and minted.name == "Rouge Coco"


def test_a_slug_held_by_another_brand_refuses_the_line_and_every_row_under_it(db):
    db.add(Brand(id=2, slug="dior", name="Dior"))
    db.flush()
    db.add(ProductLine(id=9, brand_id=2, key="rouge", slug="chanel-rouge-dior", name="Rouge"))
    db.commit()
    data = _file(db)
    for p in data["proposals"]:
        p["sheet_line_ref"] = "line:new:chanel-rouge-dior"
    data["proposals"][0]["entity"]["key"] = "line:new:chanel-rouge-dior"
    data["proposals"][1]["value"] = "line:new:chanel-rouge-dior"
    proposals.load(db, data)
    result = proposals.approve(db, "chanel", PASS, scope={"proposal_uids": [_uid(db, "name", "product_line"), _uid(db, "product_line")]}, by="rian")
    assert [r["code"] for r in result["refusals"]] == ["SLUG_TAKEN", "SLUG_TAKEN"]
    assert "Dior" in result["refusals"][0]["summary"] and result["counts"]["approved"] == 0
    assert db.get(ProductVariant, 4).product_line_id == 3


def test_a_variant_merged_since_the_load_receives_its_decision_on_the_survivor(db):
    """Spec test 17: the membership is re-resolved from its natural key at approval."""
    from app.services.decisions import writer

    proposals.load(db, _file(db))
    with writer.batch(db, "route", "individual", "rian") as b:
        writer.record(b, "product_variant", 4, "merged_into", 2, reason="one lipstick")
    assert db.get(ProductVariant, 4).merged_into_id == 2
    result = proposals.approve(db, "chanel", PASS, scope={"proposal_uids": [_uid(db, "name", "product_line"), _uid(db, "product_line")]}, by="rian")
    assert result["counts"]["approved"] == 2, result
    member = db.scalar(select(Decision).where(Decision.field == "product_line", Decision.origin == "proposal"))
    assert member.entity_id == 2 and db.get(ProductVariant, 2).product_line_id == 1


def test_the_merge_runs_after_the_memberships_it_depends_on(db):
    proposals.load(db, _file(db))
    uids = [_uid(db, "decision"), _uid(db, "product_line"), _uid(db, "name", "product_line")]
    result = proposals.approve(db, "chanel", PASS, scope={"proposal_uids": uids}, by="rian")
    assert result["counts"]["approved"] == 3, result
    rows = list(db.scalars(select(Decision).where(Decision.caused_by_id.is_(None), Decision.origin == "proposal").order_by(Decision.id)))
    assert [r.field for r in rows] == ["name", "product_line", "alias_of", "decision"], "the absorbed line's alias before the merge"
    assert db.get(ProductVariant, 4).merged_into_id == 1


def test_a_rejected_pair_is_keep_separate_with_its_pass_kept_and_any_other_reject_writes_nothing(db):
    proposals.load(db, _file(db))
    with pytest.raises(Refused):
        proposals.approve(db, "chanel", PASS, scope=None, reject=[_uid(db, "decision")], by="rian")
    result = proposals.approve(db, "chanel", PASS, scope=None, reject=[_uid(db, "decision"), _uid(db, "attribute:color")],
                               note="two different shades despite the shared number", by="rian")
    assert result["counts"]["rejected"] == 2
    pair = db.scalar(select(Proposal).where(Proposal.field == "decision"))
    decision = db.get(Decision, pair.decision_id)
    assert decision.value["decision"] == "separate" and decision.origin == "person" and decision.mode == "individual"
    assert decision.pass_id == pair.pass_id and decision.proposal_id == pair.id
    assert db.scalar(select(Suggestion)).decision == "separate"
    assert db.scalar(select(func.count()).select_from(Decision).where(Decision.field == "attribute:color")) == 0
    assert db.scalar(select(Proposal).where(Proposal.field == "attribute:color")).status == "rejected"


def test_a_correction_is_a_persons_individual_act_with_the_proposal_kept(db):
    proposals.load(db, _file(db))
    uid = _uid(db, "attribute:color")
    result = proposals.approve(db, "chanel", PASS, scope=None, corrections={uid: "99 Pirate Rouge"}, by="rian")
    assert result["counts"]["approved"] == 1
    row = db.scalar(select(Decision).where(Decision.field == "attribute:color"))
    assert row.origin == "person" and row.mode == "individual" and row.detail["corrected"] is True
    assert row.detail["proposed_value"] == "99 Pirate" and row.proposal_id is not None and row.value == "99 pirate rouge"


def test_the_gates_refuse_while_a_collection_runs(db):
    """Spec test 22: approve and undo-batch refuse while a collection runs; force overrides."""
    proposals.load(db, _file(db))
    uid = _uid(db, "attribute:color")
    first = proposals.approve(db, "chanel", PASS, scope={"proposal_uids": [uid]}, by="rian")
    db.add(Source(id=1, slug="attenza", name="Attenza"))
    db.flush()
    db.add(CollectionRun(source_id=1, status="running", started_at=datetime.now(UTC)))
    db.commit()
    with pytest.raises(Refused) as refused:
        proposals.approve(db, "chanel", PASS, scope="all", by="rian")
    assert refused.value.code == "COLLECTION_RUNNING"
    with pytest.raises(Refused) as refused:
        proposals.undo_batch(db, first["batch_uid"], "rian", "wrong field")
    assert refused.value.code == "COLLECTION_RUNNING"
    assert proposals.undo_batch(db, first["batch_uid"], "rian", "wrong field", force=True)["decisions"] >= 1


def test_undo_a_batch_then_approve_again_writes_a_new_decision(db):
    proposals.load(db, _file(db))
    uid = _uid(db, "attribute:color")
    first = proposals.approve(db, "chanel", PASS, scope={"proposal_uids": [uid]}, by="rian")
    undone = proposals.undo_batch(db, first["batch_uid"], "rian", "approved too early")
    assert undone["decisions"] == 1
    assert (1, "attribute:color") not in effective(db, "product_variant", [1])
    again = proposals.approve(db, "chanel", PASS, scope={"proposal_uids": [uid]}, by="rian")
    assert again["counts"]["approved"] == 1
    p = db.scalar(select(Proposal).where(Proposal.field == "attribute:color"))
    assert p.status == "approved" and effective(db, "product_variant", [1])[(1, "attribute:color")].id == p.decision_id
    assert len(p.detail["undone_decision_ids"]) == 1
    single = proposals.undo_decision(db, p.decision_id, "rian", "one row, not the batch")
    assert single["decision_uid"] and (1, "attribute:color") not in effective(db, "product_variant", [1])


def test_an_absorbed_line_is_aliased_once_empty_and_waits_while_a_member_is_left(db):
    data = _file(db)
    data["proposals"][0]["entity"]["detail"]["absorbs"] = [_key(db, ProductLine, 3), _key(db, ProductLine, 2)]
    proposals.load(db, data)
    result = proposals.approve(db, "chanel", PASS, scope={"proposal_uids": [_uid(db, "name", "product_line"), _uid(db, "product_line")]}, by="rian")
    states = {a["line"]: a["state"] for a in result["absorbs"]}
    assert states[_key(db, ProductLine, 3)] == "aliased" and states[_key(db, ProductLine, 2)] == "waiting"
    assert db.get(ProductLine, 3).alias_of_id == 1 and db.get(ProductLine, 2).alias_of_id is None
    alias = db.scalar(select(Decision).where(Decision.field == "alias_of"))
    assert alias.origin == "proposal" and alias.caused_by_id is None, "an absorb exports and replays like any approved row"


def test_the_derived_absorbs_name_every_line_whose_members_all_move(db):
    proposals.load(db, _file(db))
    header = db.scalar(select(Proposal).where(Proposal.entity_type == "product_line"))
    assert header.detail["absorbs_derived"] == [_key(db, ProductLine, 3)]


def _client(db, monkeypatch):
    api = FastAPI()
    api.include_router(review_router.router)
    def borrowed():
        yield db

    api.dependency_overrides[get_db] = borrowed
    monkeypatch.setattr(identity, "actor", lambda request: SimpleNamespace(id=1, username="rian", display_name="rian"))
    monkeypatch.setattr(audit_log, "record", lambda *a, **k: None)
    return TestClient(api)


def test_the_routes_answer_in_their_shapes(db, monkeypatch):
    proposals.load(db, _file(db))
    c = _client(db, monkeypatch)
    sheets = c.get("/api/review/sheets")
    assert sheets.status_code == 200 and sheets.json()[0]["brand_slug"] == "chanel"
    sheet = c.get("/api/review/sheets/chanel")
    # No pass named: the whole brand, every live pass, and pass_name says "not narrowed".
    assert sheet.status_code == 200 and sheet.json()["pass_name"] is None
    assert {g["pass_name"] for g in sheet.json()["groups"]} == {PASS}, "each group names the pass that asked it"
    narrowed = c.get("/api/review/sheets/chanel", params={"pass": PASS})
    assert narrowed.status_code == 200 and narrowed.json()["pass_name"] == PASS
    assert c.get("/api/review/sheets/nobody").status_code == 404
    bad = c.post("/api/review/sheets/chanel/approve", json={"pass": PASS, "reject": [_uid(db, "decision")]})
    assert bad.status_code == 422 and bad.json()["detail"]["error_code"] == "VALUE_INVALID"
    ok = c.post("/api/review/sheets/chanel/approve", json={"pass": PASS, "scope": {"proposal_uids": [_uid(db, "attribute:color")]}})
    assert ok.status_code == 200, ok.text
    body = ok.json()
    assert body["message"] == "1 approved, 0 held, 0 refused" and body["batch_uid"]
    assert c.get(f"/api/review/batches/{body['batch_uid']}").json()["decisions"][0]["field"] == "attribute:color"
    ledger = c.get("/api/review/decisions", params={"entity": "product_variant:1"}).json()
    assert len(ledger) == 1
    undo = c.post(f"/api/review/decisions/{ledger[0]['id']}/undo", json={"reason": "wrong field"})
    assert undo.status_code == 200 and undo.json()["decision_uid"]
    again = c.post(f"/api/review/batches/{body['batch_uid']}/undo", json={"reason": "again"})
    assert again.status_code == 200, "a batch whose one row is already undone is skipped, not refused"
    assert again.json()["skipped"]


def test_a_member_cannot_fold_a_line_its_header_kept_separate(db, monkeypatch):
    """What it cost: a header and the members under it are two routes to ONE outcome, and nothing
    tied them. Keeping "Million Gold absorbs One Million Gold" separate and then approving the two
    membership rows moved the very variants the rejection was about: the fold happened anyway, with
    a rejection recorded beside it. `LINE_NOT_APPROVED` did not catch it, because that guard only
    fires for a line that does not exist yet and every target here already existed."""
    data = _file(db)
    proposals.load(db, data)
    name = data["pass"]["name"]
    header = db.scalar(select(Proposal).where(Proposal.entity_type == "product_line", Proposal.field == "name"))
    member = db.scalar(select(Proposal).where(Proposal.field == "product_line",
                                              Proposal.sheet_line_ref == header.natural_key))
    assert member is not None, "the fixture must hold a member under that header"

    proposals.approve(db, "chanel", name, scope=None, by="rian", reject=[str(header.uid)],
                      note="these are two product lines")
    out = proposals.approve(db, "chanel", name, scope={"proposal_uids": [str(member.uid)]}, by="rian")
    assert out["counts"]["refused"] == 1
    assert out["refusals"][0]["code"] == "HEADER_ANSWERED_OTHERWISE"
    assert db.scalar(select(Proposal).where(Proposal.uid == member.uid)).status == "open", "left for a person"


def test_a_member_is_refused_while_its_header_is_only_deferred(db, monkeypatch):
    """Deferred is not answered, and applying half the consequence answers it for him."""
    data = _file(db)
    proposals.load(db, data)
    name = data["pass"]["name"]
    header = db.scalar(select(Proposal).where(Proposal.entity_type == "product_line", Proposal.field == "name"))
    member = db.scalar(select(Proposal).where(Proposal.field == "product_line",
                                              Proposal.sheet_line_ref == header.natural_key))
    proposals.approve(db, "chanel", name, scope=None, by="rian", defer=[str(header.uid)], note="need to look")
    out = proposals.approve(db, "chanel", name, scope={"proposal_uids": [str(member.uid)]}, by="rian")
    assert out["refusals"][0]["code"] == "HEADER_ANSWERED_OTHERWISE"