Files
railiance-fabric/tests/test_graph_explorer.py

252 lines
10 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.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 {layer["id"] for layer in manifest["layers"]} >= {"server", "deployment"}
filter_labels = {field["id"]: field["label"] for field in manifest["filter"]["fields"]}
assert filter_labels["layer"] == "Node Type"
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"
)
deployment = next(element for element in nodes if element["data"]["kind"] == "Deployment")
server = next(element for element in nodes if element["data"]["kind"] == "Server")
runs_on = next(edge for edge in edges if edge["data"]["edgeType"] == "runs_on")
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")
assert registered_only["data"]["reviewState"] == "candidate"
assert registered_only["data"]["unresolved"] is True
assert deployment["data"]["layer"] == "deployment"
assert server["data"]["layer"] == "server"
assert runs_on["data"]["layoutIdealLength"] < cross_repo_edge["data"]["layoutIdealLength"]
assert runs_on["data"]["layoutElasticity"] > cross_repo_edge["data"]["layoutElasticity"]
assert same_repo_edge["data"]["layoutIdealLength"] < cross_repo_edge["data"]["layoutIdealLength"]
assert any(edge["data"]["edgeType"] == "declares" for edge in edges)
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
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": [],
"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="node-type-filter"' in page
assert 'id="edge-type-filter"' in page
assert 'id="selection-anchor"' in page
assert 'id="help-popup"' in page
assert "updateSelectionAnchor" in page
assert "updateLabelVisibility" 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 "Interface consumers" in page
assert "Dependency path" 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)