"""The machine paths honour the ledger (Stream K2; spec §3.1, §5; tests 8, 9, 11). What it cost:
`suggest.generate` wrote `superseded` into the person's `decision` column, so the desk could
not tell a ruling from a closure; the automatic fold read no Keep separate at all (plan W9, a
defect found 16 Sep), and a rules fold could take a decided variant as its loser and lose its
rows. SQLite kit, no network."""
from __future__ import annotations

import pytest
from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.models import (Account, Base, Brand, Decision, LEDGER_TABLES, Listing, Merge, ProductLine, ProductVariant, Retailer, Shop,
                        Suggestion, AttributeAlias, Award)
from app.models.places import Place, ShopPlace
from app.services import keying, merge_session, merges, suggest
from app.services.decisions import natural_keys, writer

TABLES = [Account.__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


@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"))
        brand = Brand(slug="jw", name="Johnnie Walker"); retailer = Retailer(slug="r", name="R")
        s.add_all([brand, retailer]); s.flush()
        line = ProductLine(brand_id=brand.id, key="blue label", slug="jw-blue-label", name="Blue Label"); s.add(line); s.flush()
        for n in (1, 2, 3):
            v = ProductVariant(id=n, name="Blue Label 1L", brand="Johnnie Walker", brand_id=brand.id, vertical="liquor", match_key="jw|blue-label||1000ml",
                               product_line_id=line.id, quantity_value=1000, quantity_unit="ml", quantity_state="stated", quantity_ml=1000, form="single", attributes={})
            shop = Shop(id=n, retailer_id=retailer.id, code=f"S{n}", iata=f"A0{n}", name=f"Shop {n}", currency="EUR")
            s.add_all([v, shop]); s.flush()
            s.add(Listing(variant_id=n, shop_id=n, source_sku=f"sku{n}"))
        s.commit()
        yield s
    keying.invalidate()


def pair_key(db, a, b):
    return natural_keys.pair_key("product", a, b, sides=(f"variant:{db.get(ProductVariant, a).uid}", f"variant:{db.get(ProductVariant, b).uid}"))[0]


def test_a_pair_kept_separate_never_folds_and_follows_a_side_through_a_merge(db):
    # Three variants share one key with no barcode: the rules would fold them.
    with writer.batch(db, "desk", "individual", RIAN, commit=True, tail=False) as b:
        writer.record(b, "suggestion", pair_key(db, 1, 2), "decision", {"decision": "separate", "survivor": None, "name": None, "note": None})
    keying.invalidate()
    counts = merges.merge_duplicates(db)
    db.flush()
    assert counts["groups"] == 0 and counts["candidate_groups"] == 1
    assert db.scalar(select(Suggestion).where(Suggestion.reason.like("%kept_separate%"))) is not None
    # 2 merges into 3 by a person: the veto follows to {1, 3} and a carried Keep separate exists.
    with writer.batch(db, "route", "individual", RIAN, commit=True, tail=False) as b:
        writer.record(b, "product_variant", 2, "merged_into", 3, reason="same bottle")
    keying.invalidate()
    maps = keying.load_maps(db)
    assert maps.separated == {frozenset((1, 3))}
    assert db.scalar(select(Suggestion).where(Suggestion.left_id == 1, Suggestion.right_id == 3, Suggestion.decision == "separate")) is not None
    counts = merges.merge_duplicates(db)
    db.flush()
    assert counts["groups"] == 0 and db.get(ProductVariant, 1).merged_into_id is None and db.get(ProductVariant, 3).merged_into_id is None


def test_a_decided_variant_never_auto_folds(db):
    with writer.batch(db, "route", "individual", RIAN, commit=True, tail=False) as b:
        writer.record(b, "product_variant", 2, "name", "Blue Label Litre", reason="typed")
    keying.invalidate()
    counts = merges.merge_duplicates(db)
    db.flush()
    assert counts["groups"] == 0 and counts["candidate_groups"] == 1
    assert all(v.merged_into_id is None for v in db.scalars(select(ProductVariant)))
    reason = db.scalar(select(Suggestion.reason).where(Suggestion.level == "product"))
    assert "decided_member" in reason and "a person's decision" in suggest.product_reason(reason)


def test_generate_closes_with_closed_reason_only_and_a_closed_pair_can_still_be_decided(db, monkeypatch):
    # A rule pair whose rule no longer offers it: withdrawn by closure, never by a decision.
    db.add(Suggestion(level="product", left_id=1, right_id=2, reason="name_within", score=0.5, detail={"why": "x"}))
    db.commit()
    monkeypatch.setattr(suggest, "_facts", lambda db: ([], [], []))
    counts = suggest.generate(db)
    row = db.scalar(select(Suggestion).where(Suggestion.left_id == 1, Suggestion.right_id == 2))
    assert counts["withdrawn"] == 1 and row.closed_reason == "withdrawn" and row.decision is None
    assert db.scalar(select(Suggestion).where(Suggestion.is_open())) is None
    # The same row can later be decided: a person's Keep separate reopens nothing and closes it for good.
    merge_session.keep_separate(db, row.id, decided_by=RIAN, note="not one bottle")
    assert row.decision == "separate" and row.decision_id is not None
    assert db.scalar(select(Decision).where(Decision.entity_type == "suggestion", Decision.entity_id == row.id)) is not None
    # A pre-minted `decided` pair the writer made is left alone by generate, and a superseded side closes by closure.
    assert suggest.generate(db)["superseded"] == 0
    with writer.batch(db, "route", "individual", RIAN, commit=True, tail=False) as b:
        writer.record(b, "product_variant", 3, "merged_into", 1, reason="one bottle")
    db.add(Suggestion(level="product", left_id=2, right_id=3, reason="name_within", score=0.5, detail={"why": "y"}))
    db.commit()
    assert suggest.generate(db)["superseded"] == 1
    closed = db.scalar(select(Suggestion).where(Suggestion.left_id == 2, Suggestion.right_id == 3))
    assert closed.closed_reason == "superseded" and closed.decision is None