"""Finance ledger read surface — summary / pivot / timeseries / movers /
completeness / digest, over a materialised period.

Reuses the four-stream ``seeded`` fixture (cloud EC2 $100 + Bedrock AI
$40, software 700 CNY → $70, procurement $500 USD) from
``test_finance_rollup``.
"""
from __future__ import annotations

from decimal import Decimal

import pytest

from app.core.coordinators import finance_spend_coordinator as coord
from app.domains.finance import read_service

# ``seeded`` is a pytest fixture reused from the rollup suite; every test
# below takes it as a parameter, which ruff reads as a redefinition of the
# imported name (F811) — the documented pytest idiom, not a shadowing bug.
from tests.test_finance.test_finance_rollup import PERIOD, seeded  # noqa: F401


@pytest.mark.asyncio
async def test_summary_separates_bases_and_surfaces_unallocated(seeded):
    db = seeded
    await coord.materialize_period(db, PERIOD)
    await db.flush()
    s = await read_service.get_summary(db, PERIOD)
    assert Decimal(s["consumption_total_usd"]) == Decimal("210.00")  # 100+70+40
    assert Decimal(s["procurement_total_usd"]) == Decimal("500.00")
    assert s["is_provisional"] is True  # not closed
    cts = {c["cost_type"]: c for c in s["by_cost_type"]}
    assert Decimal(cts["cloud"]["usd"]) == Decimal("100.00")
    assert Decimal(cts["ai"]["usd"]) == Decimal("40.00")
    assert Decimal(cts["software"]["usd"]) == Decimal("70.00")


@pytest.mark.asyncio
async def test_summary_excludes_hardware_cash_from_consumption(db_session):
    """Regression: the hardware CASH lens (purchase month) must NOT be summed
    into consumption_total / the cost-type mix — that double-counts the asset
    against its depreciation (violates the never-sum-across-bases invariant)."""
    from app.domains.finance.models import FinanceSpendFact

    db = db_session

    def f(ct, basis, usd):
        return FinanceSpendFact(
            fact_key=f"cashtest|{ct}|{basis}|{usd}", period=PERIOD, cost_type=ct,
            basis=basis, amount_native=Decimal(usd), currency_native="USD",
            amount_usd=Decimal(usd), fx_basis="native_usd", allocation_source="rule",
            source_domain=ct, rollup_fingerprint="cashtest", is_provisional=False,
        )

    db.add(f("cloud", "consumption", "100"))
    db.add(f("hardware", "consumption", "7"))   # depreciation → consumption
    db.add(f("hardware", "cash", "300"))        # purchase cash → EXCLUDED
    db.add(f("procurement", "procurement", "50"))
    await db.flush()

    s = await read_service.get_summary(db, PERIOD)
    # 100 + 7 only — the $300 cash purchase is not in consumption.
    assert Decimal(s["consumption_total_usd"]) == Decimal("107.00")
    assert Decimal(s["procurement_total_usd"]) == Decimal("50.00")
    cts = {c["cost_type"]: Decimal(c["usd"]) for c in s["by_cost_type"]}
    assert cts["hardware"] == Decimal("7.00")  # depreciation only, not 307


@pytest.mark.asyncio
async def test_pivot_excludes_hardware_cash_by_default(db_session):
    """Regression: a no-basis pivot/timeseries must drop the hardware CASH
    lens so it can't sum a hardware asset against its own depreciation. A
    non-overlapping basis (procurement) stays visible; an explicit basis=cash
    drill-in still works."""
    from app.domains.finance.models import FinanceSpendFact

    db = db_session

    def f(ct, basis, usd, key):
        return FinanceSpendFact(
            fact_key=key, period=PERIOD, cost_type=ct, basis=basis,
            amount_native=Decimal(usd), currency_native="USD",
            amount_usd=Decimal(usd), fx_basis="native_usd",
            allocation_source="rule", source_domain=ct,
            rollup_fingerprint="cashpivot", is_provisional=False,
        )

    db.add(f("hardware", "consumption", "7", "cp|hw|cons"))
    db.add(f("hardware", "cash", "300", "cp|hw|cash"))
    db.add(f("procurement", "procurement", "50", "cp|proc"))
    await db.flush()

    # Default (no basis): hardware = depreciation only (cash dropped, NOT 307);
    # procurement (non-overlapping basis) stays visible.
    p = await read_service.pivot(db, PERIOD, ["cost_type"], {})
    by = {r["cost_type"]: Decimal(r["usd"]) for r in p["rows"]}
    assert by["hardware"] == Decimal("7.00")
    assert by["procurement"] == Decimal("50.00")
    assert Decimal(p["total_usd"]) == Decimal("57.00")

    # Trend line excludes cash too.
    ts = await read_service.timeseries(db, PERIOD, PERIOD, None, {})
    assert Decimal(ts["series"][0]["usd"]) == Decimal("57.00")

    # Explicit basis=cash → the cash lens is still drillable.
    pc = await read_service.pivot(db, PERIOD, ["cost_type"], {"basis": "cash"})
    byc = {r["cost_type"]: Decimal(r["usd"]) for r in pc["rows"]}
    assert byc["hardware"] == Decimal("300.00")

    # Splitting BY basis keeps every basis (deliberate breakdown view).
    pb = await read_service.pivot(db, PERIOD, ["basis"], {})
    byb = {r["basis"]: Decimal(r["usd"]) for r in pb["rows"]}
    assert byb["cash"] == Decimal("300.00")
    assert byb["consumption"] == Decimal("7.00")


@pytest.mark.asyncio
async def test_pivot_by_cost_type_totals(seeded):
    db = seeded
    await coord.materialize_period(db, PERIOD)
    await db.flush()
    p = await read_service.pivot(db, PERIOD, ["cost_type"], {})
    by = {r["cost_type"]: Decimal(r["usd"]) for r in p["rows"]}
    assert by["cloud"] == Decimal("100.00")
    assert by["ai"] == Decimal("40.00")
    assert by["software"] == Decimal("70.00")
    assert by["procurement"] == Decimal("500.00")


@pytest.mark.asyncio
async def test_pivot_department_filter_consumption_only(seeded):
    db = seeded
    await coord.materialize_period(db, PERIOD)
    await db.flush()
    # Engineering should see only its cloud spend ($100, EC2); Bedrock is
    # AI cost_type (also Engineering) — filter by cost_type to isolate.
    p = await read_service.pivot(
        db, PERIOD, ["cost_type"],
        {"department": "Engineering", "basis": "consumption"},
    )
    by = {r["cost_type"]: Decimal(r["usd"]) for r in p["rows"]}
    assert by.get("cloud") == Decimal("100.00")
    assert by.get("ai") == Decimal("40.00")
    assert "software" not in by  # software dept is Design, not Engineering


@pytest.mark.asyncio
async def test_movers_picks_up_new_spend(seeded):
    db = seeded
    await coord.materialize_period(db, PERIOD)
    await db.flush()
    # Previous month has nothing → every department is an increase.
    m = await read_service.movers(db, PERIOD, "department", top=10)
    keys = {i["key"] for i in m["increases"]}
    assert "Engineering" in keys
    assert all(Decimal(i["prev_usd"]) == 0 for i in m["increases"])


@pytest.mark.asyncio
async def test_completeness_lists_sources(seeded):
    db = seeded
    await coord.materialize_period(db, PERIOD)
    await db.flush()
    c = await read_service.completeness(db, PERIOD)
    domains = {s["source_domain"] for s in c["sources"]}
    assert {"cmdb", "asset", "ai_spending", "procurement"} <= domains
    assert c["is_provisional"] is True


@pytest.mark.asyncio
async def test_timeseries_ascending_and_zero_filled(seeded):
    db = seeded
    await coord.materialize_period(db, PERIOD)
    await db.flush()
    r = await read_service.timeseries(db, "2026-03", "2026-05", None, {})
    assert r["periods"] == ["2026-03", "2026-04", "2026-05"]
    by = {p["period"]: Decimal(p["usd"]) for p in r["series"]}
    assert by["2026-03"] == Decimal("0.00")
    # 2026-05 sums ALL amount_usd (no basis filter): 100+70+40+500.
    assert by["2026-05"] == Decimal("710.00")


@pytest.mark.asyncio
async def test_timeseries_accepts_reversed_endpoints(seeded):
    db = seeded
    await coord.materialize_period(db, PERIOD)
    await db.flush()
    # Endpoints in descending order must still produce the ascending range
    # (regression: the loop used to walk away from the target → 400).
    r = await read_service.timeseries(db, "2026-05", "2026-03", None, {})
    assert r["periods"] == ["2026-03", "2026-04", "2026-05"]


@pytest.mark.asyncio
async def test_digest_is_deterministic_narrative(seeded):
    db = seeded
    await coord.materialize_period(db, PERIOD)
    await db.flush()
    d = await read_service.digest(db, PERIOD)
    assert d["period"] == PERIOD
    assert any("消费总额" in line for line in d["lines"])
    # Procurement called out as separate basis.
    assert any("采购" in line for line in d["lines"])