resource-control/tests/test_financial_exchange.py
tegwick 2c2a6073ff feat(portfolio): complete RESOURCE-WP-0003 T06 optimization cases and T07 reporting
T06: optimization-case schema, fail-closed evaluator, cadence and decision
template. Every option including the baseline must present all ten decision
fields; one unknown blocks the comparison. Validated on the storage case
(Hetzner computes and loses to Scaleway by EUR 29.14/month on operator labour;
Host Europe blocks on four named gaps) and on the non-storage reef-railiance
k3s rightsizing case (low utilization is real, but nothing is costable while
the railiance01 price is unknown).

T07: portfolio report over coverage, lifecycle, utilization, cost, renewals,
risks, open cases, and next actions, derived only from committed evidence.
Portfolio spend is reported null rather than as a partial sum, unattributed
cost is a named list rather than a spread, and unmeasurable resources are
reported rather than dropped.

RESOURCE-WP-0003 is finished; both cases remain blocked_on_evidence against
live delegated records in other repositories. RESOURCE-WP-0002 is untouched.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-14 09:28:44 +02:00

60 lines
3.3 KiB
Python

import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parents[1] / "tools"))
from financial_exchange import forecast_records, reconcile, validate_booked_cost
class FinancialExchangeTest(unittest.TestCase):
def setUp(self):
self.forecast = {
"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}]
}
def test_forecast_export_separates_costs_and_is_stable(self):
first = forecast_records(self.forecast, resource_id="resource:platform_audit_storage", source_ref="forecast.json")
second = forecast_records(self.forecast, resource_id="resource:platform_audit_storage", source_ref="forecast.json")
self.assertEqual(first, second)
self.assertEqual("2.89", first[0]["costs"]["infrastructure"])
self.assertEqual("60.00", first[0]["costs"]["internal_labor"])
self.assertEqual("usage_observation" not in first[0]["record_type"], True)
def test_reconciles_only_attributable_booked_cost(self):
forecast = forecast_records(self.forecast, resource_id="resource:platform_audit_storage", source_ref="forecast.json")
booked = [{
"schema_version": "0.1", "record_type": "booked_cost", "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:1",
"content_fingerprint": "sha256:x", "provider": "scaleway",
"accounting_period": "2026-09", "currency": "EUR", "gross_amount": "3.10",
"adjustment_amount": "0.00", "effective_amount": "3.10", "tax_status": "unknown",
"tax_amount": None, "cost_attribution_key": "platform:audit-storage",
"source_evidence_ref": "invoice:1", "recorded_at": "2026-10-01T00:00:00Z"
}]
rows = reconcile(forecast, booked)
self.assertEqual("reconciled", rows[0]["status"])
self.assertEqual("0.21", rows[0]["variance"])
self.assertEqual(["fact:1"], rows[0]["financial_fact_ids"])
def test_rejects_invalid_booked_arithmetic(self):
with self.assertRaisesRegex(ValueError, "effective_amount"):
validate_booked_cost({
"schema_version": "0.1", "record_type": "booked_cost", "financial_fact_id": "fact:1",
"adjustment_kind": "charge", "source_document_id": "doc", "source_line_id": "line",
"content_fingerprint": "hash", "provider": "provider", "accounting_period": "2026-09",
"currency": "EUR", "gross_amount": "1.00", "adjustment_amount": "0.00",
"effective_amount": "2.00", "source_evidence_ref": "doc", "recorded_at": "2026-10-01T00:00:00Z"
})
if __name__ == "__main__":
unittest.main()