tegwick
09ef7fa967
Record the FIN-WP-0007 split after checking resource-control: they keep intelligence resource identity and provision-level class I metering; session tokens stay with State Hub; work-effectiveness is a reporting join. Missing token counts are now unknown, not zero.
313 lines
10 KiB
Python
313 lines
10 KiB
Python
import sqlite3
|
|
import os
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from fin_hub.services.evaluate import evaluate_runway
|
|
from fin_hub.services.evidence import build_runway_report, seed_fixture_ledger, write_runway_evidence
|
|
from fin_hub.services.ledger import (
|
|
client_margin_report,
|
|
import_csv,
|
|
monthly_burn_series,
|
|
monthly_summary,
|
|
record_engagement_price,
|
|
reverse_financial_fact,
|
|
set_opening_balance,
|
|
)
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
def test_ledger_import_anthropic_cost_only_does_not_require_tokens(tmp_path: Path):
|
|
ledger = tmp_path / "ledger.db"
|
|
source = tmp_path / "anthropic-cost-only.csv"
|
|
source.write_text(
|
|
"model,cost,currency,usage_date\nclaude-opus-4-20250514,20.00,USD,2026-08-01\n",
|
|
encoding="utf-8",
|
|
)
|
|
result = import_csv(source, "anthropic", ledger_path=ledger)
|
|
assert result.rows_imported == 1
|
|
rollups = monthly_summary(ledger_path=ledger)
|
|
assert rollups[0].currency == "USD"
|
|
assert rollups[0].total == pytest.approx(20.0)
|
|
|
|
|
|
def test_ledger_import_and_monthly_summary(tmp_path: Path):
|
|
ledger = tmp_path / "ledger.db"
|
|
set_opening_balance(ledger, 12000.0)
|
|
|
|
cloud = import_csv(FIXTURES / "cloud-costs.csv", "cloud", ledger_path=ledger)
|
|
host = import_csv(FIXTURES / "hosteurope.csv", "hosteurope", ledger_path=ledger)
|
|
|
|
assert cloud.rows_imported == 3
|
|
assert host.rows_imported == 2
|
|
|
|
rollups = monthly_summary(ledger_path=ledger)
|
|
eur_totals = {rollup.period_month: rollup.total for rollup in rollups if rollup.currency == "EUR"}
|
|
assert eur_totals["2026-06"] == pytest.approx(85.5 + 99.8)
|
|
|
|
burns = monthly_burn_series(ledger_path=ledger, currency="EUR")
|
|
assert burns == [pytest.approx(185.3)]
|
|
|
|
|
|
def test_ledger_import_skips_unchanged_file(tmp_path: Path):
|
|
ledger = tmp_path / "ledger.db"
|
|
first = import_csv(FIXTURES / "cloud-costs.csv", "cloud", ledger_path=ledger)
|
|
second = import_csv(FIXTURES / "cloud-costs.csv", "cloud", ledger_path=ledger)
|
|
assert first.rows_imported == 3
|
|
assert second.skipped is True
|
|
assert second.rows_imported == 0
|
|
|
|
|
|
def test_ledger_import_is_idempotent_when_file_is_renamed_or_touched(tmp_path: Path):
|
|
ledger = tmp_path / "ledger.db"
|
|
source = tmp_path / "costs.csv"
|
|
shutil.copy(FIXTURES / "cloud-costs.csv", source)
|
|
renamed = tmp_path / "renamed.csv"
|
|
|
|
first = import_csv(source, "cloud", ledger_path=ledger)
|
|
os.utime(source, (source.stat().st_atime, source.stat().st_mtime + 1))
|
|
touched = import_csv(source, "cloud", ledger_path=ledger)
|
|
shutil.copy(source, renamed)
|
|
delivered_again = import_csv(renamed, "cloud", ledger_path=ledger)
|
|
|
|
assert first.rows_imported == 3
|
|
assert touched.rows_imported == 0
|
|
assert delivered_again.rows_imported == 0
|
|
assert monthly_summary(ledger_path=ledger)[0].entry_count == 3
|
|
|
|
|
|
def test_changed_fact_requires_and_applies_one_correction(tmp_path: Path):
|
|
ledger = tmp_path / "ledger.db"
|
|
source = tmp_path / "costs.csv"
|
|
source.write_text(
|
|
"service,amount,currency,period_month\ncompute,10.00,EUR,2026-07\n",
|
|
encoding="utf-8",
|
|
)
|
|
import_csv(source, "cloud", ledger_path=ledger)
|
|
source.write_text(
|
|
"service,amount,currency,period_month\ncompute,12.00,EUR,2026-07\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="explicit correction"):
|
|
import_csv(source, "cloud", ledger_path=ledger)
|
|
corrected = import_csv(source, "cloud", ledger_path=ledger, force=True)
|
|
duplicate = import_csv(source, "cloud", ledger_path=ledger, force=True)
|
|
|
|
assert corrected.rows_imported == 1
|
|
assert duplicate.rows_imported == 0
|
|
rollup = monthly_summary(ledger_path=ledger)[0]
|
|
assert rollup.total == 12.0
|
|
assert rollup.entry_count == 1
|
|
with sqlite3.connect(ledger) as conn:
|
|
facts = conn.execute(
|
|
"SELECT correction_of, is_current FROM ledger_entries ORDER BY id"
|
|
).fetchall()
|
|
assert facts[0][1] == 0
|
|
assert facts[1][0] is not None
|
|
assert facts[1][1] == 1
|
|
|
|
|
|
def test_financial_fact_reversal_is_append_only_and_zeroes_effective_total(tmp_path: Path):
|
|
ledger = tmp_path / "ledger.db"
|
|
source = tmp_path / "costs.csv"
|
|
source.write_text(
|
|
"service,amount,currency,period_month\ncompute,10.00,EUR,2026-07\n",
|
|
encoding="utf-8",
|
|
)
|
|
import_csv(source, "cloud", ledger_path=ledger)
|
|
with sqlite3.connect(ledger) as conn:
|
|
fact_id = conn.execute(
|
|
"SELECT financial_fact_id FROM ledger_entries"
|
|
).fetchone()[0]
|
|
|
|
reversal_id = reverse_financial_fact(
|
|
fact_id, source="provider-credit-note", ledger_path=ledger
|
|
)
|
|
|
|
rollup = monthly_summary(ledger_path=ledger)[0]
|
|
assert rollup.total == 0.0
|
|
with sqlite3.connect(ledger) as conn:
|
|
rows = conn.execute(
|
|
"SELECT financial_fact_id, correction_of, adjustment_kind, is_current "
|
|
"FROM ledger_entries ORDER BY id"
|
|
).fetchall()
|
|
assert rows == [
|
|
(fact_id, None, "charge", 0),
|
|
(reversal_id, fact_id, "reversal", 1),
|
|
]
|
|
|
|
|
|
def test_ledger_import_preserves_client_attribution(tmp_path: Path):
|
|
source = tmp_path / "attributed.csv"
|
|
source.write_text(
|
|
"product,amount,currency,invoice_date,client_id,application_id,app_instance_id\n"
|
|
"Managed cluster,42.00,EUR,2026-07-01,acme,portal,prod-01\n",
|
|
encoding="utf-8",
|
|
)
|
|
ledger = tmp_path / "ledger.db"
|
|
|
|
import_csv(source, "hosteurope", ledger_path=ledger)
|
|
|
|
with sqlite3.connect(ledger) as conn:
|
|
row = conn.execute(
|
|
"SELECT client_id, application_id, app_instance_id, cost_attribution_key "
|
|
"FROM ledger_entries"
|
|
).fetchone()
|
|
assert row == (
|
|
"acme",
|
|
"portal",
|
|
"prod-01",
|
|
"client:acme|app:portal|instance:prod-01",
|
|
)
|
|
|
|
|
|
def test_existing_ledger_schema_is_migrated_additively(tmp_path: Path):
|
|
ledger = tmp_path / "legacy.db"
|
|
with sqlite3.connect(ledger) as conn:
|
|
conn.execute(
|
|
"CREATE TABLE ledger_entries ("
|
|
"id INTEGER PRIMARY KEY, source_type TEXT NOT NULL, category TEXT NOT NULL, "
|
|
"label TEXT NOT NULL, amount REAL NOT NULL, currency TEXT NOT NULL, "
|
|
"period_month TEXT NOT NULL, incurred_on TEXT, source_path TEXT NOT NULL, "
|
|
"imported_at TEXT NOT NULL)"
|
|
)
|
|
|
|
monthly_summary(ledger_path=ledger)
|
|
|
|
with sqlite3.connect(ledger) as conn:
|
|
columns = {row[1] for row in conn.execute("PRAGMA table_info(ledger_entries)")}
|
|
assert {
|
|
"client_id",
|
|
"application_id",
|
|
"app_instance_id",
|
|
"cost_attribution_key",
|
|
}.issubset(columns)
|
|
|
|
|
|
def test_engagement_price_and_margin_report(tmp_path: Path):
|
|
source = tmp_path / "attributed.csv"
|
|
source.write_text(
|
|
"product,amount,currency,invoice_date,client_id,application_id,app_instance_id\n"
|
|
"Managed cluster,42.00,EUR,2026-07-01,acme,portal,prod-01\n",
|
|
encoding="utf-8",
|
|
)
|
|
ledger = tmp_path / "ledger.db"
|
|
import_csv(source, "hosteurope", ledger_path=ledger)
|
|
price = record_engagement_price(
|
|
client_id="acme",
|
|
application_id="portal",
|
|
app_instance_id="prod-01",
|
|
period_month="2026-07",
|
|
amount=100.0,
|
|
currency="eur",
|
|
source="agreement-2026-01",
|
|
ledger_path=ledger,
|
|
)
|
|
|
|
margins = client_margin_report(ledger_path=ledger)
|
|
|
|
assert margins[0].price_id == price.id
|
|
assert margins[0].revenue == 100.0
|
|
assert margins[0].attributed_cost == 42.0
|
|
assert margins[0].margin == 58.0
|
|
assert margins[0].currency == "EUR"
|
|
|
|
|
|
def test_engagement_price_revision_supersedes_prior_price(tmp_path: Path):
|
|
ledger = tmp_path / "ledger.db"
|
|
original = record_engagement_price(
|
|
client_id="acme",
|
|
application_id="portal",
|
|
app_instance_id="prod-01",
|
|
period_month="2026-07",
|
|
amount=100.0,
|
|
currency="EUR",
|
|
source="agreement-v1",
|
|
ledger_path=ledger,
|
|
)
|
|
revision = record_engagement_price(
|
|
client_id="acme",
|
|
application_id="portal",
|
|
app_instance_id="prod-01",
|
|
period_month="2026-07",
|
|
amount=120.0,
|
|
currency="EUR",
|
|
source="agreement-v2",
|
|
revision_of=original.id,
|
|
ledger_path=ledger,
|
|
)
|
|
|
|
margins = client_margin_report(ledger_path=ledger)
|
|
|
|
assert [(row.price_id, row.revenue) for row in margins] == [(revision.id, 120.0)]
|
|
|
|
|
|
def test_engagement_price_requires_explicit_revision(tmp_path: Path):
|
|
ledger = tmp_path / "ledger.db"
|
|
kwargs = {
|
|
"client_id": "acme",
|
|
"application_id": "portal",
|
|
"app_instance_id": "prod-01",
|
|
"period_month": "2026-07",
|
|
"amount": 100.0,
|
|
"currency": "EUR",
|
|
"source": "agreement",
|
|
"ledger_path": ledger,
|
|
}
|
|
original = record_engagement_price(**kwargs)
|
|
|
|
with pytest.raises(ValueError, match=original.id):
|
|
record_engagement_price(**kwargs)
|
|
|
|
|
|
def test_margin_does_not_join_cost_in_another_currency(tmp_path: Path):
|
|
source = tmp_path / "attributed.csv"
|
|
source.write_text(
|
|
"product,amount,currency,invoice_date,client_id,application_id,app_instance_id\n"
|
|
"Managed cluster,42.00,USD,2026-07-01,acme,portal,prod-01\n",
|
|
encoding="utf-8",
|
|
)
|
|
ledger = tmp_path / "ledger.db"
|
|
import_csv(source, "hosteurope", ledger_path=ledger)
|
|
record_engagement_price(
|
|
client_id="acme",
|
|
application_id="portal",
|
|
app_instance_id="prod-01",
|
|
period_month="2026-07",
|
|
amount=100.0,
|
|
currency="EUR",
|
|
source="agreement",
|
|
ledger_path=ledger,
|
|
)
|
|
|
|
margin = client_margin_report(ledger_path=ledger)[0]
|
|
assert margin.attributed_cost == 0.0
|
|
assert margin.margin == 100.0
|
|
|
|
|
|
def test_evaluate_runway_from_ledger(tmp_path: Path):
|
|
ledger = tmp_path / "ledger.db"
|
|
seed_fixture_ledger(ledger_path=ledger, opening_balance=12000.0)
|
|
report = evaluate_runway(ledger_path=ledger, alert_threshold_months=3.0)
|
|
assert report["runway"]["current_balance"] == 12000.0
|
|
assert report["runway"]["monthly_burn"] > 0
|
|
|
|
|
|
def test_build_runway_report_and_write_evidence(tmp_path: Path):
|
|
ledger = tmp_path / "ledger.db"
|
|
seed_fixture_ledger(ledger_path=ledger, opening_balance=12000.0)
|
|
text = build_runway_report(ledger_path=ledger)
|
|
assert "Runway Dogfood Evidence" in text
|
|
assert "2026-06" in text
|
|
|
|
output = write_runway_evidence(
|
|
ledger_path=ledger,
|
|
output_dir=tmp_path / "evidence",
|
|
opening_balance=12000.0,
|
|
)
|
|
assert output.exists()
|
|
assert "runway-dogfood-" in output.name
|