Files
railiance-fabric/tests/test_registry.py

649 lines
24 KiB
Python

from __future__ import annotations
import json
import threading
import urllib.request
from http.server import ThreadingHTTPServer
from pathlib import Path
from railiance_fabric.cli import main as cli_main
from railiance_fabric.graph import build_graph
from railiance_fabric.financial import materialize_financial_graph_export
from railiance_fabric.financial_baseline import financial_export_from_legacy
from railiance_fabric.registry import (
RESET_CONFIRMATION_TOKEN,
RegistryError,
RegistryStore,
backstage_projection,
blast_radius,
consumers,
library_xregistry_projection,
providers,
unresolved_dependencies,
validate_graph_export,
xregistry_projection,
)
from railiance_fabric.schema_validation import draft202012_validator
from railiance_fabric.server import RegistryHandler
def test_registry_accepts_snapshot_and_queries_graph(tmp_path: Path) -> None:
store = RegistryStore(tmp_path / "registry.sqlite3")
store.init_schema()
store.upsert_repository(
{
"slug": "railiance-fabric",
"name": "Railiance Fabric",
"remote_url": "https://example.invalid/railiance-fabric.git",
}
)
graph = build_graph([Path(".")])
snapshot = store.add_snapshot(
"railiance-fabric",
{
"commit": "test-commit",
"generated_at": "2026-05-17T00:00:00Z",
"graph": graph.to_export(),
},
)
changed_export = graph.to_export()
changed_export["nodes"] = [
node
for node in changed_export["nodes"]
if node["id"] != "repo-scoping.scope-generator"
]
changed_snapshot = store.add_snapshot(
"railiance-fabric",
{
"commit": "test-commit-2",
"generated_at": "2026-05-17T00:01:00Z",
"graph": changed_export,
},
)
combined = store.combined_graph()
artifact = store.add_artifact(
{
"repo_slug": "railiance-fabric",
"target_id": "flex-auth.api.http-api",
"target_kind": "InterfaceDeclaration",
"artifact_type": "openapi",
"name": "flex-auth OpenAPI",
"uri": "docs/contracts/flex-auth.openapi.yaml",
"media_type": "application/vnd.oai.openapi+yaml",
"version": "0.1.0",
"metadata": {"source": "test"},
}
)
libraries = store.ingest_cyclonedx("railiance-fabric", _cyclonedx_bom())
assert snapshot["repo_slug"] == "railiance-fabric"
assert changed_snapshot["commit"] == "test-commit-2"
assert artifact["artifact_type"] == "openapi"
assert libraries["component_count"] == 2
assert store.list_libraries(name="jsonschema")[0]["purl"] == "pkg:pypi/jsonschema@4.18.0"
inventory = store.repository_inventory("railiance-fabric")
assert inventory["counts"]["snapshots"] == 2
assert inventory["counts"]["libraries"] == 2
diff = store.snapshot_diff("railiance-fabric")
assert diff["from"]["commit"] == "test-commit"
assert diff["to"]["commit"] == "test-commit-2"
assert diff["graph"]["removed_nodes"][0]["id"] == "repo-scoping.scope-generator"
assert store.search("jsonschema")["libraries"][0]["name"] == "jsonschema"
assert store.graph_node_detail("flex-auth.api.http-api")["artifacts"][0]["name"] == "flex-auth OpenAPI"
assert providers(combined, "runtime-secrets")[0]["provider_id"] == "railiance-platform.openbao.runtime-secrets"
assert {match["status"] for match in consumers(combined, "railiance-platform.openbao.kv-v2")} >= {"exact"}
assert unresolved_dependencies(combined) == []
assert blast_radius(combined, "openbao-kv-v2-mount")
assert any(item["kind"] == "Component" for item in backstage_projection(combined)["items"])
assert "services" in xregistry_projection(combined)["groups"]
assert "libraries" in library_xregistry_projection(store.list_libraries())["groups"]
def test_registry_http_service_serves_queries(tmp_path: Path) -> None:
store = RegistryStore(tmp_path / "registry.sqlite3")
store.init_schema()
store.upsert_repository({"slug": "railiance-fabric", "name": "Railiance Fabric"})
store.add_snapshot(
"railiance-fabric",
{
"commit": "test-commit",
"generated_at": "2026-05-17T00:00:00Z",
"graph": build_graph([Path(".")]).to_export(),
},
)
class Handler(RegistryHandler):
pass
Handler.store = store
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
base_url = f"http://127.0.0.1:{server.server_port}"
assert cli_main(
[
"registry",
"sync",
"--registry-url",
base_url,
"--repo-slug",
"railiance-fabric",
"--commit",
"test-cli",
".",
]
) == 0
assert store.latest_snapshot("railiance-fabric")["commit"] == "test-cli"
second_export = build_graph([Path(".")]).to_export()
second_export["nodes"] = second_export["nodes"][:-1]
_post_json(
f"{base_url}/repositories/railiance-fabric/snapshots",
{
"commit": "test-cli-2",
"generated_at": "2026-05-17T00:02:00Z",
"graph": second_export,
},
)
with urllib.request.urlopen(f"{base_url}/health", timeout=5) as response:
assert json.loads(response.read())["status"] == "ok"
with urllib.request.urlopen(f"{base_url}/status", timeout=5) as response:
status_payload = json.loads(response.read())
with urllib.request.urlopen(
f"{base_url}/repositories/railiance-fabric/snapshots",
timeout=5,
) as response:
snapshots_payload = json.loads(response.read())
with urllib.request.urlopen(
f"{base_url}/repositories/railiance-fabric/snapshots/diff",
timeout=5,
) as response:
drift_payload = json.loads(response.read())
with urllib.request.urlopen(
f"{base_url}/graph/providers?capability_type=runtime-secrets",
timeout=5,
) as response:
providers_payload = json.loads(response.read())
artifact_payload = _post_json(
f"{base_url}/artifacts",
{
"repo_slug": "railiance-fabric",
"target_id": "railiance-platform.openbao.kv-v2",
"target_kind": "InterfaceDeclaration",
"artifact_type": "openapi",
"name": "OpenBao KV API",
"uri": "https://example.invalid/openbao.yaml",
},
)
sbom_path = tmp_path / "bom.json"
sbom_path.write_text(json.dumps(_cyclonedx_bom()), encoding="utf-8")
assert cli_main(
[
"registry",
"ingest-cyclonedx",
str(sbom_path),
"--registry-url",
base_url,
"--repo-slug",
"railiance-fabric",
]
) == 0
library_payload = _post_json(
f"{base_url}/repositories/railiance-fabric/libraries/cyclonedx",
_cyclonedx_bom(),
)
with urllib.request.urlopen(
f"{base_url}/artifacts?target_id=railiance-platform.openbao.kv-v2",
timeout=5,
) as response:
artifacts_payload = json.loads(response.read())
with urllib.request.urlopen(
f"{base_url}/repositories/railiance-fabric/libraries",
timeout=5,
) as response:
libraries_payload = json.loads(response.read())
with urllib.request.urlopen(
f"{base_url}/repositories/railiance-fabric/inventory",
timeout=5,
) as response:
inventory_payload = json.loads(response.read())
with urllib.request.urlopen(f"{base_url}/search?q=jsonschema", timeout=5) as response:
search_payload = json.loads(response.read())
with urllib.request.urlopen(f"{base_url}/exports/backstage", timeout=5) as response:
backstage_payload = json.loads(response.read())
with urllib.request.urlopen(f"{base_url}/exports/xregistry", timeout=5) as response:
xregistry_payload = json.loads(response.read())
with urllib.request.urlopen(f"{base_url}/exports/libraries/xregistry", timeout=5) as response:
library_projection_payload = json.loads(response.read())
assert providers_payload[0]["provider_id"] == "railiance-platform.openbao.runtime-secrets"
assert snapshots_payload[0]["commit"] == "test-cli-2"
assert status_payload["counts"]["repositories"] == 1
assert status_payload["counts"]["snapshots"] == 3
assert drift_payload["graph"]["removed_nodes"]
assert artifact_payload["name"] == "OpenBao KV API"
assert artifacts_payload[0]["artifact_type"] == "openapi"
assert library_payload["component_count"] == 2
assert libraries_payload[0]["name"] == "jsonschema"
assert inventory_payload["counts"]["snapshots"] == 3
assert search_payload["libraries"][0]["name"] == "jsonschema"
assert backstage_payload["kind"] == "BackstageCatalogProjection"
assert "interfaces" in xregistry_payload["groups"]
assert "libraries" in library_projection_payload["groups"]
finally:
server.shutdown()
server.server_close()
thread.join(timeout=5)
def test_graph_export_validation_rejects_unflagged_display_edges() -> None:
graph = {
"apiVersion": "railiance.fabric/v1alpha1",
"kind": "FabricGraphExport",
"nodes": [],
"edges": [
{
"from": "repo.fixture",
"to": "fixture.service",
"type": "declares",
"canonical_type": "part_of",
"canon_anchor": "model/devsecops",
"mapping_fit": "partial",
"display_only": False,
"evidence_state": "declared",
}
],
}
try:
validate_graph_export(graph)
except RegistryError as exc:
assert "display-only edge type" in exc.message
else:
raise AssertionError("expected RegistryError for unflagged display-only edge")
def test_state_hub_export_schema_accepts_legacy_and_financial_shapes() -> None:
validator = draft202012_validator(Path("schemas/state-hub-export.schema.yaml"))
legacy_graph = {
"apiVersion": "railiance.fabric/v1alpha1",
"kind": "FabricGraphExport",
"nodes": [],
"edges": [],
}
financial_graph = materialize_financial_graph_export(_financial_graph())
assert list(validator.iter_errors(legacy_graph)) == []
assert list(validator.iter_errors(financial_graph)) == []
def test_financial_graph_export_requires_resolvable_owner() -> None:
graph = _financial_graph()
del graph["nodes"][0]["ownership"]
try:
validate_graph_export(graph)
except RegistryError as exc:
assert "ownership must be an object" in exc.message
else:
raise AssertionError("expected RegistryError for accepted node without ownership")
def test_registry_accepts_financial_graph_and_materializes_vnext_fields(tmp_path: Path) -> None:
store = RegistryStore(tmp_path / "registry.sqlite3")
store.init_schema()
store.upsert_repository({"slug": "state-hub", "name": "State Hub"})
snapshot = store.add_snapshot(
"state-hub",
{
"commit": "financial-vnext",
"generated_at": "2026-05-24T00:00:00Z",
"graph": _financial_graph(),
},
)
graph = snapshot["graph"]
edge = graph["edges"][0]
assert graph["apiVersion"] == "railiance.fabric/v1alpha2"
assert graph["schema_version"] == "financial-fabric-v1"
assert graph["nodes"][0]["evidence"]["review_state"] == "accepted"
assert edge["relationship_category"] == "utility"
assert edge["boundary"]["crosses_fabric_boundary"] is False
assert edge["boundary"]["crosses_subfabric_boundary"] is True
combined = store.combined_graph()
assert combined["apiVersion"] == "railiance.fabric/v1alpha2"
assert combined["actors"][0]["id"] == "actor.coulomb.tenant"
assert combined["fabrics"][1]["id"] == "subfabric.railiance.tenant.coulomb"
assert combined["edges"][0]["relationship_category"] == "utility"
def test_current_graph_projects_to_financial_baseline() -> None:
legacy_graph = build_graph([Path(".")]).to_export()
financial_graph = financial_export_from_legacy(legacy_graph)
validate_graph_export(financial_graph)
validator = draft202012_validator(Path("schemas/state-hub-export.schema.yaml"))
assert list(validator.iter_errors(financial_graph)) == []
assert financial_graph["netkingdom"]["id"] == "railiance.netkingdom"
assert financial_graph["fabrics"][0]["id"] == "fabric.railiance.primary"
assert financial_graph["generated_at"].endswith("Z")
assert financial_graph["unresolved"] == []
assert all(node["containment"]["fabric_id"] == "fabric.railiance.primary" for node in financial_graph["nodes"])
assert all(node["ownership"]["owner_actor_id"] == "actor.railiance.primary-lord" for node in financial_graph["nodes"])
def test_export_cli_can_print_financial_baseline(capsys) -> None:
assert cli_main(["export", "--format", "financial", "."]) == 0
payload = json.loads(capsys.readouterr().out)
validate_graph_export(payload)
assert payload["apiVersion"] == "railiance.fabric/v1alpha2"
assert payload["compatibility"]["projected_from_apiVersion"] == "railiance.fabric/v1alpha1"
def test_registry_reset_archive_and_guarded_reset(tmp_path: Path) -> None:
store = RegistryStore(tmp_path / "registry.sqlite3")
store.init_schema()
store.upsert_repository({"slug": "fixture-repo", "name": "Fixture Repo"})
store.add_snapshot(
"fixture-repo",
{
"commit": "abc123",
"generated_at": "2026-05-23T00:00:00Z",
"graph": build_graph([Path(".")]).to_export(),
},
)
archive = store.reset_archive()
assert archive["kind"] == "RegistryResetArchive"
assert archive["counts"]["repositories"] == 1
assert archive["counts"]["snapshots"] == 1
assert archive["snapshots"][0]["commit"] == "abc123"
try:
store.reset_graph_data(
{
"confirm": "nope",
"reason": "test reset",
"archive_sha256": "abc123",
}
)
except RegistryError as exc:
assert RESET_CONFIRMATION_TOKEN in exc.message
else:
raise AssertionError("expected RegistryError for missing reset confirmation")
event = store.reset_graph_data(
{
"confirm": RESET_CONFIRMATION_TOKEN,
"reason": "test reset",
"archive_path": str(tmp_path / "archive.json"),
"archive_sha256": "abc123",
}
)
assert event["dropped_counts"]["snapshots"] == 1
assert event["repositories_preserved"] == 1
assert store.status()["counts"]["repositories"] == 1
assert store.status()["counts"]["snapshots"] == 0
assert store.status()["counts"]["registry_reset_events"] == 1
def test_registry_cli_exports_archive_before_reset(tmp_path: Path, capsys) -> None:
store = RegistryStore(tmp_path / "registry.sqlite3")
store.init_schema()
store.upsert_repository({"slug": "fixture-repo", "name": "Fixture Repo"})
store.add_snapshot(
"fixture-repo",
{
"commit": "abc123",
"generated_at": "2026-05-23T00:00:00Z",
"graph": build_graph([Path(".")]).to_export(),
},
)
class Handler(RegistryHandler):
pass
Handler.store = store
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
archive_path = tmp_path / "reset-archive.json"
assert cli_main(
[
"registry",
"reset-graph-data",
"--registry-url",
f"http://127.0.0.1:{server.server_port}",
"--archive",
str(archive_path),
"--confirm",
RESET_CONFIRMATION_TOKEN,
"--reason",
"test reset",
]
) == 0
output = capsys.readouterr().out
archive = json.loads(archive_path.read_text(encoding="utf-8"))
assert "reset event" in output
assert archive["counts"]["snapshots"] == 1
assert store.status()["counts"]["snapshots"] == 0
assert store.status()["counts"]["repositories"] == 1
finally:
server.shutdown()
server.server_close()
thread.join(timeout=5)
def test_registry_sync_manifest_registers_multiple_repos(tmp_path: Path) -> None:
store = RegistryStore(tmp_path / "registry.sqlite3")
store.init_schema()
class Handler(RegistryHandler):
pass
Handler.store = store
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
base_url = f"http://127.0.0.1:{server.server_port}"
sbom_path = tmp_path / "bom.json"
sbom_path.write_text(json.dumps(_cyclonedx_bom()), encoding="utf-8")
manifest_path = tmp_path / "manifest.yaml"
manifest_path.write_text(
"\n".join(
[
"apiVersion: railiance.fabric/v1alpha1",
"kind: RegistryOnboardingManifest",
"repositories:",
" - slug: railiance-fabric",
" name: Railiance Fabric",
f" path: {Path('.').resolve()}",
" commit: manifest-commit",
f" sbom: {sbom_path.name}",
" - slug: missing-repo",
" name: Missing Repo",
f" path: {tmp_path / 'missing-repo'}",
"",
]
),
encoding="utf-8",
)
assert cli_main(
[
"registry",
"sync-manifest",
str(manifest_path),
"--registry-url",
base_url,
]
) == 0
repositories = {repo["slug"]: repo for repo in store.list_repositories()}
assert set(repositories) == {"missing-repo", "railiance-fabric"}
assert store.latest_snapshot("railiance-fabric")["commit"] == "manifest-commit"
assert store.list_snapshots("missing-repo") == []
assert store.list_libraries(repo_slug="railiance-fabric")[0]["name"] == "jsonschema"
finally:
server.shutdown()
server.server_close()
thread.join(timeout=5)
def _post_json(url: str, payload: dict) -> dict:
request = urllib.request.Request(
url,
data=json.dumps(payload).encode("utf-8"),
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(request, timeout=5) as response:
return json.loads(response.read())
def _financial_graph() -> dict:
return {
"apiVersion": "railiance.fabric/v1alpha2",
"kind": "FabricGraphExport",
"schema_version": "financial-fabric-v1",
"generated_at": "2026-05-24T00:00:00Z",
"netkingdom": {
"id": "railiance.netkingdom",
"name": "Railiance Netkingdom",
"king_actor_id": "actor.railiance.king",
},
"actors": [
{"id": "actor.railiance.king", "kind": "FabricActor", "role": "king", "name": "Railiance King"},
{"id": "actor.railiance.primary-lord", "kind": "FabricActor", "role": "lord", "name": "Railiance Lord"},
{"id": "actor.coulomb.tenant", "kind": "FabricActor", "role": "tenant", "name": "Coulomb Tenant"},
],
"fabrics": [
{
"id": "fabric.railiance.primary",
"kind": "Fabric",
"name": "Railiance Primary Fabric",
"netkingdom_id": "railiance.netkingdom",
"lord_actor_id": "actor.railiance.primary-lord",
"status": "active",
},
{
"id": "subfabric.railiance.tenant.coulomb",
"kind": "Subfabric",
"name": "Coulomb Tenant Subfabric",
"netkingdom_id": "railiance.netkingdom",
"parent_fabric_id": "fabric.railiance.primary",
"tenant_actor_id": "actor.coulomb.tenant",
"status": "planned",
},
],
"nodes": [
{
"id": "state-hub.http",
"kind": "UtilityInterface",
"name": "State Hub HTTP API",
"repo": "state-hub",
"domain": "custodian",
"lifecycle": "active",
"containment": {
"netkingdom_id": "railiance.netkingdom",
"fabric_id": "fabric.railiance.primary",
"subfabric_id": None,
"environment": "local",
},
"ownership": {
"owner_actor_id": "actor.railiance.primary-lord",
"owner_role": "lord",
"resolution": "inherited",
"inherited_from": "fabric.railiance.primary",
},
},
{
"id": "coulomb.automation-client",
"kind": "Service",
"name": "Coulomb Automation Client",
"repo": "coulomb-automation",
"domain": "railiance",
"lifecycle": "planned",
"containment": {
"netkingdom_id": "railiance.netkingdom",
"fabric_id": "fabric.railiance.primary",
"subfabric_id": "subfabric.railiance.tenant.coulomb",
"environment": "local",
},
"ownership": {
"owner_actor_id": "actor.coulomb.tenant",
"owner_role": "tenant",
"resolution": "explicit",
},
"accounting": {
"cost_center_id": "cc.coulomb.automation",
"allocation_model": "direct",
},
},
],
"edges": [
{
"id": "utility:state-hub-http:coulomb-client",
"from": "state-hub.http",
"to": "coulomb.automation-client",
"type": "provides_utility_to",
"provider": {
"owner_actor_id": "actor.railiance.primary-lord",
"fabric_id": "fabric.railiance.primary",
"subfabric_id": None,
},
"consumer": {
"owner_actor_id": "actor.coulomb.tenant",
"fabric_id": "fabric.railiance.primary",
"subfabric_id": "subfabric.railiance.tenant.coulomb",
},
"utility": {
"utility_type": "coordination_api",
"contract_id": "state-hub.http",
"payment_schema_id": "payment.internal-tenant-access",
"metering_basis": "unknown",
"business_model": "tenant_utility",
},
"accounting": {
"provider_profit_center_id": "pc.tenant-utilities",
"consumer_cost_center_id": "cc.coulomb.automation",
"allocation_model": "usage_weighted",
},
}
],
}
def _cyclonedx_bom() -> dict:
return {
"bomFormat": "CycloneDX",
"specVersion": "1.6",
"version": 1,
"components": [
{
"bom-ref": "pkg:pypi/jsonschema@4.18.0",
"type": "library",
"name": "jsonschema",
"version": "4.18.0",
"purl": "pkg:pypi/jsonschema@4.18.0",
"licenses": [{"license": {"id": "MIT"}}],
}
],
"services": [
{
"bom-ref": "urn:service:state-hub",
"name": "state-hub-api",
"version": "0.1.0",
"endpoints": ["http://127.0.0.1:8000"],
}
],
}