Files
railiance-fabric/tests/test_graph_explorer.py
2026-05-19 00:16:57 +02:00

240 lines
9.5 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="node-type-filter"' in page
assert 'id="edge-type-filter"' in page
assert 'id="selection-anchor"' in page
assert "updateSelectionAnchor" in page
assert "Node Types" in page
assert "Edge Types" 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)