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.graph_explorer import ( fabric_graph_explorer_manifest, fabric_graph_explorer_payload, ) from railiance_fabric.registry import RegistryStore from railiance_fabric.schema_validation import draft202012_validator from railiance_fabric.server import RegistryHandler def test_graph_explorer_manifest_and_payload_validate() -> None: graph = build_graph([Path(".")]).to_export() manifest = fabric_graph_explorer_manifest() payload = fabric_graph_explorer_payload( graph, [ { "slug": "railiance-fabric", "name": "Railiance Fabric", "state_hub_repo_id": "2c0de614-e468-4eb6-8157-470649ac8c05", }, { "slug": "registered-only", "name": "Registered Only", }, ], {"railiance-fabric"}, ) _validate_schema("graph-explorer-manifest.schema.yaml", manifest) _validate_schema("graph-explorer-payload.schema.yaml", payload) assert manifest["profile_persistence"] == "local" assert manifest["shareable_state"]["profile_id"] is True assert set(manifest["filter"]["actions"]) >= {"show", "hide", "blur", "highlight", "remove"} assert {layer["id"] for layer in manifest["layers"]} >= { "server", "runtime_service", "application", "network", "domain", "deployment", } filter_labels = {field["id"]: field["label"] for field in manifest["filter"]["fields"]} assert filter_labels["layer"] == "Node Type" assert filter_labels["deploymentEnvironment"] == "Deployment Environment" assert filter_labels["accessZone"] == "Access Zone" assert {"by-deployment-environment", "by-deployment-scenario", "by-routing-authority", "by-access-zone"} <= { mode["id"] for mode in manifest["modes"] } nodes = [element for element in payload["elements"] if "source" not in element["data"]] edges = [element for element in payload["elements"] if "source" in element["data"]] registered_only = next( element for element in nodes if element["data"]["id"] == "repo:registered-only" ) nodes_by_id = {element["data"]["id"]: element for element in nodes} runs_on = next(edge for edge in edges if edge["data"]["edgeType"] == "runs_on") deployment = nodes_by_id[runs_on["data"]["source"]] server = nodes_by_id[runs_on["data"]["target"]] network_port = next(element for element in nodes if element["data"]["kind"] == "NetworkPort") same_repo_edge = next(edge for edge in edges if edge["data"].get("sameRepo") is True) cross_repo_edge = next(edge for edge in edges if edge["data"].get("layoutAffinity") == "cross-repo") declares_edge = next(edge for edge in edges if edge["data"]["edgeType"] == "declares") assert registered_only["data"]["reviewState"] == "candidate" assert registered_only["data"]["unresolved"] is True assert deployment["data"]["layer"] == "deployment" assert server["data"]["layer"] == "server" assert ":" not in server["data"]["label"] assert network_port["data"]["layer"] == "network" assert network_port["data"]["label"].endswith("/tcp") assert ( len( [ edge for edge in edges if edge["data"]["edgeType"] == "runs_on" and edge["data"]["source"] == deployment["data"]["id"] and edge["data"]["target"] == server["data"]["id"] ] ) == 1 ) assert runs_on["data"]["layoutIdealLength"] < cross_repo_edge["data"]["layoutIdealLength"] assert runs_on["data"]["layoutElasticity"] > cross_repo_edge["data"]["layoutElasticity"] assert runs_on["data"]["displayOnly"] is True assert runs_on["data"]["canonicalType"] == "deploys" assert same_repo_edge["data"]["layoutIdealLength"] < cross_repo_edge["data"]["layoutIdealLength"] assert declares_edge["data"]["displayOnly"] is True assert any(node["data"]["sourceReferences"] for node in nodes if node["data"]["kind"] != "Repository") assert payload["metrics"]["deployment_node_count"] >= 1 assert payload["metrics"]["server_node_count"] >= 1 assert payload["metrics"]["registered_repo_count"] == 2 assert payload["metrics"]["unresolved_count"] == 0 assert "removed nodes are removed" in payload["filter"]["connected_edge_behavior"] def test_graph_explorer_collapses_discovered_repository_nodes() -> None: graph = { "apiVersion": "railiance.fabric/v1alpha1", "kind": "FabricGraphExport", "generated_at": "2026-05-21T00:00:00Z", "source": {"repo": "fixture-repo", "commit": "abc123"}, "nodes": [ { "id": "discovery:fixture-repo:repository:fixture-repo", "kind": "Repository", "name": "Fixture Repo", "repo": "fixture-repo", "domain": "testing", "lifecycle": "active", "attributes": {"discovery_origin": "deterministic"}, }, { "id": "discovery:fixture-repo:library:fixture-service", "kind": "Library", "name": "fixture-service", "repo": "fixture-repo", "domain": "testing", "lifecycle": "active", "attributes": {"language": "python"}, }, ], "edges": [ { "from": "discovery:fixture-repo:repository:fixture-repo", "to": "discovery:fixture-repo:library:fixture-service", "type": "declares_package", } ], } payload = fabric_graph_explorer_payload( graph, [{"slug": "fixture-repo", "name": "Fixture Repo"}], {"fixture-repo"}, ) nodes = [element for element in payload["elements"] if "source" not in element["data"]] edges = [element for element in payload["elements"] if "source" in element["data"]] repository_nodes = [node for node in nodes if node["data"]["kind"] == "Repository"] declares_package = next(edge for edge in edges if edge["data"]["edgeType"] == "declares_package") assert [node["data"]["id"] for node in repository_nodes] == ["repo:fixture-repo"] assert declares_package["data"]["source"] == "repo:fixture-repo" assert declares_package["data"]["target"] == "discovery:fixture-repo:library:fixture-service" assert declares_package["data"]["canonicalType"] == "built_from" assert declares_package["data"]["displayOnly"] is False def test_graph_explorer_presents_legacy_server_nodes_as_runtime_entities() -> None: graph = { "apiVersion": "railiance.fabric/v1alpha1", "kind": "FabricGraphExport", "nodes": [ { "id": "fixture.server.gitea.example.test", "kind": "Server", "name": "gitea.example.test", "repo": "fixture-repo", "domain": "testing", "lifecycle": "active", "attributes": { "host": "gitea.example.test", "server_type": "ingress-host", "deployment_overlay": { "deployment_environment": "test", "deployment_scenario": "coulombcore", "routing_authority": "kubernetes", "access_zone": "early-access", "policy_authority": "netkingdom-iam", "exposure_class": "early-access", "route_evidence": {"hostname": "gitea.example.test", "port": 443, "protocol": "tcp"}, }, }, }, { "id": "fixture.server.gitea.default.svc.cluster.local", "kind": "Server", "name": "gitea.default.svc.cluster.local", "repo": "fixture-repo", "domain": "testing", "lifecycle": "active", "attributes": {"host": "gitea.default.svc.cluster.local", "server_type": "kubernetes-service-dns"}, }, { "id": "fixture.domain.gitea.example.test", "kind": "DomainName", "name": "gitea.example.test", "repo": "fixture-repo", "domain": "testing", "lifecycle": "active", "attributes": {"domain": "gitea.example.test"}, }, { "id": "fixture.port.gitea.default.svc.cluster.local-3000-tcp", "kind": "NetworkPort", "name": "gitea.default.svc.cluster.local:3000/tcp", "repo": "fixture-repo", "domain": "testing", "lifecycle": "active", "attributes": {"host": "gitea.default.svc.cluster.local", "port": 3000, "protocol": "tcp"}, }, ], "edges": [ { "from": "fixture.domain.gitea.example.test", "to": "fixture.server.gitea.example.test", "type": "resolves_to", }, { "from": "fixture.server.gitea.default.svc.cluster.local", "to": "fixture.port.gitea.default.svc.cluster.local-3000-tcp", "type": "opens_port", }, ], } payload = fabric_graph_explorer_payload(graph, [{"slug": "fixture-repo", "name": "Fixture Repo"}], {"fixture-repo"}) nodes_by_id = { element["data"]["id"]: element["data"] for element in payload["elements"] if "source" not in element["data"] } assert nodes_by_id["fixture.server.gitea.example.test"]["kind"] == "ApplicationEndpoint" assert nodes_by_id["fixture.server.gitea.example.test"]["layer"] == "application" assert nodes_by_id["fixture.server.gitea.example.test"]["deploymentEnvironment"] == "test" assert nodes_by_id["fixture.server.gitea.example.test"]["deploymentScenario"] == "coulombcore" assert nodes_by_id["fixture.server.gitea.example.test"]["accessZone"] == "early-access" assert nodes_by_id["fixture.server.gitea.example.test"]["policyAuthority"] == "netkingdom-iam" assert nodes_by_id["fixture.server.gitea.example.test"]["routeHostname"] == "gitea.example.test" assert nodes_by_id["fixture.server.gitea.example.test"]["routePort"] == 443 assert nodes_by_id["fixture.server.gitea.default.svc.cluster.local"]["kind"] == "RuntimeService" assert nodes_by_id["fixture.server.gitea.default.svc.cluster.local"]["layer"] == "runtime_service" edge_types = { (element["data"]["source"], element["data"]["target"]): element["data"]["edgeType"] for element in payload["elements"] if "source" in element["data"] } assert edge_types[("fixture.domain.gitea.example.test", "fixture.server.gitea.example.test")] == "names_endpoint" assert ( edge_types[ ( "fixture.server.gitea.default.svc.cluster.local", "fixture.port.gitea.default.svc.cluster.local-3000-tcp", ) ] == "listens_on" ) def test_cli_exports_graph_explorer_payload(capsys) -> None: assert cli_main(["export", "--format", "graph-explorer"]) == 0 payload = json.loads(capsys.readouterr().out) _validate_schema("graph-explorer-payload.schema.yaml", payload) assert payload["kind"] == "GraphExplorerPayload" assert any( element["data"].get("sourceReferences") for element in payload["elements"] if "source" not in element["data"] ) def test_graph_explorer_payload_accepts_repo_scoping_shape() -> None: payload = { "apiVersion": "railiance.fabric/v1alpha1", "kind": "GraphExplorerPayload", "manifest_id": "repo-scoping.dependency-graph", "mode": "full", "profile": { "id": 1, "repository_id": 1, "name": "Evidence Audit", "description": "Show supporting evidence.", "default_mode": "full", "filter_rules": [ {"name": "Blur facts", "action": "blur", "match": {"layer": "fact"}} ], "manual_overrides": {"feature:1": "show"}, }, "filter": { "rules": [ {"action": "highlight", "match": {"layer": "fact", "reviewState": "candidate"}}, {"action": "remove", "match": {"layer": "stale_fact"}}, ], "manual_overrides": {}, "orphaned_overrides": [], }, "elements": [ { "data": { "id": "fact:document:README.md", "stableKey": "fact:document:README.md", "kind": "fact", "layer": "fact", "label": "README.md", "displayState": "blur", "reviewState": "accepted", "freshnessState": "current", "confidence": 0.8, "visualSize": 50, "sourceReferences": [{"path": "README.md", "kind": "document"}], }, "classes": "fact display-blur", }, { "data": { "id": "capability:1", "stableKey": "capability:1", "kind": "capability", "layer": "capability", "label": "Registry Capabilities", "displayState": "show", "reviewState": "accepted", }, }, { "data": { "id": "edge:fact:capability", "stableKey": "edge:fact:capability", "kind": "edge", "layer": "dependency", "label": "supports", "source": "fact:document:README.md", "target": "capability:1", "edgeType": "supports", "dependencyType": "supports", "strength": "strong", "edgeWidth": 5, "sameLayer": False, "displayState": "show", }, "classes": "supports strong", }, ], "hidden_elements": [], } _validate_schema("graph-explorer-payload.schema.yaml", payload) def test_registry_serves_graph_explorer_exports(tmp_path: Path) -> None: store = RegistryStore(tmp_path / "registry.sqlite3") store.init_schema() store.upsert_repository( { "slug": "railiance-fabric", "name": "Railiance Fabric", "state_hub_repo_id": "2c0de614-e468-4eb6-8157-470649ac8c05", } ) store.upsert_repository({"slug": "registered-only", "name": "Registered Only"}) store.add_snapshot( "railiance-fabric", { "commit": "test-commit", "generated_at": "2026-05-18T00: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}" with urllib.request.urlopen( f"{base_url}/exports/graph-explorer/manifest", timeout=5, ) as response: manifest = json.loads(response.read()) with urllib.request.urlopen(f"{base_url}/exports/graph-explorer", timeout=5) as response: payload = json.loads(response.read()) with urllib.request.urlopen(f"{base_url}/ui/graph-explorer", timeout=5) as response: page = response.read().decode("utf-8") content_type = response.headers["Content-Type"] _validate_schema("graph-explorer-manifest.schema.yaml", manifest) _validate_schema("graph-explorer-payload.schema.yaml", payload) assert manifest["id"] == "railiance-fabric.registry-map" assert payload["metrics"]["registered_only_repo_count"] == 1 assert content_type.startswith("text/html") assert 'id="graph-canvas"' in page assert 'id="mode-select"' in page assert 'id="layout-select"' in page assert 'id="label-select"' in page assert 'id="zone-overlay"' in page assert 'id="zone-boundary-toggle"' in page assert 'id="zone-group-select"' in page assert 'id="zone-layout-select"' in page assert 'id="node-type-filter"' in page assert 'id="edge-type-filter"' in page assert 'id="rule-panel"' in page assert 'id="rule-target"' in page assert 'id="rule-list"' in page assert 'id="selection-anchor"' in page assert 'id="help-popup"' in page assert "updateSelectionAnchor" in page assert "updateLabelVisibility" in page assert "ruleActionFor" in page assert "ruleRemovalSignature" in page assert "zoneModeFields" in page assert "zoneForData" in page assert "defaultZoneDefinitions" in page assert "resolveZoneInstances" in page assert "zoneDiagnosticCodes" in page assert "ZONE_EMPTY_SEED_SET" in page assert "ZONE_NODE_SEEDED_BY_MULTIPLE_ZONES" in page assert "ZONE_EDGE_CROSSES_ZONE_BOUNDARY" in page assert "zone.diagnostics" in page assert "isZoneContextOnlyEdge" in page assert "isZoneConnectivityEdge" in page assert "currentZoneViewState" in page assert "normalizeZoneViewState" in page assert "zoneDefinitionSet" in page assert "zoneDefinitionSets" in page assert "fabric-default" in page assert "activeZoneLayoutAlgorithm" in page assert "zoneLayoutAlgorithms" in page assert "normalizeZoneLayoutAlgorithm" in page assert "zoneLayoutAlgorithm" in page assert "zoneLayout" in page assert "compact-grid" in page assert "circle" in page assert "zoneContainerState" in page assert "ensureZoneContainer" in page assert "packZoneContainers" in page assert "applyZoneContainerLayout" in page assert "layoutZoneSubgraph" in page assert "serializedZoneContainers" in page assert "normalizedZoneContainers" in page assert "collapsedZoneSnapshots" in page assert "zoneDragState" in page assert "startZoneDrag" in page assert "moveZoneDrag" in page assert "finishZoneDrag" in page assert "Moved zone" in page assert "cursor: grab" in page assert "background: transparent" in page assert "box-shadow: none" in page assert "collapseZone" in page assert "expandZone" in page assert "zoneCollapseNodeId" in page assert "Collapse Zone" in page assert "Expand Zone" in page assert 'normalize: "deploymentEnvironment"' in page assert "zoneBoundsForNodes" in page assert "renderZoneOverlay" in page assert "renderZoneDetails" in page assert "dev-tegwick" in page assert 'normalized === "staging"' in page assert "zoneBoundaries" in page assert "zoneGrouping" in page assert "renderMapOverview" in page assert "route without policy authority" in page assert "deploymentEnvironment" in page assert "routingAuthority" in page assert "accessZone" in page assert "Remove and redraw" in page assert "Rules are applied top to bottom" in page assert "showHelp" in page assert "label-hidden" in page assert "Loading graph..." in page assert "No graph entities were returned by the registry." in page assert "Could not load graph explorer data" in page assert "Node Types" in page assert "Edge Types" in page assert 'data-help-title="Node Types"' in page assert 'data-help-title="Saved Views"' in page assert "Remove will redraw" in page assert "Registered only" in page assert 'id="layer-filter"' not in page assert '"border-width": 4' not in page assert 'id="profile-select"' in page assert 'id="profile-name"' in page assert 'id="orientation-list"' in page assert 'id="orientation-actions"' in page assert "Service dependency chain" in page assert "Interface consumers" in page assert "Interface impact" in page assert "applyOrientationContext" in page assert "Focus Context" in page assert "Highlight Context" in page assert "Remove Other" in page assert "viewStateSummary" in page assert "cytoscape.min.js" in page assert "layoutIdealLength" in page assert "layoutElasticity" in page assert "/exports/graph-explorer/manifest" in page assert 'data-override="hide"' in page assert 'data-profile-action="save"' in page assert 'data-profile-action="copy"' in page assert "railiance.fabric.graphExplorer.profiles" in page assert "URLSearchParams" in page finally: server.shutdown() server.server_close() thread.join(timeout=5) def _validate_schema(schema_name: str, payload: dict[str, object]) -> None: validator = draft202012_validator(Path("schemas") / schema_name) validator.validate(payload)