fin-hub/tests/test_fabric_cutover.py
tegwick debe67edf0
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s
feat: complete fabric authority cutover contract
Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a053ff-1d6f-7fe2-ac1c-a6eb40a42a0c
2026-08-31 22:18:39 +02:00

69 lines
2.4 KiB
Python

from __future__ import annotations
from copy import deepcopy
from fin_hub.fabric_cutover import evaluate_fabric_cutover, fabric_content_hash
def _authority() -> dict:
return {
"apiVersion": "railiance.fabric/v1alpha2",
"kind": "FabricGraphExport",
"schema_version": "financial-fabric-v1",
"generated_at": "2026-08-31T12:00:00Z",
"source": {
"producer": "railiance-fabric",
"commit": "snapshot-set:sha256:abc",
"path": "registry://accepted-snapshots",
},
"actors": [{"id": "actor.king"}],
"fabrics": [{"id": "fabric.main"}],
"nodes": [{"id": "service.one"}],
"edges": [],
"unresolved": [],
}
def _projection(authority: dict) -> dict:
return {
"schema_version": authority["schema_version"],
"node_count": 1,
"edge_count": 0,
"actor_count": 1,
"fabric_count": 1,
"unresolved_count": 0,
"latest_import": {
"api_version": authority["apiVersion"],
"schema_version": authority["schema_version"],
"export_kind": authority["kind"],
"content_hash": fabric_content_hash(authority),
"source_commit": authority["source"]["commit"],
"source_path": authority["source"]["path"],
"exported_at": authority["generated_at"],
},
}
def test_cutover_gate_accepts_exact_counts_hash_and_provenance() -> None:
authority = _authority()
report = evaluate_fabric_cutover(authority, _projection(authority))
assert report["ready"] is True
assert report["failures"] == []
def test_cutover_gate_rejects_count_or_hash_drift() -> None:
authority = _authority()
projection = _projection(authority)
projection["node_count"] = 2
projection["latest_import"]["content_hash"] = "stale"
report = evaluate_fabric_cutover(authority, projection)
assert report["ready"] is False
assert {"node_count", "content_hash"}.issubset(report["failures"])
def test_cutover_gate_rejects_unversioned_authority_provenance() -> None:
authority = deepcopy(_authority())
authority["source"]["commit"] = "working-tree"
report = evaluate_fabric_cutover(authority, _projection(authority))
assert report["ready"] is False
assert "authority:source.commit must identify an immutable authority revision" in report["failures"]