tegwick
d343888a41
Store mixed measured/estimated/superseded aggregates as session-token evidence. Map kinds onto the evidence-basis vocabulary, keep unknown months unknown, and refuse to post the snapshot as booked spend or resource-control usage.
121 lines
4.4 KiB
Python
121 lines
4.4 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from fin_hub.services.exchange import ingest_planning_evidence
|
|
from fin_hub.services.ledger import import_csv, monthly_summary
|
|
from fin_hub.services.session_tokens import (
|
|
ingest_session_token_evidence,
|
|
ingest_session_token_file,
|
|
list_current_session_token_evidence,
|
|
summarize_session_token_evidence,
|
|
)
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
def test_ingest_mixed_aggregates_and_summarize(tmp_path: Path):
|
|
ledger = tmp_path / "ledger.db"
|
|
record = ingest_session_token_file(
|
|
FIXTURES / "session-tokens-2026-08.json", ledger_path=ledger
|
|
)
|
|
again = ingest_session_token_file(
|
|
FIXTURES / "session-tokens-2026-08.json", ledger_path=ledger
|
|
)
|
|
assert again.record_id == record.record_id
|
|
summary = summarize_session_token_evidence(record)
|
|
assert summary.unknown_residual is False
|
|
assert summary.tokens_measured == 150
|
|
assert summary.tokens_derived == 30
|
|
assert summary.tokens_estimated == 1500
|
|
assert "excluded" not in summary.by_basis
|
|
assert set(summary.by_basis) == {"measured", "derived", "estimated"}
|
|
assert summary.by_basis["derived"].method == "state-hub-task-share"
|
|
stored = list_current_session_token_evidence(ledger_path=ledger)
|
|
assert [item.record_id for item in stored] == [record.record_id]
|
|
assert monthly_summary(ledger_path=ledger) == []
|
|
|
|
|
|
def test_empty_month_is_unknown_residual_not_zero(tmp_path: Path):
|
|
ledger = tmp_path / "ledger.db"
|
|
record = ingest_session_token_evidence(
|
|
{
|
|
"record_id": "session-tokens:state-hub:2026-07",
|
|
"period_start": "2026-07-01",
|
|
"period_end": "2026-07-31",
|
|
"captured_at": "2026-08-15T12:00:00Z",
|
|
"source_evidence": ["state-hub:/token-events/aggregate/?month=2026-07"],
|
|
"totals": {
|
|
"superseded": {
|
|
"tokens_in": 9,
|
|
"tokens_out": 1,
|
|
"event_count": 1,
|
|
"confidence": "0",
|
|
}
|
|
},
|
|
},
|
|
ledger_path=ledger,
|
|
)
|
|
summary = summarize_session_token_evidence(record)
|
|
assert summary.unknown_residual is True
|
|
assert summary.tokens_measured is None
|
|
assert summary.tokens_derived is None
|
|
assert summary.tokens_estimated is None
|
|
assert summary.coverage is None
|
|
|
|
|
|
def test_allocated_events_require_a_method():
|
|
with pytest.raises(ValueError, match="method"):
|
|
ingest_session_token_evidence(
|
|
{
|
|
"record_id": "session-tokens:bad",
|
|
"period_start": "2026-08-01",
|
|
"period_end": "2026-08-31",
|
|
"captured_at": "2026-08-15T12:00:00Z",
|
|
"source_evidence": ["state-hub"],
|
|
"totals": {
|
|
"allocated": {
|
|
"tokens_in": 1,
|
|
"tokens_out": 1,
|
|
"event_count": 1,
|
|
}
|
|
},
|
|
},
|
|
ledger_path=Path("/tmp/unused"),
|
|
)
|
|
|
|
|
|
def test_session_tokens_cannot_be_posted_as_planning_usage(tmp_path: Path):
|
|
ledger = tmp_path / "ledger.db"
|
|
payload = json.loads((FIXTURES / "session-tokens-2026-08.json").read_text())
|
|
with pytest.raises(ValueError, match="not resource-control"):
|
|
ingest_planning_evidence(payload, ledger_path=ledger)
|
|
with pytest.raises(ValueError, match="usage_observation"):
|
|
ingest_session_token_evidence(
|
|
{
|
|
"record_type": "usage_observation",
|
|
"record_id": "usage:1",
|
|
"period_start": "2026-08-01",
|
|
"period_end": "2026-08-31",
|
|
"resource_id": "resource:invented",
|
|
"source_evidence": ["nope"],
|
|
"created_at": "2026-08-15T12:00:00Z",
|
|
"measures": [{"name": "tokens", "value": "1", "unit": "token"}],
|
|
},
|
|
ledger_path=ledger,
|
|
)
|
|
|
|
|
|
def test_session_tokens_cannot_be_imported_as_booked_spend(tmp_path: Path):
|
|
ledger = tmp_path / "ledger.db"
|
|
with pytest.raises(ValueError, match="Unknown source type"):
|
|
import_csv(FIXTURES / "session-tokens-2026-08.json", "session-tokens", ledger_path=ledger)
|
|
with pytest.raises(ValueError, match="booked spend"):
|
|
ingest_session_token_evidence(
|
|
{
|
|
"record_type": "booked_cost",
|
|
"financial_fact_id": "fact:1",
|
|
},
|
|
ledger_path=ledger,
|
|
)
|