Merge feature/workbench-w3-artifacts into unify/skills-20261112
Conflicts kept both sides: runner.py docstring (W2 board changes and W3 gate changes), signing.py constants (W2 STATUSES and W3 GATE_*), index.html.j2 styles (W1 diff and W3 ancestry). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Kevin Veen-BirkenbachandClaude Opus 5.5 committed
commit
f5b2352f19
33 files changed
+1997
-106
No files matched your search
@@ -0,0 +1,273 @@
|
||||
{
|
||||
"generated_at": "2026-11-12T11:14:56+00:00",
|
||||
"host_pid_namespace": true,
|
||||
"delete_ready": false,
|
||||
"delete_reason": "no GitHub token in the Secret Service",
|
||||
"repos": [
|
||||
{
|
||||
"path": "@ROOT@/repos/github.com/anon/web",
|
||||
"remote": "git@github.com:anon/web.git",
|
||||
"default": "main",
|
||||
"missing": false,
|
||||
"known": true,
|
||||
"forge": "unavailable: no GitHub token in the Secret Service",
|
||||
"scanned_at": "2026-11-12T11:14:56+00:00",
|
||||
"warnings": [
|
||||
"forge unavailable: no GitHub token in the Secret Service"
|
||||
],
|
||||
"items": [
|
||||
{
|
||||
"branch": "chore/ancestor-local",
|
||||
"worktree": null,
|
||||
"head": null,
|
||||
"detached": false,
|
||||
"locked": null,
|
||||
"worktree_state": null,
|
||||
"staged": null,
|
||||
"unstaged": null,
|
||||
"untracked": null,
|
||||
"upstream": null,
|
||||
"ahead": null,
|
||||
"behind": null,
|
||||
"gone": false,
|
||||
"local_sha": "1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b",
|
||||
"remotes": {},
|
||||
"compose_projects": [],
|
||||
"images": [],
|
||||
"state": "merged-clean",
|
||||
"merged": true,
|
||||
"dirty": 0,
|
||||
"merged_by": "ancestor",
|
||||
"merged_into": "origin/main",
|
||||
"unpushed": null,
|
||||
"last_activity": 1783873160,
|
||||
"size_bytes": null,
|
||||
"skip": "forge unavailable",
|
||||
"actions": [
|
||||
"local-branch"
|
||||
],
|
||||
"reason": "merged by ancestor",
|
||||
"result": null,
|
||||
"flow": null
|
||||
},
|
||||
{
|
||||
"branch": "fix/remote-merged",
|
||||
"worktree": null,
|
||||
"head": null,
|
||||
"detached": false,
|
||||
"locked": null,
|
||||
"worktree_state": null,
|
||||
"staged": null,
|
||||
"unstaged": null,
|
||||
"untracked": null,
|
||||
"upstream": "origin/fix/remote-merged",
|
||||
"ahead": 0,
|
||||
"behind": 0,
|
||||
"gone": false,
|
||||
"local_sha": "2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c",
|
||||
"remotes": {
|
||||
"origin": "2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c"
|
||||
},
|
||||
"compose_projects": [],
|
||||
"images": [],
|
||||
"state": "merged-clean",
|
||||
"merged": true,
|
||||
"dirty": 0,
|
||||
"merged_by": "ancestor",
|
||||
"merged_into": "origin/main",
|
||||
"unpushed": null,
|
||||
"last_activity": 1783873160,
|
||||
"size_bytes": null,
|
||||
"skip": "forge unavailable",
|
||||
"actions": [
|
||||
"local-branch",
|
||||
"remote-branch:origin"
|
||||
],
|
||||
"reason": "merged by ancestor",
|
||||
"result": null,
|
||||
"flow": null
|
||||
},
|
||||
{
|
||||
"branch": "feature/open",
|
||||
"worktree": null,
|
||||
"head": null,
|
||||
"detached": false,
|
||||
"locked": null,
|
||||
"worktree_state": null,
|
||||
"staged": null,
|
||||
"unstaged": null,
|
||||
"untracked": null,
|
||||
"upstream": null,
|
||||
"ahead": null,
|
||||
"behind": null,
|
||||
"gone": false,
|
||||
"local_sha": "3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d",
|
||||
"remotes": {},
|
||||
"compose_projects": [],
|
||||
"images": [],
|
||||
"state": "unmerged",
|
||||
"merged": false,
|
||||
"dirty": 0,
|
||||
"merged_by": null,
|
||||
"merged_into": null,
|
||||
"unpushed": 2,
|
||||
"last_activity": 1783873160,
|
||||
"size_bytes": null,
|
||||
"skip": "forge unavailable",
|
||||
"actions": [],
|
||||
"reason": "unmerged: 2 commits not in origin/main",
|
||||
"result": null,
|
||||
"flow": null
|
||||
},
|
||||
{
|
||||
"branch": "main",
|
||||
"worktree": "@ROOT@/repos/github.com/anon/web",
|
||||
"head": "4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e",
|
||||
"detached": false,
|
||||
"locked": null,
|
||||
"worktree_state": "clean",
|
||||
"staged": 0,
|
||||
"unstaged": 0,
|
||||
"untracked": 0,
|
||||
"upstream": "origin/main",
|
||||
"ahead": 0,
|
||||
"behind": 0,
|
||||
"gone": false,
|
||||
"local_sha": "4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e",
|
||||
"remotes": {
|
||||
"origin": "4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e"
|
||||
},
|
||||
"compose_projects": [],
|
||||
"images": [],
|
||||
"state": "protected",
|
||||
"merged": true,
|
||||
"dirty": 0,
|
||||
"merged_by": null,
|
||||
"merged_into": null,
|
||||
"unpushed": null,
|
||||
"last_activity": 1783873160,
|
||||
"size_bytes": null,
|
||||
"skip": null,
|
||||
"actions": [],
|
||||
"reason": "default or protected branch",
|
||||
"result": null,
|
||||
"flow": null
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"path": "@ROOT@/repos/git.example.org/anon/notes",
|
||||
"remote": "ssh://git@git.example.org/anon/notes.git",
|
||||
"default": "master",
|
||||
"missing": false,
|
||||
"known": true,
|
||||
"forge": "none",
|
||||
"scanned_at": "2026-11-12T11:14:56+00:00",
|
||||
"warnings": [],
|
||||
"items": [
|
||||
{
|
||||
"branch": "topic/ancestor-local",
|
||||
"worktree": null,
|
||||
"head": null,
|
||||
"detached": false,
|
||||
"locked": null,
|
||||
"worktree_state": null,
|
||||
"staged": null,
|
||||
"unstaged": null,
|
||||
"untracked": null,
|
||||
"upstream": null,
|
||||
"ahead": null,
|
||||
"behind": null,
|
||||
"gone": false,
|
||||
"local_sha": "5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f",
|
||||
"remotes": {},
|
||||
"compose_projects": [],
|
||||
"images": [],
|
||||
"state": "merged-clean",
|
||||
"merged": true,
|
||||
"dirty": 0,
|
||||
"merged_by": "ancestor",
|
||||
"merged_into": "origin/master",
|
||||
"unpushed": null,
|
||||
"last_activity": 1783873160,
|
||||
"size_bytes": null,
|
||||
"skip": null,
|
||||
"actions": [
|
||||
"local-branch"
|
||||
],
|
||||
"reason": "merged by ancestor",
|
||||
"result": null,
|
||||
"flow": null
|
||||
},
|
||||
{
|
||||
"branch": "topic/patch-local",
|
||||
"worktree": null,
|
||||
"head": null,
|
||||
"detached": false,
|
||||
"locked": null,
|
||||
"worktree_state": null,
|
||||
"staged": null,
|
||||
"unstaged": null,
|
||||
"untracked": null,
|
||||
"upstream": null,
|
||||
"ahead": null,
|
||||
"behind": null,
|
||||
"gone": false,
|
||||
"local_sha": "6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a",
|
||||
"remotes": {},
|
||||
"compose_projects": [],
|
||||
"images": [],
|
||||
"state": "merged-clean",
|
||||
"merged": true,
|
||||
"dirty": 0,
|
||||
"merged_by": "patch",
|
||||
"merged_into": "origin/master",
|
||||
"unpushed": null,
|
||||
"last_activity": 1783873160,
|
||||
"size_bytes": null,
|
||||
"skip": null,
|
||||
"actions": [
|
||||
"local-branch"
|
||||
],
|
||||
"reason": "merged by patch",
|
||||
"result": null,
|
||||
"flow": null
|
||||
},
|
||||
{
|
||||
"branch": "master",
|
||||
"worktree": "@ROOT@/repos/git.example.org/anon/notes",
|
||||
"head": "7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b",
|
||||
"detached": false,
|
||||
"locked": null,
|
||||
"worktree_state": "clean",
|
||||
"staged": 0,
|
||||
"unstaged": 0,
|
||||
"untracked": 0,
|
||||
"upstream": "origin/master",
|
||||
"ahead": 0,
|
||||
"behind": 0,
|
||||
"gone": false,
|
||||
"local_sha": "7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b",
|
||||
"remotes": {
|
||||
"origin": "7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b"
|
||||
},
|
||||
"compose_projects": [],
|
||||
"images": [],
|
||||
"state": "protected",
|
||||
"merged": true,
|
||||
"dirty": 0,
|
||||
"merged_by": null,
|
||||
"merged_into": null,
|
||||
"unpushed": null,
|
||||
"last_activity": 1783873160,
|
||||
"size_bytes": null,
|
||||
"skip": null,
|
||||
"actions": [],
|
||||
"reason": "default or protected branch",
|
||||
"result": null,
|
||||
"flow": null
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -1818,7 +1818,7 @@ class TestOrchestratorE2E(HubCase):
|
||||
self.assertIn(["twig", "true"], by_hash["#agents/tree"]["spawn"])
|
||||
self.assertIn(["twig", "false"], by_hash["#agents/tree"]["spawn"])
|
||||
artifacts = by_hash["#artifacts/tree"]
|
||||
self.assertEqual(artifacts["deletes"], [["Delete all merged", True, "Delete all merged blocked: Secret Service locked"], ["Delete", True, "Delete blocked: Secret Service locked"]])
|
||||
self.assertEqual(artifacts["deletes"], [["Delete all merged", True, "Delete all merged blocked: Secret Service locked"]])
|
||||
for state in ("merged by #12 into main", "upstream gone", "protected", "1 commits unpushed", "clean", "not committed: 0 staged, 0 changed, 1 new", "gc skips: ignored files beyond known caches: notes.txt"):
|
||||
with self.subTest(state=state):
|
||||
self.assertIn(state, artifacts["states"])
|
||||
@@ -2125,6 +2125,7 @@ class TestOrchestratorE2E(HubCase):
|
||||
def test_chromium_deletes_a_merged_branch_after_dry_run_and_signature(self):
|
||||
tree = json.loads(self.tree_file.read_text(encoding="utf-8"))
|
||||
tree["delete_ready"], tree["delete_reason"] = True, ""
|
||||
next(item for item in tree["repos"][0]["items"] if item["branch"] == "feature/done").update(worktree=None, worktree_state=None, actions=["local-branch", "remote-branch"], compose_projects=[], images=[])
|
||||
self.tree_file.write_text(json.dumps(tree), encoding="utf-8")
|
||||
demo = f"{self.root}/repos/github.com/me/demo"
|
||||
Path(demo).mkdir(parents=True)
|
||||
@@ -2132,8 +2133,7 @@ class TestOrchestratorE2E(HubCase):
|
||||
fake.parent.mkdir()
|
||||
fake.write_text(FAKE_PKGMGR.format(python=sys.executable), encoding="utf-8")
|
||||
fake.chmod(0o755)
|
||||
worktree = f"{self.root}/worktrees/github.com/me/demo/feature-done"
|
||||
(fake.parent / "plan.json").write_text(json.dumps([{"branch": "feature/done", "worktree": worktree, "local_sha": DONE_TIP, "head": DONE_TIP, "state": "merged-clean", "skip": None}]), encoding="utf-8")
|
||||
(fake.parent / "plan.json").write_text(json.dumps([{"branch": "feature/done", "worktree": None, "local_sha": DONE_TIP, "head": DONE_TIP, "state": "merged-clean", "skip": None, "actions": ["local-branch"]}]), encoding="utf-8")
|
||||
following = json.loads(json.dumps(tree))
|
||||
following["repos"][0]["items"] = [item for item in following["repos"][0]["items"] if item["branch"] != "feature/done"]
|
||||
(fake.parent / "next-tree.json").write_text(json.dumps(following), encoding="utf-8")
|
||||
@@ -2158,7 +2158,7 @@ class TestOrchestratorE2E(HubCase):
|
||||
self.assertEqual(opened["title"], "Delete feature/done")
|
||||
self.assertEqual(len(opened["dry"]), 1)
|
||||
self.assertTrue(opened["dry"][0].startswith("feature/done @ 5b1e6fd3a7c0: merged-clean, merged by #12"), opened["dry"])
|
||||
self.assertIn("stays: remote-branch (local only)", opened["dry"][0])
|
||||
self.assertIn("stays: remote-branch (local only; the host refuses a plan that deletes it)", opened["dry"][0])
|
||||
self.assertEqual((confirmed["result"], confirmed["text"]), ("done", "feature/donedeleted"), confirmed)
|
||||
self.assertEqual(confirmed["toasts"], ["Delete: donefeature/done: deleted"])
|
||||
self.assertEqual(after["past"], ["done", "deleted"])
|
||||
@@ -2167,8 +2167,8 @@ class TestOrchestratorE2E(HubCase):
|
||||
self.assertEqual([[(item["branch"], item["status"]) for item in result["items"]] for result in written], [[("feature/done", "done")]])
|
||||
calls = [json.loads(line)["argv"] for line in (fake.parent / "calls.jsonl").read_text(encoding="utf-8").splitlines()]
|
||||
self.assertEqual(calls, [
|
||||
["branch", "cleanup", "--json", "--local", "--branch", "feature/done"],
|
||||
["branch", "cleanup", "--apply", "--json", "--local", "--branch", "feature/done"],
|
||||
["branch", "cleanup", "--json", "--branch", "feature/done"],
|
||||
["branch", "cleanup", "--apply", "--json", "--branch", "feature/done"],
|
||||
["branch", "tree", "--format", "json", "--output", str(self.tree_file)],
|
||||
])
|
||||
|
||||
|
||||
@@ -0,0 +1,215 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
HERE = Path(__file__).resolve().parent
|
||||
sys.path[:0] = [str(HERE.parent), str(HERE)]
|
||||
|
||||
import test_orchestrator_e2e as base
|
||||
from hub_case import HubCase
|
||||
from hub_support import REPO_ROOT, iso
|
||||
|
||||
OUT = Path(os.environ.get("E2E_OUT", "/tmp/workbench-w3-shots"))
|
||||
GATES_SPEC = importlib.util.spec_from_file_location("kvb_gates_for_e2e", REPO_ROOT / "skills" / "kvb-check-gate" / "scripts" / "gates.py")
|
||||
gates = importlib.util.module_from_spec(GATES_SPEC)
|
||||
GATES_SPEC.loader.exec_module(gates)
|
||||
KEPT_TIP = "0c9d8e7f6a5b4c3d2e1f0a9b8c7d6e5f4a3b2c1d"
|
||||
CLEAN_ALL = """(async () => {@HELPERS@
|
||||
const every = await until(() => document.querySelector('#main button[aria-label="Delete all merged"]'), 8000);
|
||||
await until(() => document.querySelector("#main svg.ancestry"), 8000);
|
||||
const token = all("#main .delete-all span.state").map(name);
|
||||
const store = document.querySelector('#main button[aria-label="Store token"]');
|
||||
const before = every ? [every.disabled, every.title] : null;
|
||||
every.click();
|
||||
const sheet = await until(() => document.querySelector("#sheet:not([hidden]) li.dry-run") && document.getElementById("sheet"), 8000);
|
||||
const rows = all("li.dry-run", sheet).map((row) => [text(row.querySelector("code")), row.dataset.token || (row.dataset.held ? "held" : null), text(row.querySelector("pre")).split("\\n")[0]]);
|
||||
const confirm = all("#sheet button").find((button) => name(button) === "Delete");
|
||||
confirm.click();
|
||||
const outcome = await until(() => sheet.querySelector("[data-cleanup]") || all("#sheet p.muted").find((node) => text(node).startsWith("Not signed")), 15000);
|
||||
const tokenClick = store ? (store.click(), await until(() => all("#toasts .toast").map(text).find((line) => line.startsWith("Token")), 8000)) : null;
|
||||
return {before, token, store: store ? [store.title, store.getAttribute("aria-label")] : null, rows, outcome: outcome ? (outcome.dataset.cleanup ? "queued" : "unsigned") : null, tokenClick,
|
||||
wide: document.documentElement.scrollWidth > document.documentElement.clientWidth};
|
||||
})()"""
|
||||
ANCESTRY = """(async () => {@HELPERS@
|
||||
const svg = await until(() => document.querySelector('#main svg.ancestry[aria-label="Ancestry of github.com/me/demo"]'), 8000);
|
||||
if (!svg) {
|
||||
return {error: "no ancestry"};
|
||||
}
|
||||
const nodes = all("g.ancestry-node", svg).map((node) => [node.dataset.branch, Number(node.dataset.generation)]);
|
||||
const edges = all("path.ancestry-edge", svg).map((edge) => [edge.dataset.parent, edge.dataset.child, edge.dataset.source, getComputedStyle(edge).strokeDasharray !== "none"]);
|
||||
const marks = (branch) => {
|
||||
const row = all("#main .branch-row").find((entry) => text(entry.querySelector(".branch-name strong")) === branch);
|
||||
return row ? all(".gate-mark", row).map((button) => [button.dataset.action, button.dataset.level, button.classList.contains("inherited"), button.title, Number(getComputedStyle(button).opacity) < 1]) : null;
|
||||
};
|
||||
return {nodes, edges, done: marks("feature/done"), main: marks("main"), wide: document.documentElement.scrollWidth > document.documentElement.clientWidth};
|
||||
})()"""
|
||||
STAGE = """(async () => {@HELPERS@
|
||||
const svg = await until(() => document.querySelector("#main svg.ancestry"), 8000);
|
||||
svg.scrollIntoView({block: "center", inline: "center"});
|
||||
await sleep(200);
|
||||
const center = (branch) => {
|
||||
const node = svg.querySelector(`g.ancestry-node[data-branch="${CSS.escape(branch)}"] circle`);
|
||||
const box = node.getBoundingClientRect();
|
||||
return {x: box.left + box.width / 2, y: box.top + box.height / 2, node};
|
||||
};
|
||||
const from = center("feature/kept");
|
||||
const onto = center("feature");
|
||||
from.node.dispatchEvent(new PointerEvent("pointerdown", {bubbles: true, button: 0, clientX: from.x, clientY: from.y, pointerType: "mouse", isPrimary: true}));
|
||||
document.dispatchEvent(new PointerEvent("pointerup", {bubbles: true, button: 0, clientX: onto.x, clientY: onto.y, pointerType: "mouse", isPrimary: true}));
|
||||
const pending = await until(() => document.querySelector('#main path.ancestry-edge.pending[data-child="feature/kept"][data-parent="feature"]'), 4000);
|
||||
const row = all("#main .branch-row").find((entry) => text(entry.querySelector(".branch-name strong")) === "feature");
|
||||
const commit = () => row.isConnected ? row.querySelector('.gate-mark[data-action="commit"]') : all("#main .branch-row").find((entry) => text(entry.querySelector(".branch-name strong")) === "feature").querySelector('.gate-mark[data-action="commit"]');
|
||||
commit().click();
|
||||
await sleep(100);
|
||||
commit().click();
|
||||
const staged = await until(() => document.querySelector('#main .gate-mark.pending[data-action="commit"][data-level="quick"]'), 4000);
|
||||
const apply = await until(() => document.querySelector('#main button[aria-label="Apply"]'), 4000);
|
||||
const count = apply ? text(apply.parentElement.querySelector(".count")) : null;
|
||||
apply.click();
|
||||
const toast = await until(() => all("#toasts .toast").map(text).find((line) => line.startsWith("Gates") || line.startsWith("Not signed") || line.includes("passkey")), 15000);
|
||||
return {pending: Boolean(pending), staged: Boolean(staged), title: staged ? staged.title : null, count, apply: apply ? [apply.title, apply.getAttribute("aria-label")] : null, toast,
|
||||
left: all("#main .gate-mark.pending").length + all("#main path.ancestry-edge.pending").length};
|
||||
})()"""
|
||||
TOKENLESS = """(async () => {@HELPERS@
|
||||
const every = await until(() => document.querySelector('#main button[aria-label="Delete all merged"]'), 8000);
|
||||
await until(() => document.querySelector("#main .branch-row"), 8000);
|
||||
const store = document.querySelector('#main button[aria-label="Store token"]');
|
||||
const rows = all("#main .branch-row").map((row) => [text(row.querySelector(".branch-name strong")), Boolean(row.querySelector('span.state[aria-label="token needed"]')), Boolean(row.querySelector('button[aria-label="Delete"]'))]);
|
||||
return {every: [every.disabled, every.title], token: all("#main .delete-all span.state").map(name), store: store ? store.getAttribute("aria-label") : null, rows};
|
||||
})()"""
|
||||
AGENTS = """(async () => {@HELPERS@
|
||||
const svg = await until(() => document.querySelector('#main svg.ancestry[aria-label="Ancestry: operator, chair, sessions, subagents"]'), 8000);
|
||||
if (!svg) {
|
||||
return {error: "no agent ancestry"};
|
||||
}
|
||||
await until(() => svg.querySelector('g.ancestry-node[data-kind="twig"]') || document.querySelector('#main svg.ancestry g[data-kind="twig"]'), 8000);
|
||||
const fresh = document.querySelector("#main svg.ancestry");
|
||||
const nodes = all("g.ancestry-node", fresh).map((node) => [node.dataset.kind, Number(node.dataset.generation), node.getBoundingClientRect().top]);
|
||||
return {nodes, edges: all("path.ancestry-edge", fresh).length, list: all("#main .spawn-tree li").length, wide: document.documentElement.scrollWidth > document.documentElement.clientWidth};
|
||||
})()"""
|
||||
|
||||
|
||||
def probe(template: str) -> str:
|
||||
return template.replace("@HELPERS@", base.HELPERS)
|
||||
|
||||
|
||||
@unittest.skipUnless(shutil.which(os.environ.get("CHROMIUM", "chromium")), "no Chromium")
|
||||
class TestWorkbenchArtifacts(HubCase):
|
||||
drive = base.TestOrchestratorE2E.drive
|
||||
authenticator = base.TestOrchestratorE2E.authenticator
|
||||
subagents = base.TestOrchestratorE2E.subagents
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
OUT.mkdir(parents=True, exist_ok=True)
|
||||
now = time.time()
|
||||
root = self.write_tree()
|
||||
data = json.loads(self.tree_file.read_text(encoding="utf-8"))
|
||||
next(item for item in data["repos"][0]["items"] if item["branch"] == "feature/kept")["skip"] = None
|
||||
self.tree_file.write_text(json.dumps(data), encoding="utf-8")
|
||||
self.repo = f"{root}/repos/github.com/me/demo"
|
||||
db = gates.connect(self.kept / "gates.db")
|
||||
db.executemany("INSERT INTO rules VALUES (?, ?, ?, ?, ?)", [(self.repo, "main", "commit", "test", 0.0), (self.repo, "feature/done", "pr", "e2e", 0.0)])
|
||||
db.executemany("INSERT INTO parents VALUES (?, ?, ?, ?)", [(self.repo, "feature", "main", "recorded"), (self.repo, "feature/done", "feature", "derived"), (self.repo, "feature/done", "feature/kept", "operator"), (self.repo, "feature/kept", "main", "derived")])
|
||||
db.commit()
|
||||
db.close()
|
||||
self.claim("s1", "Skills session")
|
||||
self.write_council({
|
||||
"work/s2/claim.json": {"office": "work/s2", "session": "s2", "runtime": "claude", "name": "Stage", "since": iso(now - 7200), "beat": now - 60, "work": "x"},
|
||||
"chair/claim.json": {"office": "chair", "session": "s1", "runtime": "claude", "name": "Skills session", "since": iso(now - 7200), "beat": now, "work": "allocate"},
|
||||
})
|
||||
self.subagents(now)
|
||||
self.base = f"http://{self.host}/"
|
||||
|
||||
def steps(self, browser: str) -> list[dict]:
|
||||
steps = []
|
||||
for width, height in base.WIDTHS:
|
||||
shot = str(OUT / f"{browser}-{width}")
|
||||
steps += [
|
||||
{"width": width, "height": height, "url": f"{self.base}?{width}#artifacts/tree", "wait": 1200, "probe": probe(ANCESTRY), "shot": f"{shot}-ancestry.png"},
|
||||
{"width": width, "height": height, "url": None, "wait": 0, "probe": probe(STAGE), "shot": f"{shot}-staged.png"},
|
||||
{"width": width, "height": height, "url": f"{self.base}?{width}-clean#artifacts/tree", "wait": 1200, "probe": probe(CLEAN_ALL), "shot": f"{shot}-clean-all.png"},
|
||||
{"width": width, "height": height, "url": f"{self.base}?{width}#agents/tree", "wait": 1200, "probe": probe(AGENTS), "shot": f"{shot}-agents-ancestry.png"},
|
||||
]
|
||||
return steps
|
||||
|
||||
def check(self, data: dict, browser: str, signs: bool) -> None:
|
||||
results = iter(data["results"])
|
||||
for width, _ in base.WIDTHS:
|
||||
ancestry, staged, clean, agents = next(results), next(results), next(results), next(results)
|
||||
with self.subTest(browser=browser, width=width):
|
||||
for found in (ancestry, agents):
|
||||
self.assertNotIn("error", found, found)
|
||||
nodes = dict(ancestry["nodes"])
|
||||
self.assertEqual(nodes, {"main": 0, "feature": 1, "feature/kept": 1, "feature/done": 2})
|
||||
self.assertEqual(sorted(ancestry["edges"]), [
|
||||
["feature", "feature/done", "derived", True], ["feature/kept", "feature/done", "operator", False], ["main", "feature", "recorded", False], ["main", "feature/kept", "derived", True]])
|
||||
self.assertEqual([mark[:3] for mark in ancestry["done"]], [["commit", "test", True], ["merge", "none", True], ["push", "none", True], ["pr", "e2e", False]])
|
||||
self.assertEqual([mark[3] for mark in ancestry["done"]][:1], ["commit: test · from main"])
|
||||
self.assertEqual([mark[4] for mark in ancestry["done"]], [True, True, True, False])
|
||||
self.assertEqual(ancestry["main"][0][:3], ["commit", "test", False])
|
||||
self.assertEqual((staged["pending"], staged["staged"], staged["count"]), (True, True, "2"), staged)
|
||||
self.assertEqual(staged["apply"], ["Apply 2 gate changes with one passkey signature", "Apply"])
|
||||
self.assertEqual(staged["left"], 0 if signs else 2, staged)
|
||||
self.assertEqual(clean["before"], [False, "Delete all merged: 1 branches after dry run and signature"])
|
||||
self.assertEqual(clean["token"], ["1 held by a worktree: use branch gc"])
|
||||
self.assertEqual(clean["store"], ["Store token: add the operator row that stores the GitHub token in the Secret Service", "Store token"])
|
||||
self.assertEqual([row[:2] for row in clean["rows"]], [["feature/kept", None], ["feature/done", "held"]])
|
||||
self.assertEqual(clean["outcome"], "queued" if signs else "unsigned")
|
||||
self.assertTrue(clean["tokenClick"].startswith("Token: operator row added"), clean)
|
||||
kinds = {}
|
||||
for kind, generation, _ in agents["nodes"]:
|
||||
kinds.setdefault(kind, set()).add(generation)
|
||||
self.assertEqual((kinds["operator"], kinds["chair"]), ({0}, {1}))
|
||||
self.assertTrue(all(generation >= 2 for generation in kinds.get("session", set()) | kinds.get("subagent", set()) | kinds.get("twig", set())), agents)
|
||||
self.assertIn("twig", kinds)
|
||||
tops = {kind: min(top for k, _, top in agents["nodes"] if k == kind) for kind in kinds}
|
||||
self.assertLess(tops["operator"], tops["chair"])
|
||||
self.assertGreater(agents["list"], 0)
|
||||
self.assertFalse(any(found["wide"] for found in (ancestry, clean, agents)), (ancestry["wide"], clean["wide"], agents["wide"]))
|
||||
jobs = self.runner_jobs()
|
||||
gate_jobs = [job["argv"] for job in jobs if job.get("kind") == "gate"]
|
||||
expected = [["parents", "add", "--repo", self.repo, "--branch", "feature/kept", "--parent", "feature"], ["rule", "set", "--repo", self.repo, "--branch", "feature", "--action", "commit", "--level", "quick"]]
|
||||
self.assertEqual(sorted(gate_jobs), sorted(expected * len(base.WIDTHS)) if signs else [])
|
||||
spooled = [json.loads(path.read_text(encoding="utf-8")) for path in sorted((self.cleanup / "jobs").glob("*.json"))]
|
||||
self.assertEqual([[item["branch"] for item in job["items"]] for job in spooled], [["feature/kept"]] * len(base.WIDTHS) if signs else [])
|
||||
self.assertTrue(all(job["items"][0]["tip"] == KEPT_TIP for job in spooled))
|
||||
tokens = [args for script, args in self.scripts.calls if script.endswith("operator.py") and args[:2] == ["add", "pkgmgr-github-token"]]
|
||||
self.assertEqual(len(tokens), len(base.WIDTHS))
|
||||
|
||||
def test_chromium_shows_ancestry_in_both_tabs_stages_a_parent_and_a_gate_and_signs_clean_all_merged_at_three_widths(self):
|
||||
data = self.drive(self.steps("chromium"), self.authenticator())
|
||||
(OUT / "chromium-artifacts.json").write_text(json.dumps(data, indent=1), encoding="utf-8")
|
||||
self.assertEqual([error for error in data["errors"] if "/favicon.ico" not in error], [])
|
||||
self.check(data, "chromium", True)
|
||||
|
||||
def test_chromium_marks_every_forge_unavailable_branch_token_needed_and_offers_store_token_on_the_real_tokenless_tree(self):
|
||||
self.write_fixture("branch-tree-tokenless.json")
|
||||
data = json.loads(self.tree_file.read_text(encoding="utf-8"))
|
||||
data["repos"] = [repo for repo in data["repos"] if repo["forge"] != "none"]
|
||||
self.tree_file.write_text(json.dumps(data), encoding="utf-8")
|
||||
width, height = base.WIDTHS[0]
|
||||
found = self.drive([{"width": width, "height": height, "url": f"{self.base}?tokenless#artifacts/tree", "wait": 1200, "probe": probe(TOKENLESS), "shot": str(OUT / "chromium-tokenless.png")}], self.authenticator())
|
||||
self.assertEqual([error for error in found["errors"] if "/favicon.ico" not in error], [])
|
||||
shown = found["results"][0]
|
||||
self.assertEqual(shown["every"], [True, "Delete all merged blocked: no GitHub token in the Secret Service"])
|
||||
self.assertEqual((shown["token"], shown["store"]), (["2 token needed"], "Store token"))
|
||||
self.assertEqual(sorted(shown["rows"]), [["chore/ancestor-local", True, False], ["feature/open", False, False], ["fix/remote-merged", True, False], ["main", False, False]])
|
||||
|
||||
@unittest.skipUnless(shutil.which(os.environ.get("FIREFOX", "firefox")), "no Firefox")
|
||||
def test_firefox_shows_ancestry_in_both_tabs_and_hands_nothing_over_unsigned_at_three_widths(self):
|
||||
data = self.drive(self.steps("firefox"), None, "bidi.mjs", os.environ.get("FIREFOX", "firefox"))
|
||||
(OUT / "firefox-artifacts.json").write_text(json.dumps(data, indent=1), encoding="utf-8")
|
||||
self.assertEqual([error for error in data["errors"] if "/favicon.ico" not in error], [])
|
||||
self.check(data, "firefox", False)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+7
-1
@@ -232,8 +232,14 @@ class HubCase(unittest.TestCase):
|
||||
path.write_text(text, encoding="utf-8")
|
||||
|
||||
def write_tree(self) -> Path:
|
||||
return self.write_fixture("branch-tree.json")
|
||||
|
||||
def write_fixture(self, fixture: str) -> Path:
|
||||
"""Args:
|
||||
fixture: the tree file under tests/ to write, @ROOT@ replaced.
|
||||
"""
|
||||
root = self.outside / "scan"
|
||||
sample = (Path(__file__).parent / "branch-tree.json").read_text(encoding="utf-8")
|
||||
sample = (Path(__file__).parent / fixture).read_text(encoding="utf-8")
|
||||
self.tree_file.write_text(sample.replace("@ROOT@", str(root)), encoding="utf-8")
|
||||
return root
|
||||
|
||||
|
||||
@@ -83,6 +83,8 @@ ESTIMATE = REPO_ROOT / "skills" / "kvb-org-queue-estimate" / "scripts" / "estima
|
||||
|
||||
|
||||
class FakeScripts:
|
||||
skills = REPO_ROOT / "skills"
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.calls: list[tuple[str, list[str]]] = []
|
||||
self.code = 0
|
||||
|
||||
@@ -0,0 +1,144 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import unittest
|
||||
|
||||
from hub_case import HubCase
|
||||
from hub_support import HUB, signing
|
||||
|
||||
ANCESTRY = HUB / "templates" / "ancestry.js"
|
||||
PAGE = HUB / "templates" / "page-ancestry.js"
|
||||
STYLE = HUB / "templates" / "style-ancestry.css"
|
||||
REPO = "/home/op/Repositories/demo"
|
||||
# Param: argv[1] is ancestry.js; prints the layouts, gate marks and argv the page builds.
|
||||
PROBE = """
|
||||
const A = require(process.argv[1]);
|
||||
const node = (id) => ({id, label: id});
|
||||
const shape = (plan) => ({
|
||||
rows: plan.nodes.map((entry) => [entry.id, entry.generation]),
|
||||
ys: Object.fromEntries(plan.nodes.map((entry) => [entry.id, entry.y])),
|
||||
xs: Object.fromEntries(plan.nodes.map((entry) => [entry.id, entry.x])),
|
||||
links: plan.links.map((link) => [link.parent, link.child, link.derived, link.y1 < link.y2]),
|
||||
size: [plan.width, plan.height],
|
||||
});
|
||||
const family = shape(A.layout(["main", "feature/a", "feature/b", "feature/ab", "lonely"].map(node), [
|
||||
{child: "feature/a", parent: "main", source: "recorded"},
|
||||
{child: "feature/b", parent: "main", source: "derived"},
|
||||
{child: "feature/ab", parent: "feature/a", source: "operator"},
|
||||
{child: "feature/ab", parent: "feature/b", source: "derived"},
|
||||
{child: "ghost", parent: "main", source: "recorded"},
|
||||
]));
|
||||
const looped = shape(A.layout(["x", "y", "z"].map(node), [{child: "x", parent: "y"}, {child: "y", parent: "x"}, {child: "z", parent: "z"}]));
|
||||
const chain = shape(A.layout(["main", "mid", "leaf"].map(node), [{child: "leaf", parent: "main"}, {child: "leaf", parent: "mid"}, {child: "mid", parent: "main"}]));
|
||||
const roots = Array.from({length: 20}, (_, index) => `b${String(index).padStart(2, "0")}`);
|
||||
const wide = shape(A.layout([...roots, "child"].map(node), [{child: "child", parent: "b00"}]));
|
||||
const shown = {resolved: {commit: {level: "test", source: "feature/a"}, merge: {level: "quality-high", source: "main"}, push: {level: "none", source: null}, pr: {level: "e2e", source: "feature/a"}}, rules: {commit: "test", pr: "e2e"}};
|
||||
console.log(JSON.stringify({
|
||||
family, looped, chain, wide, spacing: A.SPACING,
|
||||
marks: A.gateMarks("feature/a", shown),
|
||||
bare: A.gateMarks("lonely", null),
|
||||
cycle: [null, "none", "quick", "test", "quality-high", "e2e"].map(A.nextRule),
|
||||
argv: [A.ruleArgv("/r", "b", "push", "test"), A.ruleArgv("/r", "b", "push", null), A.parentArgv("/r", "b", "p", false), A.parentArgv("/r", "b", "p", true)],
|
||||
icons: A.LEVEL_ICONS,
|
||||
}));
|
||||
"""
|
||||
|
||||
|
||||
class TestAncestryLayout(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
done = subprocess.run(["node", "-e", PROBE, str(ANCESTRY)], capture_output=True, text=True, timeout=60, check=True)
|
||||
cls.data = json.loads(done.stdout)
|
||||
|
||||
def test_generations_run_top_down_and_a_node_with_several_parents_sits_below_the_deepest(self):
|
||||
family = self.data["family"]
|
||||
self.assertEqual(family["rows"], [["main", 0], ["feature/a", 1], ["feature/b", 1], ["feature/ab", 2], ["lonely", 0]])
|
||||
self.assertLess(family["ys"]["main"], family["ys"]["feature/a"])
|
||||
self.assertLess(family["ys"]["feature/a"], family["ys"]["feature/ab"])
|
||||
self.assertEqual(family["ys"]["feature/a"], family["ys"]["feature/b"])
|
||||
self.assertEqual(self.data["chain"]["rows"], [["main", 0], ["mid", 1], ["leaf", 2]])
|
||||
|
||||
def test_every_parent_edge_is_drawn_downwards_and_only_derived_ones_are_marked_dashed(self):
|
||||
self.assertEqual(self.data["family"]["links"], [
|
||||
["main", "feature/a", False, True],
|
||||
["main", "feature/b", True, True],
|
||||
["feature/a", "feature/ab", False, True],
|
||||
["feature/b", "feature/ab", True, True],
|
||||
])
|
||||
|
||||
def test_a_child_sits_between_its_parents_and_unknown_or_looping_edges_do_not_break_the_layout(self):
|
||||
xs = self.data["family"]["xs"]
|
||||
self.assertTrue(min(xs["feature/a"], xs["feature/b"]) <= xs["feature/ab"] <= max(xs["feature/a"], xs["feature/b"]))
|
||||
looped = self.data["looped"]
|
||||
self.assertEqual(sorted(generation for _, generation in looped["rows"]), [0, 0, 1])
|
||||
self.assertEqual([link[:2] for link in looped["links"] if link[0] == link[1]], [])
|
||||
|
||||
def test_a_wide_generation_wraps_into_bands_and_the_next_generation_starts_below_them(self):
|
||||
wide, spacing = self.data["wide"], self.data["spacing"]
|
||||
self.assertEqual(wide["size"], [spacing["wrap"] * spacing["column"], 4 * spacing["row"]])
|
||||
roots = [name for name, generation in wide["rows"] if generation == 0]
|
||||
self.assertEqual(len(roots), 20)
|
||||
self.assertEqual(len({(wide["xs"][name], wide["ys"][name]) for name in roots}), 20)
|
||||
self.assertEqual(sorted({wide["ys"][name] for name in roots}), [spacing["row"] * band + spacing["row"] / 2 for band in range(3)])
|
||||
self.assertEqual(wide["ys"]["child"], 3.5 * spacing["row"])
|
||||
self.assertTrue(all(0 < x < wide["size"][0] for x in wide["xs"].values()))
|
||||
|
||||
|
||||
class TestGateMarks(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
done = subprocess.run(["node", "-e", PROBE, str(ANCESTRY)], capture_output=True, text=True, timeout=60, check=True)
|
||||
cls.data = json.loads(done.stdout)
|
||||
|
||||
def test_each_action_shows_its_resolved_level_with_own_rules_solid_and_inherited_faint_with_the_source(self):
|
||||
marks = {mark["action"]: mark for mark in self.data["marks"]}
|
||||
self.assertEqual(list(marks), ["commit", "merge", "push", "pr"])
|
||||
self.assertEqual([(marks[a]["level"], marks[a]["own"], marks[a]["inherited"]) for a in marks], [("test", True, False), ("quality-high", False, True), ("none", False, True), ("e2e", True, False)])
|
||||
self.assertEqual([marks[a]["title"] for a in marks], ["commit: test · own rule", "merge: quality-high · from main", "push: none · no rule", "pr: e2e · own rule"])
|
||||
self.assertEqual([marks[a]["icon"] for a in marks], ["bi-check2-circle", "bi-shield-check", "bi-circle", "bi-browser-chrome"])
|
||||
|
||||
def test_a_branch_gates_py_does_not_know_shows_none_inherited_for_every_action(self):
|
||||
self.assertEqual([(mark["level"], mark["inherited"], mark["source"]) for mark in self.data["bare"]], [("none", True, "")] * 4)
|
||||
|
||||
def test_a_click_cycles_the_own_rule_through_every_level_and_back_to_inherited(self):
|
||||
self.assertEqual(self.data["cycle"], ["none", "quick", "test", "quality-high", "e2e", None])
|
||||
|
||||
def test_the_staged_argv_is_exactly_what_signing_gate_argv_accepts(self):
|
||||
for argv in self.data["argv"]:
|
||||
with self.subTest(argv=argv):
|
||||
self.assertEqual(signing.gate_argv(argv), argv)
|
||||
self.assertEqual(self.data["argv"][1], ["rule", "drop", "--repo", "/r", "--branch", "b", "--action", "push"])
|
||||
|
||||
def test_the_level_icons_are_the_page_levels_of_gates_py(self):
|
||||
self.assertEqual(list(self.data["icons"]), list(signing.GATE_LEVELS))
|
||||
|
||||
|
||||
class TestAncestryAssets(unittest.TestCase):
|
||||
def test_the_layout_module_parses_on_its_own_and_derived_edges_are_dashed(self):
|
||||
done = subprocess.run(["node", "--check", str(ANCESTRY)], capture_output=True, text=True, timeout=60, check=False)
|
||||
self.assertEqual(done.returncode, 0, done.stderr)
|
||||
self.assertIn("svg .ancestry-edge.derived { stroke-dasharray: 5 5; }", STYLE.read_text(encoding="utf-8"))
|
||||
self.assertIn(".gate-mark.inherited { opacity: .4; }", STYLE.read_text(encoding="utf-8"))
|
||||
|
||||
def test_every_new_control_is_an_icon_with_title_and_aria_label(self):
|
||||
page = PAGE.read_text(encoding="utf-8")
|
||||
for control in re.findall(r'iconBtn\("(bi-[a-z0-9-]+)", "([^"]+)"', page):
|
||||
with self.subTest(control=control):
|
||||
self.assertTrue(control[1])
|
||||
self.assertIn('...named(`${mark.action} gate`, title)', page)
|
||||
self.assertIn('role: "button", "aria-label": node.label', page)
|
||||
|
||||
|
||||
class TestHubAncestry(HubCase):
|
||||
def test_the_page_ships_the_layout_module_and_the_gate_level_icons(self):
|
||||
html = self.index()
|
||||
self.assertIn(ANCESTRY.read_text(encoding="utf-8").strip(), html)
|
||||
for name in ("bi-circle", "bi-lightning-charge", "bi-check2-circle", "bi-shield-check", "bi-browser-chrome", "bi-key", "bi-key-fill", "bi-check2-all"):
|
||||
with self.subTest(name=name):
|
||||
self.assertIn(f'<symbol id="{name}"', html)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -69,6 +69,16 @@ class TestHubIcons(HubCase):
|
||||
|
||||
|
||||
class TestHubArtifacts(HubCase):
|
||||
def test_a_merge_pkgmgr_names_into_the_remote_tracking_default_points_at_the_default_branch_node(self):
|
||||
self.write_fixture("branch-tree-tokenless.json")
|
||||
status, _, body = self.get("/api/tree")
|
||||
self.assertEqual(status, 200, body)
|
||||
merges = {repo["name"]: repo["merges"] for repo in json.loads(body)["repos"]}
|
||||
self.assertEqual(merges, {
|
||||
"github.com/anon/web": [{"from": "chore/ancestor-local", "into": "main", "by": "ancestor"}, {"from": "fix/remote-merged", "into": "main", "by": "ancestor"}],
|
||||
"git.example.org/anon/notes": [{"from": "topic/ancestor-local", "into": "master", "by": "ancestor"}, {"from": "topic/patch-local", "into": "master", "by": "patch"}],
|
||||
})
|
||||
|
||||
def test_the_tree_carries_artefact_leaves_overlaps_merges_and_lanes(self):
|
||||
root = self.write_tree()
|
||||
demo = f"{root}/worktrees/github.com/me/demo"
|
||||
|
||||
+131
-19
@@ -1,8 +1,11 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import secrets
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
@@ -11,16 +14,18 @@ import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from hub_case import HubCase
|
||||
from hub_support import HUB, ORIGIN, key, signing, verifier
|
||||
from hub_support import GIT_ENV, HUB, ORIGIN, key, signing, verifier
|
||||
|
||||
CLEANUP = HUB / "client" / "cleanup.py"
|
||||
CLEANUP_SPEC = importlib.util.spec_from_file_location("kvb_orchestrator_cleanup_runner", CLEANUP)
|
||||
host = importlib.util.module_from_spec(CLEANUP_SPEC)
|
||||
CLEANUP_SPEC.loader.exec_module(host)
|
||||
INSTALLED_PKGMGR = shutil.which("pkgmgr")
|
||||
DONE_TIP = "5b1e6fd3a7c0e2d94f8a61b3c2d7e9f0a1b2c3d4"
|
||||
OTHER_TIP = "1111111111111111111111111111111111111111"
|
||||
# Param: format with python=<interpreter>; plan.json beside it is the dry-run plan, next-tree.json the rewritten tree.
|
||||
# Param: format with python=<interpreter>; parses argv as pkgmgr's branch_cmd.py does (argparse exits 2 on any other flag) and prints the plan as gc.py _render does, {"host_pid_namespace", "repos": [{"repo", "default", "merged_into", "warnings", "items"}]}; plan.json beside it is the plan's items, next-tree.json the rewritten tree, outcome.txt the result --apply reports per item (done without it), exit 0 also for a skipped item.
|
||||
FAKE_PKGMGR = """#!{python}
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
@@ -29,22 +34,48 @@ from pathlib import Path
|
||||
here = Path(__file__).parent
|
||||
with (here / "calls.jsonl").open("a", encoding="utf-8") as log:
|
||||
log.write(json.dumps({{"argv": sys.argv[1:], "cwd": os.getcwd()}}) + "\\n")
|
||||
if sys.argv[1:3] == ["branch", "tree"]:
|
||||
parser = argparse.ArgumentParser(prog="pkgmgr")
|
||||
branch = parser.add_subparsers(dest="command", required=True).add_parser("branch").add_subparsers(dest="subcommand", required=True)
|
||||
tree = branch.add_parser("tree")
|
||||
tree.add_argument("--format", choices=("json", "yaml"), default="json")
|
||||
tree.add_argument("--output")
|
||||
cleanup = branch.add_parser("cleanup")
|
||||
for flag in ("--apply", "--json", "--all", "--sizes"):
|
||||
cleanup.add_argument(flag, action="store_true")
|
||||
for option in ("--branch", "--exclude", "--merged-into"):
|
||||
cleanup.add_argument(option, action="append", default=[])
|
||||
cleanup.add_argument("--min-age", type=float)
|
||||
args = parser.parse_args()
|
||||
if args.subcommand == "tree":
|
||||
following = here / "next-tree.json"
|
||||
rewritten = following.read_text(encoding="utf-8") if following.exists() else json.dumps({{"rewritten": True}})
|
||||
Path(sys.argv[sys.argv.index("--output") + 1]).write_text(rewritten, encoding="utf-8")
|
||||
Path(args.output).write_text(rewritten, encoding="utf-8")
|
||||
raise SystemExit(0)
|
||||
print(json.dumps({{"items": json.loads((here / "plan.json").read_text(encoding="utf-8"))}}))
|
||||
items = json.loads((here / "plan.json").read_text(encoding="utf-8"))
|
||||
if args.apply:
|
||||
outcome = here / "outcome.txt"
|
||||
result = outcome.read_text(encoding="utf-8") if outcome.exists() else "done"
|
||||
items = [{{**item, "result": result}} for item in items]
|
||||
plan = {{"host_pid_namespace": True, "repos": [{{"repo": os.getcwd(), "default": "origin/main", "merged_into": [], "warnings": [], "items": items}}]}}
|
||||
print(json.dumps(plan) if args.json else "\\n".join(str(item["branch"]) for item in items))
|
||||
"""
|
||||
|
||||
|
||||
class TestHubCleanup(HubCase):
|
||||
def ready(self, ready: bool) -> dict:
|
||||
def ready(self, ready: bool, held: bool) -> dict:
|
||||
"""Args:
|
||||
ready: delete_ready of the tree file.
|
||||
held: keep feature/done in its worktree; otherwise it is a local
|
||||
branch without one.
|
||||
"""
|
||||
root = self.write_tree()
|
||||
data = json.loads(self.tree_file.read_text(encoding="utf-8"))
|
||||
data["delete_ready"], data["delete_reason"] = ready, "" if ready else "Secret Service locked"
|
||||
done = next(item for item in data["repos"][0]["items"] if item["branch"] == "feature/done")
|
||||
if not held:
|
||||
done.update(worktree=None, worktree_state=None, actions=["local-branch", "remote-branch"], compose_projects=[], images=[])
|
||||
self.tree_file.write_text(json.dumps(data), encoding="utf-8")
|
||||
return signing.cleanup_item(f"{root}/repos/github.com/me/demo", "feature/done", f"{root}/worktrees/github.com/me/demo/feature-done", DONE_TIP)
|
||||
return signing.cleanup_item(f"{root}/repos/github.com/me/demo", "feature/done", done["worktree"] or "", DONE_TIP)
|
||||
|
||||
def branches(self) -> dict:
|
||||
status, _, body = self.get("/api/tree")
|
||||
@@ -57,27 +88,24 @@ class TestHubCleanup(HubCase):
|
||||
return status, body
|
||||
|
||||
def test_the_tree_offers_a_delete_only_for_merged_clean_branches_without_skip_reason(self):
|
||||
item = self.ready(False)
|
||||
item = self.ready(False, False)
|
||||
branches = self.branches()
|
||||
done = branches["feature/done"]
|
||||
self.assertEqual(done["cleanup"], item)
|
||||
self.assertEqual((done["cleanup"], done["held"]), (item, ""))
|
||||
self.assertEqual(done["plan"], [
|
||||
"feature/done @ 5b1e6fd3a7c0: merged-clean, merged by #12",
|
||||
f"removes the worktree {item['worktree']}",
|
||||
"deletes the local branch feature/done",
|
||||
"stays: remote-branch (local only)",
|
||||
"stops the Compose stack demo-feature-done without -v",
|
||||
"removes the image demo:feature-done",
|
||||
"stays: remote-branch (local only; the host refuses a plan that deletes it)",
|
||||
])
|
||||
self.assertEqual((done["merged_by"], done["merged_into"], done["gone"]), ("#12", "main", True))
|
||||
self.assertEqual({name: entry["cleanup"] for name, entry in branches.items() if name != "feature/done"}, {"feature": None, "feature/kept": None, "main": None})
|
||||
self.assertEqual(branches["feature/kept"]["skip"], "ignored files beyond known caches: notes.txt")
|
||||
|
||||
def test_signing_a_delete_needs_delete_ready_and_exactly_an_offered_item(self):
|
||||
item = self.ready(False)
|
||||
item = self.ready(False, False)
|
||||
status, body = self.sign([item])
|
||||
self.assertEqual((status, body.decode()), (400, "delete blocked: Secret Service locked\n"))
|
||||
item = self.ready(True)
|
||||
item = self.ready(True, False)
|
||||
kept = {**item, "branch": "feature/kept", "worktree": "", "tip": "0c9d8e7f6a5b4c3d2e1f0a9b8c7d6e5f4a3b2c1d"}
|
||||
for name, items in (("changed tip", [{**item, "tip": OTHER_TIP}]), ("skip reason", [kept]), ("pattern", [{**item, "branch": "feature/*"}]), ("twice", [item, item]), ("extra key", [{**item, "command": "rm -rf /"}])):
|
||||
with self.subTest(name=name):
|
||||
@@ -86,8 +114,37 @@ class TestHubCleanup(HubCase):
|
||||
self.assertEqual(status, 200, body)
|
||||
self.assertEqual(json.loads(body)["payload"]["items"], [signing.item_hash(item)])
|
||||
|
||||
def test_a_branch_held_by_a_worktree_offers_no_delete_and_names_branch_gc(self):
|
||||
item = self.ready(True, True)
|
||||
done = self.branches()["feature/done"]
|
||||
held = f"held by worktree {item['worktree']}: pkgmgr branch cleanup skips it, use branch gc (kvb-git-branch-gc, pkgmgr gc)"
|
||||
self.assertEqual((done["cleanup"], done["held"], done["plan"][1]), (None, held, held))
|
||||
status, body = self.sign([item])
|
||||
self.assertEqual((status, body.decode()), (400, f"feature/done in {item['repository']} is no merged-clean branch without skip reason at that tip in the tree file\n"))
|
||||
page = (HUB / "templates" / "page-tree.js").read_text(encoding="utf-8")
|
||||
self.assertIn('if (entry.held) {\n return h("span", {class: "hint held", title: entry.held}, chip("bi-folder-symlink", entry.held, "neutral"));', page)
|
||||
self.assertIn("deleteDialog(items, locked, held)", page)
|
||||
self.assertIn("held by a worktree: not deleted here", (HUB / "templates" / "page-cleanup.js").read_text(encoding="utf-8"))
|
||||
|
||||
@unittest.skipUnless(INSTALLED_PKGMGR, "no pkgmgr installed")
|
||||
def test_every_volume_text_states_the_compose_down_argv_of_the_installed_pkgmgr(self):
|
||||
found = subprocess.run([str(Path(INSTALLED_PKGMGR).parent / "python"), "-I", "-c", "import pkgmgr.actions.branch.gc as gc; print(gc.__file__)"], capture_output=True, text=True, timeout=60, check=True)
|
||||
calls = [node for node in ast.walk(ast.parse(Path(found.stdout.strip()).read_text(encoding="utf-8"))) if isinstance(node, ast.Call) and getattr(node.func, "id", None) == "_docker" and node.args and isinstance(node.args[0], ast.List)]
|
||||
downs = [["docker", *(part.value if isinstance(part, ast.Constant) else "{project}" for part in call.args[0].elts)] for call in calls if any(isinstance(part, ast.Constant) and part.value == "down" for part in call.args[0].elts)]
|
||||
self.assertEqual(len(downs), 1, downs)
|
||||
command = " ".join(downs[0]).format(project="demo-feature-done")
|
||||
self.ready(True, True)
|
||||
plan = self.branches()["feature/done"]["plan"]
|
||||
self.assertIn(f"branch gc would run {command}, which deletes the volumes of demo-feature-done", plan)
|
||||
self.assertIn("-v", downs[0])
|
||||
dialog = (HUB / "templates" / "page-cleanup.js").read_text(encoding="utf-8")
|
||||
self.assertIn("never touches a worktree, Compose stack or volume", dialog)
|
||||
self.assertIn("its docker compose down -v deletes the stack's volumes", dialog)
|
||||
texts = [*plan, dialog, (HUB / "kvb_orchestrator" / "tree.py").read_text(encoding="utf-8")]
|
||||
self.assertFalse([text for text in texts if "without -v" in text])
|
||||
|
||||
def test_a_signed_delete_becomes_a_job_without_any_command(self):
|
||||
item = self.ready(True)
|
||||
item = self.ready(True, False)
|
||||
signer = self.pin()
|
||||
proof = signer.proof(self.prepare([item])["payload"], self.origin)
|
||||
self.assertEqual(self.post_json("/api/cleanup", {"items": [item], "proof": None}, self.guard(self.host))[0], 400)
|
||||
@@ -127,7 +184,7 @@ class TestHostCleanup(unittest.TestCase):
|
||||
self.calls = self.fake.parent / "calls.jsonl"
|
||||
self.tree = base / "tree" / "branch-tree.json"
|
||||
self.tree.parent.mkdir()
|
||||
self.tree.write_text(json.dumps({"repos": [{"path": str(self.repo), "known": True}, {"path": str(base / "stranger"), "known": False}]}), encoding="utf-8")
|
||||
self.tree.write_text(json.dumps({"delete_ready": True, "repos": [{"path": str(self.repo), "known": True}, {"path": str(base / "stranger"), "known": False}]}), encoding="utf-8")
|
||||
self.approvers = base / "config" / "approvers.json"
|
||||
self.trust = verifier.Trust(approvers=self.approvers, replay=base / "replay" / "cleanup.json", origin=ORIGIN)
|
||||
self.signer = key(-7, "main")
|
||||
@@ -154,7 +211,7 @@ class TestHostCleanup(unittest.TestCase):
|
||||
def test_a_signed_merged_clean_item_is_dry_run_then_deleted_by_exact_name(self):
|
||||
result, calls = self.run_job([self.item], self.proof([self.item], self.signer), [self.entry()])
|
||||
self.assertEqual([(entry["status"], entry["reason"]) for entry in result["items"]], [("done", "")])
|
||||
selector = ["--json", "--local", "--branch", "feature/done"]
|
||||
selector = ["--json", "--branch", "feature/done"]
|
||||
self.assertEqual(calls, [
|
||||
{"argv": ["branch", "cleanup", *selector], "cwd": str(self.repo)},
|
||||
{"argv": ["branch", "cleanup", "--apply", *selector], "cwd": str(self.repo)},
|
||||
@@ -163,6 +220,22 @@ class TestHostCleanup(unittest.TestCase):
|
||||
self.assertEqual(json.loads(self.tree.read_text(encoding="utf-8")), {"rewritten": True})
|
||||
self.assertEqual(result["tree"], {"code": 0, "error": ""})
|
||||
|
||||
def test_an_apply_that_exits_0_but_skips_or_reports_no_result_is_failed_with_pkgmgrs_reason_and_keeps_the_tree(self):
|
||||
outcome = self.fake.parent / "outcome.txt"
|
||||
before = self.tree.read_text(encoding="utf-8")
|
||||
cases = (("skipped", "skipped: changed since plan (dirty: 1 paths)", "skipped: changed since plan (dirty: 1 paths)"), ("no result", "", "pkgmgr reported no result for feature/done"))
|
||||
for name, result, reason in cases:
|
||||
with self.subTest(name=name):
|
||||
outcome.write_text(result, encoding="utf-8")
|
||||
done, calls = self.run_job([self.item], self.proof([self.item], self.signer), [self.entry()])
|
||||
self.assertEqual([(entry["status"], entry["reason"], entry["code"]) for entry in done["items"]], [("failed", reason, 0)])
|
||||
self.assertEqual([call["argv"][:3] for call in calls], [["branch", "cleanup", "--json"], ["branch", "cleanup", "--apply"]])
|
||||
self.assertNotIn("tree", done)
|
||||
self.assertEqual(self.tree.read_text(encoding="utf-8"), before)
|
||||
outcome.write_text("done; image kept: demo:feature-done", encoding="utf-8")
|
||||
done, _ = self.run_job([self.item], self.proof([self.item], self.signer), [self.entry()])
|
||||
self.assertEqual([(entry["status"], entry["reason"]) for entry in done["items"]], [("done", "")])
|
||||
|
||||
def test_every_doubtful_item_is_refused_and_changes_nothing(self):
|
||||
pattern = {**self.item, "branch": "feature/*"}
|
||||
option = {**self.item, "branch": "--all"}
|
||||
@@ -181,13 +254,14 @@ class TestHostCleanup(unittest.TestCase):
|
||||
("skip reason", [self.item], self.proof([self.item], self.signer), [self.entry(skip="ignored files")], "gc skip reason", 1),
|
||||
("other worktree", [self.item], self.proof([self.item], self.signer), [self.entry(worktree="/elsewhere")], "worktree changed", 1),
|
||||
("pattern match in pkgmgr", [self.item], self.proof([self.item], self.signer), [self.entry(), self.entry(branch="feature/done-too")], "other branches too", 1),
|
||||
("remote branch delete", [self.item], self.proof([self.item], self.signer), [self.entry(actions=["local-branch", "remote-branch:origin"])], "a signed delete is local only", 1),
|
||||
)
|
||||
for name, items, proof, plan, reason, dry_runs in cases:
|
||||
with self.subTest(name=name):
|
||||
result, calls = self.run_job(items, proof, plan)
|
||||
self.assertEqual([entry["status"] for entry in result["items"]], ["refused"])
|
||||
self.assertIn(reason, result["items"][0]["reason"])
|
||||
self.assertEqual([call["argv"] for call in calls], [["branch", "cleanup", "--json", "--local", "--branch", items[0]["branch"]]] * dry_runs)
|
||||
self.assertEqual([call["argv"] for call in calls], [["branch", "cleanup", "--json", "--branch", items[0]["branch"]]] * dry_runs)
|
||||
self.assertNotIn("tree", result)
|
||||
|
||||
def test_a_signature_deletes_once_and_a_malformed_job_nothing(self):
|
||||
@@ -217,6 +291,44 @@ class TestHostCleanup(unittest.TestCase):
|
||||
self.assertEqual((self.results / f"{job_id}.json").stat().st_mode & 0o777, 0o600)
|
||||
|
||||
|
||||
@unittest.skipUnless(INSTALLED_PKGMGR, "no pkgmgr installed")
|
||||
class TestInstalledPkgmgrContract(unittest.TestCase):
|
||||
def git(self, cwd: Path, *args: str) -> str:
|
||||
return subprocess.run(["git", *args], cwd=cwd, env={**os.environ, **GIT_ENV}, check=True, capture_output=True, text=True).stdout.strip()
|
||||
|
||||
def test_the_installed_pkgmgr_takes_the_runners_dry_run_argv_and_prints_a_plan_the_runner_parses_and_accepts(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
base = Path(tmp)
|
||||
origin, repo = base / "origin.git", base / "repo"
|
||||
self.git(base, "init", "-q", "--bare", "-b", "main", str(origin))
|
||||
self.git(base, "init", "-q", "-b", "main", str(repo))
|
||||
self.git(repo, "remote", "add", "origin", origin.as_uri())
|
||||
self.git(repo, "commit", "-q", "--allow-empty", "-m", "one")
|
||||
self.git(repo, "checkout", "-q", "-b", "side")
|
||||
self.git(repo, "commit", "-q", "--allow-empty", "-m", "side")
|
||||
tip = self.git(repo, "rev-parse", "HEAD")
|
||||
self.git(repo, "checkout", "-q", "main")
|
||||
self.git(repo, "merge", "-q", "--no-ff", "-m", "merge side", "side")
|
||||
self.git(origin, "fetch", "-q", str(repo), "main:main")
|
||||
self.git(repo, "fetch", "-q", "origin")
|
||||
self.git(repo, "remote", "set-head", "origin", "main")
|
||||
code, stdout, stderr = host.run(Path(INSTALLED_PKGMGR), host.plan_argv("side", False), str(repo))
|
||||
self.assertEqual(code, 0, stderr)
|
||||
plan = host.plan_items(stdout)
|
||||
self.assertIsNotNone(plan, stdout[:500])
|
||||
self.assertEqual([(entry["branch"], entry["state"], entry["merged_by"], entry["merged_into"], entry["skip"], entry["actions"]) for entry in plan], [("side", "merged-clean", "ancestor", "origin/main", None, ["local-branch"])])
|
||||
self.assertIsNone(host.mismatch(plan[0], {"tip": tip, "worktree": ""}))
|
||||
self.assertTrue(signing.token_free(plan[0], "main"))
|
||||
self.assertEqual(self.git(repo, "rev-parse", "refs/heads/side"), tip)
|
||||
held = base / "held"
|
||||
self.git(repo, "worktree", "add", "-q", str(held), "side")
|
||||
code, stdout, stderr = host.run(Path(INSTALLED_PKGMGR), host.plan_argv("side", False), str(repo))
|
||||
self.assertEqual(code, 0, stderr)
|
||||
plan = host.plan_items(stdout)
|
||||
self.assertEqual([(entry["worktree"], entry["actions"], bool(entry["skip"])) for entry in plan], [(os.path.realpath(held), [], True)], plan)
|
||||
self.assertIn("gc skip reason", host.mismatch(plan[0], {"tip": tip, "worktree": os.path.realpath(held)}))
|
||||
|
||||
|
||||
class TestHostCleanupGitConfig(unittest.TestCase):
|
||||
def test_pkgmgr_runs_no_hook_the_repository_config_names(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
|
||||
@@ -0,0 +1,241 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import secrets
|
||||
import shlex
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from hub_case import HubCase
|
||||
from hub_support import GIT_ENV, HUB, ORIGIN, REPO_ROOT, key, signing, verifier
|
||||
|
||||
CLEANUP_SPEC = importlib.util.spec_from_file_location("kvb_orchestrator_cleanup_token_runner", HUB / "client" / "cleanup.py")
|
||||
host = importlib.util.module_from_spec(CLEANUP_SPEC)
|
||||
CLEANUP_SPEC.loader.exec_module(host)
|
||||
OPERATOR_SPEC = importlib.util.spec_from_file_location("kvb_operator_for_token", REPO_ROOT / "skills" / "kvb-org-operator" / "scripts" / "operator.py")
|
||||
operator_script = importlib.util.module_from_spec(OPERATOR_SPEC)
|
||||
OPERATOR_SPEC.loader.exec_module(operator_script)
|
||||
KEPT_TIP = "0c9d8e7f6a5b4c3d2e1f0a9b8c7d6e5f4a3b2c1d"
|
||||
DONE_TIP = "5b1e6fd3a7c0e2d94f8a61b3c2d7e9f0a1b2c3d4"
|
||||
# Param: format with python=<interpreter>; plans one branch of the real repository in its cwd as pkgmgr 2.4.0 `branch cleanup` does without a GitHub token: argv parsed as branch_cmd.py does (exit 2 on any other flag), merged_into the remote HEAD ref (gc.py symbolic-ref --short), an origin on github.com skips every item with "forge unavailable" (gc.py _skip), actions local-branch plus remote-branch:<remote> (gc.py _actions), --apply deletes only an item without skip (Item.eligible) and the output is gc.py _render's {"host_pid_namespace", "repos": [{"repo", "default", "merged_into", "warnings", "items"}]}.
|
||||
SCRATCH_PKGMGR = """#!{python}
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
|
||||
parser = argparse.ArgumentParser(prog="pkgmgr")
|
||||
branch = parser.add_subparsers(dest="command", required=True).add_parser("branch").add_subparsers(dest="subcommand", required=True)
|
||||
tree = branch.add_parser("tree")
|
||||
tree.add_argument("--format", choices=("json", "yaml"), default="json")
|
||||
tree.add_argument("--output")
|
||||
cleanup = branch.add_parser("cleanup")
|
||||
for flag in ("--apply", "--json", "--all", "--sizes"):
|
||||
cleanup.add_argument(flag, action="store_true")
|
||||
for option in ("--branch", "--exclude", "--merged-into"):
|
||||
cleanup.add_argument(option, action="append", default=[])
|
||||
cleanup.add_argument("--min-age", type=float)
|
||||
args = parser.parse_args()
|
||||
if args.subcommand == "tree":
|
||||
raise SystemExit(0)
|
||||
|
||||
|
||||
def git(*argv):
|
||||
return subprocess.run(["git", *argv], capture_output=True, text=True, check=False)
|
||||
|
||||
|
||||
name = args.branch[0]
|
||||
tip = git("rev-parse", "--verify", "refs/heads/" + name).stdout.strip()
|
||||
default = git("symbolic-ref", "--short", "refs/remotes/origin/HEAD").stdout.strip()
|
||||
merged = bool(tip) and git("merge-base", "--is-ancestor", tip, default).returncode == 0
|
||||
listed = git("for-each-ref", "refs/remotes", "--format=%(refname:lstrip=2) %(objectname)").stdout.split("\\n")
|
||||
remotes = {{line.split("/", 1)[0]: line.split(" ")[1] for line in listed if line and line.split(" ")[0].split("/", 1)[1] == name}}
|
||||
skip = "forge unavailable" if "github.com" in git("remote", "get-url", "origin").stdout else None
|
||||
actions = ["local-branch", *("remote-branch:" + remote for remote in remotes)] if merged else []
|
||||
entry = {{"branch": name, "worktree": None, "local_sha": tip, "head": None, "state": "merged-clean" if merged else "unmerged", "skip": skip,
|
||||
"merged_by": "ancestor" if merged else None, "merged_into": default if merged else None, "remotes": remotes, "actions": actions, "result": None}}
|
||||
if args.apply and skip is None:
|
||||
git("branch", "-D", "--", name)
|
||||
entry["result"] = "done"
|
||||
print(json.dumps({{"host_pid_namespace": True, "repos": [{{"repo": os.getcwd(), "default": default, "merged_into": [], "warnings": [], "items": [entry]}}]}}))
|
||||
"""
|
||||
|
||||
|
||||
class TestHubCleanupToken(HubCase):
|
||||
def tree(self, ready: bool, kept_skip: str | None) -> Path:
|
||||
root = self.write_tree()
|
||||
data = json.loads(self.tree_file.read_text(encoding="utf-8"))
|
||||
data["delete_ready"], data["delete_reason"] = ready, "" if ready else "no GitHub token in the Secret Service"
|
||||
kept = next(item for item in data["repos"][0]["items"] if item["branch"] == "feature/kept")
|
||||
kept["skip"] = kept_skip
|
||||
done = next(item for item in data["repos"][0]["items"] if item["branch"] == "feature/done")
|
||||
done.update(worktree=None, worktree_state=None, actions=["local-branch"], compose_projects=[], images=[])
|
||||
self.tree_file.write_text(json.dumps(data), encoding="utf-8")
|
||||
return root
|
||||
|
||||
def items(self, root: Path) -> tuple[dict, dict]:
|
||||
repo = f"{root}/repos/github.com/me/demo"
|
||||
return (signing.cleanup_item(repo, "feature/kept", "", KEPT_TIP),
|
||||
signing.cleanup_item(repo, "feature/done", "", DONE_TIP))
|
||||
|
||||
def branches(self) -> dict:
|
||||
status, _, body = self.get("/api/tree")
|
||||
self.assertEqual(status, 200, body)
|
||||
demo = next(repo for repo in json.loads(body)["repos"] if repo["name"] == "github.com/me/demo")
|
||||
return {entry["name"]: entry for entry in demo["branches"]}
|
||||
|
||||
def test_only_a_local_branch_merged_by_ancestry_into_the_default_branch_without_remote_is_token_free(self):
|
||||
self.tree(False, None)
|
||||
branches = self.branches()
|
||||
self.assertEqual({name: entry["token_free"] for name, entry in branches.items()}, {"feature": False, "feature/done": False, "feature/kept": True, "main": False})
|
||||
entry = {"merged_by": "ancestor", "merged_into": "main", "remotes": {}}
|
||||
cases = (("ancestry, no remote", entry, "main", True), ("pull request", {**entry, "merged_by": "#12"}, "main", False),
|
||||
("remote branch", {**entry, "remotes": {"origin": "a" * 40}}, "main", False), ("other line", {**entry, "merged_into": "store-i18n"}, "main", False),
|
||||
("no default", entry, None, False), ("line merge", {**entry, "merged_by": "line:main"}, "main", False),
|
||||
("pkgmgr names the remote HEAD", {**entry, "merged_into": "origin/main"}, "main", True), ("other remote line", {**entry, "merged_into": "origin/store-i18n"}, "main", False),
|
||||
("nested name", {**entry, "merged_into": "origin/feature/main"}, "main", False), ("no merged_into", {**entry, "merged_into": None}, "main", False))
|
||||
for name, candidate, default, free in cases:
|
||||
with self.subTest(name=name):
|
||||
self.assertEqual(signing.token_free(candidate, default), free)
|
||||
|
||||
def test_without_the_token_the_dry_run_offers_the_token_free_branch_and_blocks_the_rest(self):
|
||||
kept, done = self.items(self.tree(False, None))
|
||||
status, _, body = self.post_json("/api/sign", {"items": [kept]}, self.guard(self.host))
|
||||
self.assertEqual(status, 200, body)
|
||||
self.assertEqual(json.loads(body)["payload"]["items"], [signing.item_hash(kept)])
|
||||
for items in ([done], [kept, done]):
|
||||
with self.subTest(items=[item["branch"] for item in items]):
|
||||
status, _, body = self.post_json("/api/sign", {"items": items}, self.guard(self.host))
|
||||
self.assertEqual((status, body.decode()), (400, "delete blocked: no GitHub token in the Secret Service\n"))
|
||||
signer = self.pin()
|
||||
proof = signer.proof(self.prepare([kept])["payload"], self.origin)
|
||||
status, _, body = self.post_json("/api/cleanup", {"items": [kept], "proof": proof}, self.guard(self.host))
|
||||
self.assertEqual(status, 202, body)
|
||||
|
||||
def test_on_the_real_tokenless_tree_the_ancestor_merged_local_branch_is_token_free_and_only_the_forge_free_one_is_offered(self):
|
||||
root = self.write_fixture("branch-tree-tokenless.json")
|
||||
status, _, body = self.get("/api/tree")
|
||||
self.assertEqual(status, 200, body)
|
||||
repos = {repo["name"]: {entry["name"]: entry for entry in repo["branches"]} for repo in json.loads(body)["repos"]}
|
||||
shown = {name: {branch: (entry["token_free"], entry["cleanup"] is not None, entry["skip"]) for branch, entry in branches.items()} for name, branches in repos.items()}
|
||||
self.assertEqual(shown, {
|
||||
"github.com/anon/web": {"chore/ancestor-local": (True, False, "forge unavailable"), "fix/remote-merged": (False, False, "forge unavailable"), "feature/open": (False, False, "forge unavailable"), "main": (False, False, None)},
|
||||
"git.example.org/anon/notes": {"topic/ancestor-local": (True, True, None), "topic/patch-local": (False, True, None), "master": (False, False, None)},
|
||||
})
|
||||
notes = repos["git.example.org/anon/notes"]
|
||||
free, patch = notes["topic/ancestor-local"]["cleanup"], notes["topic/patch-local"]["cleanup"]
|
||||
self.assertEqual(free, signing.cleanup_item(f"{root}/repos/git.example.org/anon/notes", "topic/ancestor-local", "", "5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f"))
|
||||
status, _, body = self.post_json("/api/sign", {"items": [free]}, self.guard(self.host))
|
||||
self.assertEqual(status, 200, body)
|
||||
self.assertEqual(json.loads(body)["payload"]["items"], [signing.item_hash(free)])
|
||||
status, _, body = self.post_json("/api/sign", {"items": [patch]}, self.guard(self.host))
|
||||
self.assertEqual((status, body.decode()), (400, "delete blocked: no GitHub token in the Secret Service\n"))
|
||||
|
||||
def test_with_the_token_every_merged_clean_branch_without_skip_reason_is_offered(self):
|
||||
kept, done = self.items(self.tree(True, None))
|
||||
status, _, body = self.post_json("/api/sign", {"items": [kept, done]}, self.guard(self.host))
|
||||
self.assertEqual(status, 200, body)
|
||||
self.assertEqual(sorted(json.loads(body)["payload"]["items"]), sorted(signing.item_hash(item) for item in (kept, done)))
|
||||
kept, _ = self.items(self.tree(True, "ignored files beyond known caches: notes.txt"))
|
||||
self.assertEqual(self.post_json("/api/sign", {"items": [kept]}, self.guard(self.host))[0], 400)
|
||||
|
||||
def test_store_token_adds_one_operator_row_whose_command_stores_the_token_under_pkgmgrs_key(self):
|
||||
status, _, body = self.post_json("/api/cleanup/token", {}, self.guard(self.host))
|
||||
self.assertEqual((status, json.loads(body)), (201, {"task": "operator/pkgmgr-github-token", "status": "queued"}))
|
||||
self.assertEqual(len(self.scripts.calls), 1)
|
||||
script, args = self.scripts.calls[0]
|
||||
self.assertEqual((script, args[:4]), ("kvb-org-operator/scripts/operator.py", ["add", "pkgmgr-github-token", "--dir", "/"]))
|
||||
command = args[args.index("--command") + 1]
|
||||
self.assertEqual(shlex.split(command), ["env", "secret-tool", "store", "--label=pkgmgr-github", "service", "pkgmgr:github", "username", "github.com|-", "</dev/tty"])
|
||||
self.assertEqual(args[args.index("--check") + 1], "manual")
|
||||
self.assertIsNone(operator_script.unrunnable(command, "/"))
|
||||
self.assertEqual(self.post_json("/api/cleanup/token", {"extra": 1}, self.guard(self.host))[0], 400)
|
||||
self.assertEqual(self.post_json("/api/cleanup/token", {}, {"Host": self.host})[0], 403)
|
||||
self.assertEqual(self.get("/api/cleanup/token")[0], 405)
|
||||
|
||||
|
||||
class TestHostCleanupWithoutToken(unittest.TestCase):
|
||||
def setUp(self):
|
||||
tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(tmp.cleanup)
|
||||
base = Path(tmp.name)
|
||||
self.repo = base / "repo"
|
||||
self.git("init", "-q", "-b", "main", str(self.repo), cwd=base)
|
||||
self.git("commit", "-q", "--allow-empty", "-m", "one")
|
||||
self.git("branch", "local-merged")
|
||||
self.git("branch", "remote-merged")
|
||||
self.git("branch", "squash-candidate")
|
||||
self.git("commit", "-q", "--allow-empty", "-m", "two")
|
||||
self.git("checkout", "-q", "squash-candidate")
|
||||
self.git("commit", "-q", "--allow-empty", "-m", "side")
|
||||
self.git("checkout", "-q", "main")
|
||||
self.git("remote", "add", "origin", "ssh://git@git.example.org/anon/repo.git")
|
||||
self.git("update-ref", "refs/remotes/origin/main", self.tip("main"))
|
||||
self.git("symbolic-ref", "refs/remotes/origin/HEAD", "refs/remotes/origin/main")
|
||||
self.git("update-ref", "refs/remotes/origin/remote-merged", self.tip("remote-merged"))
|
||||
self.jobs, self.results = base / "jobs", base / "results"
|
||||
self.jobs.mkdir()
|
||||
self.results.mkdir()
|
||||
self.fake = base / "bin" / "pkgmgr"
|
||||
self.fake.parent.mkdir()
|
||||
self.fake.write_text(SCRATCH_PKGMGR.format(python=sys.executable), encoding="utf-8")
|
||||
self.fake.chmod(0o755)
|
||||
self.tree = base / "tree" / "branch-tree.json"
|
||||
self.tree.parent.mkdir()
|
||||
self.tree.write_text(json.dumps({"delete_ready": False, "delete_reason": "no GitHub token in the Secret Service", "repos": [{"path": str(self.repo), "known": True, "default": "main"}]}), encoding="utf-8")
|
||||
self.approvers = base / "config" / "approvers.json"
|
||||
self.trust = verifier.Trust(approvers=self.approvers, replay=base / "replay" / "cleanup.json", origin=ORIGIN)
|
||||
self.signer = key(-7, "main")
|
||||
self.signer.pin(self.approvers, ORIGIN)
|
||||
|
||||
def git(self, *args: str, cwd: Path | None = None) -> str:
|
||||
return subprocess.run(["git", *args], cwd=cwd or self.repo, env={**os.environ, **GIT_ENV}, check=True, capture_output=True, text=True).stdout.strip()
|
||||
|
||||
def tip(self, branch: str) -> str:
|
||||
return self.git("rev-parse", f"refs/heads/{branch}")
|
||||
|
||||
def branches(self) -> set[str]:
|
||||
return set(self.git("for-each-ref", "refs/heads", "--format=%(refname:short)").split())
|
||||
|
||||
def run_job(self, names: list[str]) -> dict:
|
||||
items = [{"repository": str(self.repo), "branch": name, "worktree": "", "tip": self.tip(name)} for name in names]
|
||||
hashes = [signing.item_hash(signing.cleanup_item(**item)) for item in items]
|
||||
proof = self.signer.proof(signing.approval_payload(hashes, secrets.token_hex(16), int(time.time()) + 600), ORIGIN)
|
||||
job_id = secrets.token_hex(8)
|
||||
(self.jobs / f"{job_id}.json").write_text(json.dumps({"v": 1, "id": job_id, "created": "x", "items": items, "proof": proof}), encoding="utf-8")
|
||||
self.assertEqual(host.cycle(self.jobs, self.results, host.Cleanup(self.trust, self.tree, self.fake)), 1)
|
||||
result = json.loads((self.results / f"{job_id}.json").read_text(encoding="utf-8"))
|
||||
return {entry["branch"]: entry for entry in result["items"]}
|
||||
|
||||
def test_the_executor_deletes_a_local_merged_branch_without_token_and_refuses_a_remote_one(self):
|
||||
result = self.run_job(["local-merged", "remote-merged"])
|
||||
self.assertEqual(result["local-merged"]["status"], "done", result["local-merged"]["reason"])
|
||||
self.assertEqual(result["remote-merged"]["status"], "refused")
|
||||
self.assertIn("a signed delete is local only", result["remote-merged"]["reason"])
|
||||
self.assertEqual(self.branches(), {"main", "remote-merged", "squash-candidate"})
|
||||
|
||||
def test_without_token_an_unmerged_squash_candidate_stays_and_with_the_token_a_plan_that_deletes_the_remote_branch_is_refused(self):
|
||||
result = self.run_job(["squash-candidate"])
|
||||
self.assertEqual(result["squash-candidate"]["status"], "refused")
|
||||
self.assertIn("not merged-clean", result["squash-candidate"]["reason"])
|
||||
data = json.loads(self.tree.read_text(encoding="utf-8"))
|
||||
self.tree.write_text(json.dumps({**data, "delete_ready": True}), encoding="utf-8")
|
||||
remote = self.run_job(["remote-merged"])["remote-merged"]
|
||||
self.assertEqual((remote["status"], remote["reason"]), ("refused", "the plan also deletes remote-branch:origin; a signed delete is local only"))
|
||||
self.assertEqual(self.branches(), {"main", "local-merged", "remote-merged", "squash-candidate"})
|
||||
|
||||
def test_without_token_pkgmgr_skips_every_item_of_a_github_repository_so_the_executor_refuses_and_keeps_even_the_token_free_branch(self):
|
||||
self.git("remote", "set-url", "origin", "git@github.com:anon/repo.git")
|
||||
result = self.run_job(["local-merged"])["local-merged"]
|
||||
self.assertEqual((result["status"], result["reason"]), ("refused", "gc skip reason: forge unavailable"))
|
||||
self.assertIn("local-merged", self.branches())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,221 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import unittest
|
||||
from contextlib import closing
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
from hub_case import HubCase
|
||||
from hub_support import GIT_ENV, REPO_ROOT, RUNNER, runner, signing
|
||||
|
||||
GATES = REPO_ROOT / "skills" / "kvb-check-gate" / "scripts" / "gates.py"
|
||||
GATES_SPEC = importlib.util.spec_from_file_location("kvb_gates_for_hub", GATES)
|
||||
gates = importlib.util.module_from_spec(GATES_SPEC)
|
||||
GATES_SPEC.loader.exec_module(gates)
|
||||
REPO = "/home/op/Repositories/demo"
|
||||
|
||||
|
||||
def rule(action: str, level: str | None, branch: str = "feature/x", repo: str = REPO) -> list[str]:
|
||||
head = ["--repo", repo, "--branch", branch, "--action", action]
|
||||
return ["rule", "drop", *head] if level is None else ["rule", "set", *head, "--level", level]
|
||||
|
||||
|
||||
def parent(name: str, drop: bool = False, branch: str = "feature/x", repo: str = REPO) -> list[str]:
|
||||
return ["parents", "drop" if drop else "add", "--repo", repo, "--branch", branch, "--parent", name]
|
||||
|
||||
|
||||
class TestGateArgv(unittest.TestCase):
|
||||
def test_exactly_rule_set_drop_and_parents_add_drop_of_one_branch_pass(self):
|
||||
for argv in (rule("commit", "test"), rule("pr", "e2e"), rule("merge", None), parent("main"), parent("feature/base", drop=True), rule("push", "none", branch="fix/a-b_c.1")):
|
||||
with self.subTest(argv=argv):
|
||||
self.assertEqual(signing.gate_argv(argv), argv)
|
||||
self.assertEqual(signing.gate_item(argv), {"kind": "gate", "argv": argv})
|
||||
|
||||
def test_every_other_argv_is_refused(self):
|
||||
cases = {
|
||||
"not a list": "rule set",
|
||||
"no texts": ["rule", "set", 1],
|
||||
"resolve": ["resolve", "--repo", REPO, "--branch", "b", "--action", "commit"],
|
||||
"derive": ["parents", "derive", "--repo", REPO],
|
||||
"receipt": ["receipt", "add", "--repo", REPO, "--tree", "a" * 40, "--level", "e2e", "--log", "/x"],
|
||||
"show": ["show", "--json"],
|
||||
"db flag": ["--db", "/tmp/x.db", *rule("commit", "test")],
|
||||
"source flag": [*parent("main"), "--source", "recorded"],
|
||||
"reordered": ["rule", "set", "--branch", "feature/x", "--repo", REPO, "--action", "commit", "--level", "test"],
|
||||
"missing level": rule("commit", "test")[:-1],
|
||||
"level on drop": [*rule("commit", None), "--level", "test"],
|
||||
"unknown level": rule("commit", "strict"),
|
||||
"unknown action": rule("deploy", "test"),
|
||||
"relative repo": rule("commit", "test", repo="repo"),
|
||||
"dotdot repo": rule("commit", "test", repo="/home/op/../root"),
|
||||
"control repo": rule("commit", "test", repo="/home/op/\nrepo"),
|
||||
"option branch": rule("commit", "test", branch="--level"),
|
||||
"pattern branch": rule("commit", "test", branch="feature/*"),
|
||||
"spaced branch": rule("commit", "test", branch="a b"),
|
||||
"dotdot branch": rule("commit", "test", branch="a..b"),
|
||||
"trailing slash": rule("commit", "test", branch="feature/"),
|
||||
"own parent": parent("feature/x"),
|
||||
"empty parent": parent(""),
|
||||
}
|
||||
for name, argv in cases.items():
|
||||
with self.subTest(name=name):
|
||||
self.assertRaises(signing.Refused, signing.gate_argv, argv)
|
||||
|
||||
|
||||
class TestHubGates(HubCase):
|
||||
def stage(self, argvs: list[list[str]]) -> list[dict]:
|
||||
return [{"kind": "gate", "argv": argv} for argv in argvs]
|
||||
|
||||
def test_a_signed_gate_batch_becomes_one_runner_job_per_change(self):
|
||||
items = self.stage([rule("commit", "test"), parent("main")])
|
||||
payload = self.prepare(items)["payload"]
|
||||
self.assertEqual(sorted(payload["items"]), sorted(signing.item_hash(signing.gate_item(item["argv"])) for item in items))
|
||||
signer = self.pin()
|
||||
proof = signer.proof(payload, self.origin)
|
||||
self.assertEqual(self.post_json("/api/gates", {"items": items, "proof": None}, self.guard(self.host))[0], 400)
|
||||
self.assertEqual(self.post_json("/api/gates", {"items": items, "proof": proof}, {"Host": self.host})[0], 403)
|
||||
status, _, body = self.post_json("/api/gates", {"items": items, "proof": proof}, self.guard(self.host))
|
||||
self.assertEqual(status, 202, body)
|
||||
self.assertEqual(len(json.loads(body)["jobs"]), 2)
|
||||
jobs = self.runner_jobs()
|
||||
self.assertEqual(sorted((job["kind"], job["argv"], job["command"].split()[0]) for job in jobs), sorted(("gate", item["argv"], "gates.py") for item in items))
|
||||
self.assertEqual(self.get("/api/gates")[0], 405)
|
||||
|
||||
def test_the_hub_refuses_a_malformed_or_forged_gate_change_before_it_queues(self):
|
||||
for name, items in (("receipt", self.stage([["receipt", "add", "--repo", REPO, "--tree", "a" * 40, "--level", "e2e", "--log", "/x"]])), ("own parent", self.stage([parent("feature/x")])), ("twice", self.stage([parent("main"), parent("main")])), ("extra key", [{"kind": "gate", "argv": parent("main"), "id": "x"}])):
|
||||
with self.subTest(name=name):
|
||||
status, _, body = self.post_json("/api/sign", {"items": items}, self.guard(self.host))
|
||||
self.assertEqual(status, 400, body)
|
||||
signer = self.pin()
|
||||
proof = signer.proof(self.prepare(self.stage([rule("commit", "test")]))["payload"], self.origin)
|
||||
status, _, body = self.post_json("/api/gates", {"items": self.stage([rule("commit", "e2e")]), "proof": proof}, self.guard(self.host))
|
||||
self.assertEqual(status, 400, body)
|
||||
self.assertEqual(self.runner_jobs(), [])
|
||||
|
||||
|
||||
class TestRunnerGates(HubCase):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.home = self.outside / "home"
|
||||
self.home.mkdir()
|
||||
self.repo = self.outside / "repo"
|
||||
subprocess.run(["git", "init", "-q", "-b", "main", str(self.repo)], env={**os.environ, **GIT_ENV}, check=True)
|
||||
self.db = self.home / ".local" / "state" / "agents-council" / "gates.db"
|
||||
|
||||
def run_gates_runner(self, flags: list[str]) -> None:
|
||||
trust = ["--approvers", str(self.approvers), "--replay", str(self.replay / "runner.json")]
|
||||
command = [sys.executable, str(RUNNER), "--port", str(self.port), "--interval", "0.1", *trust, *flags, "--once"]
|
||||
done = subprocess.run(command, capture_output=True, text=True, timeout=60, check=False, env={**os.environ, "HOME": str(self.home)})
|
||||
self.assertEqual(done.returncode, 0, done.stderr)
|
||||
|
||||
def queue(self, argvs: list[list[str]]) -> list[str]:
|
||||
items = [{"kind": "gate", "argv": argv} for argv in argvs]
|
||||
proof = self.signed(items, self.pin())
|
||||
status, _, body = self.post_json("/api/gates", {"items": items, "proof": proof}, self.guard(self.host))
|
||||
self.assertEqual(status, 202, body)
|
||||
return json.loads(body)["jobs"]
|
||||
|
||||
def finished(self) -> list[tuple[str, int | None]]:
|
||||
jobs = sorted(self.httpd.hub.jobs.listed(), key=lambda job: job["created_ts"])
|
||||
return [(job["status"], job["code"]) for job in jobs]
|
||||
|
||||
def test_the_runner_writes_a_signed_rule_and_parent_into_the_default_gates_db_of_the_operator(self):
|
||||
repo = str(self.repo.resolve())
|
||||
self.queue([rule("commit", "quality-high", branch="main", repo=repo), parent("main", branch="feature/x", repo=repo)])
|
||||
self.run_gates_runner(["--gates", str(GATES)])
|
||||
self.assertEqual(self.finished(), [("done", 0), ("done", 0)])
|
||||
with closing(sqlite3.connect(self.db)) as db:
|
||||
self.assertEqual(gates.resolve(db, repo, "feature/x", "commit"), ("quality-high", "main"))
|
||||
self.assertEqual(gates.parents_of(db, repo, "feature/x"), [("main", "operator")])
|
||||
self.assertEqual([call for call in self.scripts.calls if call[0].endswith("operator.py")], [])
|
||||
|
||||
def test_without_gates_the_runner_refuses_and_writes_nothing(self):
|
||||
self.queue([rule("commit", "test", repo=str(self.repo))])
|
||||
self.run_gates_runner([])
|
||||
jobs = self.httpd.hub.jobs.listed()
|
||||
self.assertEqual([(job["status"], job["output"]) for job in jobs], [("refused", runner.UNGATED)])
|
||||
self.assertFalse(self.db.exists())
|
||||
|
||||
def test_the_runner_refuses_an_argv_that_differs_from_the_signed_one(self):
|
||||
self.queue([rule("commit", "test", repo=str(self.repo))])
|
||||
job = self.httpd.hub.jobs.listed()[0]
|
||||
self.httpd.hub.jobs.change(job["id"], {"argv": rule("commit", "none", repo=str(self.repo))})
|
||||
self.run_gates_runner(["--gates", str(GATES)])
|
||||
refused = self.httpd.hub.jobs.listed()[0]
|
||||
self.assertEqual(refused["status"], "refused")
|
||||
self.assertIn("does not list this item", refused["output"])
|
||||
self.assertFalse(self.db.exists())
|
||||
|
||||
def test_gate_refusal_checks_the_argv_shape_before_any_signature(self):
|
||||
reason = runner.gate_refusal({"argv": ["receipt", "add"], "proof": None}, mock.Mock(), time.time())
|
||||
self.assertIn("gates.py rule set|drop or parents add|drop", reason)
|
||||
with self.assertRaises(signing.Refused):
|
||||
runner.gate({"argv": ["show", "--json"]}, GATES, print)
|
||||
|
||||
|
||||
class TestHubGateView(HubCase):
|
||||
def write_db(self, repo: str) -> Path:
|
||||
db_path = self.kept / "gates.db"
|
||||
db = gates.connect(db_path)
|
||||
db.executemany("INSERT INTO rules VALUES (?, ?, ?, ?, ?)", [(repo, "main", "commit", "test", 0.0), (repo, "main", "merge", "quality-high", 0.0), (repo, "feature", "pr", "e2e", 0.0), (repo, "feature/kept", "merge", "quick", 0.0)])
|
||||
db.executemany("INSERT INTO parents VALUES (?, ?, ?, ?)", [(repo, "feature", "main", "recorded"), (repo, "feature/done", "feature", "derived"), (repo, "feature/done", "feature/kept", "operator"), (repo, "feature/kept", "main", "derived")])
|
||||
db.commit()
|
||||
db.close()
|
||||
return db_path
|
||||
|
||||
def test_the_tree_carries_gates_py_show_json_for_every_branch_from_the_read_only_database(self):
|
||||
root = self.write_tree()
|
||||
repo = f"{root}/repos/github.com/me/demo"
|
||||
db_path = self.write_db(repo)
|
||||
before = db_path.read_bytes()
|
||||
db_path.chmod(0o444)
|
||||
self.kept.chmod(0o555)
|
||||
self.addCleanup(self.kept.chmod, 0o755)
|
||||
status, _, body = self.get("/api/tree")
|
||||
self.assertEqual(status, 200, body)
|
||||
shown = json.loads(body)["gates"]
|
||||
self.assertEqual(sorted(shown[repo]), ["feature", "feature/done", "feature/kept", "main"])
|
||||
with closing(sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)) as db, mock.patch.object(gates, "branches", return_value={name: "0" * 40 for name in shown[repo]}):
|
||||
self.assertEqual(shown[repo], gates.show(db, [repo])[repo])
|
||||
done = shown[repo]["feature/done"]
|
||||
self.assertEqual(done["parents"], [{"parent": "feature", "source": "derived"}, {"parent": "feature/kept", "source": "operator"}])
|
||||
self.assertEqual(done["resolved"], {"commit": {"level": "test", "source": "main"}, "merge": {"level": "quality-high", "source": "main"}, "push": {"level": "none", "source": None}, "pr": {"level": "e2e", "source": "feature"}})
|
||||
self.assertEqual(db_path.read_bytes(), before)
|
||||
self.assertEqual(sorted(path.name for path in self.kept.iterdir()), ["gates.db"])
|
||||
|
||||
def test_without_a_database_or_with_a_broken_one_the_tree_carries_no_gates(self):
|
||||
self.write_tree()
|
||||
self.assertEqual(json.loads(self.get("/api/tree")[2])["gates"], {})
|
||||
(self.kept / "gates.db").write_bytes(b"not sqlite")
|
||||
self.assertEqual(json.loads(self.get("/api/tree")[2])["gates"], {})
|
||||
|
||||
def test_a_stored_level_outside_levels_or_an_older_gates_py_without_parents_of_keeps_the_tree_up_without_gates(self):
|
||||
root = self.write_tree()
|
||||
repo = f"{root}/repos/github.com/me/demo"
|
||||
db = gates.connect(self.write_db(repo))
|
||||
db.execute("UPDATE rules SET level = 'bogus' WHERE branch = 'main' AND action = 'commit'")
|
||||
db.commit()
|
||||
status, _, body = self.get("/api/tree")
|
||||
self.assertEqual((status, json.loads(body)["gates"] if status == 200 else body), (200, {}))
|
||||
db.execute("UPDATE rules SET level = 'test' WHERE branch = 'main' AND action = 'commit'")
|
||||
db.commit()
|
||||
db.close()
|
||||
self.assertEqual(json.loads(self.get("/api/tree")[2])["gates"][repo]["main"]["rules"]["commit"], "test")
|
||||
older = self.kept.parent / "older-skills"
|
||||
script = older / "kvb-check-gate" / "scripts" / "gates.py"
|
||||
script.parent.mkdir(parents=True)
|
||||
script.write_text('ACTIONS = ("commit",)\n\n\ndef resolve(db, repo, branch, action):\n return ("none", None)\n', encoding="utf-8")
|
||||
self.scripts.skills = older
|
||||
status, _, body = self.get("/api/tree")
|
||||
self.assertEqual((status, json.loads(body)["gates"] if status == 200 else body), (200, {}))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user