DocForge2/tests/test_visualization.py

from __future__ import annotations

import json
import shutil
import subprocess
import tempfile
import threading
import time
import unittest
import urllib.error
import urllib.parse
import urllib.request
from contextlib import contextmanager
from pathlib import Path

from docforge.errors import DocForgeError
from docforge.index import ProjectIndex
from docforge.project import Project
from docforge.viewer_manager import ViewerManager, ViewerManagerClient
from docforge.visualization import (
    _GRAPH_BROWSER_CSS,
    _GRAPH_BROWSER_HTML,
    _GRAPH_BROWSER_JAVASCRIPT,
    VISUALIZATION_TEMPLATE,
    VisualizationIndexSnapshot,
    VisualizationRunner,
)

ROOT = Path(__file__).resolve().parents[1]
FIXTURES = ROOT / "tests" / "fixtures"


class VisualizationTests(unittest.TestCase):
    def copy_fixture(self, name: str, destination: Path) -> Path:
        root = destination / name
        shutil.copytree(FIXTURES / name, root)
        return root

    @contextmanager
    def running_manager(
        self,
        state_path: Path,
        *,
        idle_timeout_seconds: float = 60,
        check_interval_seconds: float = 0.02,
    ):
        manager = ViewerManager(
            state_path,
            idle_timeout_seconds=idle_timeout_seconds,
            check_interval_seconds=check_interval_seconds,
        )
        thread = threading.Thread(target=manager.serve_forever, daemon=True)
        thread.start()
        deadline = time.monotonic() + 2
        while not state_path.exists() and time.monotonic() < deadline:
            time.sleep(0.01)
        self.assertTrue(state_path.exists())
        try:
            yield manager
        finally:
            manager.shutdown()
            thread.join(timeout=2)

    def test_overview_and_neighborhood_are_deterministic_and_bounded(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            index = ProjectIndex(Project.open(root))
            index.build()
            snapshot = VisualizationIndexSnapshot(index, index.check())

            overview = snapshot.overview()
            first = snapshot.node("guide.workflow", depth=2, limit=2)
            second = snapshot.node("guide.workflow", depth=2, limit=2)
            filtered = snapshot.filter_nodes(category="tag", value="canonical", limit=2)
            source = snapshot.source("guide.workflow")

            self.assertEqual(3, overview["node_count"])
            self.assertEqual(2, overview["edge_count"])
            self.assertEqual(
                [
                    {"value": "guide", "count": 2},
                    {"value": "proof", "count": 1},
                ],
                overview["families"],
            )
            self.assertEqual(first, second)
            self.assertEqual("guide.workflow", first["root"])
            self.assertEqual(1, filtered["count"])
            self.assertEqual(1, filtered["total"])
            self.assertFalse(filtered["truncated"])
            self.assertEqual("guide.foundation", filtered["results"][0]["node_id"])
            self.assertLessEqual(len(first["edges"]), 2)
            self.assertEqual("docs/content/workflow.md", source["source_path"])
            self.assertIn("Editors change canonical nodes", source["content"])
            self.assertIn(
                "guide.workflow",
                {node["node_id"] for node in first["nodes"]},
            )

            with self.assertRaisesRegex(DocForgeError, "safety boundary"):
                snapshot.node("guide.workflow", depth=1, limit=401)
            with self.assertRaisesRegex(DocForgeError, "category is unsupported"):
                snapshot.filter_nodes(category="relation", value="depends_on", limit=2)

    @unittest.skipUnless(shutil.which("node"), "Node.js is required for browser script validation")
    def test_browser_javascript_is_valid(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "graph-browser.js"
            path.write_text(_GRAPH_BROWSER_JAVASCRIPT, encoding="utf-8")
            result = subprocess.run(
                ["node", "--check", str(path)],
                check=False,
                capture_output=True,
                text=True,
            )
        self.assertEqual("", result.stderr)
        self.assertEqual(0, result.returncode)

    def test_browser_contains_hiding_source_navigation_and_scrollable_inspector(self) -> None:
        self.assertIn('id="restore-hidden"', _GRAPH_BROWSER_HTML)
        self.assertIn('id="open-node-source"', _GRAPH_BROWSER_HTML)
        self.assertIn('id="hide-node"', _GRAPH_BROWSER_HTML)
        self.assertIn('id="source-dialog"', _GRAPH_BROWSER_HTML)
        self.assertIn("state.hiddenNodes.add(nodeId)", _GRAPH_BROWSER_JAVASCRIPT)
        self.assertIn(
            "grid-template-rows: auto minmax(0, 1fr) auto",
            _GRAPH_BROWSER_CSS,
        )

    @unittest.skipUnless(shutil.which("node"), "Node.js is required for topology validation")
    def test_topology_roles_hops_and_shading_are_deterministic(self) -> None:
        script = _GRAPH_BROWSER_JAVASCRIPT
        viewport_logic = script.split("function viewportForPositions", 1)[1].split(
            "function selectNode", 1
        )[0]
        relation_constants = script.split("const relationStyles", 1)[1].split("const $", 1)[0]
        relation_logic = script.split("function relationHash", 1)[1].split(
            "function applyViewport", 1
        )[0]
        topology_logic = script.split("function analyzeTopology", 1)[1].split(
            "function renderNeighborhood", 1
        )[0]
        harness = (
            "const defaultViewport = Object.freeze({x: -600, y: -410, width: 1200, height: 820});\n"
            "const relationStyles"
            + relation_constants
            + "function relationHash"
            + relation_logic
            + "function viewportForPositions"
            + viewport_logic
            + "function analyzeTopology"
            + topology_logic
            + """
const data = {
  root: "primary",
  depth: 2,
  nodes: [
    {node_id: "primary"},
    {node_id: "child-one"},
    {node_id: "child-two"},
    {node_id: "incoming"},
  ],
  edges: [
    {source_id: "primary", relation: "contains", target_id: "child-one"},
    {source_id: "child-one", relation: "contains", target_id: "child-two"},
    {source_id: "incoming", relation: "references", target_id: "primary"},
  ],
};
const topology = analyzeTopology(data);
const positions = layoutNodes(data.nodes, data.root, topology);
const fail = (message) => { throw new Error(message); };
if (topology.get("primary").role !== "primary") fail("root role");
if (topology.get("child-one").role !== "child") fail("direct child role");
if (topology.get("child-two").role !== "child") fail("descendant role");
if (topology.get("incoming").role !== "edge") fail("incoming edge role");
if (topology.get("child-two").hop !== 2) fail("descendant hop");
if (Math.hypot(positions.get("child-two").x, positions.get("child-two").y)
    <= Math.hypot(positions.get("child-one").x, positions.get("child-one").y)) {
  fail("hop rings");
}
if (nodePalette("child", 2).fill === nodePalette("child", 1).fill) fail("hop shading");
const singleViewport = viewportForPositions(new Map([["primary", {x: 0, y: 0}]]));
if (singleViewport.width !== 440) fail("single-node fit");
if (singleViewport.x !== -220 || singleViewport.y !== -singleViewport.height / 2) {
  fail("single-node centered");
}
const centered = viewportCenteredOn(
  positions.get("child-two"),
  {x: 20, y: 30, width: 500, height: 300},
);
if (centered.width !== 500 || centered.height !== 300) fail("center preserves zoom");
if (centered.x !== positions.get("child-two").x - 250) fail("center x");
if (centered.y !== positions.get("child-two").y - 150) fail("center y");
if (relationStyle("calls").family !== "Execution") fail("calls family");
if (relationStyle("unknown_relation").family !== "Other") fail("fallback relation family");
const flowData = {
  root: "primary",
  nodes: [
    {node_id: "primary"},
    {node_id: "package"},
    {node_id: "module"},
    {node_id: "test-class"},
    {node_id: "test-file"},
    {node_id: "evidence"},
  ],
  edges: [
    {source_id: "test-class", relation: "contains", target_id: "primary"},
    {source_id: "test-file", relation: "contains", target_id: "test-class"},
    {source_id: "module", relation: "contains", target_id: "test-file"},
    {source_id: "package", relation: "contains", target_id: "module"},
    {source_id: "evidence", relation: "verifies", target_id: "primary"},
  ],
};
const flow = buildFlowGraph(flowData);
const flowIds = new Set(flow.nodes.map((node) => node.node_id));
if (!flowIds.has("package") || !flowIds.has("module") || !flowIds.has("test-class")
    || !flowIds.has("test-file") || !flowIds.has("evidence")) fail("full lineage membership");
const flowPositions = layoutFlow(flow.nodes, flow.root, flow.topology);
if (flowPositions.get("primary").x !== 0) fail("flow destination position");
if (flowPositions.get("package").x >= flowPositions.get("primary").x) {
  fail("flow upstream direction");
}
const containmentEdge = flow.edges.find((edge) => edge.relation === "contains"
  && edge.source_id === "test-class");
if (!containmentEdge || containmentEdge.target_id !== "primary") {
  fail("containment ancestry direction");
}
"""
        )
        result = subprocess.run(
            ["node", "-e", harness],
            check=False,
            capture_output=True,
            text=True,
        )
        self.assertEqual("", result.stderr)
        self.assertEqual(0, result.returncode)

    def test_runner_serves_only_token_bound_read_only_graph_endpoints(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            index = ProjectIndex(Project.open(root))
            index.build()
            runner = VisualizationRunner(index)
            try:
                first = runner.start(node_id="guide.workflow", depth=2)
                second = runner.start(query="canonical nodes")
                first_url = urllib.parse.urlparse(str(first["url"]))
                second_url = urllib.parse.urlparse(str(second["url"]))
                base = f"{first_url.scheme}://{first_url.netloc}{first_url.path}"

                self.assertEqual(VISUALIZATION_TEMPLATE, first["template"])
                self.assertTrue(first["read_only"])
                self.assertEqual("browser_lease", first["lifetime"]["policy"])
                self.assertFalse(runner._thread.daemon)
                self.assertEqual(first_url.netloc, second_url.netloc)
                self.assertEqual(first_url.path, second_url.path)

                with urllib.request.urlopen(base, timeout=2) as response:
                    html = response.read().decode("utf-8")
                    headers = response.headers
                with urllib.request.urlopen(f"{base}assets/graph.css", timeout=2) as response:
                    css = response.read().decode("utf-8")
                    self.assertEqual("text/css; charset=utf-8", response.headers["Content-Type"])
                with urllib.request.urlopen(f"{base}assets/graph.js", timeout=2) as response:
                    javascript = response.read().decode("utf-8")
                    self.assertEqual(
                        "text/javascript; charset=utf-8",
                        response.headers["Content-Type"],
                    )
                self.assertIn("DocForge graph", html)
                self.assertIn('href="assets/graph.css"', html)
                self.assertIn('src="assets/graph.js"', html)
                self.assertIn('id="zoom-in"', html)
                self.assertIn('id="zoom-out"', html)
                self.assertIn('id="reset-view"', html)
                self.assertIn('id="node-dialog"', html)
                self.assertIn('id="node-card"', html)
                self.assertIn('id="explore-node"', html)
                self.assertIn('id="explore-card-node"', html)
                self.assertIn("positionNodeCard(dialog, event)", javascript)
                self.assertIn(
                    'renderDetails($("node-card-details"), data.node, data, true, false)',
                    javascript,
                )
                self.assertIn("dialog.show();", javascript)
                self.assertIn("position: fixed; inset: auto; margin: 0", css)
                self.assertIn("background: rgba(7, 18, 29, .82)", css)
                self.assertIn("width: fit-content; height: fit-content", css)
                self.assertIn("max-width: min(760px, calc(100vw - 32px))", css)
                self.assertIn('id="left-resizer"', html)
                self.assertIn('id="right-resizer"', html)
                self.assertIn('id="view-nodes"', html)
                self.assertIn('id="view-flow"', html)
                self.assertIn('id="neighborhood-sections"', html)
                self.assertIn('id="relationship-key"', html)
                self.assertIn('id="relationship-key-list"', html)
                self.assertNotIn('id="details"', html)
                self.assertIn(".empty[hidden] { display: none; }", css)
                self.assertIn("html, body { height: 100%; overflow: hidden; }", css)
                self.assertIn("Space centers selection", html)
                self.assertIn("resize: both", css)
                self.assertNotIn("backdrop-filter", css)
                self.assertIn('addEventListener("wheel"', javascript)
                self.assertIn('addEventListener("pointermove"', javascript)
                self.assertIn("inspectNode(node.node_id)", javascript)
                self.assertIn("showNodeCard(node.node_id)", javascript)
                self.assertIn('addEventListener("contextmenu"', javascript)
                self.assertIn("dialog.showModal()", javascript)
                self.assertIn("await loadNode(nodeId)", javascript)
                self.assertIn("beginDialogDrag", javascript)
                self.assertIn('setupPanelResizer("left")', javascript)
                self.assertIn('setupPanelResizer("right")', javascript)
                self.assertIn("Focus node", javascript)
                self.assertIn("Outgoing paths", javascript)
                self.assertIn("Incoming & lateral", javascript)
                self.assertNotIn(">Children<", html)
                self.assertIn("distanceShade", javascript)
                self.assertIn("viewportForPositions", javascript)
                self.assertIn("centerSelectedNode", javascript)
                self.assertIn("relationStyles", javascript)
                self.assertIn("appendRelationMarker", javascript)
                self.assertIn("renderRelationshipKey", javascript)
                self.assertIn("buildFlowGraph", javascript)
                self.assertIn("layoutFlow", javascript)
                self.assertIn(
                    'api(`${showingFlow ? "lineage" : "node"}?${params}`)',
                    javascript,
                )
                self.assertIn("filterByDescriptor", javascript)
                self.assertIn("api(`filter?${params}`)", javascript)
                self.assertIn('setViewMode("flow")', javascript)
                self.assertIn('event.code !== "Space"', javascript)
                self.assertIn('class: "selection-ring"', javascript)
                self.assertIn("renewViewerLease", javascript)
                pointerdown = javascript.split(
                    '$("graph").addEventListener("pointerdown"',
                    1,
                )[1].split('$("graph").addEventListener("pointermove"', 1)[0]
                pointermove = javascript.split(
                    '$("graph").addEventListener("pointermove"',
                    1,
                )[1].split("function endPan", 1)[0]
                self.assertNotIn("setPointerCapture", pointerdown)
                self.assertIn("setPointerCapture", pointermove)
                self.assertIn("right-click full inspector", html)
                self.assertIn("default-src 'none'", headers["Content-Security-Policy"])
                self.assertIn("script-src 'self'", headers["Content-Security-Policy"])
                self.assertIn("style-src 'self'", headers["Content-Security-Policy"])
                self.assertNotIn("unsafe-inline", headers["Content-Security-Policy"])
                self.assertEqual("no-store", headers["Cache-Control"])
                self.assertEqual("DENY", headers["X-Frame-Options"])

                with self.assertRaises(urllib.error.HTTPError) as missing_asset:
                    urllib.request.urlopen(f"{base}assets/missing.js", timeout=2)
                self.assertEqual(404, missing_asset.exception.code)
                missing_asset.exception.close()

                with urllib.request.urlopen(f"{base}api/overview", timeout=2) as response:
                    overview = json.load(response)
                self.assertEqual("alpha-docs", overview["project_id"])
                self.assertEqual(3, overview["node_count"])
                self.assertEqual(20, overview["max_results"])

                with urllib.request.urlopen(f"{base}api/heartbeat", timeout=2) as response:
                    heartbeat = json.load(response)
                self.assertEqual("alive", heartbeat["viewer"])
                self.assertEqual(runner.lease_seconds, heartbeat["lease_seconds"])

                search_query = urllib.parse.urlencode(
                    {"q": "canonical nodes", "family": "", "limit": overview["max_results"]}
                )
                with urllib.request.urlopen(
                    f"{base}api/search?{search_query}", timeout=2
                ) as response:
                    search = json.load(response)
                self.assertEqual("alpha-docs", search["project_id"])
                self.assertGreaterEqual(search["count"], 1)

                filter_query = urllib.parse.urlencode(
                    {"category": "tag", "value": "canonical", "limit": overview["max_results"]}
                )
                with urllib.request.urlopen(
                    f"{base}api/filter?{filter_query}", timeout=2
                ) as response:
                    filtered = json.load(response)
                self.assertEqual("alpha-docs", filtered["project_id"])
                self.assertEqual("tag", filtered["category"])
                self.assertEqual("canonical", filtered["value"])
                self.assertEqual(1, filtered["total"])
                self.assertEqual("guide.foundation", filtered["results"][0]["node_id"])

                node_query = urllib.parse.urlencode(
                    {"id": "guide.workflow", "depth": 1, "limit": 20}
                )
                with urllib.request.urlopen(f"{base}api/node?{node_query}", timeout=2) as response:
                    node = json.load(response)
                self.assertEqual("guide.workflow", node["node"]["node_id"])
                self.assertEqual("guide.workflow", node["root"])

                lineage_query = urllib.parse.urlencode({"id": "guide.workflow", "limit": 1000})
                with urllib.request.urlopen(
                    f"{base}api/lineage?{lineage_query}", timeout=2
                ) as response:
                    lineage = json.load(response)
                self.assertTrue(lineage["lineage"])
                self.assertEqual("guide.workflow", lineage["root"])
                self.assertIn(
                    {
                        "source_id": "proof.validation",
                        "relation": "proves",
                        "target_id": "guide.workflow",
                    },
                    lineage["edges"],
                )

                wrong_token = f"{first_url.scheme}://{first_url.netloc}/wrong-token/api/overview"
                with self.assertRaises(urllib.error.HTTPError) as missing:
                    urllib.request.urlopen(wrong_token, timeout=2)
                self.assertEqual(404, missing.exception.code)
                missing.exception.close()

                request = urllib.request.Request(
                    f"{base}api/overview",
                    data=b"{}",
                    method="POST",
                )
                with self.assertRaises(urllib.error.HTTPError) as rejected:
                    urllib.request.urlopen(request, timeout=2)
                self.assertEqual(405, rejected.exception.code)
                try:
                    self.assertEqual(
                        "method_not_allowed",
                        json.loads(rejected.exception.read())["error"]["code"],
                    )
                finally:
                    rejected.exception.close()
            finally:
                runner.stop()

    def test_browser_lease_keeps_listener_alive_then_closes_it(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            index = ProjectIndex(Project.open(root))
            index.build()
            runner = VisualizationRunner(
                index,
                initial_grace_seconds=0.2,
                lease_seconds=0.2,
                monitor_interval_seconds=0.02,
            )
            try:
                result = runner.start()
                parsed = urllib.parse.urlparse(str(result["url"]))
                base = f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
                heartbeat = f"{base}api/heartbeat"
                with urllib.request.urlopen(heartbeat, timeout=2) as response:
                    self.assertEqual("alive", json.load(response)["viewer"])
                time.sleep(0.12)
                with urllib.request.urlopen(heartbeat, timeout=2) as response:
                    self.assertEqual("alive", json.load(response)["viewer"])

                deadline = time.monotonic() + 2
                while runner._server is not None and time.monotonic() < deadline:
                    time.sleep(0.02)
                self.assertIsNone(runner._server)
                with self.assertRaises(OSError):
                    urllib.request.urlopen(base, timeout=0.2)
            finally:
                runner.stop()

    def test_manager_reuses_workers_and_applies_explicit_or_idle_shutdown(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            index = ProjectIndex(Project.open(root))
            index.build()
            state_path = Path(directory) / "viewer-manager.json"
            with self.running_manager(
                state_path, idle_timeout_seconds=0.25, check_interval_seconds=0.02
            ):
                client = ViewerManagerClient(index, state_path=state_path)
                first = client.start()
                url = str(first["url"])
                self.assertEqual("managed_idle", first["lifetime"]["policy"])
                self.assertTrue(url.startswith("http://127.0.0.1:"))
                with urllib.request.urlopen(url, timeout=2) as response:
                    self.assertEqual(200, response.status)

                reused = client.start(depth=2)
                self.assertTrue(reused["reused"])
                self.assertEqual(url.split("?", 1)[0], str(reused["url"]).split("?", 1)[0])
                self.assertEqual("running", client.status()["state"])

                time.sleep(0.1)
                heartbeat = url.split("?", 1)[0] + "api/heartbeat"
                with urllib.request.urlopen(heartbeat, timeout=2) as response:
                    self.assertEqual("alive", json.load(response)["viewer"])
                time.sleep(0.1)
                self.assertEqual("running", client.status()["state"])

                time.sleep(0.35)
                deadline = time.monotonic() + 2
                while client.status()["state"] == "running" and time.monotonic() < deadline:
                    time.sleep(0.02)
                self.assertEqual("not_running", client.status()["state"])

                restarted = client.start()
                stopped = client.stop()
                self.assertEqual("stopped", stopped["state"])
                with self.assertRaises(OSError):
                    urllib.request.urlopen(str(restarted["url"]), timeout=0.2)

    def test_runner_rejects_ambiguous_targets_and_changed_index_snapshot(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            index = ProjectIndex(Project.open(root))
            index.build()
            runner = VisualizationRunner(index)
            try:
                with self.assertRaisesRegex(DocForgeError, "either one exact"):
                    runner.start(node_id="guide.workflow", query="workflow")
                result = runner.start()
                parsed = urllib.parse.urlparse(str(result["url"]))
                with index.path.open("ab") as handle:
                    handle.write(b"\n")
                endpoint = f"{parsed.scheme}://{parsed.netloc}{parsed.path}api/overview"
                with self.assertRaises(urllib.error.HTTPError) as stale:
                    urllib.request.urlopen(endpoint, timeout=2)
                self.assertEqual(409, stale.exception.code)
                try:
                    self.assertEqual(
                        "visualization_stale",
                        json.loads(stale.exception.read())["error"]["code"],
                    )
                finally:
                    stale.exception.close()
            finally:
                runner.stop()

    def test_two_visualizations_remain_project_bound(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            parent = Path(directory)
            alpha_root = self.copy_fixture("alpha", parent / "alpha")
            beta_root = self.copy_fixture("beta", parent / "beta")
            alpha_index = ProjectIndex(Project.open(alpha_root))
            beta_index = ProjectIndex(Project.open(beta_root))
            alpha_index.build()
            beta_index.build()
            alpha = VisualizationRunner(alpha_index)
            beta = VisualizationRunner(beta_index)
            try:
                alpha_url = urllib.parse.urlparse(str(alpha.start()["url"]))
                beta_url = urllib.parse.urlparse(str(beta.start()["url"]))
                self.assertNotEqual(alpha_url.netloc, beta_url.netloc)
                for parsed, expected in (
                    (alpha_url, "alpha-docs"),
                    (beta_url, "beta-notes"),
                ):
                    endpoint = f"{parsed.scheme}://{parsed.netloc}{parsed.path}api/overview"
                    with urllib.request.urlopen(endpoint, timeout=2) as response:
                        self.assertEqual(expected, json.load(response)["project_id"])
            finally:
                alpha.stop()
                beta.stop()


if __name__ == "__main__":
    unittest.main()