fin-hub/tests/test_ai_plan.py

from pathlib import Path

import pytest

from fin_hub.ingest.ai_plan import parse_ai_plan_csv
from fin_hub.services.evaluate import evaluate_runway
from fin_hub.services.exchange import booked_cost_projection
from fin_hub.services.ledger import (
    import_csv,
    list_current_commitments,
    monthly_summary,
    set_opening_balance,
)

FIXTURES = Path(__file__).parent / "fixtures"


def test_parse_ai_plan_fixture_keeps_subscription_and_topup_separate():
    rows = parse_ai_plan_csv(FIXTURES / "ai-plans.csv")
    assert len(rows) == 2
    assert rows[0].provider == "anthropic"
    assert rows[0].charge_kind == "subscription"
    assert rows[0].ongoing is True
    assert rows[0].provider_account is None
    assert rows[1].provider == "openai"
    assert rows[1].charge_kind == "usage_topup"
    assert rows[1].ongoing is False
    assert rows[1].provider_account == "org-demo"
    assert rows[0].incurred_on.isoformat() == "2026-08-01"


def test_parse_ai_plan_rejects_session_level_rows(tmp_path: Path):
    source = tmp_path / "sessions.csv"
    source.write_text(
        "provider,plan,charge_kind,amount,currency,period_month,session_id\n"
        "anthropic,claude-max,subscription,200.00,EUR,2026-08,sess-1\n",
        encoding="utf-8",
    )
    with pytest.raises(ValueError, match="session-level"):
        parse_ai_plan_csv(source)


def test_import_ai_plans_books_once_and_commits_only_subscriptions(tmp_path: Path):
    ledger = tmp_path / "ledger.db"
    first = import_csv(FIXTURES / "ai-plans.csv", "ai-plan", ledger_path=ledger)
    second = import_csv(FIXTURES / "ai-plans.csv", "ai-plan", ledger_path=ledger)
    assert first.rows_imported == 2
    assert second.skipped is True

    rollup = monthly_summary(ledger_path=ledger)[0]
    assert rollup.currency == "EUR"
    assert rollup.period_month == "2026-08"
    assert rollup.total == pytest.approx(240.0)
    assert rollup.entry_count == 2

    commitments = list_current_commitments(ledger_path=ledger)
    assert len(commitments) == 1
    assert commitments[0].provider == "anthropic"
    assert commitments[0].plan == "claude-max"
    assert commitments[0].amount == 200
    assert commitments[0].financial_fact_id is not None


def test_ai_plan_correction_is_append_only(tmp_path: Path):
    ledger = tmp_path / "ledger.db"
    source = tmp_path / "plans.csv"
    source.write_text(
        "provider,plan,charge_kind,amount,currency,period_month,ongoing\n"
        "anthropic,claude-max,subscription,200.00,EUR,2026-08,true\n",
        encoding="utf-8",
    )
    import_csv(source, "ai-plan", ledger_path=ledger)
    source.write_text(
        "provider,plan,charge_kind,amount,currency,period_month,ongoing\n"
        "anthropic,claude-max,subscription,220.00,EUR,2026-08,true\n",
        encoding="utf-8",
    )
    with pytest.raises(ValueError, match="explicit correction"):
        import_csv(source, "ai-plan", ledger_path=ledger)
    corrected = import_csv(source, "ai-plan", ledger_path=ledger, force=True)
    assert corrected.rows_imported == 1
    assert monthly_summary(ledger_path=ledger)[0].total == pytest.approx(220.0)
    commitments = list_current_commitments(ledger_path=ledger)
    assert len(commitments) == 1
    assert commitments[0].amount == 220
    assert commitments[0].revision_of is not None


def test_ending_a_subscription_closes_the_commitment(tmp_path: Path):
    ledger = tmp_path / "ledger.db"
    source = tmp_path / "plans.csv"
    source.write_text(
        "provider,plan,charge_kind,amount,currency,period_month,ongoing\n"
        "anthropic,claude-max,subscription,200.00,EUR,2026-08,true\n",
        encoding="utf-8",
    )
    import_csv(source, "ai-plan", ledger_path=ledger)
    source.write_text(
        "provider,plan,charge_kind,amount,currency,period_month,ongoing\n"
        "anthropic,claude-max,subscription,200.00,EUR,2026-09,false\n",
        encoding="utf-8",
    )
    import_csv(source, "ai-plan", ledger_path=ledger)
    assert list_current_commitments(ledger_path=ledger) == []
    months = {row.period_month: row.total for row in monthly_summary(ledger_path=ledger)}
    assert months["2026-08"] == pytest.approx(200.0)
    assert months["2026-09"] == pytest.approx(200.0)


def test_ai_plan_facts_project_without_invented_resource_id(tmp_path: Path):
    ledger = tmp_path / "ledger.db"
    import_csv(FIXTURES / "ai-plans.csv", "ai-plan", ledger_path=ledger)
    facts = booked_cost_projection(ledger_path=ledger)
    by_provider = {fact.provider: fact for fact in facts}
    assert set(by_provider) == {"anthropic", "openai"}
    assert by_provider["anthropic"].service_id == "claude-max"
    assert by_provider["anthropic"].resource_id is None
    assert by_provider["anthropic"].tax_status == "unknown"
    assert by_provider["openai"].service_id == "api-prepaid"
    assert by_provider["openai"].effective_amount == 40


def test_ai_plan_commitment_appears_on_runway_when_no_other_burns(tmp_path: Path):
    ledger = tmp_path / "ledger.db"
    set_opening_balance(ledger, 12000.0, currency="EUR")
    source = tmp_path / "future-only.csv"
    source.write_text(
        "provider,plan,charge_kind,amount,currency,period_month,ongoing\n"
        "xai,grok-heavy,subscription,300.00,USD,2026-09,true\n",
        encoding="utf-8",
    )
    import_csv(source, "ai-plan", ledger_path=ledger)
    empty_eur = evaluate_runway(ledger_path=ledger, currency="EUR")
    assert empty_eur["commitments"] == []
    usd = evaluate_runway(ledger_path=ledger, currency="USD")
    assert usd["runway"]["monthly_burn"] == pytest.approx(300.0)
    assert usd["commitments"][0]["plan"] == "grok-heavy"


def test_booked_ai_plan_invoices_drive_runway_without_double_counting(tmp_path: Path):
    ledger = tmp_path / "ledger.db"
    set_opening_balance(ledger, 12000.0)
    import_csv(FIXTURES / "ai-plans.csv", "ai-plan", ledger_path=ledger)
    report = evaluate_runway(ledger_path=ledger)
    assert report["runway"]["monthly_burn"] == pytest.approx(240.0)
    assert len(report["commitments"]) == 1
    assert report["commitments"][0]["amount"] == 200