feat(orchestrator): Agents and Artifacts views with spawn tree, orbit, 3D and signed branch deletes
Phase 4 of requirement 001. Agents and Artifacts each offer Orbit, Chord, Swimlane, Tree and 3D: the spawn tree (page-spawn.js) and the artifacts orbit (page-artifacts.js, kvb_orchestrator/artifacts.py) replace the "folgt" placeholders, 3D runs on the vendored 3d-force-graph and three.js (MIT, LICENSE files kept, served from the image under the unchanged CSP), and a Bootstrap Icons sprite (icons.py, MIT) replaces the emoji glyphs, also in the meditation page. Signed branch deletes: the Tree view offers a delete only for merged-clean branches without a gc skip reason; the operator signs the exact item with the passkey, the hub writes a job without any command (kvb_orchestrator/cleanup.py), and the host path unit runs client/cleanup.py, which verifies the signature, dry-runs pkgmgr branch cleanup for that exact branch, refuses patterns, option names, changed tips, other worktrees and plans naming other branches, deletes, and rewrites the branch tree. pkgmgr runs with core.hooksPath=/dev/null, core.fsmonitor=false and the config hook events disabled through GIT_CONFIG_COUNT, because the cwd is a repository agents can write and pkgmgr releases before 4c8b137 delete without scan mode (test with a positive control that both hook kinds fire without it). 3D labels go into the tooltip as a text node: the vendored tooltip renders a string label as HTML. Tests: 503 OK, Chromium (CDP) and Firefox (BiDi) walks over every screen at three widths. MEDUSA acceptances (kvb-think-criticize skeptic 2026-11-07): WEB-FLASK-004 view.py:135 (file template, autoescape on, state via tojson), MEDUSA-MM-SCAN-058 server.py:69/:85 (substring pla in --templates and Places), RH005 client/cleanup.py:199 (raise SystemExit(main())), B704 icons.py:45 (sprite baked into the image, names checked against it), EXA002 page-artifacts.js:221 (openAgent only opens the drawer), STG002 page-constants.js:64 (U+2212 in a static label), PL002 page-graph3d.js:44 (CSS class space-tools; the real tooltip sink fixed), B603 client/cleanup.py:75 (argv without shell; hook hole closed). 🎯 Fix: 92 % 💥 Break: 5 % 💸 Cost: 100 % ⏱️ Time: – 🛡️ Security: 100 % 🏆 Score: 87 🚦 Gate: ✅ 90/5, cost and time <= 100, security 100 % Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Kevin Veen-BirkenbachandClaude Opus 5.5 committed
1 parent
db3c5264f2
commit
90c56cb786
64 files changed
+82829
-451
No files matched your search
@@ -44,6 +44,70 @@
|
||||
"reason": "dirty: 1 paths",
|
||||
"result": null
|
||||
},
|
||||
{
|
||||
"branch": "feature/done",
|
||||
"worktree": "@ROOT@/worktrees/github.com/me/demo/feature-done",
|
||||
"head": "5b1e6fd3a7c0e2d94f8a61b3c2d7e9f0a1b2c3d4",
|
||||
"detached": false,
|
||||
"locked": null,
|
||||
"worktree_state": "clean",
|
||||
"staged": 0,
|
||||
"unstaged": 0,
|
||||
"untracked": 0,
|
||||
"upstream": "origin/feature/done",
|
||||
"ahead": 0,
|
||||
"behind": 0,
|
||||
"gone": true,
|
||||
"local_sha": "5b1e6fd3a7c0e2d94f8a61b3c2d7e9f0a1b2c3d4",
|
||||
"remotes": {},
|
||||
"compose_projects": [
|
||||
"demo-feature-done"
|
||||
],
|
||||
"images": [
|
||||
"demo:feature-done"
|
||||
],
|
||||
"state": "merged-clean",
|
||||
"merged_by": "#12",
|
||||
"merged_into": "main",
|
||||
"size_bytes": 4096,
|
||||
"skip": null,
|
||||
"actions": [
|
||||
"worktree",
|
||||
"local-branch",
|
||||
"remote-branch"
|
||||
],
|
||||
"reason": "merged into main by #12",
|
||||
"result": null
|
||||
},
|
||||
{
|
||||
"branch": "feature/kept",
|
||||
"worktree": null,
|
||||
"head": null,
|
||||
"detached": false,
|
||||
"locked": null,
|
||||
"worktree_state": "none",
|
||||
"staged": null,
|
||||
"unstaged": null,
|
||||
"untracked": null,
|
||||
"upstream": null,
|
||||
"ahead": null,
|
||||
"behind": null,
|
||||
"gone": false,
|
||||
"local_sha": "0c9d8e7f6a5b4c3d2e1f0a9b8c7d6e5f4a3b2c1d",
|
||||
"remotes": {},
|
||||
"compose_projects": [],
|
||||
"images": [],
|
||||
"state": "merged-clean",
|
||||
"merged_by": "ancestor",
|
||||
"merged_into": "main",
|
||||
"size_bytes": null,
|
||||
"skip": "ignored files beyond known caches: notes.txt",
|
||||
"actions": [
|
||||
"local-branch"
|
||||
],
|
||||
"reason": "merged into main",
|
||||
"result": null
|
||||
},
|
||||
{
|
||||
"branch": "main",
|
||||
"worktree": "@ROOT@/repos/github.com/me/demo",
|
||||
|
||||
@@ -0,0 +1,96 @@
|
||||
import {spawn} from "node:child_process";
|
||||
import {mkdtempSync, readFileSync, rmSync, writeFileSync} from "node:fs";
|
||||
import {createServer} from "node:net";
|
||||
import {tmpdir} from "node:os";
|
||||
import {join} from "node:path";
|
||||
|
||||
// Args: <firefox binary> <plan.json>; the plan lists steps {width, height, url, wait, probe, shot} like cdp.mjs.
|
||||
// Prints {results, errors}: one probe value per step and every JavaScript error the page logged.
|
||||
const [binary, planFile] = process.argv.slice(2);
|
||||
const plan = JSON.parse(readFileSync(planFile, "utf8"));
|
||||
const profile = mkdtempSync(join(tmpdir(), "bidi-"));
|
||||
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
|
||||
const port = await new Promise((resolve) => {
|
||||
const probe = createServer().listen(0, "127.0.0.1", () => {
|
||||
const {port: free} = probe.address();
|
||||
probe.close(() => resolve(free));
|
||||
});
|
||||
});
|
||||
const browser = spawn(binary, ["--headless", "--no-remote", "--profile", profile, "--remote-debugging-port", String(port), "about:blank"], {stdio: ["ignore", "ignore", "pipe"]});
|
||||
|
||||
const address = await new Promise((resolve, reject) => {
|
||||
let text = "";
|
||||
const timer = setTimeout(() => reject(new Error(`no BiDi address: ${text.slice(-500)}`)), 60000);
|
||||
browser.stderr.on("data", (chunk) => {
|
||||
text += chunk;
|
||||
const found = /WebDriver BiDi listening on (ws:\/\/\S+)/.exec(text);
|
||||
if (found) {
|
||||
clearTimeout(timer);
|
||||
resolve(found[1]);
|
||||
}
|
||||
});
|
||||
browser.on("exit", (code) => reject(new Error(`browser exited with ${code}: ${text.slice(-500)}`)));
|
||||
});
|
||||
|
||||
const socket = new WebSocket(`${address}/session`);
|
||||
await new Promise((resolve, reject) => {
|
||||
socket.onopen = resolve;
|
||||
socket.onerror = reject;
|
||||
});
|
||||
const waiting = new Map();
|
||||
const errors = [];
|
||||
let next = 0;
|
||||
socket.onmessage = (event) => {
|
||||
const data = JSON.parse(event.data);
|
||||
if (data.method === "log.entryAdded" && data.params.level === "error" && data.params.type === "javascript") {
|
||||
errors.push(data.params.text);
|
||||
}
|
||||
if (data.id && waiting.has(data.id)) {
|
||||
const {resolve, reject} = waiting.get(data.id);
|
||||
waiting.delete(data.id);
|
||||
if (data.type === "error") {
|
||||
reject(new Error(`${data.error}: ${data.message}`));
|
||||
} else {
|
||||
resolve(data.result);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
function send(method, params) {
|
||||
next += 1;
|
||||
const id = next;
|
||||
socket.send(JSON.stringify({id, method, params}));
|
||||
return new Promise((resolve, reject) => waiting.set(id, {resolve, reject}));
|
||||
}
|
||||
|
||||
const results = [];
|
||||
try {
|
||||
await send("session.new", {capabilities: {}});
|
||||
await send("session.subscribe", {events: ["log.entryAdded"]});
|
||||
const {contexts} = await send("browsingContext.getTree", {});
|
||||
const context = contexts[0].context;
|
||||
for (const step of plan.steps) {
|
||||
await send("browsingContext.setViewport", {context, viewport: {width: step.width, height: step.height}});
|
||||
if (step.url) {
|
||||
await send("browsingContext.navigate", {context, url: step.url, wait: "complete"});
|
||||
}
|
||||
await sleep(step.wait);
|
||||
let value = null;
|
||||
if (step.probe) {
|
||||
const answer = await send("script.evaluate", {expression: `(async () => JSON.stringify(await ${step.probe}))()`, target: {context}, awaitPromise: true, resultOwnership: "none"});
|
||||
value = answer.type === "success" ? JSON.parse(answer.result.value) : {error: answer.exceptionDetails.text};
|
||||
}
|
||||
if (step.shot) {
|
||||
const {data} = await send("browsingContext.captureScreenshot", {context});
|
||||
writeFileSync(step.shot, Buffer.from(data, "base64"));
|
||||
}
|
||||
results.push(value);
|
||||
}
|
||||
await send("browser.close", {}).catch(() => null);
|
||||
} finally {
|
||||
socket.close();
|
||||
browser.kill();
|
||||
await sleep(300);
|
||||
rmSync(profile, {recursive: true, force: true});
|
||||
}
|
||||
console.log(JSON.stringify({results, errors}));
|
||||
+10
-2
@@ -3,13 +3,15 @@ import {mkdtempSync, readFileSync, rmSync, writeFileSync} from "node:fs";
|
||||
import {tmpdir} from "node:os";
|
||||
import {join} from "node:path";
|
||||
|
||||
// Args: <chromium binary> <plan.json>; the plan lists steps {width, height, url, wait, probe, shot}.
|
||||
// Args: <chromium binary> <plan.json>; the plan lists steps {width, height, url, wait, probe, shot, media} and an optional
|
||||
// authenticator {credentialId, privateKey, rpId}: base64 id and PKCS#8 key of a passkey the page may sign with.
|
||||
// Prints {results, errors}: one probe value per step and every uncaught page exception.
|
||||
const [binary, planFile] = process.argv.slice(2);
|
||||
const plan = JSON.parse(readFileSync(planFile, "utf8"));
|
||||
const profile = mkdtempSync(join(tmpdir(), "cdp-"));
|
||||
// --no-sandbox: the agent sandbox already forbids the user namespaces Chromium's own sandbox needs.
|
||||
const flags = ["--headless", "--no-sandbox", "--disable-gpu", "--no-first-run", "--no-default-browser-check", "--hide-scrollbars", `--user-data-dir=${profile}`, "--remote-debugging-port=0", "about:blank"];
|
||||
// --enable-unsafe-swiftshader: without a GPU, headless Chromium gives WebGL only through SwiftShader, which it no longer enables by itself.
|
||||
const flags = ["--headless", "--no-sandbox", "--disable-gpu", "--use-angle=swiftshader", "--enable-unsafe-swiftshader", "--no-first-run", "--no-default-browser-check", "--hide-scrollbars", `--user-data-dir=${profile}`, "--remote-debugging-port=0", "about:blank"];
|
||||
const browser = spawn(binary, flags, {stdio: ["ignore", "ignore", "pipe"]});
|
||||
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
|
||||
|
||||
@@ -65,8 +67,14 @@ try {
|
||||
const {sessionId} = await send("Target.attachToTarget", {targetId, flatten: true});
|
||||
const page = (method, params) => send(method, params, sessionId);
|
||||
await page("Runtime.enable", {});
|
||||
if (plan.authenticator) {
|
||||
await page("WebAuthn.enable", {enableUI: false});
|
||||
const {authenticatorId} = await page("WebAuthn.addVirtualAuthenticator", {options: {protocol: "ctap2", transport: "internal", hasResidentKey: true, hasUserVerification: true, isUserVerified: true, automaticPresenceSimulation: true}});
|
||||
await page("WebAuthn.addCredential", {authenticatorId, credential: {...plan.authenticator, isResidentCredential: false, signCount: 0}});
|
||||
}
|
||||
for (const step of plan.steps) {
|
||||
await page("Emulation.setDeviceMetricsOverride", {width: step.width, height: step.height, deviceScaleFactor: 1, mobile: false});
|
||||
await page("Emulation.setEmulatedMedia", {features: step.media || []});
|
||||
if (step.url) {
|
||||
await page("Page.navigate", {url: step.url});
|
||||
}
|
||||
|
||||
@@ -1,57 +1,87 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
|
||||
HERE = Path(__file__).resolve().parent
|
||||
sys.path.insert(0, str(HERE.parent))
|
||||
|
||||
from hub_case import HubCase
|
||||
from hub_support import ASK, iso
|
||||
from hub_support import ASK, iso, verifier
|
||||
from test_hub_cleanup import DONE_TIP, FAKE_PKGMGR, host
|
||||
|
||||
OUT = Path(os.environ.get("E2E_OUT", "/tmp/orchestrator-e2e"))
|
||||
OUT = Path(os.environ.get("E2E_OUT", "/tmp/orchestrator-phase4-shots"))
|
||||
WIDTHS = ((1920, 1080), (1366, 768), (390, 844))
|
||||
SCREENS = ["stream", "stream/inbox", "stream/rules", *(f"{tab}/{view}" for tab in ("agents", "artifacts") for view in ("orbit", "chord", "swimlane", "tree", "3d"))]
|
||||
PROBE = """(() => {
|
||||
const all = (selector) => [...document.querySelectorAll(selector)];
|
||||
VIEWS = ("orbit", "chord", "swimlane", "tree", "3d")
|
||||
SCREENS = ["stream", "stream/inbox", "stream/rules", *(f"{tab}/{view}" for tab in ("agents", "artifacts") for view in VIEWS)]
|
||||
HELPERS = """
|
||||
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
|
||||
const until = async (check, ms) => {
|
||||
const end = Date.now() + ms;
|
||||
while (Date.now() < end) {
|
||||
const found = check();
|
||||
if (found) {
|
||||
return found;
|
||||
}
|
||||
await sleep(200);
|
||||
}
|
||||
return null;
|
||||
};
|
||||
const all = (selector, root) => [...(root || document).querySelectorAll(selector)];
|
||||
const text = (node) => (node ? node.textContent.trim() : null);
|
||||
const label = (button) => text(button.firstChild);
|
||||
const name = (node) => (node ? node.getAttribute("aria-label") || text(node) : null);
|
||||
"""
|
||||
PROBE = """(async () => {@HELPERS@
|
||||
await until(() => !document.querySelector("#main .space-note") || !text(document.querySelector("#main .space-note")).startsWith("Lade"), 8000);
|
||||
const badge = (button) => {
|
||||
const found = button.querySelector(".badge");
|
||||
return found && !found.hidden ? text(found) : null;
|
||||
};
|
||||
const icons = all("svg.bi").filter((svg) => !svg.closest(".sprite"));
|
||||
const banner = document.getElementById("mayday");
|
||||
return {
|
||||
hash: location.hash,
|
||||
title: document.title,
|
||||
h1: text(document.querySelector("h1")),
|
||||
header: all("header.top button").map((button) => button.id),
|
||||
tabs: all("#tabs > .tab-group > button").map(label),
|
||||
subs: all("#tabs .subtabs button").map(label),
|
||||
tabs: all("#tabs > .tab-group > button").map(name),
|
||||
subs: all("#tabs .subtabs button").map(name),
|
||||
badges: Object.fromEntries(all("#tabs button").map((button) => [button.id, badge(button)])),
|
||||
banner: banner.hidden ? null : text(banner),
|
||||
rail: Boolean(document.getElementById("rail")),
|
||||
views: all('#main [aria-label="Ansicht"] button').map(text),
|
||||
chips: all('#main [aria-label="Kategorien"] button').map(text),
|
||||
menus: all("#main details.menu > summary").map(text),
|
||||
views: all('#main [aria-label="Ansicht"] button').map(name),
|
||||
layouts: all('#main [aria-label="Anordnung"] button').map(name),
|
||||
chips: all('#main [aria-label="Kategorien"] button').map(name),
|
||||
menus: all("#main details.menu > summary").map(name),
|
||||
time: text(document.querySelector("#time > summary")),
|
||||
replay: Boolean(document.querySelector("#time .replay input[type=range]")),
|
||||
custom: Boolean(document.querySelector('#main details.menu input[type="datetime-local"]')),
|
||||
toolbar: all('#main [role="toolbar"] > *').map((node) => (node.tagName === "DETAILS" ? text(node.querySelector("summary")) : text(node))),
|
||||
toolbar: all('#main [role="toolbar"] > *').map((node) => name(node.tagName === "DETAILS" ? node.querySelector("summary") : node.querySelector("button") || node)),
|
||||
toAll: all("#to-all input[type=radio]").map((input) => [input.value, input.checked]),
|
||||
inbox: all("#main .inbox-list > li").length,
|
||||
icons: icons.length,
|
||||
unlabelled: icons.map((svg) => svg.parentElement).filter((owner) => !text(owner) && !(owner.getAttribute("title") && owner.getAttribute("aria-label"))).map((owner) => owner.outerHTML.slice(0, 160)),
|
||||
broken: all("svg.bi use").filter((use) => !document.getElementById(use.getAttribute("href").slice(1))).map((use) => use.getAttribute("href")),
|
||||
states: all("#main span.state").map(name),
|
||||
deletes: all('#main button[aria-label="Löschen"], #main button[aria-label="Alle gemergten löschen"]').map((button) => [name(button), button.disabled, button.title]),
|
||||
spawn: all("#main .spawn-tree li").map((item) => [item.dataset.kind, item.dataset.running || null]),
|
||||
marks: all("#main svg.chart .node").length,
|
||||
note: text(document.querySelector("#main .space-note")),
|
||||
canvas: Boolean(document.querySelector("#main .graph3d canvas")),
|
||||
wide: document.documentElement.scrollWidth > window.innerWidth + 1,
|
||||
};
|
||||
})()"""
|
||||
ACTIONS = """(async () => {
|
||||
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
|
||||
})()""".replace("@HELPERS@", HELPERS)
|
||||
ACTIONS = """(async () => {@HELPERS@
|
||||
const press = async (key) => {
|
||||
document.dispatchEvent(new KeyboardEvent("keydown", {key, bubbles: true}));
|
||||
await sleep(300);
|
||||
@@ -60,22 +90,79 @@ ACTIONS = """(async () => {
|
||||
const found = {keys: [await press("2"), await press("3"), await press("1")]};
|
||||
document.getElementById("settings-open").click();
|
||||
await sleep(300);
|
||||
found.ambience = [...document.querySelectorAll(".sheet-box form > section:first-child button")].map((button) => button.id);
|
||||
found.speech = [...document.querySelectorAll('.sheet-box [aria-label="Klassen"] input')].map((input) => [input.parentElement.textContent.trim(), input.checked, input.disabled]);
|
||||
found.ambience = all(".sheet-box form > section:first-child button").map((button) => button.id);
|
||||
found.speech = all('.sheet-box [aria-label="Klassen"] input').map((input) => [input.parentElement.textContent.trim(), input.checked, input.disabled]);
|
||||
found.settingsUnlabelled = all(".sheet-box svg.bi").map((svg) => svg.parentElement).filter((owner) => !text(owner) && !(owner.getAttribute("title") && owner.getAttribute("aria-label"))).length;
|
||||
document.querySelector(".sheet-head button").click();
|
||||
document.getElementById("palette-open").click();
|
||||
await sleep(300);
|
||||
found.palette = [...document.querySelectorAll(".palette-list li > span:first-child")].map((span) => span.textContent.trim());
|
||||
found.palette = all(".palette-list li > span:first-child").map(text);
|
||||
document.dispatchEvent(new KeyboardEvent("keydown", {key: "Escape", bubbles: true}));
|
||||
await sleep(200);
|
||||
document.querySelector('#main [aria-label="Kategorien"] button:nth-child(2)').click();
|
||||
await sleep(900);
|
||||
const someone = document.querySelector('#main button.linkish[title$="öffnen"]');
|
||||
document.querySelector('#main button.linkish[title$="öffnen"]').click();
|
||||
await sleep(500);
|
||||
found.drawer = all("#drawer input[type=radio]").map((input) => [input.value, input.checked]);
|
||||
found.drawerUnlabelled = all("#drawer svg.bi").map((svg) => svg.parentElement).filter((owner) => !text(owner) && !(owner.getAttribute("title") && owner.getAttribute("aria-label"))).length;
|
||||
return found;
|
||||
})()""".replace("@HELPERS@", HELPERS)
|
||||
PICK_3D = """(async () => {@HELPERS@
|
||||
const list = await until(() => document.querySelector("#main .space-list"), 8000);
|
||||
if (!list) {
|
||||
return {error: text(document.querySelector("#main .graph"))};
|
||||
}
|
||||
list.open = true;
|
||||
const chip = all("#main .space-list [data-node]").find((node) => node.dataset.node.startsWith("@PREFIX@"));
|
||||
if (!chip) {
|
||||
return {error: all("#main .space-list [data-node]").map((node) => node.dataset.node)};
|
||||
}
|
||||
chip.click();
|
||||
await sleep(600);
|
||||
return {panel: text(document.querySelector("#main aside .panel-title")), label: chip.textContent};
|
||||
})()""".replace("@HELPERS@", HELPERS)
|
||||
CALL = """(async () => {@HELPERS@
|
||||
document.querySelector('#main [aria-label="Kategorien"] button[aria-label="Funk"]').click();
|
||||
const someone = await until(() => document.querySelector('#main button.linkish[title$="öffnen"]'), 8000);
|
||||
someone.click();
|
||||
await sleep(500);
|
||||
found.drawer = [...document.querySelectorAll("#drawer input[type=radio]")].map((input) => [input.value, input.checked]);
|
||||
return found;
|
||||
})()"""
|
||||
document.querySelector('#drawer input[type=radio][value="call"]').click();
|
||||
const field = document.querySelector("#drawer textarea");
|
||||
field.value = "E2E call ping";
|
||||
field.dispatchEvent(new Event("input", {bubbles: true}));
|
||||
document.querySelector('#drawer button[type="submit"]').click();
|
||||
const entry = await until(() => all("#main .entry").find((row) => row.textContent.includes("Gelesen") && row.textContent.includes("E2E call ping")), 20000);
|
||||
return {to: text(someone), receipt: entry ? text(entry) : null, sent: all("#main .entry").filter((row) => row.textContent.includes("E2E call ping")).length};
|
||||
})()""".replace("@HELPERS@", HELPERS)
|
||||
OPEN_DELETE = """(async () => {@HELPERS@
|
||||
const card = await until(() => all("#main details.repo").find((details) => details.textContent.includes("github.com/me/demo")), 8000);
|
||||
if (!card) {
|
||||
return {error: text(document.getElementById("main"))};
|
||||
}
|
||||
card.open = true;
|
||||
await sleep(300);
|
||||
const every = document.querySelector('#main button[aria-label="Alle gemergten löschen"]');
|
||||
const buttons = all('button[aria-label="Löschen"]', card);
|
||||
const states = all("span.state", card).map(name);
|
||||
buttons.find((button) => !button.disabled).click();
|
||||
await sleep(500);
|
||||
return {states, every: [every.disabled, every.title], buttons: buttons.map((button) => [button.disabled, button.title]), title: text(document.querySelector(".sheet-head .panel-title")), dry: all('.sheet-box [aria-label="Trockenlauf"] pre').map(text)};
|
||||
})()""".replace("@HELPERS@", HELPERS)
|
||||
CONFIRM = """(async () => {@HELPERS@
|
||||
all(".sheet-box button").find((button) => name(button) === "Löschen").click();
|
||||
const done = await until(() => document.querySelector(".sheet-box [data-result]"), 30000);
|
||||
const job = document.querySelector(".sheet-box [data-cleanup]");
|
||||
return {job: job ? job.dataset.cleanup : null, result: done ? done.dataset.result : null, text: text(done), toasts: all("#toasts .toast").map(text)};
|
||||
})()""".replace("@HELPERS@", HELPERS)
|
||||
AFTER = """(async () => {@HELPERS@
|
||||
document.querySelector(".sheet-head button").click();
|
||||
const past = await until(() => all("#main details.repo").find((details) => details.textContent.includes("Zuletzt gelöscht")), 15000);
|
||||
return {past: past ? [past.querySelector("[data-result]").dataset.result, name(past.querySelector("span.state"))] : null, branches: all("#main details.repo").filter((details) => details !== past).flatMap((details) => all(".branch-name strong", details)).map(text)};
|
||||
})()""".replace("@HELPERS@", HELPERS)
|
||||
|
||||
|
||||
def labelled(probe: dict) -> dict:
|
||||
return {key: probe[key] for key in ("unlabelled", "broken")}
|
||||
|
||||
|
||||
@unittest.skipUnless(shutil.which(os.environ.get("CHROMIUM", "chromium")), "no Chromium")
|
||||
@@ -97,24 +184,55 @@ class TestOrchestratorE2E(HubCase):
|
||||
{"ts": now - 60, "from": "Docs", "to": [], "text": "MAYDAY disk full", "class": "mayday", "v": 2},
|
||||
]),
|
||||
})
|
||||
self.subagents(now)
|
||||
self.request("feat: orchestrator", "")
|
||||
self.create_ask(ASK)
|
||||
self.write_tree()
|
||||
self.root = self.write_tree()
|
||||
self.base = f"http://{self.host}/"
|
||||
|
||||
def drive(self, steps: list[dict]) -> dict:
|
||||
def subagents(self, now: float) -> None:
|
||||
def line(ts: float) -> dict:
|
||||
return {"type": "user", "sessionId": "s1", "timestamp": iso(ts), "message": {"role": "user", "content": "x"}}
|
||||
|
||||
folder = self.projects / "-repo" / "s1" / "subagents"
|
||||
folder.mkdir(parents=True)
|
||||
(self.projects / "-repo" / "s1.jsonl").write_text(self.jsonl([line(now - 3600), line(now - 30)]), encoding="utf-8")
|
||||
for ident, finished in (("abc", True), ("def", False)):
|
||||
(folder / f"agent-{ident}.meta.json").write_text(json.dumps({"agentType": "Explore", "description": f"look {ident}", "spawnDepth": 1, "finished": finished}), encoding="utf-8")
|
||||
(folder / f"agent-{ident}.jsonl").write_text(self.jsonl([line(now - 1800), line(now - 60)]), encoding="utf-8")
|
||||
|
||||
def drive(self, steps: list[dict], authenticator: dict | None, driver: str = "cdp.mjs", binary: str = os.environ.get("CHROMIUM", "chromium")) -> dict:
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
|
||||
json.dump({"steps": steps}, handle)
|
||||
json.dump({"steps": steps, "authenticator": authenticator}, handle)
|
||||
self.addCleanup(os.unlink, handle.name)
|
||||
done = subprocess.run(["node", str(HERE / "cdp.mjs"), os.environ.get("CHROMIUM", "chromium"), handle.name], capture_output=True, text=True, timeout=600, check=False)
|
||||
done = subprocess.run(["node", str(HERE / driver), binary, handle.name], capture_output=True, text=True, timeout=900, check=False)
|
||||
self.assertEqual(done.returncode, 0, done.stderr[-2000:])
|
||||
return json.loads(done.stdout)
|
||||
|
||||
def beside(self, work: Callable[[], None]) -> None:
|
||||
stop = threading.Event()
|
||||
|
||||
def loop() -> None:
|
||||
while not stop.wait(0.2):
|
||||
work()
|
||||
|
||||
thread = threading.Thread(target=loop, daemon=True)
|
||||
thread.start()
|
||||
self.addCleanup(thread.join, 10)
|
||||
self.addCleanup(stop.set)
|
||||
|
||||
def shot(self, name: str) -> str:
|
||||
return str(OUT / f"chromium-{name}.png")
|
||||
|
||||
def test_chromium_walks_every_screen_at_three_widths(self):
|
||||
steps = [{"width": width, "height": height, "url": f"{self.base}#{screen}", "wait": 900, "probe": PROBE, "shot": str(OUT / f"chromium-{width}-{screen.replace('/', '-')}.png")} for width, height in WIDTHS for screen in SCREENS]
|
||||
steps = [{"width": width, "height": height, "url": f"{self.base}#{screen}", "wait": 900, "probe": PROBE, "shot": self.shot(f"{width}-{screen.replace('/', '-')}")} for width, height in WIDTHS for screen in SCREENS]
|
||||
steps += [{"width": 1366, "height": 768, "url": f"{self.base}#{old}", "wait": 600, "probe": PROBE, "shot": None} for old in ("graph", "graph/chord", "tree")]
|
||||
steps.append({"width": 1366, "height": 768, "url": f"{self.base}#stream", "wait": 900, "probe": ACTIONS, "shot": None})
|
||||
data = self.drive(steps)
|
||||
steps.append({"width": 1366, "height": 768, "url": f"{self.base}#artifacts/3d", "wait": 900, "probe": PICK_3D.replace("@PREFIX@", "branch:"), "shot": self.shot("1366-artifacts-3d-picked")})
|
||||
steps.append({"width": 1366, "height": 768, "url": f"{self.base}#agents/3d", "wait": 900, "probe": PICK_3D.replace("@PREFIX@", "s2"), "shot": self.shot("1366-agents-3d-picked")})
|
||||
steps.append({"width": 1366, "height": 768, "url": f"{self.base}#artifacts/3d", "wait": 900, "probe": PROBE, "shot": self.shot("1366-artifacts-3d-reduced-motion"), "media": [{"name": "prefers-reduced-motion", "value": "reduce"}]})
|
||||
data = self.drive(steps, None)
|
||||
(OUT / "chromium-walk.json").write_text(json.dumps(data, indent=1), encoding="utf-8")
|
||||
self.assertEqual(data["errors"], [])
|
||||
probes = data["results"][: len(WIDTHS) * len(SCREENS)]
|
||||
for probe, step in zip(probes, steps, strict=False):
|
||||
@@ -124,46 +242,149 @@ class TestOrchestratorE2E(HubCase):
|
||||
self.assertEqual(probe["header"], ["palette-open", "settings-open"])
|
||||
self.assertEqual((probe["tabs"], probe["subs"]), (["Stream", "Agents", "Artifacts"], ["Inbox", "Regeln"]))
|
||||
self.assertEqual(probe["badges"], {"tab-stream": None, "tab-inbox": "2", "tab-rules": None, "tab-agents": "1", "tab-artifacts": None})
|
||||
self.assertEqual(probe["banner"], "MAYDAY · Docs: disk fullaufheben")
|
||||
self.assertEqual(probe["banner"], "MAYDAY · Docs: disk full")
|
||||
self.assertFalse(probe["rail"])
|
||||
self.assertEqual((probe["time"], probe["replay"]), ("24 h ▾", True))
|
||||
self.assertGreater(probe["icons"], 8)
|
||||
self.assertEqual(labelled(probe), {"unlabelled": [], "broken": []})
|
||||
self.assertFalse(probe["wide"], "the page scrolls sideways")
|
||||
by_hash = {probe["hash"]: probe for probe in probes}
|
||||
stream = by_hash["#stream"]
|
||||
self.assertEqual([probe["chips"] for probe in probes if probe["hash"] == "#stream"], [["Für dich", "Funk", "Arbeit", "Transkripte"]] * len(WIDTHS))
|
||||
self.assertEqual((stream["menus"], stream["custom"]), (["Filter ▾"], True))
|
||||
self.assertEqual((stream["menus"], stream["custom"]), (["Filter"], True))
|
||||
self.assertEqual(by_hash["#stream/inbox"]["inbox"], 2)
|
||||
for tab in ("agents", "artifacts"):
|
||||
with self.subTest(tab=tab):
|
||||
self.assertEqual(by_hash[f"#{tab}/orbit"]["views"], ["Orbit", "Chord", "Swimlane", "Tree", "3D"])
|
||||
self.assertEqual(by_hash[f"#{tab}/3d"]["layouts"], ["Kraft", "Zeit", "Spawn-Tiefe"])
|
||||
self.assertTrue(by_hash[f"#{tab}/3d"]["canvas"], by_hash[f"#{tab}/3d"]["note"])
|
||||
self.assertRegex(by_hash[f"#{tab}/3d"]["note"], r"^\d+ Knoten · \d+ Kanten · Kraft$")
|
||||
agents = by_hash["#agents/orbit"]
|
||||
self.assertEqual(agents["toolbar"], ["Funk an alle ▾", "Mayday aufheben", "Priorität", "Tag/Nacht ▾"])
|
||||
self.assertEqual(agents["toolbar"], ["Funk an alle", "Mayday aufheben", "Priorität", "Tag/Nacht"])
|
||||
self.assertEqual(agents["toAll"], [["newsletter", True], ["broadcast", False], ["pan-pan", False], ["security", False], ["mayday", False]])
|
||||
redirects = data["results"][len(probes) : len(probes) + 3]
|
||||
self.assertEqual([probe["hash"] for probe in redirects], ["#agents/orbit", "#agents/chord", "#artifacts/tree"])
|
||||
actions = data["results"][-1]
|
||||
self.assertEqual(by_hash["#agents/tree"]["spawn"][0], ["operator", None])
|
||||
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"], [["Alle gemergten löschen", True, "Alle gemergten löschen gesperrt: Secret Service locked"], ["Löschen", True, "Löschen gesperrt: Secret Service locked"]])
|
||||
for state in ("gemergt durch #12 in main", "Upstream weg", "geschützt", "1 Commits ungepusht", "sauber", "nicht committet: 0 gestaged, 0 geändert, 1 neu", "gc überspringt: ignored files beyond known caches: notes.txt"):
|
||||
with self.subTest(state=state):
|
||||
self.assertIn(state, artifacts["states"])
|
||||
for view in ("orbit", "chord"):
|
||||
with self.subTest(view=view):
|
||||
self.assertGreaterEqual(by_hash[f"#artifacts/{view}"]["marks"], 4)
|
||||
extra = data["results"][len(probes) :]
|
||||
self.assertEqual([probe["hash"] for probe in extra[:3]], ["#agents/orbit", "#agents/chord", "#artifacts/tree"])
|
||||
actions = extra[3]
|
||||
self.assertNotIn("error", actions)
|
||||
self.assertEqual(actions["keys"], ["#agents/chord", "#artifacts/tree", "#stream"])
|
||||
self.assertEqual(actions["ambience"], ["sky-look", "music-toggle", "leitstand", "notify"])
|
||||
self.assertEqual((actions["settingsUnlabelled"], actions["drawerUnlabelled"]), (0, 0))
|
||||
self.assertEqual([name for name, _, _ in actions["speech"]], ["Post", "Call", "Newsletter", "Broadcast", "MAYDAY", "PAN-PAN", "SECURITE"])
|
||||
self.assertEqual([entry[1:] for entry in actions["speech"] if entry[0] == "MAYDAY"], [[True, True]])
|
||||
for tool in ("Funk an alle …", "Mayday aufheben", "Priorität …", "Abend", "Regeln …", "Einstellungen …", "Himmel ansehen", "Leitstand umschalten", "Benachrichtigungen aktivieren"):
|
||||
with self.subTest(tool=tool):
|
||||
self.assertTrue(any(tool in entry for entry in actions["palette"]), actions["palette"])
|
||||
self.assertEqual(actions["drawer"], [["post", True], ["call", False]])
|
||||
artifact_pick, agent_pick, calm = extra[4:]
|
||||
self.assertEqual(artifact_pick["panel"], artifact_pick["label"])
|
||||
self.assertEqual(agent_pick["panel"], agent_pick["label"])
|
||||
self.assertRegex(calm["note"], r"Kraft · ohne Bewegung \(prefers-reduced-motion\)$")
|
||||
|
||||
def test_chromium_sends_a_call_from_the_agent_panel_and_shows_its_receipt(self):
|
||||
handled = []
|
||||
|
||||
def answer() -> None:
|
||||
for script, args in self.scripts.calls[len(handled) :]:
|
||||
handled.append(script)
|
||||
if script.endswith("radio.py") and args[:2] == ["send", "call"]:
|
||||
to, text, now = args[args.index("--to") + 1], args[-1], time.time()
|
||||
rows = [{"ts": now, "from": "Operator", "to": [to], "text": text, "class": "call", "v": 2}, {"ts": now + 0.5, "from": to, "to": ["Operator"], "text": text, "class": "receipt", "v": 2, "call": now}]
|
||||
with (self.council / "news.jsonl").open("a", encoding="utf-8") as handle:
|
||||
handle.write(self.jsonl(rows))
|
||||
|
||||
self.beside(answer)
|
||||
data = self.drive([{"width": 1366, "height": 768, "url": f"{self.base}#stream", "wait": 900, "probe": CALL, "shot": self.shot("1366-call-receipt")}], None)
|
||||
self.assertEqual(data["errors"], [])
|
||||
result = data["results"][0]
|
||||
self.assertNotIn("error", result)
|
||||
self.assertIsNotNone(result["receipt"], result)
|
||||
self.assertIn("Gelesen", result["receipt"])
|
||||
|
||||
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, ""
|
||||
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)
|
||||
fake = self.outside / "bin" / "pkgmgr"
|
||||
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")
|
||||
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")
|
||||
signer = self.pin()
|
||||
trust = verifier.Trust(approvers=self.approvers, replay=self.replay / "cleanup.json", origin=self.origin)
|
||||
runner = host.Cleanup(trust, self.tree_file, fake)
|
||||
jobs, results = self.cleanup / "jobs", self.cleanup / "results"
|
||||
self.beside(lambda: any(jobs.iterdir()) and host.cycle(jobs, results, runner))
|
||||
private = "".join(line for line in signer.pem.read_text(encoding="utf-8").splitlines() if not line.startswith("-----"))
|
||||
authenticator = {"credentialId": base64.b64encode(signer.raw_id).decode(), "privateKey": private, "rpId": "localhost"}
|
||||
local = f"http://localhost:{self.port}/"
|
||||
steps = [
|
||||
{"width": 1366, "height": 768, "url": f"{local}#artifacts/tree", "wait": 1200, "probe": OPEN_DELETE, "shot": self.shot("1366-delete-dry-run")},
|
||||
{"width": 1366, "height": 768, "url": None, "wait": 0, "probe": CONFIRM, "shot": self.shot("1366-delete-result")},
|
||||
{"width": 1366, "height": 768, "url": None, "wait": 0, "probe": AFTER, "shot": self.shot("1366-delete-tree-after")},
|
||||
]
|
||||
data = self.drive(steps, authenticator)
|
||||
(OUT / "chromium-delete.json").write_text(json.dumps(data, indent=1), encoding="utf-8")
|
||||
self.assertEqual(data["errors"], [])
|
||||
opened, confirmed, after = data["results"]
|
||||
self.assertNotIn("error", opened)
|
||||
self.assertEqual(opened["every"], [False, "Alle gemergten löschen: 1 Branches nach Trockenlauf und Signatur"])
|
||||
self.assertEqual(opened["buttons"], [[False, "Löschen: feature/done lokal entfernen, nach Trockenlauf und Signatur"]])
|
||||
self.assertEqual(opened["title"], "feature/done löschen")
|
||||
self.assertEqual(len(opened["dry"]), 1)
|
||||
self.assertTrue(opened["dry"][0].startswith("feature/done @ 5b1e6fd3a7c0: merged-clean, gemergt durch #12"), opened["dry"])
|
||||
self.assertIn("bleibt: remote-branch (nur lokal)", opened["dry"][0])
|
||||
self.assertEqual((confirmed["result"], confirmed["text"]), ("done", "feature/donegelöscht"), confirmed)
|
||||
self.assertEqual(confirmed["toasts"], ["Löschen: erledigtfeature/done: gelöscht"])
|
||||
self.assertEqual(after["past"], ["done", "gelöscht"])
|
||||
self.assertNotIn("feature/done", after["branches"])
|
||||
written = [json.loads(path.read_text(encoding="utf-8")) for path in results.glob("*.json")]
|
||||
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", "tree", "--format", "json", "--output", str(self.tree_file)],
|
||||
])
|
||||
|
||||
@unittest.skipUnless(shutil.which(os.environ.get("FIREFOX", "firefox")), "no Firefox")
|
||||
def test_firefox_renders_every_screen_at_three_widths(self):
|
||||
profile = Path(tempfile.mkdtemp(prefix="firefox-e2e-"))
|
||||
self.addCleanup(shutil.rmtree, profile, True)
|
||||
for width, height in WIDTHS:
|
||||
for screen in SCREENS:
|
||||
shot = OUT / f"firefox-{width}-{screen.replace('/', '-')}.png"
|
||||
shot.unlink(missing_ok=True)
|
||||
with self.subTest(width=width, screen=screen):
|
||||
command = [os.environ.get("FIREFOX", "firefox"), "--headless", "--no-remote", "--profile", str(profile), "--window-size", f"{width},{height}", "--screenshot", str(shot), f"{self.base}#{screen}"]
|
||||
subprocess.run(command, capture_output=True, timeout=120, check=False)
|
||||
self.assertTrue(shot.is_file() and shot.stat().st_size > 1000, shot)
|
||||
def test_firefox_walks_every_screen_at_three_widths(self):
|
||||
shots = [(width, height, screen, OUT / f"firefox-{width}-{screen.replace('/', '-')}.png") for width, height in WIDTHS for screen in SCREENS]
|
||||
for *_, shot in shots:
|
||||
shot.unlink(missing_ok=True)
|
||||
steps = [{"width": width, "height": height, "url": f"{self.base}#{screen}", "wait": 900, "probe": PROBE, "shot": str(shot)} for width, height, screen, shot in shots]
|
||||
data = self.drive(steps, None, "bidi.mjs", os.environ.get("FIREFOX", "firefox"))
|
||||
(OUT / "firefox-walk.json").write_text(json.dumps(data, indent=1), encoding="utf-8")
|
||||
blocked = [error for error in data["errors"] if error.startswith("Content-Security-Policy")]
|
||||
self.assertEqual([error for error in data["errors"] if error not in blocked], [])
|
||||
# 3d-force-graph injects its own <style>, which the unchanged CSP blocks; the page carries those rules itself.
|
||||
self.assertEqual([error for error in blocked if "(style-src-elem)" not in error and "/favicon.ico" not in error], [])
|
||||
for probe, (width, _, screen, shot) in zip(data["results"], shots, strict=True):
|
||||
with self.subTest(width=width, screen=screen):
|
||||
self.assertNotIn("error", probe)
|
||||
self.assertEqual(probe["hash"], f"#{screen}")
|
||||
self.assertEqual((probe["tabs"], probe["subs"]), (["Stream", "Agents", "Artifacts"], ["Inbox", "Regeln"]))
|
||||
self.assertEqual(labelled(probe), {"unlabelled": [], "broken": []})
|
||||
self.assertFalse(probe["wide"], "the page scrolls sideways")
|
||||
self.assertTrue(shot.is_file() and shot.stat().st_size > 1000, shot)
|
||||
if screen.endswith("/3d"):
|
||||
self.assertTrue(probe["canvas"], probe["note"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
+2
-1
@@ -42,6 +42,7 @@ class HubCase(unittest.TestCase):
|
||||
self.projects = base / "projects"
|
||||
self.tree_file = base / "tree" / "branch-tree.json"
|
||||
self.timers = base / "timers"
|
||||
self.cleanup = base / "state" / "cleanup"
|
||||
self.approvers = base / "config" / "approvers.json"
|
||||
self.replay = base / "replay"
|
||||
for path in (self.pages, self.state, self.outside, self.projects, self.tree_file.parent, self.timers):
|
||||
@@ -63,7 +64,7 @@ class HubCase(unittest.TestCase):
|
||||
transcripts=self.projects,
|
||||
scripts=self.scripts,
|
||||
templates=HUB / "templates",
|
||||
places=server.Places(tree_file=self.tree_file, timers=self.timers),
|
||||
places=server.Places(tree_file=self.tree_file, timers=self.timers, vendor=HUB / "vendor", cleanup=self.cleanup),
|
||||
model=server.Model(url="", model="", key=""),
|
||||
engines=self.engines,
|
||||
approvers=self.approvers,
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
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'<svg xmlns="http://www.w3.org/2000/svg" class="sprite" aria-hidden="true" focusable="false">(.*?)</svg>', html, re.DOTALL)
|
||||
self.assertIsNotNone(sprite)
|
||||
inlined = re.findall(r'<symbol id="bi-([a-z0-9-]+)" viewBox="0 0 16 16">', 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'<use href="#bi-{name}"></use>', 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()
|
||||
@@ -0,0 +1,250 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
import secrets
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from hub_case import HubCase
|
||||
from hub_support import 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)
|
||||
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.
|
||||
FAKE_PKGMGR = """#!{python}
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
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"]:
|
||||
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")
|
||||
raise SystemExit(0)
|
||||
print(json.dumps({{"items": json.loads((here / "plan.json").read_text(encoding="utf-8"))}}))
|
||||
"""
|
||||
|
||||
|
||||
class TestHubCleanup(HubCase):
|
||||
def ready(self, ready: bool) -> dict:
|
||||
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"
|
||||
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)
|
||||
|
||||
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 sign(self, items: list[dict]) -> tuple[int, bytes]:
|
||||
status, _, body = self.post_json("/api/sign", {"items": items}, self.guard(self.host))
|
||||
return status, body
|
||||
|
||||
def test_the_tree_offers_a_delete_only_for_merged_clean_branches_without_skip_reason(self):
|
||||
item = self.ready(False)
|
||||
branches = self.branches()
|
||||
done = branches["feature/done"]
|
||||
self.assertEqual(done["cleanup"], item)
|
||||
self.assertEqual(done["plan"], [
|
||||
"feature/done @ 5b1e6fd3a7c0: merged-clean, gemergt durch #12",
|
||||
f"entfernt den Worktree {item['worktree']}",
|
||||
"löscht den lokalen Branch feature/done",
|
||||
"bleibt: remote-branch (nur lokal)",
|
||||
"stoppt den Compose-Stack demo-feature-done ohne -v",
|
||||
"entfernt das Image demo:feature-done",
|
||||
])
|
||||
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)
|
||||
status, body = self.sign([item])
|
||||
self.assertEqual((status, body.decode()), (400, "löschen gesperrt: Secret Service locked\n"))
|
||||
item = self.ready(True)
|
||||
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):
|
||||
self.assertEqual(self.sign(items)[0], 400)
|
||||
status, body = self.sign([item])
|
||||
self.assertEqual(status, 200, body)
|
||||
self.assertEqual(json.loads(body)["payload"]["items"], [signing.item_hash(item)])
|
||||
|
||||
def test_a_signed_delete_becomes_a_job_without_any_command(self):
|
||||
item = self.ready(True)
|
||||
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)
|
||||
self.assertEqual(self.post_json("/api/cleanup", {"items": [item], "proof": proof}, {"Host": self.host})[0], 403)
|
||||
status, _, body = self.post_json("/api/cleanup", {"items": [item], "proof": proof}, self.guard(self.host))
|
||||
self.assertEqual(status, 202, body)
|
||||
job_id = json.loads(body)["id"]
|
||||
written = self.cleanup / "jobs" / f"{job_id}.json"
|
||||
job = json.loads(written.read_text(encoding="utf-8"))
|
||||
self.assertEqual(sorted(job), ["created", "id", "items", "proof", "v"])
|
||||
self.assertEqual(job["items"], [{key_name: item[key_name] for key_name in ("repository", "branch", "worktree", "tip")}])
|
||||
self.assertEqual(job["proof"], proof)
|
||||
self.assertNotIn("command", written.read_text(encoding="utf-8"))
|
||||
self.assertEqual((written.stat().st_mode & 0o777, (self.cleanup / "jobs").stat().st_mode & 0o777), (0o600, 0o700))
|
||||
self.assertEqual(json.loads(self.get(f"/api/cleanup/{job_id}")[2])["status"], "queued")
|
||||
self.assertEqual(self.get("/api/cleanup/0000000000000000")[0], 404)
|
||||
result = {"id": job_id, "finished": time.time(), "items": [{**job["items"][0], "status": "done", "reason": ""}]}
|
||||
(self.cleanup / "results" / f"{job_id}.json").write_text(json.dumps(result), encoding="utf-8")
|
||||
self.assertEqual(json.loads(self.get(f"/api/cleanup/{job_id}")[2])["status"], "done")
|
||||
self.assertEqual(json.loads(self.get("/api/tree")[2])["cleanups"], [result])
|
||||
|
||||
|
||||
class TestHostCleanup(unittest.TestCase):
|
||||
def setUp(self):
|
||||
tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(tmp.cleanup)
|
||||
base = Path(tmp.name)
|
||||
self.repo = base / "repos" / "demo"
|
||||
self.repo.mkdir(parents=True)
|
||||
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(FAKE_PKGMGR.format(python=sys.executable), encoding="utf-8")
|
||||
self.fake.chmod(0o755)
|
||||
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.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)
|
||||
self.item = {"repository": str(self.repo), "branch": "feature/done", "worktree": f"{self.repo}/.worktrees/feature-done", "tip": DONE_TIP}
|
||||
|
||||
def proof(self, items: list[dict], signer) -> dict:
|
||||
hashes = [signing.item_hash(signing.cleanup_item(**item)) for item in items]
|
||||
return signer.proof(signing.approval_payload(hashes, secrets.token_hex(16), int(time.time()) + 600), ORIGIN)
|
||||
|
||||
def entry(self, **changes: object) -> dict:
|
||||
return {"branch": self.item["branch"], "worktree": self.item["worktree"], "local_sha": DONE_TIP, "head": DONE_TIP, "state": "merged-clean", "skip": None, **changes}
|
||||
|
||||
def run_job(self, items: list[dict], proof: object, plan: list[dict]) -> tuple[dict, list[dict]]:
|
||||
(self.fake.parent / "plan.json").write_text(json.dumps(plan), encoding="utf-8")
|
||||
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)
|
||||
self.assertEqual(list(self.jobs.iterdir()), [])
|
||||
calls = [json.loads(line) for line in self.calls.read_text(encoding="utf-8").splitlines()] if self.calls.exists() else []
|
||||
self.calls.unlink(missing_ok=True)
|
||||
return json.loads((self.results / f"{job_id}.json").read_text(encoding="utf-8")), calls
|
||||
|
||||
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"]
|
||||
self.assertEqual(calls, [
|
||||
{"argv": ["branch", "cleanup", *selector], "cwd": str(self.repo)},
|
||||
{"argv": ["branch", "cleanup", "--apply", *selector], "cwd": str(self.repo)},
|
||||
{"argv": ["branch", "tree", "--format", "json", "--output", str(self.tree)], "cwd": str(self.tree.parent)},
|
||||
])
|
||||
self.assertEqual(json.loads(self.tree.read_text(encoding="utf-8")), {"rewritten": True})
|
||||
self.assertEqual(result["tree"], {"code": 0, "error": ""})
|
||||
|
||||
def test_every_doubtful_item_is_refused_and_changes_nothing(self):
|
||||
pattern = {**self.item, "branch": "feature/*"}
|
||||
option = {**self.item, "branch": "--all"}
|
||||
stranger = {**self.item, "repository": str(self.repo.parent.parent / "stranger")}
|
||||
rule = self.signer.proof(signing.rule_payload(secrets.token_hex(8), ["review"], "", "*", "*", int(time.time()) + 600), ORIGIN)
|
||||
cases = (
|
||||
("bad signature", [self.item], self.proof([self.item], key(-7, "intruder")), [self.entry()], "no valid operator signature", 0),
|
||||
("missing signature", [self.item], None, [self.entry()], "no signature", 0),
|
||||
("rule instead of approval", [self.item], rule, [self.entry()], "no valid operator signature", 0),
|
||||
("other item signed", [self.item], self.proof([{**self.item, "tip": OTHER_TIP}], self.signer), [self.entry()], "does not list this item", 0),
|
||||
("unknown repository", [stranger], self.proof([stranger], self.signer), [self.entry()], "unknown repository", 0),
|
||||
("pattern characters", [pattern], self.proof([pattern], self.signer), [self.entry(branch="feature/*")], "pattern or option characters", 0),
|
||||
("option name", [option], self.proof([option], self.signer), [self.entry(branch="--all")], "pattern or option characters", 0),
|
||||
("changed tip", [self.item], self.proof([self.item], self.signer), [self.entry(local_sha=OTHER_TIP)], "tip changed", 1),
|
||||
("not merged-clean", [self.item], self.proof([self.item], self.signer), [self.entry(state="dirty")], "state is dirty", 1),
|
||||
("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),
|
||||
)
|
||||
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.assertNotIn("tree", result)
|
||||
|
||||
def test_a_signature_deletes_once_and_a_malformed_job_nothing(self):
|
||||
proof = self.proof([self.item], self.signer)
|
||||
self.assertEqual(self.run_job([self.item], proof, [self.entry()])[0]["items"][0]["status"], "done")
|
||||
again, calls = self.run_job([self.item], proof, [self.entry()])
|
||||
self.assertEqual((again["items"][0]["status"], calls), ("refused", []))
|
||||
self.assertIn("already used", again["items"][0]["reason"])
|
||||
for name, items in (("command smuggled in", [{**self.item, "command": "rm -rf ~"}]), ("relative repository", [{**self.item, "repository": "demo"}]), ("no items", [])):
|
||||
with self.subTest(name=name):
|
||||
result, calls = self.run_job(items, proof, [self.entry()])
|
||||
self.assertEqual((result["items"], calls), ([], []))
|
||||
self.assertIn("malformed job", result["error"])
|
||||
|
||||
def test_the_unit_runs_the_script_once_over_the_job_directory(self):
|
||||
(self.fake.parent / "plan.json").write_text(json.dumps([self.entry()]), encoding="utf-8")
|
||||
job_id = secrets.token_hex(8)
|
||||
(self.jobs / f"{job_id}.json").write_text(json.dumps({"v": 1, "id": job_id, "created": "x", "items": [self.item], "proof": self.proof([self.item], self.signer)}), encoding="utf-8")
|
||||
(self.jobs / "stray.tmp").write_text("x", encoding="utf-8")
|
||||
trust = ["--approvers", str(self.approvers), "--replay", str(self.trust.replay)]
|
||||
command = [sys.executable, str(CLEANUP), "--port", ORIGIN.rsplit(":", 1)[1], "--jobs", str(self.jobs), "--results", str(self.results), *trust, "--tree", str(self.tree), "--pkgmgr", str(self.fake)]
|
||||
done = subprocess.run(command, capture_output=True, text=True, timeout=60, check=False)
|
||||
self.assertEqual(done.returncode, 0, done.stderr)
|
||||
self.assertEqual(list(self.jobs.iterdir()), [])
|
||||
result = json.loads((self.results / f"{job_id}.json").read_text(encoding="utf-8"))
|
||||
self.assertEqual(result["items"][0]["status"], "done")
|
||||
self.assertEqual((self.results / f"{job_id}.json").stat().st_mode & 0o777, 0o600)
|
||||
|
||||
|
||||
class TestHostCleanupGitConfig(unittest.TestCase):
|
||||
def test_pkgmgr_runs_no_hook_the_repository_config_names(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
base = Path(tmp)
|
||||
repo, hooks, marker = base / "repo", base / "hooks", base / "hook-ran"
|
||||
hooks.mkdir()
|
||||
(hooks / "reference-transaction").write_text(f"#!/bin/sh\necho path >> {marker}\n", encoding="utf-8")
|
||||
(hooks / "reference-transaction").chmod(0o755)
|
||||
git = ["git", "-c", "user.name=t", "-c", "user.email=t@example.invalid", "-c", "commit.gpgsign=false"]
|
||||
subprocess.run([*git, "init", "-q", str(repo)], check=True)
|
||||
subprocess.run([*git, "commit", "-q", "--allow-empty", "-m", "x"], cwd=repo, check=True)
|
||||
subprocess.run(["git", "branch", "control"], cwd=repo, check=True)
|
||||
subprocess.run(["git", "branch", "doomed"], cwd=repo, check=True)
|
||||
subprocess.run(["git", "config", "core.hooksPath", str(hooks)], cwd=repo, check=True)
|
||||
subprocess.run(["git", "config", "hook.spy.event", "reference-transaction"], cwd=repo, check=True)
|
||||
subprocess.run(["git", "config", "hook.spy.command", f"echo config >> {marker}"], cwd=repo, check=True)
|
||||
subprocess.run(["git", "branch", "-D", "control"], cwd=repo, check=True, capture_output=True)
|
||||
self.assertEqual({line.split()[0] for line in marker.read_text(encoding="utf-8").splitlines()}, {"path", "config"})
|
||||
marker.unlink()
|
||||
fake = base / "pkgmgr"
|
||||
fake.write_text(f"#!{sys.executable}\nimport subprocess\nsubprocess.run(['git', 'branch', '-D', '--', 'doomed'], check=True)\n", encoding="utf-8")
|
||||
fake.chmod(0o755)
|
||||
code, _, error = host.run(fake, ["branch", "cleanup", "--apply"], str(repo))
|
||||
self.assertEqual(code, 0, error)
|
||||
self.assertFalse(marker.exists())
|
||||
branches = subprocess.run(["git", "branch", "--format=%(refname:short)"], cwd=repo, capture_output=True, text=True, check=True).stdout.split()
|
||||
self.assertNotIn("doomed", branches)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -175,7 +175,8 @@ class TestHubSpeech(HubCase):
|
||||
args = {
|
||||
"--bind": "127.0.0.1", "--listen-port": "0", "--public-port": "1", "--council": str(self.council), "--pages": str(self.pages), "--state": str(self.state),
|
||||
"--host-state": str(HOST_STATE), "--skills": str(self.outside), "--transcripts": str(self.projects), "--templates": str(HUB / "templates"),
|
||||
"--tree-file": str(self.tree_file), "--timers": str(self.timers), "--llm-key-file": os.devnull, "--llm-url": "", "--llm-model": "",
|
||||
"--tree-file": str(self.tree_file), "--timers": str(self.timers), "--vendor": str(HUB / "vendor"), "--cleanup": str(self.cleanup),
|
||||
"--llm-key-file": os.devnull, "--llm-url": "", "--llm-model": "",
|
||||
"--piper": str(self.engines.piper), "--voices": str(self.engines.voices), "--operator-voice": "en_US-test-high:9", "--whisper": str(self.engines.whisper),
|
||||
"--whisper-model": str(self.engines.whisper_model), "--scratch": str(self.engines.scratch), "--approvers": str(self.approvers),
|
||||
}
|
||||
@@ -184,7 +185,7 @@ class TestHubSpeech(HubCase):
|
||||
argv = [part for pair in given.items() for part in pair]
|
||||
return subprocess.run([sys.executable, str(HUB / "server.py"), *argv], capture_output=True, text=True, timeout=60, check=False)
|
||||
|
||||
for name in ("--piper", "--voices", "--operator-voice", "--whisper", "--whisper-model", "--scratch", "--approvers"):
|
||||
for name in ("--piper", "--voices", "--operator-voice", "--whisper", "--whisper-model", "--scratch", "--approvers", "--vendor", "--cleanup"):
|
||||
with self.subTest(name=name):
|
||||
done = run({key: value for key, value in args.items() if key != name})
|
||||
self.assertEqual(done.returncode, 2, done.stderr)
|
||||
|
||||
@@ -83,6 +83,26 @@ class TestPageMatchesScript(unittest.TestCase):
|
||||
found = re.findall(r'^\s*"?(\w+)"?:', block, re.MULTILINE)
|
||||
self.assertEqual(sorted(found), sorted(module.CHOICES[choices]))
|
||||
|
||||
def test_buttons_show_vendored_icons_with_their_former_text(self):
|
||||
page = PAGE.read_text(encoding="utf-8")
|
||||
sprite = (REPO_ROOT / "orchestrator" / "vendor" / "bootstrap-icons" / "bootstrap-icons.svg").read_text(encoding="utf-8")
|
||||
vendored = dict(re.findall(r'<symbol class="bi bi-[a-z0-9-]+" viewBox="0 0 16 16" id="([a-z0-9-]+)">(.*?)</symbol>', sprite, re.DOTALL))
|
||||
embedded = dict(re.findall(r'<symbol id="bi-([a-z0-9-]+)" viewBox="0 0 16 16">(.*?)</symbol>', page, re.DOTALL))
|
||||
used = re.findall(r'<use href="#bi-([a-z0-9-]+)">', page)
|
||||
self.assertTrue(used)
|
||||
self.assertEqual(sorted(embedded), sorted(set(used)))
|
||||
for name in used:
|
||||
with self.subTest(icon=name):
|
||||
self.assertEqual(embedded[name], vendored[name])
|
||||
buttons = re.findall(r"<button\b([^>]*)>(.*?)</button>", page, re.DOTALL)
|
||||
self.assertEqual([re.sub(r"<[^>]+>", "", inner).strip() for _, inner in buttons], [""])
|
||||
for attributes, inner in buttons:
|
||||
self.assertIn('<use href="#bi-play-fill">', inner)
|
||||
self.assertIn('title="Start"', attributes)
|
||||
self.assertIn('aria-label="Start"', attributes)
|
||||
self.assertNotIn("☀", page)
|
||||
self.assertIsNone(re.search(r'(?:src|href)="(?:https?:)?//', page))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user