operator
adfcd2e602
- box passkey [show|fetch] (+ muse passkey): documents VM-only passkey (/srv/box/passkey.txt, fallback /etc/netvm/passkey.txt on 34.139.37.135), probes VM over SSH with graceful fallback; --json supported. No secrets on bl. - approvals: request_key_approval / check_node_key_request; KEY_APPROVAL status surfaced in `box approvals check`; allow/deny resolve + audit to box-ctl.jsonl; never auto-approved. New `box approvals request-key <node> --reason`. - box lookup (summary|fleet|threads|unread|approvals|key|docs) and docs-lookup engine with lookup_internal/ database (docs_internal symlink). - muse-tmux: non-TTY attach falls back to scrollback capture; prune NameError fix. - box/muse passthrough for tmux/muse/docs; thread list/view alias + prefix resolve. - Docs: AGENTS.md, AGENT-TOOLING.md, BOX-WEB-SURFACE-GUIDE.md, README. - Tests: key-approval + passkey tests; sync stale sidechat UUIDs and manifest name. - .gitignore runtime trackers (subagent-sessions, conversation-nudge-tracker).
225 lines
8.7 KiB
Python
225 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
|
"""test_docs_lookup.py — Unit tests for docs_internal database and lookup engine."""
|
|
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
LOOKUP_INTERNAL = REPO_ROOT / "lookup_internal"
|
|
DOCS_INTERNAL = REPO_ROOT / "docs_internal"
|
|
DOCS_LOOKUP = REPO_ROOT / "bin" / "docs-lookup.py"
|
|
SUPER_CLI = REPO_ROOT / "bin" / "super-cli.py"
|
|
|
|
|
|
class TestDocsInternalDatabase(unittest.TestCase):
|
|
def test_manifest_and_symlink(self):
|
|
self.assertTrue(LOOKUP_INTERNAL.is_dir(), "lookup_internal must exist as directory")
|
|
self.assertTrue(DOCS_INTERNAL.exists(), "docs_internal must exist")
|
|
self.assertTrue(DOCS_INTERNAL.is_symlink(), "docs_internal must be a symlink to lookup_internal")
|
|
|
|
manifest_path = LOOKUP_INTERNAL / "manifest.json"
|
|
self.assertTrue(manifest_path.is_file(), "manifest.json must exist")
|
|
data = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
self.assertIn(data.get("name"), ("lookup_internal", "docs_internal"))
|
|
self.assertIn("collections", data)
|
|
collections = data["collections"]
|
|
self.assertGreaterEqual(len(collections), 5)
|
|
for col in collections:
|
|
self.assertTrue((LOOKUP_INTERNAL / col["json_file"]).is_file(), f"{col['json_file']} missing")
|
|
self.assertTrue((LOOKUP_INTERNAL / col["md_file"]).is_file(), f"{col['md_file']} missing")
|
|
|
|
|
|
def test_sentence_structures(self):
|
|
p = DOCS_INTERNAL / "sentence_structure.json"
|
|
self.assertTrue(p.is_file())
|
|
data = json.loads(p.read_text(encoding="utf-8"))
|
|
structs = data.get("structures", {})
|
|
self.assertIn("work_order", structs)
|
|
self.assertIn("ack", structs)
|
|
self.assertIn("result", structs)
|
|
self.assertIn("claim", structs)
|
|
self.assertIn("tool_call", structs)
|
|
self.assertIn("nudge", structs)
|
|
|
|
for name, entry in structs.items():
|
|
self.assertIn("protocol_tag", entry)
|
|
self.assertIn("template", entry)
|
|
self.assertIn("required_fields", entry)
|
|
self.assertIn("description", entry)
|
|
self.assertIn("example", entry)
|
|
|
|
def test_regex_patterns_and_fixtures(self):
|
|
p = DOCS_INTERNAL / "regex_patterns.json"
|
|
self.assertTrue(p.is_file())
|
|
data = json.loads(p.read_text(encoding="utf-8"))
|
|
patterns = data.get("patterns", {})
|
|
self.assertIn("work_order", patterns)
|
|
self.assertIn("ack", patterns)
|
|
self.assertIn("verb", patterns)
|
|
self.assertIn("result", patterns)
|
|
self.assertIn("tool_call", patterns)
|
|
|
|
for pat_name, entry in patterns.items():
|
|
raw_pat = entry["pattern"]
|
|
flag_names = entry.get("flags", [])
|
|
flags = 0
|
|
for fn in flag_names:
|
|
flags |= getattr(re, fn)
|
|
# Must compile cleanly
|
|
compiled = re.compile(raw_pat, flags)
|
|
|
|
# Test valid samples
|
|
samples = entry.get("test_samples", {})
|
|
for valid_str in samples.get("valid", []):
|
|
self.assertIsNotNone(
|
|
compiled.search(valid_str),
|
|
f"Pattern '{pat_name}' failed to match valid sample: '{valid_str}'"
|
|
)
|
|
|
|
# Test invalid samples
|
|
for invalid_str in samples.get("invalid", []):
|
|
self.assertIsNone(
|
|
compiled.fullmatch(invalid_str),
|
|
f"Pattern '{pat_name}' unexpectedly matched invalid sample: '{invalid_str}'"
|
|
)
|
|
|
|
def test_assistive_surfaces(self):
|
|
p = DOCS_INTERNAL / "assistive_surfaces.json"
|
|
self.assertTrue(p.is_file())
|
|
data = json.loads(p.read_text(encoding="utf-8"))
|
|
self.assertEqual(data.get("host"), "box.muse-dev.online")
|
|
self.assertIn("views", data)
|
|
views = data["views"]
|
|
self.assertIn("dashboard", views)
|
|
self.assertIn("dms", views)
|
|
self.assertIn("jobs", views)
|
|
self.assertIn("loops", views)
|
|
|
|
# Check dashboard surface details
|
|
dash = views["dashboard"]
|
|
self.assertEqual(dash["dom_pane_selector"], "#tab-dashboard")
|
|
self.assertEqual(dash["key_elements"]["grid"], "#fleet-grid")
|
|
self.assertEqual(dash["api_endpoints"][0]["path"], "/api/box/fleet")
|
|
|
|
def test_cli_tools_reference(self):
|
|
p = DOCS_INTERNAL / "cli_tools.json"
|
|
self.assertTrue(p.is_file())
|
|
data = json.loads(p.read_text(encoding="utf-8"))
|
|
self.assertIn("binaries", data)
|
|
self.assertIn("super", data["binaries"])
|
|
self.assertIn("box", data["binaries"])
|
|
self.assertIn("domains", data)
|
|
domains = data["domains"]
|
|
self.assertIn("fleet", domains)
|
|
self.assertIn("dm", domains)
|
|
self.assertIn("job", domains)
|
|
self.assertIn("docs", domains)
|
|
|
|
|
|
class TestDocsLookupCLI(unittest.TestCase):
|
|
def run_cmd(self, args: list) -> subprocess.CompletedProcess:
|
|
return subprocess.run(
|
|
[sys.executable, str(DOCS_LOOKUP)] + args,
|
|
capture_output=True,
|
|
text=True
|
|
)
|
|
|
|
def test_cli_overview_json(self):
|
|
res = self.run_cmd(["overview", "--json"])
|
|
self.assertEqual(res.returncode, 0, res.stderr)
|
|
out = json.loads(res.stdout)
|
|
self.assertEqual(out.get("name"), "docs_internal")
|
|
|
|
def test_cli_search(self):
|
|
res = self.run_cmd(["search", "work order", "--json"])
|
|
self.assertEqual(res.returncode, 0, res.stderr)
|
|
out = json.loads(res.stdout)
|
|
self.assertGreater(out.get("count", 0), 0)
|
|
|
|
def test_cli_surfaces(self):
|
|
res = self.run_cmd(["surfaces", "jobs", "--json"])
|
|
self.assertEqual(res.returncode, 0, res.stderr)
|
|
out = json.loads(res.stdout)
|
|
self.assertIn("jobs", out)
|
|
self.assertEqual(out["jobs"]["dom_tab_selector"], ".tab-btn[data-tab=\"jobs\"]")
|
|
|
|
def test_cli_sentence(self):
|
|
res = self.run_cmd(["sentence", "work_order", "--json"])
|
|
self.assertEqual(res.returncode, 0, res.stderr)
|
|
out = json.loads(res.stdout)
|
|
self.assertIn("work_order", out)
|
|
self.assertEqual(out["work_order"]["protocol_tag"], "[WO:<id>]")
|
|
|
|
def test_cli_regex_test(self):
|
|
res = self.run_cmd([
|
|
"regex", "result",
|
|
"--test", "[RESULT 7fce46e0] OK 14 allowlisted ops confirmed",
|
|
"--json"
|
|
])
|
|
self.assertEqual(res.returncode, 0, res.stderr)
|
|
out = json.loads(res.stdout)
|
|
self.assertIn("result", out)
|
|
eval_res = out.get("test_evaluation", {})
|
|
self.assertTrue(eval_res.get("matched"))
|
|
self.assertEqual(eval_res["named_groups"]["job_id"], "7fce46e0")
|
|
self.assertEqual(eval_res["named_groups"]["status"], "OK")
|
|
|
|
def test_cli_parse(self):
|
|
res = self.run_cmd([
|
|
"parse",
|
|
"[WO:7fce46e0] [from super] Audit exec endpoints — Check /api/stats",
|
|
"--json"
|
|
])
|
|
self.assertEqual(res.returncode, 0, res.stderr)
|
|
out = json.loads(res.stdout)
|
|
self.assertGreaterEqual(out.get("matched_patterns_count", 0), 2)
|
|
pattern_keys = [m["pattern_key"] for m in out.get("matches", [])]
|
|
self.assertIn("work_order", pattern_keys)
|
|
|
|
def test_super_cli_integration(self):
|
|
res = subprocess.run(
|
|
[sys.executable, str(SUPER_CLI), "docs", "surfaces", "dashboard", "--json"],
|
|
capture_output=True,
|
|
text=True
|
|
)
|
|
self.assertEqual(res.returncode, 0, res.stderr)
|
|
out = json.loads(res.stdout)
|
|
self.assertIn("dashboard", out)
|
|
|
|
|
|
class TestLookupEngine(unittest.TestCase):
|
|
def test_lookup_engine_imports_and_hot_reload(self):
|
|
sys.path.insert(0, str(REPO_ROOT / "bin"))
|
|
import lookup_engine
|
|
|
|
# Verify compiled patterns
|
|
pats = lookup_engine.get_all_compiled_patterns()
|
|
self.assertGreaterEqual(len(pats), 14)
|
|
self.assertIn("result", pats)
|
|
self.assertIn("verb", pats)
|
|
|
|
res_re = lookup_engine.get_result_regex()
|
|
self.assertIsNotNone(res_re.search("[RESULT 7fce46e0] OK Done"))
|
|
|
|
verb_re = lookup_engine.get_verb_regex()
|
|
self.assertIsNotNone(verb_re.search("[CLAIM 7fce46e0]"))
|
|
|
|
# Verify soft validation
|
|
val_ok, kind, hint = lookup_engine.validate_outbound_sentence("[WO:7fce46e0] [from super] Audit — OK")
|
|
self.assertTrue(val_ok)
|
|
self.assertEqual(kind, "work_order")
|
|
|
|
val_fail, kind_f, hint_f = lookup_engine.validate_outbound_sentence("[WO:invalid] bad format")
|
|
self.assertFalse(val_fail)
|
|
self.assertEqual(kind_f, "work_order")
|
|
self.assertIn("Expected format", hint_f)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|