485 lines
22 KiB
Python
485 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
from docforge.errors import DocForgeError
|
|
from docforge.index import ProjectIndex
|
|
from docforge.project import Project
|
|
from docforge.visualization import (
|
|
_GRAPH_BROWSER_HTML,
|
|
VISUALIZATION_TEMPLATE,
|
|
PersistentVisualizationRunner,
|
|
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
|
|
|
|
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)
|
|
|
|
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.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 embedded script validation")
|
|
def test_embedded_browser_javascript_is_valid(self) -> None:
|
|
script = _GRAPH_BROWSER_HTML.split("<script>", 1)[1].split("</script>", 1)[0]
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "graph-browser.js"
|
|
path.write_text(script, 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)
|
|
|
|
@unittest.skipUnless(shutil.which("node"), "Node.js is required for topology validation")
|
|
def test_embedded_topology_roles_hops_and_shading_are_deterministic(self) -> None:
|
|
script = _GRAPH_BROWSER_HTML.split("<script>", 1)[1].split("</script>", 1)[0]
|
|
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("reads").flow !== "reverse") fail("reads flow direction");
|
|
if (relationStyle("documents").flow !== null) fail("documents excluded from flow");
|
|
if (relationStyle("unknown_relation").family !== "Other") fail("fallback relation family");
|
|
const flowData = {
|
|
root: "primary",
|
|
depth: 2,
|
|
nodes: [
|
|
{node_id: "primary"},
|
|
{node_id: "caller"},
|
|
{node_id: "dependency"},
|
|
{node_id: "reader"},
|
|
{node_id: "document"},
|
|
],
|
|
edges: [
|
|
{source_id: "caller", relation: "calls", target_id: "primary"},
|
|
{source_id: "primary", relation: "depends_on", target_id: "dependency"},
|
|
{source_id: "primary", relation: "reads", target_id: "reader"},
|
|
{source_id: "document", relation: "documents", target_id: "primary"},
|
|
],
|
|
};
|
|
const flow = buildFlowGraph(flowData);
|
|
const flowIds = new Set(flow.nodes.map((node) => node.node_id));
|
|
if (!flowIds.has("caller") || !flowIds.has("dependency") || !flowIds.has("reader")) {
|
|
fail("upstream flow membership");
|
|
}
|
|
if (flowIds.has("document")) fail("evidence leaked into flow");
|
|
const flowPositions = layoutFlow(flow.nodes, flow.root, flow.topology);
|
|
if (flowPositions.get("primary").x !== 0) fail("flow destination position");
|
|
if (flowPositions.get("caller").x >= flowPositions.get("primary").x) {
|
|
fail("flow upstream direction");
|
|
}
|
|
const dependencyEdge = flow.edges.find((edge) => edge.relation === "depends_on");
|
|
if (dependencyEdge.source_id !== "dependency" || dependencyEdge.target_id !== "primary") {
|
|
fail("dependency semantic reversal");
|
|
}
|
|
"""
|
|
)
|
|
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
|
|
self.assertIn("DocForge graph", 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('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; }", html)
|
|
self.assertIn("html, body { height: 100%; overflow: hidden; }", html)
|
|
self.assertIn("Space centers selection", html)
|
|
self.assertIn("resize: both", html)
|
|
self.assertNotIn("backdrop-filter", html)
|
|
self.assertIn('addEventListener("wheel"', html)
|
|
self.assertIn('addEventListener("pointermove"', html)
|
|
self.assertIn("inspectNode(node.node_id)", html)
|
|
self.assertIn("showNodeCard(node.node_id)", html)
|
|
self.assertIn('addEventListener("contextmenu"', html)
|
|
self.assertIn("dialog.showModal()", html)
|
|
self.assertIn("await loadNode(nodeId)", html)
|
|
self.assertIn("beginDialogDrag", html)
|
|
self.assertIn('setupPanelResizer("left")', html)
|
|
self.assertIn('setupPanelResizer("right")', html)
|
|
self.assertIn("Focus node", html)
|
|
self.assertIn("Outgoing paths", html)
|
|
self.assertIn("Incoming & lateral", html)
|
|
self.assertNotIn(">Children<", html)
|
|
self.assertIn("distanceShade", html)
|
|
self.assertIn("viewportForPositions", html)
|
|
self.assertIn("centerSelectedNode", html)
|
|
self.assertIn("relationStyles", html)
|
|
self.assertIn("appendRelationMarker", html)
|
|
self.assertIn("renderRelationshipKey", html)
|
|
self.assertIn("semanticFlowEdge", html)
|
|
self.assertIn("buildFlowGraph", html)
|
|
self.assertIn("layoutFlow", html)
|
|
self.assertIn("filterByDescriptor", html)
|
|
self.assertIn("api(`filter?${params}`)", html)
|
|
self.assertIn('setViewMode("flow")', html)
|
|
self.assertIn('event.code !== "Space"', html)
|
|
self.assertIn('class: "selection-ring"', html)
|
|
self.assertIn("renewViewerLease", html)
|
|
pointerdown = html.split('$("graph").addEventListener("pointerdown"', 1)[1].split(
|
|
'$("graph").addEventListener("pointermove"', 1
|
|
)[0]
|
|
pointermove = html.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.assertEqual("no-store", headers["Cache-Control"])
|
|
self.assertEqual("DENY", headers["X-Frame-Options"])
|
|
|
|
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"])
|
|
|
|
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_persistent_worker_survives_launcher_and_stops_only_explicitly(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture("alpha", Path(directory))
|
|
ProjectIndex(Project.open(root)).build()
|
|
script = """
|
|
import sys
|
|
from pathlib import Path
|
|
from docforge.index import ProjectIndex
|
|
from docforge.project import Project
|
|
from docforge.visualization import PersistentVisualizationRunner
|
|
|
|
runner = PersistentVisualizationRunner(ProjectIndex(Project.open(Path(sys.argv[1]))))
|
|
print(runner.start()["url"], flush=True)
|
|
time.sleep(60)
|
|
"""
|
|
script = "import time\n" + script
|
|
with subprocess.Popen(
|
|
[sys.executable, "-c", script, str(root)],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
) as launcher:
|
|
assert launcher.stdout is not None
|
|
url = launcher.stdout.readline().strip()
|
|
launcher.terminate()
|
|
launcher.wait(timeout=2)
|
|
self.assertLess(launcher.returncode, 0)
|
|
self.assertTrue(url.startswith("http://127.0.0.1:"))
|
|
with urllib.request.urlopen(url, timeout=2) as response:
|
|
self.assertEqual(200, response.status)
|
|
|
|
time.sleep(0.6)
|
|
with urllib.request.urlopen(url, timeout=2) as response:
|
|
self.assertEqual(200, response.status)
|
|
|
|
runner = PersistentVisualizationRunner(ProjectIndex(Project.open(root)))
|
|
try:
|
|
reused = runner.start()
|
|
self.assertTrue(reused["reused"])
|
|
self.assertEqual(url.split("?", 1)[0], str(reused["url"]).split("?", 1)[0])
|
|
stopped = runner.stop()
|
|
self.assertEqual("stopped", stopped["state"])
|
|
with self.assertRaises(OSError):
|
|
urllib.request.urlopen(url, timeout=0.2)
|
|
finally:
|
|
runner.stop()
|
|
|
|
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()
|