207 lines
6.7 KiB
Python
207 lines
6.7 KiB
Python
from datetime import date, datetime, timezone
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from fin_hub.models.engagement_price import EngagementPrice, _validate_engagement_price
|
|
from fin_hub.schemas.exchange import (
|
|
AllocationEvidence,
|
|
BookedCostEvidence,
|
|
ForecastEvidence,
|
|
)
|
|
from fin_hub.services.exchange import (
|
|
booked_cost_projection,
|
|
ingest_planning_evidence,
|
|
ingest_resource_forecast,
|
|
)
|
|
from fin_hub.services.ledger import import_csv
|
|
|
|
|
|
def _common_planning() -> dict:
|
|
return {
|
|
"schema_version": "0.1",
|
|
"record_id": "forecast:platform-audit-storage:2026-08",
|
|
"revision_of": None,
|
|
"resource_id": "resource:platform_audit_storage",
|
|
"service_id": "object-storage",
|
|
"workload_id": "platform-pg",
|
|
"tenant_id": None,
|
|
"environment": "production",
|
|
"cost_attribution_key": "platform:audit-storage",
|
|
"period_start": "2026-09-01",
|
|
"period_end": "2027-08-31",
|
|
"currency": "eur",
|
|
"source_evidence": ["resource-control:data/forecasts/platform-audit-storage"],
|
|
"created_at": "2026-08-10T17:10:00Z",
|
|
}
|
|
|
|
|
|
def test_booked_cost_money_relationships_and_decimal_json():
|
|
record = BookedCostEvidence(
|
|
financial_fact_id="fact:1",
|
|
correction_of=None,
|
|
adjustment_kind="charge",
|
|
source_type="provider_invoice",
|
|
source_document_id="invoice:1",
|
|
source_line_id="invoice:1:line:1",
|
|
content_fingerprint="sha256:abc",
|
|
provider="provider",
|
|
accounting_period="2026-07",
|
|
currency="eur",
|
|
net_amount="10.005",
|
|
discount_amount="1.00",
|
|
tax_status="known",
|
|
tax_amount="1.90",
|
|
gross_amount="10.90",
|
|
adjustment_amount="0",
|
|
effective_amount="10.90",
|
|
source_evidence_ref="invoice:1",
|
|
recorded_at=datetime.now(timezone.utc),
|
|
)
|
|
|
|
assert record.net_amount == Decimal("10.00")
|
|
assert record.currency == "EUR"
|
|
assert '"gross_amount":"10.90"' in record.model_dump_json()
|
|
|
|
|
|
def test_booked_cost_rejects_invalid_relationships():
|
|
with pytest.raises(ValidationError, match="gross_amount"):
|
|
BookedCostEvidence(
|
|
financial_fact_id="fact:1",
|
|
adjustment_kind="charge",
|
|
source_type="invoice",
|
|
source_document_id="doc",
|
|
source_line_id="line",
|
|
content_fingerprint="hash",
|
|
provider="provider",
|
|
accounting_period="2026-07",
|
|
currency="EUR",
|
|
net_amount="10",
|
|
discount_amount="0",
|
|
tax_status="known",
|
|
tax_amount="1.90",
|
|
gross_amount="12",
|
|
effective_amount="12",
|
|
source_evidence_ref="doc",
|
|
recorded_at=datetime.now(timezone.utc),
|
|
)
|
|
|
|
|
|
def test_planning_forecast_is_typed_and_idempotent(tmp_path: Path):
|
|
payload = {
|
|
**_common_planning(),
|
|
"record_type": "forecast",
|
|
"scenario": "base",
|
|
"forecast_version": "2026-08-base",
|
|
"costs": {
|
|
"infrastructure": "2.89",
|
|
"internal_labor": "60",
|
|
"external_services": "0",
|
|
"setup": "0",
|
|
"other": "0",
|
|
},
|
|
"uncertainty": "provider price excludes unknown tax",
|
|
"assumptions": ["180 GB stored"],
|
|
}
|
|
ledger = tmp_path / "ledger.db"
|
|
|
|
first = ingest_planning_evidence(payload, ledger_path=ledger)
|
|
second = ingest_planning_evidence(payload, ledger_path=ledger)
|
|
|
|
assert isinstance(first, ForecastEvidence)
|
|
assert first.costs.total == Decimal("62.89")
|
|
assert second.record_id == first.record_id
|
|
|
|
|
|
def test_resource_control_backup_forecast_enters_planning_store(tmp_path: Path):
|
|
payload = {
|
|
"schema_version": "0.1",
|
|
"record_type": "forecast",
|
|
"workload": "platform-pg",
|
|
"cost_attribution_key": "platform:audit-storage",
|
|
"provider_id": "scaleway-standard-multi-az",
|
|
"created_at": "2026-08-10T17:10:00Z",
|
|
"scenario": "base",
|
|
"forecast_ref": None,
|
|
"rows": [
|
|
{
|
|
"period": "2026-09",
|
|
"database_gb": 5,
|
|
"stored_gb": 180,
|
|
"wal_gb": 30,
|
|
"restore_egress_gb": 5,
|
|
"write_requests": 2500,
|
|
"read_requests": 1000,
|
|
"infrastructure_eur": 2.89,
|
|
"internal_labor_hours": 1,
|
|
"internal_labor_eur": 60,
|
|
"total_eur": 62.89,
|
|
}
|
|
],
|
|
}
|
|
records = ingest_resource_forecast(
|
|
payload,
|
|
resource_id="resource:platform_audit_storage",
|
|
source_evidence_ref="resource-control:data/forecasts/platform-audit-storage",
|
|
ledger_path=tmp_path / "ledger.db",
|
|
)
|
|
|
|
assert len(records) == 1
|
|
assert isinstance(records[0], ForecastEvidence)
|
|
assert records[0].costs.total == Decimal("62.89")
|
|
|
|
|
|
def test_allocation_requires_shares_and_residual_to_reconcile():
|
|
payload = {
|
|
**_common_planning(),
|
|
"record_type": "allocation",
|
|
"record_id": "allocation:1",
|
|
"financial_fact_ids": ["fact:1"],
|
|
"method": "namespace-cpu-v1",
|
|
"allocated_amount": "10.00",
|
|
"shares": [{"target_key": "client:acme", "share": "0.75"}],
|
|
"residual_share": "0.20",
|
|
}
|
|
with pytest.raises(ValidationError, match="must equal 1"):
|
|
AllocationEvidence.model_validate(payload)
|
|
|
|
|
|
def test_booked_cost_projection_uses_current_corrected_fact(tmp_path: Path):
|
|
source = tmp_path / "cost.csv"
|
|
ledger = tmp_path / "ledger.db"
|
|
source.write_text(
|
|
"product,amount,currency,invoice_date,environment\n"
|
|
"Server,10.00,EUR,2026-07-01,production\n",
|
|
encoding="utf-8",
|
|
)
|
|
import_csv(source, "hosteurope", ledger_path=ledger)
|
|
source.write_text(
|
|
"product,amount,currency,invoice_date,environment\n"
|
|
"Server,12.00,EUR,2026-07-01,production\n",
|
|
encoding="utf-8",
|
|
)
|
|
import_csv(source, "hosteurope", ledger_path=ledger, force=True)
|
|
|
|
records = booked_cost_projection(ledger_path=ledger)
|
|
|
|
assert len(records) == 1
|
|
assert records[0].effective_amount == Decimal("12.00")
|
|
assert records[0].correction_of is not None
|
|
assert records[0].environment == "production"
|
|
|
|
|
|
def test_sqlalchemy_price_uses_shared_invariants():
|
|
price = EngagementPrice(
|
|
client_id="acme",
|
|
application_id="portal",
|
|
app_instance_id="prod-01",
|
|
cost_attribution_key="wrong",
|
|
effective_from=date(2026, 7, 1),
|
|
amount=Decimal("100.001"),
|
|
currency="eur",
|
|
source="agreement",
|
|
)
|
|
with pytest.raises(ValueError, match="does not match"):
|
|
_validate_engagement_price(None, None, price)
|