129 lines
4.4 KiB
Python
129 lines
4.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
|
|
|
|
|
|
def _booked_fact(tmp_path: Path, *, amount: str = "100.01", currency: str = "EUR"):
|
|
ledger = tmp_path / "ledger.db"
|
|
source = tmp_path / "shared.csv"
|
|
source.write_text(
|
|
"product,amount,currency,invoice_date,environment\n"
|
|
f"Shared cluster,{amount},{currency},2026-07-01,production\n",
|
|
encoding="utf-8",
|
|
)
|
|
import_csv(source, "hosteurope", 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]
|
|
return ledger, fact_id
|
|
|
|
|
|
def _allocation(fact_id: str, **overrides) -> dict:
|
|
payload = {
|
|
"schema_version": "0.1",
|
|
"record_type": "allocation",
|
|
"record_id": "allocation:shared-cluster:2026-07:v1",
|
|
"revision_of": None,
|
|
"resource_id": "resource:shared_cluster",
|
|
"service_id": "shared-cluster",
|
|
"workload_id": None,
|
|
"tenant_id": None,
|
|
"environment": "production",
|
|
"cost_attribution_key": "platform:shared-cluster",
|
|
"period_start": "2026-07-01",
|
|
"period_end": "2026-07-31",
|
|
"currency": "EUR",
|
|
"source_evidence": ["resource-control:allocation:namespace-cpu-v1"],
|
|
"created_at": "2026-08-11T09:00:00Z",
|
|
"financial_fact_ids": [fact_id],
|
|
"method": "namespace-cpu-v1",
|
|
"allocated_amount": "100.01",
|
|
"shares": [
|
|
{"target_key": "client:acme|app:portal|instance:prod-01", "share": "0.3333"},
|
|
{"target_key": "client:beta|app:portal|instance:prod-01", "share": "0.3333"},
|
|
],
|
|
"residual_share": "0.3334",
|
|
}
|
|
payload.update(overrides)
|
|
return payload
|
|
|
|
|
|
def test_shared_allocation_reconciles_targets_rounding_and_residual(tmp_path: Path):
|
|
ledger, fact_id = _booked_fact(tmp_path)
|
|
ingest_planning_evidence(_allocation(fact_id), ledger_path=ledger)
|
|
|
|
report = shared_cost_allocations(ledger_path=ledger)[0]
|
|
|
|
assert report.booked_amount == Decimal("100.01")
|
|
assert sum((target.amount for target in report.targets), report.residual_amount) == Decimal(
|
|
"100.01"
|
|
)
|
|
assert report.residual_amount > Decimal("0")
|
|
assert report.method == "namespace-cpu-v1"
|
|
assert report.source_evidence
|
|
|
|
|
|
def test_revised_allocation_supersedes_prior_shares(tmp_path: Path):
|
|
ledger, fact_id = _booked_fact(tmp_path)
|
|
original = _allocation(fact_id)
|
|
ingest_planning_evidence(original, ledger_path=ledger)
|
|
revised = _allocation(
|
|
fact_id,
|
|
record_id="allocation:shared-cluster:2026-07:v2",
|
|
revision_of=original["record_id"],
|
|
shares=[
|
|
{"target_key": "client:acme|app:portal|instance:prod-01", "share": "0.75"}
|
|
],
|
|
residual_share="0.25",
|
|
)
|
|
ingest_planning_evidence(revised, ledger_path=ledger)
|
|
|
|
reports = shared_cost_allocations(ledger_path=ledger)
|
|
|
|
assert len(reports) == 1
|
|
assert reports[0].allocation_id == revised["record_id"]
|
|
assert reports[0].targets[0].share == Decimal("0.75")
|
|
|
|
|
|
def test_allocation_rejects_missing_fact(tmp_path: Path):
|
|
ledger = tmp_path / "ledger.db"
|
|
ingest_planning_evidence(
|
|
_allocation("fact:missing"),
|
|
ledger_path=ledger,
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="missing/current fact"):
|
|
shared_cost_allocations(ledger_path=ledger)
|
|
|
|
|
|
def test_allocation_rejects_currency_or_amount_mismatch(tmp_path: Path):
|
|
ledger, fact_id = _booked_fact(tmp_path)
|
|
ingest_planning_evidence(
|
|
_allocation(fact_id, currency="USD"),
|
|
ledger_path=ledger,
|
|
)
|
|
with pytest.raises(ValueError, match="currency"):
|
|
shared_cost_allocations(ledger_path=ledger)
|
|
|
|
|
|
def test_fact_cannot_be_allocated_twice(tmp_path: Path):
|
|
ledger, fact_id = _booked_fact(tmp_path)
|
|
first = _allocation(fact_id)
|
|
second = _allocation(
|
|
fact_id,
|
|
record_id="allocation:duplicate",
|
|
shares=[{"target_key": "client:acme", "share": "1"}],
|
|
residual_share="0",
|
|
)
|
|
ingest_planning_evidence(first, ledger_path=ledger)
|
|
ingest_planning_evidence(second, ledger_path=ledger)
|
|
|
|
with pytest.raises(ValueError, match="allocated by both"):
|
|
shared_cost_allocations(ledger_path=ledger)
|