from __future__ import annotations import hashlib import json import re import tempfile import unittest from pathlib import Path from hub_case import HubCase from hub_support import HUB, PAGE from kvb_orchestrator import artifacts, exchange, icons from kvb_orchestrator.errors import ToolError VENDOR = HUB / "vendor" TEMPLATES = HUB / "templates" VENDORED = { "bootstrap-icons/bootstrap-icons.svg": ("fe7b61303ee8b160aa2d596dabbb70f4ffeaf8ea24891004e9b1d0d2877638b1", "Copyright (c) 2019-2024 The Bootstrap Authors"), "3d-force-graph/3d-force-graph.min.js": ("13088340aad408156064fd120c3bc3a47c93a3cc45b8f132d6284b20e2223be9", "Copyright (c) 2017 Vasco Asturiano"), "three/three.module.js": ("9052042d676cb0fdc1ddfefe193053f34b7ac0513a616fdac4535d49987812ea", "Copyright © 2010-2026 three.js authors"), "three/three.core.js": ("9edde002b066a9a05676a6127f67735b62baf399bdea529f2f7e31657da769e6", "Copyright © 2010-2026 three.js authors"), } VERSIONS = {"3d-force-graph/3d-force-graph.min.js": "Version 1.80.1", "three/three.core.js": "REVISION = '186'"} class TestHubIcons(HubCase): def test_the_page_inlines_exactly_the_icons_it_uses_from_the_vendored_sprite(self): html = self.index() sprite = re.search(r'', html, re.DOTALL) self.assertIsNotNone(sprite) inlined = re.findall(r'', sprite.group(1)) used = icons.used(TEMPLATES) self.assertGreater(len(used), 40) self.assertEqual(inlined, used) known = icons.symbols(VENDOR / icons.SPRITE) self.assertLessEqual(set(used), set(known)) for name in ("command", "gear"): with self.subTest(header=name): self.assertIn(f'', html) self.assertIsNone(re.search(r'(?:src|href)="(?:https?:)?//', html)) self.assertNotIn("cdn.", html) def test_an_icon_missing_from_the_sprite_stops_the_start(self): with tempfile.TemporaryDirectory() as folder: (Path(folder) / "page.js").write_text('const x = icon("bi-not-an-icon-ever");', encoding="utf-8") with self.assertRaisesRegex(ToolError, "not-an-icon-ever"): icons.sheet(VENDOR, Path(folder)) def test_vendored_files_are_pinned_and_carry_their_mit_licence(self): for name, (digest, holder) in VENDORED.items(): with self.subTest(file=name): path = VENDOR / name self.assertEqual(hashlib.sha256(path.read_bytes()).hexdigest(), digest) licence = (path.parent / "LICENSE").read_text(encoding="utf-8") self.assertIn("MIT License", licence) self.assertIn(holder, licence) for name, marker in VERSIONS.items(): with self.subTest(version=name): self.assertIn(marker, (VENDOR / name).read_text(encoding="utf-8")) def test_the_vendor_route_serves_only_its_allowlist_under_the_plain_csp(self): status, headers, body = self.get("/vendor/3d-force-graph.min.js") self.assertEqual((status, headers["content-type"]), (200, "text/javascript; charset=utf-8")) self.assertEqual(body, (VENDOR / "3d-force-graph" / "3d-force-graph.min.js").read_bytes()) self.assertEqual(headers["content-security-policy"], exchange.PLAIN_CSP) for path in ("/vendor/bootstrap-icons.svg", "/vendor/LICENSE", "/vendor/..%2Fserver.py", "/vendor/three"): with self.subTest(path=path): self.assertEqual(self.get(path)[0], 404) class TestHubArtifacts(HubCase): def test_the_tree_carries_artefact_leaves_overlaps_merges_and_lanes(self): root = self.write_tree() demo = f"{root}/worktrees/github.com/me/demo" feature, done = f"{demo}/feature", f"{demo}/feature-done" register = { "work/done": {"title": "Done", "status": "running", "worker": "Skills session", "workdir": done, "artifacts": [{"ref": "image:demo:feature-done", "added": "2026-11-06T10:00:00+00:00"}, {"ref": "volume:demo_data", "added": "2026-11-06T10:05:00+00:00"}]}, "work/loose": {"title": "Loose", "status": "running", "artifacts": [{"ref": "log:/tmp/loose.log", "added": "2026-11-06T11:00:00+00:00"}]}, "work/plain": {"title": "Plain", "status": "running"}, } history = [{"id": "q1", "kind": "test-nostack", "workdir": done, "start": 1000.0, "end": 1010.0, "exit": 0}, {"id": "q2", "kind": "deploy", "workdir": done, "start": 1100.0, "end": 1110.0, "exit": 0}] self.write_council({"rasci.json": register, "queue-history.jsonl": self.jsonl(history)}) touched = {} for path, branch, files in ((feature, "feature", ["a.py", "b.py"]), (done, "feature/done", ["a.py", "c.py"])): result = self.tool("review_request", {**self.arguments(f"feat: {branch}", "", self.page(f"{branch.replace('/', '-')}.html", PAGE)), "worktree": path, "branch": branch}) touched[branch] = result["structuredContent"]["id"] self.httpd.hub.reviews.stored(touched[branch], ".json").write_text(json.dumps({"files": [{"path": name} for name in files]}), encoding="utf-8") self.httpd.hub.reviews.decide(touched["feature/done"], {"status": "approved", "proof": None}) status, _, body = self.get("/api/tree") self.assertEqual(status, 200, body) data = json.loads(body) repo = next(entry for entry in data["repos"] if entry["name"] == "github.com/me/demo") branches = {entry["name"]: entry for entry in repo["branches"]} leaves = [(leaf["task"], leaf["kind"], leaf["name"]) for leaf in branches["feature/done"]["worktrees"][0]["artefacts"]] self.assertEqual(leaves, [("work/done", "image", "demo:feature-done"), ("work/done", "volume", "demo_data")]) self.assertEqual(branches["feature"]["worktrees"][0]["artefacts"], []) self.assertEqual([(task["task"], [leaf["ref"] for leaf in task["artefacts"]]) for task in data["tasks"]], [("work/loose", ["log:/tmp/loose.log"])]) self.assertEqual(repo["overlaps"], [{"a": "feature", "b": "feature/done", "count": 1, "files": ["a.py"]}]) self.assertEqual(repo["merges"], [{"from": "feature/done", "into": "main", "by": "#12"}, {"from": "feature/kept", "into": "main", "by": "ancestor"}]) self.assertEqual((branches["feature"]["files"], branches["feature/done"]["files"]), (2, 2)) kinds = [(event["kind"], event["ok"]) for event in branches["feature/done"]["events"]] self.assertEqual(kinds, [("gate", True), ("review", None), ("gate", True), ("commit", True)]) self.assertEqual([event["kind"] for event in branches["feature"]["events"]], ["review", "gate"]) self.assertEqual(data["cleanups"], []) def test_a_lane_counts_push_commands_and_failed_gates(self): path = "/w/demo/feature" history = [{"kind": "gate", "workdir": f"{path}/sub", "start": 10.0, "end": 20.0, "exit": 2}, {"kind": "test", "workdir": "/w/other", "end": 30.0, "exit": 0}] jobs = [ {"command": "gipsn", "directory": path, "status": "done", "finished": 40.0}, {"command": "make test", "directory": path, "status": "done", "finished": 50.0}, {"command": "git push origin feature", "directory": path, "status": "queued", "created_ts": 60.0}, {"command": "pushd x", "directory": path, "status": "done", "finished": 70.0}, ] events = artifacts.lane([path], [], history, jobs) self.assertEqual([(event["ts"], event["kind"], event["ok"]) for event in events], [(20.0, "gate", False), (40.0, "push", True), (60.0, "push", None)]) if __name__ == "__main__": unittest.main()