tegwick
efd1b31001
Split a booked AI-plan fact across workplans or repos using measured tokens, measured+estimated tokens, session counts, or an even split. Unmeasured remainder stays a residual. The record is not resource-control AllocationEvidence and does not create booked spend.
98 lines
4 KiB
Python
98 lines
4 KiB
Python
import sqlite3
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from fin_hub.services.allocation import shared_cost_allocations
|
|
from fin_hub.services.exchange import ingest_planning_evidence
|
|
from fin_hub.services.ledger import import_csv, monthly_summary
|
|
from fin_hub.services.reporting_allocation import (
|
|
allocate_ai_plan_report,
|
|
list_ai_plan_allocations,
|
|
)
|
|
from fin_hub.services.session_tokens import ingest_session_token_file
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
def _claude_fact(tmp_path: Path) -> tuple[Path, str]:
|
|
ledger = tmp_path / "ledger.db"
|
|
import_csv(FIXTURES / "ai-plans.csv", "ai-plan", ledger_path=ledger)
|
|
with sqlite3.connect(ledger) as conn:
|
|
fact_id = conn.execute(
|
|
"SELECT financial_fact_id FROM ledger_entries "
|
|
"WHERE source_type = 'ai-plan' AND label = 'claude-max' AND is_current = 1"
|
|
).fetchone()[0]
|
|
return ledger, fact_id
|
|
|
|
|
|
def test_measured_token_share_two_workplans_keeps_unmeasured_residual(tmp_path: Path):
|
|
ledger, fact_id = _claude_fact(tmp_path)
|
|
ingest_session_token_file(
|
|
FIXTURES / "session-tokens-two-workplans.json", ledger_path=ledger
|
|
)
|
|
report = allocate_ai_plan_report(
|
|
fact_id, method="measured_token_share", ledger_path=ledger
|
|
)
|
|
by_key = {target.target_key: target for target in report.targets}
|
|
assert report.booked_amount == Decimal("200.00")
|
|
assert report.method == "measured_token_share"
|
|
assert report.residual_reason == "unmeasured"
|
|
assert report.residual_share == Decimal("0.25")
|
|
assert report.residual_amount == Decimal("50.00")
|
|
assert by_key["workplan:workplan-a"].amount == Decimal("100.00")
|
|
assert by_key["workplan:workplan-b"].amount == Decimal("50.00")
|
|
assert sum((target.amount for target in report.targets), report.residual_amount) == Decimal(
|
|
"200.00"
|
|
)
|
|
assert monthly_summary(ledger_path=ledger)[0].total == pytest.approx(240.0)
|
|
assert shared_cost_allocations(ledger_path=ledger) == []
|
|
assert list_ai_plan_allocations(ledger_path=ledger)[0].record_id == report.record_id
|
|
|
|
|
|
def test_even_split_does_not_use_token_weights(tmp_path: Path):
|
|
ledger, fact_id = _claude_fact(tmp_path)
|
|
ingest_session_token_file(
|
|
FIXTURES / "session-tokens-two-workplans.json", ledger_path=ledger
|
|
)
|
|
report = allocate_ai_plan_report(fact_id, method="even_split", ledger_path=ledger)
|
|
amounts = sorted(target.amount for target in report.targets)
|
|
assert amounts == [Decimal("100.00"), Decimal("100.00")]
|
|
assert report.residual_share == Decimal("0.00")
|
|
assert report.residual_reason == "none"
|
|
|
|
|
|
def test_reporting_allocation_is_not_planning_allocation(tmp_path: Path):
|
|
ledger, fact_id = _claude_fact(tmp_path)
|
|
report = allocate_ai_plan_report(fact_id, method="even_split", ledger_path=ledger)
|
|
with pytest.raises(ValueError, match="not resource-control"):
|
|
ingest_planning_evidence(
|
|
{
|
|
"record_type": "reporting_allocation",
|
|
"record_id": report.record_id,
|
|
"financial_fact_id": fact_id,
|
|
"method": "even_split",
|
|
},
|
|
ledger_path=ledger,
|
|
)
|
|
with sqlite3.connect(ledger) as conn:
|
|
planning = conn.execute(
|
|
"SELECT count(*) FROM sqlite_master WHERE name = 'planning_evidence'"
|
|
).fetchone()[0]
|
|
if planning:
|
|
stored = conn.execute(
|
|
"SELECT record_type FROM planning_evidence"
|
|
).fetchall()
|
|
assert stored == []
|
|
|
|
|
|
def test_non_ai_plan_fact_is_rejected(tmp_path: Path):
|
|
ledger = tmp_path / "ledger.db"
|
|
import_csv(FIXTURES / "cloud-costs.csv", "cloud", ledger_path=ledger)
|
|
with sqlite3.connect(ledger) as conn:
|
|
fact_id = conn.execute(
|
|
"SELECT financial_fact_id FROM ledger_entries WHERE is_current = 1"
|
|
).fetchone()[0]
|
|
with pytest.raises(ValueError, match="ai-plan"):
|
|
allocate_ai_plan_report(fact_id, method="even_split", ledger_path=ledger)
|