tegwick
033ad3f41f
Report booked cost, entitlement, token coverage, and work as three series: work per measured token, work per measured+estimated token, and work per euro. Months below 50% coverage are unfit for token trends. There is no blended efficiency number.
137 lines
4.3 KiB
Python
137 lines
4.3 KiB
Python
from decimal import Decimal
|
|
from pathlib import Path
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from fin_hub.app import create_app
|
|
from fin_hub.routers.effectiveness import create_effectiveness_router
|
|
from fin_hub.services.effectiveness import cheaper_on_series, effectiveness_report
|
|
from fin_hub.services.ledger import import_csv, record_plan_entitlement
|
|
from fin_hub.services.session_tokens import ingest_session_token_evidence
|
|
|
|
|
|
def _write_plan(path: Path, *, period: str, amount: str) -> None:
|
|
path.write_text(
|
|
"provider,plan,charge_kind,amount,currency,period_month,ongoing\n"
|
|
f"anthropic,claude-max,subscription,{amount},EUR,{period},true\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def _token_snapshot(
|
|
*,
|
|
record_id: str,
|
|
period: str,
|
|
measured: int,
|
|
estimated: int,
|
|
work: int,
|
|
) -> dict:
|
|
start, end = f"{period}-01", f"{period}-28"
|
|
totals = {
|
|
"measured": {
|
|
"tokens_in": measured,
|
|
"tokens_out": 0,
|
|
"event_count": 1 if measured else 0,
|
|
"confidence": "1.0",
|
|
}
|
|
}
|
|
if estimated:
|
|
totals["estimated"] = {
|
|
"tokens_in": estimated,
|
|
"tokens_out": 0,
|
|
"event_count": 1,
|
|
"confidence": "0.35",
|
|
}
|
|
return {
|
|
"record_id": record_id,
|
|
"period_start": start,
|
|
"period_end": end,
|
|
"captured_at": "2026-08-15T12:00:00Z",
|
|
"source_evidence": [f"state-hub:/token-events/aggregate/?month={period}"],
|
|
"totals": totals,
|
|
"work": {"unit": "task", "count": work, "source": "state-hub-tasks"},
|
|
}
|
|
|
|
|
|
def test_low_coverage_month_is_unfit_and_has_no_blended_efficiency(tmp_path: Path):
|
|
ledger = tmp_path / "ledger.db"
|
|
high = tmp_path / "july.csv"
|
|
low = tmp_path / "august.csv"
|
|
_write_plan(high, period="2026-07", amount="200.00")
|
|
_write_plan(low, period="2026-08", amount="200.00")
|
|
import_csv(high, "ai-plan", ledger_path=ledger)
|
|
import_csv(low, "ai-plan", ledger_path=ledger)
|
|
record_plan_entitlement(
|
|
provider="anthropic",
|
|
plan="claude-max",
|
|
period_month="2026-07",
|
|
unit="plan",
|
|
plan_label="Max 20x",
|
|
source="vendor-plan",
|
|
ledger_path=ledger,
|
|
)
|
|
ingest_session_token_evidence(
|
|
_token_snapshot(
|
|
record_id="tokens:2026-07",
|
|
period="2026-07",
|
|
measured=200,
|
|
estimated=0,
|
|
work=2,
|
|
),
|
|
ledger_path=ledger,
|
|
)
|
|
ingest_session_token_evidence(
|
|
_token_snapshot(
|
|
record_id="tokens:2026-08",
|
|
period="2026-08",
|
|
measured=20,
|
|
estimated=180,
|
|
work=2,
|
|
),
|
|
ledger_path=ledger,
|
|
)
|
|
|
|
rows = {row.period_month: row for row in effectiveness_report(ledger_path=ledger)}
|
|
good = rows["2026-07"]
|
|
bad = rows["2026-08"]
|
|
assert good.trend_fit == "comparable"
|
|
assert bad.trend_fit == "unfit"
|
|
assert good.coverage == Decimal("1.00")
|
|
assert bad.coverage == Decimal("0.100000")
|
|
assert good.work_per_euro == bad.work_per_euro
|
|
assert good.implied_basis == "inferred"
|
|
payload = bad.as_dict()
|
|
assert "efficiency" not in payload
|
|
assert "blend" not in payload
|
|
assert cheaper_on_series(bad, good, "measured_estimated") is None
|
|
assert cheaper_on_series(bad, good, "measured") is None
|
|
assert cheaper_on_series(bad, good, "euro") is False
|
|
|
|
|
|
def test_effectiveness_http_endpoint(tmp_path: Path):
|
|
ledger = tmp_path / "ledger.db"
|
|
source = tmp_path / "july.csv"
|
|
_write_plan(source, period="2026-07", amount="200.00")
|
|
import_csv(source, "ai-plan", ledger_path=ledger)
|
|
ingest_session_token_evidence(
|
|
_token_snapshot(
|
|
record_id="tokens:2026-07",
|
|
period="2026-07",
|
|
measured=200,
|
|
estimated=0,
|
|
work=2,
|
|
),
|
|
ledger_path=ledger,
|
|
)
|
|
app = create_app()
|
|
app.include_router(
|
|
create_effectiveness_router(ledger_path_resolver=lambda _: ledger),
|
|
prefix="/test",
|
|
)
|
|
client = TestClient(app)
|
|
response = client.get("/test/effectiveness?period=2026-07")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body[0]["trend_fit"] == "comparable"
|
|
assert body[0]["work_per_euro"] == "0.010000"
|
|
assert "efficiency" not in body[0]
|