railiance-fabric/tests/test_graph_explorer.py

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)

    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"
    )

    assert registered_only["data"]["reviewState"] == "candidate"
    assert registered_only["data"]["unresolved"] is True
    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"]["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())

        _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
    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)