Files
skills/tests/test_hub_nav.py
T
Kevin Veen-BirkenbachandClaude Opus 5.5 c1310c3322 feat(orchestrator): tasks tab with register, inbox and queue
Add a Tasks tab that lists the kvb-org-rasci register next to the inbox
and the heavy-job queue, in list, board and RASCI matrix views, each
task with its roles, WSJF, open work and the merge flow of its worktree.

- Task priority pins and WSJF estimates are signed with the operator's
  passkey and applied by the runner through rasci.py; the hub verifies
  the signature itself before it queues a job or logs the change, the
  runner verifies it again and runs only the checked argv, from /, with
  a pinned PATH and a 30 s process-group limit.
- The runner refuses task changes into a council that is a symlink, not
  owned by its uid or given by a relative path, and while the installed
  council.py, nofollow.py or rasci.py differ from the tree.
- Only tasks the hub can sign whole are editable (task slug and length).
- Page lookups keyed by register text use Object.hasOwn, so a status or
  party named like an Object.prototype member cannot break the views.
- The inbox screen renders through the tasks screen, so its task filter
  shows with "Wartet auf Operator" pressed.

Tests: 604 unit tests OK, e2e 15/15 OK (Chromium and Firefox).

🎯 Fix: 93 %
💥 Break: 5 %
💸 Cost: 100 %
⏱️ Time: –
🛡️ Security: 100 %
🏆 Score: 88
🚦 Gate: ✅ 90/5, cost and time <= 100, security 100 %

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-07 23:43:36 +00:00

90 lines
5.0 KiB
Python

from __future__ import annotations
import json
import subprocess
import unittest
from hub_support import NAV
NAV_PROBE = """
const N = require(process.argv[1]);
const hashes = ["", "#stream", "#stream/inbox", "#stream/rules", "#stream/nope", "#agents/chord", "#agents/3d", "#agents", "#agents/nope",
"#artifacts/tree", "#artifacts/orbit", "#graph", "#graph/swimlane", "#graph/orbit", "#tree", "#bogus", "#agents%2Fswimlane",
"#agents/hierarchy", "#artifacts/hierarchy", "#graph/hierarchy", "#tree/hierarchy", "#stream/radio", "#radio",
"#tasks", "#tasks/board", "#tasks/rasci", "#tasks/orbit", "#agents/board", "#tasks/hierarchy"];
console.log(JSON.stringify({
tabs: N.TABS.map(([id, label]) => [id, label, (N.SUBS[id] || []).map(([sub, name]) => [sub, name])]),
views: N.VIEWS.map(([id, label]) => [id, label]),
taskViews: N.TASK_VIEWS.map(([id, label]) => [id, label]),
agents: N.viewsOf("agents").map(([id, label]) => [id, label]),
artifacts: N.viewsOf("artifacts").map(([id, label]) => [id, label]),
tasks: N.viewsOf("tasks").map(([id, label]) => [id, label]),
icon: N.ICONS.hierarchy,
routes: hashes.map((hash) => N.route(hash)),
hashes: [["stream", null], ["inbox", null], ["rules", null], ["radio", null], ["agents", "chord"], ["artifacts", "3d"], ["tasks", "board"]].map(([screen, view]) => N.hashOf(screen, view)),
tabOf: ["stream", "inbox", "rules", "radio", "agents", "artifacts", "tasks"].map((screen) => N.tabOf(screen)),
}));
"""
class TestNavModule(unittest.TestCase):
@classmethod
def setUpClass(cls):
done = subprocess.run(["node", "-e", NAV_PROBE, str(NAV)], capture_output=True, text=True, timeout=60, check=True)
cls.data = json.loads(done.stdout)
def test_the_navigation_has_four_entries_and_stream_has_inbox_rules_and_radio(self):
stream = ["stream", "Stream", [["inbox", "Inbox"], ["rules", "Regeln"], ["radio", "Radio"]]]
self.assertEqual(self.data["tabs"], [stream, ["agents", "Agents", []], ["artifacts", "Artifacts", []], ["tasks", "Tasks", []]])
def test_agents_and_artifacts_offer_five_views_in_order_and_tasks_three(self):
shared = [["orbit", "Orbit"], ["chord", "Chord"], ["swimlane", "Swimlane"], ["tree", "Tree"], ["3d", "3D"]]
self.assertEqual((self.data["views"], self.data["artifacts"]), (shared, shared))
self.assertEqual((self.data["taskViews"], self.data["tasks"]), ([["list", "Liste"], ["board", "Board"], ["rasci", "RASCI"]],) * 2)
def test_only_agents_add_the_hierarchy_after_3d(self):
self.assertEqual(self.data["agents"][-2:], [["3d", "3D"], ["hierarchy", "Hierarchie"]])
self.assertEqual(self.data["icon"], "bi-diagram-2")
def test_hashes_route_and_the_old_ones_redirect(self):
stream = {"screen": "stream", "view": None, "redirect": None}
self.assertEqual(self.data["routes"], [
stream,
stream,
{"screen": "inbox", "view": None, "redirect": None},
{"screen": "rules", "view": None, "redirect": None},
stream,
{"screen": "agents", "view": "chord", "redirect": None},
{"screen": "agents", "view": "3d", "redirect": None},
{"screen": "agents", "view": None, "redirect": None},
{"screen": "agents", "view": None, "redirect": None},
{"screen": "artifacts", "view": "tree", "redirect": None},
{"screen": "artifacts", "view": "orbit", "redirect": None},
{"screen": "agents", "view": "orbit", "redirect": "#agents/orbit"},
{"screen": "agents", "view": "swimlane", "redirect": "#agents/swimlane"},
{"screen": "agents", "view": "orbit", "redirect": "#agents/orbit"},
{"screen": "artifacts", "view": "tree", "redirect": "#artifacts/tree"},
stream,
{"screen": "agents", "view": "swimlane", "redirect": None},
{"screen": "agents", "view": "hierarchy", "redirect": None},
{"screen": "artifacts", "view": None, "redirect": None},
{"screen": "agents", "view": "hierarchy", "redirect": "#agents/hierarchy"},
{"screen": "artifacts", "view": "tree", "redirect": "#artifacts/tree"},
{"screen": "radio", "view": None, "redirect": None},
stream,
{"screen": "tasks", "view": None, "redirect": None},
{"screen": "tasks", "view": "board", "redirect": None},
{"screen": "tasks", "view": "rasci", "redirect": None},
{"screen": "tasks", "view": None, "redirect": None},
{"screen": "agents", "view": None, "redirect": None},
{"screen": "tasks", "view": None, "redirect": None},
])
def test_every_screen_has_its_own_hash_and_tab(self):
self.assertEqual(self.data["hashes"], ["#stream", "#stream/inbox", "#stream/rules", "#stream/radio", "#agents/chord", "#artifacts/3d", "#tasks/board"])
self.assertEqual(self.data["tabOf"], ["stream", "stream", "stream", "stream", "agents", "artifacts", "tasks"])
if __name__ == "__main__":
unittest.main()