diff options
Diffstat (limited to 'docs/superpowers/plans/2026-10-19-sb-builder-phase1.md')
| -rw-r--r-- | docs/superpowers/plans/2026-10-19-sb-builder-phase1.md | 2259 |
1 files changed, 0 insertions, 2259 deletions
diff --git a/docs/superpowers/plans/2026-10-19-sb-builder-phase1.md b/docs/superpowers/plans/2026-10-19-sb-builder-phase1.md deleted file mode 100644 index 15a1b5f..0000000 --- a/docs/superpowers/plans/2026-10-19-sb-builder-phase1.md +++ /dev/null @@ -1,2259 +0,0 @@ -# sb-builder Phase 1 (Core) Implementation Plan - -> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. - -**Goal:** A PyQt6 desktop GUI plus a stdlib VM helper that run `sbpull` and `sbshadow` as detachable jobs on the build VM, with a Dashboard, a Jobs tab, VM start/resume via virsh, TOML config and desktop notifications. - -**Architecture:** The GUI calls `ssh <target> sbb-remote <cmd>` through async QProcess (ControlMaster reused) and gets JSON back. `sbb-remote` runs a fixed table of job kinds, detached with `setsid` and `stdin=/dev/null`, one at a time under a flock, logging to `/var/lib/sb-builder/jobs/<id>/`; `job-follow <id> <offset>` streams a log so the GUI can drop and resume. Spec: `docs/superpowers/specs/2026-10-19-sb-builder-phase1-design.md`. - -**Tech Stack:** Python 3.11+ stdlib (`tomllib`, `fcntl`, `sqlite3`, `subprocess`), PyQt6 6.11 incl. QtDBus (stock on Slackware -current), `unittest`, bash, tmux not used. - -**Verified while planning:** all code below was run in a scratch copy: 38 tests pass; the sbshadow and sbpull changes were exercised in a throwaway repo; `sbb-remote stats` and `version` ran against the real VM data (warm 0.14s, cold 6.7s on first disk access); the real-ssh GUI path reported "sbb-remote missing" against the VM before deploy, as designed; the QtDBus `Notify` call was checked against the desktop's notification daemon (`actions` must be a typed `QStringList`, `replaces_id` a typed `uint`). - -**Rules for whoever executes this:** -- Commits are GPG-signed by global config. Never disable signing. A pause on the first commit is the smartcard PIN. -- Global pre-commit and commit-msg hooks scan for personal data. A rejection is correct until proven otherwise. -- Never stage `HANDOFF.md`. -- Anything that writes to the VM (`buildsystem`) is outward-facing: ask the user before Task 1 Step 6 and Task 5. - ---- - -## File map - -| file | responsibility | -|---|---| -| `remote/sbb-remote` | VM helper, single file, stdlib only: config, job kinds, job lifecycle, stats | -| `remote/config.example.toml` | helper config example, equal to its defaults | -| `sb_builder/__init__.py` | package, `VERSION` (must equal the helper's) | -| `sb_builder/config.py` | GUI config defaults, load, validate | -| `sb_builder/remote.py` | only code running ssh/virsh; `parse()` maps results to (data, error) | -| `sb_builder/fmt.py` | text formatting shared by the tabs | -| `sb_builder/dashboard.py` | Dashboard tab | -| `sb_builder/jobs.py` | Jobs tab | -| `sb_builder/mainwindow.py` | window, polling, status bar, VM control, notifications | -| `sb_builder/__main__.py` | entry point | -| `sb-builder` | launcher script | -| `install.sh` | local install, `--remote` helper deploy | -| `config.example.toml` | GUI config example, equal to its defaults | -| `tests/test_remote.py` | helper checks | -| `tests/test_gui.py` | GUI checks, headless | - -Outside this repo: `~/Programming/GIT/slackrepo_setup/{sbshadow,sbpull.bash,README.md}` (Task 1). - ---- - -### Task 1: Non-interactive flags in slackrepo_setup - -Jobs have no stdin, so `sbshadow`'s `Proceed?` prompt needs `--yes`/`--no`. Also `sbpull` currently returns `popd`'s status, which would make a failed `sbshadow` look like a successful job. - -**Files (repo `~/Programming/GIT/slackrepo_setup`):** -- Modify: `sbshadow`, `sbpull.bash`, `README.md` - -- [ ] **Step 1: Write the throwaway-repo check and run it against the current sbshadow to see it fail** - -```bash -T=$(mktemp -d) -cd "$T" && git init -q -b current \ - && mkdir -p cat/foo cat/bar && touch cat/foo/foo.SlackBuild cat/bar/bar.SlackBuild \ - && git add -A && git commit -qm upstream \ - && git checkout -qb danix-current \ - && mkdir -p personal/foo && touch personal/foo/foo.SlackBuild \ - && git add -A && git commit -qm overlay -SBO_REPO=$T PATH=/usr/bin:/bin bash ~/Programming/GIT/slackrepo_setup/sbshadow --no </dev/null; echo "rc=$?" -``` - -(`PATH=/usr/bin:/bin` keeps a locally installed `mkhint -F` from running. The throwaway commits are signed like any other, that is fine.) - -Expected now: `SBO_REPO` is ignored, so it fails with `cd: /var/lib/sbopkg/SBo-danix: No such file or directory` (or, on a host that has that repo, prompts and fails on EOF). `rc=1`. - -- [ ] **Step 2: Apply the change** - -Save this patch as `/tmp/slackrepo_setup.patch` and apply it from the repo root: - -````diff -diff '--color=auto' -ru a/README.md b/README.md ---- a/README.md -+++ b/README.md -@@ -22,9 +22,13 @@ - carried in an overlay, and restores (unshadows) upstream packages no longer - carried, from the pristine `current` branch. Asks once before changing - anything, commits, then runs `mkhint -F` to regenerate phantom-dep hintfiles. --- `sbpull.bash`: `sbpull {personal|pentesting}` shell function. Pulls one -- overlay subtree and runs `sbshadow`, to carry overlay changes without a full -- rebuild. Source it from root's `~/.bashrc`. -+ `--yes` applies without asking, `--no` only shows the plan (for -+ non-interactive callers such as sb-builder). `SBO_REPO` overrides the repo -+ path. -+- `sbpull.bash`: `sbpull {personal|pentesting} [--yes|--no]` shell function. -+ Pulls one overlay subtree and runs `sbshadow` (passing the flag on), to -+ carry overlay changes without a full rebuild. Source it from root's -+ `~/.bashrc`. - - ## Install - -diff '--color=auto' -ru a/sbpull.bash b/sbpull.bash ---- a/sbpull.bash -+++ b/sbpull.bash -@@ -1,4 +1,5 @@ - # sbpull: pull one overlay subtree into SBo-danix, then reconcile shadowing. -+# Usage: sbpull {personal|pentesting} [--yes|--no], the flag goes to sbshadow. - # Source from root's ~/.bashrc: . /path/to/sbpull.bash - # - # Copyright (C) 2026 Danilo M. <danix@danix.xyz> -@@ -9,7 +10,7 @@ - - sbpull() { - local repo_dir="/var/lib/sbopkg/SBo-danix" -- local target="$1" -+ local target="$1" flag="${2:-}" - local prefix branch remote - - case "$target" in -@@ -22,7 +23,7 @@ - remote="my-pentesting" - ;; - *) -- echo "sbpull: usage: sbpull {personal|pentesting}" >&2 -+ echo "sbpull: usage: sbpull {personal|pentesting} [--yes|--no]" >&2 - return 1 - ;; - esac -@@ -33,12 +34,13 @@ - - if git subtree pull --prefix="$prefix" "$remote" "$branch" --squash; then - echo "sbpull: $target subtree updated." -- sbshadow -+ sbshadow ${flag:+"$flag"} -+ local rc=$? -+ popd > /dev/null -+ return "$rc" - else - echo "sbpull: git subtree pull failed for $target, check output above." >&2 - popd > /dev/null - return 1 - fi -- -- popd > /dev/null - } -diff '--color=auto' -ru a/sbshadow b/sbshadow ---- a/sbshadow -+++ b/sbshadow -@@ -13,10 +13,24 @@ - # phantom-dep hintfiles with mkhint -F. - # - # Called by slackrepo_setup_SBo-danix.sh and by sbpull after a subtree pull. -+# -+# Usage: sbshadow [--yes|--no] -+# no flag show the plan and ask before changing anything -+# --yes apply the plan without asking -+# --no show the plan only, tree unchanged (preview) -+# SBO_REPO overrides the repo path (used to test against a throwaway repo). - - set -euo pipefail - --REPO="/var/lib/sbopkg/SBo-danix" -+answer="" -+case "${1:-}" in -+ --yes) answer=y ;; -+ --no) answer=n ;; -+ "") ;; -+ *) echo "usage: sbshadow [--yes|--no]" >&2; exit 2 ;; -+esac -+ -+REPO="${SBO_REPO:-/var/lib/sbopkg/SBo-danix}" - UPSTREAM="current" - - cd "$REPO" -@@ -47,7 +61,11 @@ - (( ${#to_restore[@]} )) && printf 'unshadow: %s\n' "${to_restore[@]}" - echo - -- read -rp "Proceed? [y/N] " confirm -+ if [[ -n $answer ]]; then -+ confirm=$answer -+ else -+ read -rp "Proceed? [y/N] " confirm -+ fi - if [[ ! "$confirm" =~ ^[Yy]$ ]]; then - echo "sbshadow: aborted, tree unchanged." - else -```` - -```bash -cd ~/Programming/GIT/slackrepo_setup && git apply /tmp/slackrepo_setup.patch && bash -n sbshadow && bash -n sbpull.bash -``` - -- [ ] **Step 3: Run the throwaway-repo checks** - -```bash -cd "$T" -SBO_REPO=$T PATH=/usr/bin:/bin bash ~/Programming/GIT/slackrepo_setup/sbshadow --no; echo "rc=$?"; ls cat -SBO_REPO=$T PATH=/usr/bin:/bin bash ~/Programming/GIT/slackrepo_setup/sbshadow </dev/null; echo "rc=$?" -SBO_REPO=$T PATH=/usr/bin:/bin bash ~/Programming/GIT/slackrepo_setup/sbshadow --bogus; echo "rc=$?" -SBO_REPO=$T PATH=/usr/bin:/bin bash ~/Programming/GIT/slackrepo_setup/sbshadow --yes; echo "rc=$?"; ls cat; git log --oneline -1 -``` - -Expected, in order: -- `--no`: prints `shadow: cat/foo`, `sbshadow: aborted, tree unchanged.`, the mkhint NOTE, `rc=0`; `ls cat` shows `bar foo`. -- no flag with stdin `/dev/null`: prints the plan, `rc=1` (read hits EOF, `set -e` exits: nothing hangs, nothing is applied). -- `--bogus`: `usage: sbshadow [--yes|--no]`, `rc=2`. -- `--yes`: `sbshadow: shadowed 1, unshadowed 0.`, `rc=0`; `ls cat` shows `bar`; last commit `shadow: reconcile upstream against personal/pentesting`. - -- [ ] **Step 4: Check sbpull passes the flag and the exit status** - -```bash -bash -c 'sbshadow() { echo "fake sbshadow [$*]"; return 3; }; git() { return 0; }; pushd() { :; }; popd() { :; }; - . ~/Programming/GIT/slackrepo_setup/sbpull.bash; sbpull personal --no'; echo "rc=$?" -bash -c 'sbshadow() { echo "fake sbshadow [$*]"; }; git() { return 0; }; pushd() { :; }; popd() { :; }; - . ~/Programming/GIT/slackrepo_setup/sbpull.bash; sbpull pentesting'; echo "rc=$?" -rm -rf "$T" -``` - -Expected: `fake sbshadow [--no]` then `rc=3`; `fake sbshadow []` then `rc=0`. - -- [ ] **Step 5: Commit in slackrepo_setup** - -```bash -cd ~/Programming/GIT/slackrepo_setup -git add sbshadow sbpull.bash README.md -git commit -F - <<'EOF' -sbshadow: add --yes/--no, sbpull: pass flag, return sbshadow status - -sbshadow --yes applies the plan without asking, --no shows it and -leaves the tree unchanged, so non-interactive callers (sb-builder runs -it as a detached job with no stdin) never hit the prompt. No flag keeps -the interactive prompt. SBO_REPO overrides the repo path for testing. - -sbpull passes the flag on, and now returns sbshadow's exit status -instead of popd's, which made a failed sbshadow look successful. - -Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> -EOF -git log --format='%h %G? %s' -1 # expect G -git push -``` - -- [ ] **Step 6: Deploy to the VM (ask the user first)** - -The HANDOFF of slackrepo_setup records the deploy locations: scripts in `/usr/local/bin`, `sbpull.bash` in `/usr/local/share/slackrepo_setup/`. - -```bash -cd ~/Programming/GIT/slackrepo_setup -ssh buildsystem 'ls -l /usr/local/bin/sbshadow /usr/local/share/slackrepo_setup/sbpull.bash' -scp sbshadow buildsystem:/usr/local/bin/sbshadow -scp sbpull.bash buildsystem:/usr/local/share/slackrepo_setup/sbpull.bash -ssh buildsystem 'chmod 755 /usr/local/bin/sbshadow; sbshadow --bogus; echo rc=$?' -``` - -Expected last line pair: `usage: sbshadow [--yes|--no]`, `rc=2`. - ---- - -### Task 2: VM helper `sbb-remote` - -**Files:** -- Create: `remote/sbb-remote`, `remote/config.example.toml` -- Test: `tests/test_remote.py` - -- [ ] **Step 1: Write the tests** - -Create `tests/test_remote.py`: - -````python -# sb-builder - self-checks for the remote helper -# Copyright (C) 2026 Danilo M. <danix@danix.xyz> -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 as -# published by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License along -# with this program; if not, write to the Free Software Foundation, Inc., -# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. - -"""Run: python3 -m unittest discover -s tests""" - -import contextlib -import fcntl -import importlib.machinery -import importlib.util -import io -import json -import os -import shutil -import sqlite3 -import tempfile -import threading -import time -import unittest -from unittest import mock - -SCRIPT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "remote", "sbb-remote") -_loader = importlib.machinery.SourceFileLoader("sbb_remote", SCRIPT) -sbb = importlib.util.module_from_spec(importlib.util.spec_from_loader("sbb_remote", _loader)) -_loader.exec_module(sbb) - - -class Base(unittest.TestCase): - def setUp(self): - self.tmp = tempfile.mkdtemp() - self.addCleanup(shutil.rmtree, self.tmp) - conf = os.path.join(self.tmp, "config.toml") - with open(conf, "w") as f: - f.write(f'state_dir = "{self.tmp}"\n' - f'repo_dir = "{self.tmp}/repo"\n' - f'slackrepo_db = "{self.tmp}/slackrepo.sqlite3"\n') - patcher = mock.patch.dict(os.environ, {"SBB_CONFIG": conf}) - patcher.start() - self.addCleanup(patcher.stop) - self.cfg = sbb.load_config() - os.makedirs(os.path.join(self.tmp, "jobs")) - - def make_job(self, argv, **meta): - jid = "20260101-000000-test" - d = os.path.join(self.tmp, "jobs", jid) - os.mkdir(d) - sbb.write_meta(d, {"id": jid, "kind": "test", "args": [], "argv": argv, - "status": "starting", "created": time.time(), **meta}) - open(os.path.join(d, "log"), "wb").close() - return jid, d - - def hold_lock(self): - f = open(os.path.join(self.tmp, "lock"), "a") - fcntl.flock(f, fcntl.LOCK_EX) - self.addCleanup(f.close) - - def run_main(self, *argv): - out = io.StringIO() - with contextlib.redirect_stdout(out): - rc = sbb.main(list(argv)) - return rc, json.loads(out.getvalue()) - - def wait_status(self, d, statuses): - for _ in range(100): - if sbb.read_meta(d)["status"] in statuses: - break - time.sleep(0.05) - return sbb.read_meta(d) - - -class ConfigTest(Base): - def write(self, text): - p = os.path.join(self.tmp, "c.toml") - with open(p, "w") as f: - f.write(text) - return p - - def test_missing_file_gives_defaults(self): - self.assertEqual(sbb.load_config(os.path.join(self.tmp, "nope.toml")), sbb.DEFAULTS) - - def test_override(self): - self.assertEqual(sbb.load_config(self.write('overlays = ["a"]\n'))["overlays"], ["a"]) - - def test_example_matches_defaults(self): - example = os.path.join(os.path.dirname(SCRIPT), "config.example.toml") - self.assertEqual(sbb.load_config(example), sbb.DEFAULTS) - - def test_unknown_key(self): - with self.assertRaisesRegex(sbb.Error, "bogus: unknown key"): - sbb.load_config(self.write("bogus = 1\n")) - - def test_wrong_type(self): - with self.assertRaisesRegex(sbb.Error, "state_dir: expected str"): - sbb.load_config(self.write("state_dir = 1\n")) - - -class KindsTest(Base): - def test_sbpull_argv(self): - self.assertEqual(sbb.KINDS["sbpull"](self.cfg, ["personal", "--no"]), - ["bash", "-c", '. "$1" && sbpull "$2" "$3"', "_", - self.cfg["sbpull_script"], "personal", "--no"]) - - def test_sbshadow_argv(self): - self.assertEqual(sbb.KINDS["sbshadow"](self.cfg, ["--yes"]), ["sbshadow", "--yes"]) - - def test_rejects_bad_args(self): - for kind, args in (("sbpull", ["evil", "--no"]), ("sbpull", ["personal"]), - ("sbpull", ["personal", "--force"]), ("sbshadow", []), - ("sbshadow", ["--yes", "--no"])): - with self.subTest(kind=kind, args=args), self.assertRaises(sbb.Error): - sbb.KINDS[kind](self.cfg, args) - - def test_unknown_kind(self): - with self.assertRaisesRegex(sbb.Error, "unknown kind"): - sbb.cmd_job_start(self.cfg, "rm", "-rf", "/") - - -class MainTest(Base): - def test_unknown_command(self): - rc, data = self.run_main("nope") - self.assertEqual(rc, 1) - self.assertIn("usage", data["error"]) - - def test_version(self): - self.assertEqual(self.run_main("version"), - (0, {"version": sbb.VERSION, "overlays": ["personal", "pentesting"]})) - - -class JobsTest(Base): - def log(self, d): - with open(os.path.join(d, "log"), "rb") as f: - return f.read() - - def test_run_ok(self): - jid, d = self.make_job(["sh", "-c", "echo hello"]) - sbb.cmd_run(self.cfg, jid) - m = sbb.read_meta(d) - self.assertEqual((m["status"], m["exit"]), ("ok", 0)) - self.assertEqual(self.log(d), b"hello\n") - - def test_run_failed(self): - jid, d = self.make_job(["sh", "-c", "exit 3"]) - sbb.cmd_run(self.cfg, jid) - m = sbb.read_meta(d) - self.assertEqual((m["status"], m["exit"]), ("failed", 3)) - - def test_run_missing_program(self): - jid, d = self.make_job(["/nonexistent/prog"]) - sbb.cmd_run(self.cfg, jid) - self.assertEqual(sbb.read_meta(d)["status"], "failed") - self.assertIn(b"sbb-remote:", self.log(d)) - - def test_kill(self): - jid, d = self.make_job(["sleep", "30"]) - t = threading.Thread(target=sbb.cmd_run, args=(self.cfg, jid)) - t.start() - self.wait_status(d, ("running",)) - self.assertEqual(sbb.cmd_job_kill(self.cfg, jid), {"ok": True}) - t.join(5) - self.assertEqual(sbb.read_meta(d)["status"], "killed") - - def test_kill_not_running(self): - jid, _ = self.make_job(["true"], status="ok") - with self.assertRaisesRegex(sbb.Error, "not running"): - sbb.cmd_job_kill(self.cfg, jid) - - def test_busy(self): - self.hold_lock() - jid, d = self.make_job(["true"]) - sbb.cmd_run(self.cfg, jid) - self.assertEqual(sbb.read_meta(d)["status"], "busy") - - def test_lost_when_lock_free(self): - self.make_job(["true"], status="running", pid=1) - self.assertEqual(sbb.cmd_jobs(self.cfg)["jobs"][0]["status"], "lost") - - def test_running_kept_while_lock_held(self): - self.hold_lock() - self.make_job(["true"], status="running", pid=1) - self.assertEqual(sbb.cmd_jobs(self.cfg)["jobs"][0]["status"], "running") - - def test_starting_grace(self): - _, d = self.make_job(["true"]) - self.assertEqual(sbb.cmd_jobs(self.cfg)["jobs"][0]["status"], "starting") - sbb.write_meta(d, dict(sbb.read_meta(d), created=time.time() - 60)) - self.assertEqual(sbb.cmd_jobs(self.cfg)["jobs"][0]["status"], "lost") - - def test_follow_from_offset(self): - jid, d = self.make_job(["true"], status="ok") - with open(os.path.join(d, "log"), "wb") as f: - f.write(b"hello world") - out = io.BytesIO() - sbb.cmd_job_follow(self.cfg, jid, "6", out=out) - self.assertEqual(out.getvalue(), b"world") - - def test_bad_job_id(self): - with self.assertRaisesRegex(sbb.Error, "bad job id"): - sbb.cmd_job_follow(self.cfg, "../../etc", "0") - - def fake_sbshadow(self): - bindir = os.path.join(self.tmp, "bin") - os.mkdir(bindir) - fake = os.path.join(bindir, "sbshadow") - with open(fake, "w") as f: - f.write('#!/bin/sh\necho "shadow: cat/$1"\n') - os.chmod(fake, 0o755) - patcher = mock.patch.dict(os.environ, {"PATH": bindir + os.pathsep + os.environ["PATH"]}) - patcher.start() - self.addCleanup(patcher.stop) - - def test_start_end_to_end(self): - self.fake_sbshadow() - rc, data = self.run_main("job-start", "sbshadow", "--no") - self.assertEqual(rc, 0) - d = os.path.join(self.tmp, "jobs", data["id"]) - self.assertEqual(self.wait_status(d, ("ok", "failed", "lost"))["status"], "ok") - self.assertEqual(self.log(d), b"shadow: cat/--no\n") - - def test_start_busy(self): - self.fake_sbshadow() - self.hold_lock() - jid, _ = self.make_job(["true"], status="running", pid=1) - self.assertEqual(self.run_main("job-start", "sbshadow", "--no"), (0, {"busy": jid})) - self.assertEqual(os.listdir(os.path.join(self.tmp, "jobs")), [jid]) - - -class StatsTest(Base): - def test_missing_sources_are_null(self): - s = sbb.cmd_stats(self.cfg) - self.assertIsNotNone(s["load"]) - self.assertIsNone(s["packages"]) - self.assertIsNone(s["slackrepo"]) - self.assertEqual(s["sync"], {"personal": None, "pentesting": None}) - json.dumps(s) - - def test_counts(self): - r = self.cfg["repo_dir"] - for p in ("cat/foo", "cat/bar", "personal/foo", "personal/baz", "pentesting/nmap"): - os.makedirs(os.path.join(r, p)) - open(os.path.join(r, p, os.path.basename(p) + ".SlackBuild"), "w").close() - os.makedirs(os.path.join(r, "cat", "notapkg")) - - def fake_git(cfg, *args): - if args[0] == "ls-tree": - return "cat/foo/foo.SlackBuild\ncat/foo/README\ncat/old/old.SlackBuild\n" - return "1792351223\n" - - with mock.patch.object(sbb, "_git", fake_git): - s = sbb.cmd_stats(self.cfg) - self.assertEqual(s["packages"], {"total": 5, "personal": 2, "pentesting": 1, "shadowed": 1}) - self.assertEqual(s["sync"]["personal"], 1792351223) - - def test_slackrepo(self): - db = sqlite3.connect(self.cfg["slackrepo_db"]) - db.executescript(""" - create table buildresults (itemid text primary key, time text, result text); - create table revisions (itemid text, dep text, deplist text, version text, - built text, rev text, os text, hintcksum text, primary key (itemid, dep)); - insert into buildresults values ('a', '1', 'ok'), ('b', '1', 'ok'), ('c', '1', 'failed'); - insert into revisions values ('cat/a', '/', '/', '1', '1', 'r', 'o', '/'), - ('cat/a', 'dep', '/', '1', '1', 'r', 'o', '/'); - """) - db.close() - self.assertEqual(sbb.cmd_stats(self.cfg)["slackrepo"], - {"results": {"ok": 2, "failed": 1}, "built": 1}) - - -if __name__ == "__main__": - unittest.main() -```` - -- [ ] **Step 2: Run them to see them fail** - -Run: `python3 -m unittest tests.test_remote` -Expected: ERROR, `FileNotFoundError` for `remote/sbb-remote`. - -- [ ] **Step 3: Write the helper** - -Create `remote/sbb-remote` and `chmod 755 remote/sbb-remote`: - -````python -#!/usr/bin/env python3 -# sbb-remote - sb-builder helper on the build VM -# Copyright (C) 2026 Danilo M. <danix@danix.xyz> -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 as -# published by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License along -# with this program; if not, write to the Free Software Foundation, Inc., -# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. - -"""Helper run over ssh by the sb-builder GUI. - -Every command prints one JSON object on stdout, except job-follow, which -streams raw log bytes. Jobs run detached (setsid, stdin=/dev/null), one at a -time under a flock, from a fixed table of kinds: the GUI sends a kind and -its arguments, this file validates them and builds the argv itself. -""" - -import fcntl -import glob -import json -import os -import re -import shutil -import signal -import sqlite3 -import subprocess -import sys -import time -import tomllib - -VERSION = "0.1.0" -DEFAULTS = { - "state_dir": "/var/lib/sb-builder", - "repo_dir": "/var/lib/sbopkg/SBo-danix", - "upstream_branch": "current", - "slackrepo_db": "/var/lib/slackrepo/SBo-danix/database_SBo-danix_x86_64.sqlite3", - "sbpull_script": "/usr/local/share/slackrepo_setup/sbpull.bash", - "overlays": ["personal", "pentesting"], -} -FLAGS = ("--yes", "--no") -JOB_ID = re.compile(r"^\d{8}-\d{6}-[a-z]+(-\d+)?$") -PKG = re.compile(r"^[^/]+/([^/]+)/\1\.SlackBuild$") -ACTIVE = ("starting", "running") -START_GRACE = 10 # seconds a job may sit in "starting" before it counts as lost - - -class Error(Exception): - pass - - -def load_config(path=None): - path = path or os.environ.get("SBB_CONFIG", "/etc/sb-builder/config.toml") - cfg = dict(DEFAULTS) - try: - with open(path, "rb") as f: - data = tomllib.load(f) - except FileNotFoundError: - return cfg - except tomllib.TOMLDecodeError as e: - raise Error(f"config: {e}") - for k, v in data.items(): - if k not in DEFAULTS: - raise Error(f"config: {k}: unknown key") - if type(v) is not type(DEFAULTS[k]): - raise Error(f"config: {k}: expected {type(DEFAULTS[k]).__name__}") - cfg[k] = v - if not all(isinstance(o, str) for o in cfg["overlays"]): - raise Error("config: overlays: expected list of strings") - return cfg - - -# ── job kinds: (cfg, args) -> argv, raising Error on bad args ─────────────── - -def _flag(args): - if len(args) != 1 or args[0] not in FLAGS: - raise Error("expected exactly one of --yes, --no") - return args[0] - - -def kind_sbpull(cfg, args): - if not args or args[0] not in cfg["overlays"]: - raise Error(f"sbpull: overlay must be one of {cfg['overlays']}") - return ["bash", "-c", '. "$1" && sbpull "$2" "$3"', "_", - cfg["sbpull_script"], args[0], _flag(args[1:])] - - -def kind_sbshadow(cfg, args): - return ["sbshadow", _flag(args)] - - -KINDS = {"sbpull": kind_sbpull, "sbshadow": kind_sbshadow} - - -# ── job state ─────────────────────────────────────────────────────────────── - -def job_dir(cfg, jid): - if not JOB_ID.match(jid or ""): - raise Error(f"bad job id: {jid}") - d = os.path.join(cfg["state_dir"], "jobs", jid) - if not os.path.isdir(d): - raise Error(f"no such job: {jid}") - return d - - -def read_meta(d): - with open(os.path.join(d, "meta.json")) as f: - return json.load(f) - - -def write_meta(d, meta): - tmp = os.path.join(d, "meta.json.tmp") - with open(tmp, "w") as f: - json.dump(meta, f) - os.replace(tmp, os.path.join(d, "meta.json")) - - -def lock_file(cfg): - return open(os.path.join(cfg["state_dir"], "lock"), "a") - - -def refresh(cfg, d, meta): - """Mark an active job lost when no runner holds the lock. - - _run writes the final status before it releases the lock, so a free lock - plus a status that is still active (re-read under the lock) means the - runner died: killed, crashed, or the VM rebooted. - """ - if meta["status"] not in ACTIVE: - return meta - if meta["status"] == "starting" and time.time() - meta["created"] < START_GRACE: - return meta - with lock_file(cfg) as lk: - try: - fcntl.flock(lk, fcntl.LOCK_EX | fcntl.LOCK_NB) - except BlockingIOError: - return meta - meta = read_meta(d) - if meta["status"] in ACTIVE: - meta["status"] = "lost" - write_meta(d, meta) - return meta - - -def all_jobs(cfg): - out = [] - for d in sorted(glob.glob(os.path.join(cfg["state_dir"], "jobs", "*")), reverse=True): - try: - out.append(refresh(cfg, d, read_meta(d))) - except (OSError, ValueError): - pass # a dir being created or removed right now - return out - - -# ── commands ──────────────────────────────────────────────────────────────── - -def cmd_version(cfg): - return {"version": VERSION, "overlays": cfg["overlays"]} - - -def cmd_jobs(cfg): - # ponytail: logs and job dirs are kept forever; prune once disk usage shows on the dashboard - return {"jobs": all_jobs(cfg)} - - -def cmd_job_start(cfg, kind=None, *args): - if kind not in KINDS: - raise Error(f"unknown kind: {kind}") - argv = KINDS[kind](cfg, list(args)) - base = time.strftime("%Y%m%d-%H%M%S-") + kind - jid, n = base, 1 - while True: - d = os.path.join(cfg["state_dir"], "jobs", jid) - try: - os.mkdir(d) - break - except FileExistsError: - n += 1 - jid = f"{base}-{n}" - write_meta(d, {"id": jid, "kind": kind, "args": list(args), "argv": argv, - "status": "starting", "created": time.time()}) - open(os.path.join(d, "log"), "wb").close() - subprocess.Popen([sys.executable, os.path.abspath(__file__), "_run", jid], - stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, start_new_session=True) - for _ in range(60): - status = read_meta(d)["status"] - if status == "busy": - shutil.rmtree(d) - return {"busy": next((m["id"] for m in all_jobs(cfg) if m["status"] == "running"), None)} - if status != "starting": - break - time.sleep(0.05) - return {"id": jid} - - -def cmd_run(cfg, jid=None): - d = job_dir(cfg, jid) - meta = read_meta(d) - with lock_file(cfg) as lk: - try: - fcntl.flock(lk, fcntl.LOCK_EX | fcntl.LOCK_NB) - except BlockingIOError: - meta["status"] = "busy" - write_meta(d, meta) - return None - with open(os.path.join(d, "log"), "ab") as log: - meta["start"] = time.time() - try: - p = subprocess.Popen(meta["argv"], stdin=subprocess.DEVNULL, stdout=log, - stderr=subprocess.STDOUT, start_new_session=True) - except OSError as e: - log.write(f"sbb-remote: {e}\n".encode()) - rc = 127 - else: - meta.update(pid=p.pid, status="running") - write_meta(d, meta) - rc = p.wait() - meta.update(end=time.time(), exit=rc, - status="ok" if rc == 0 else "killed" if rc == -signal.SIGTERM else "failed") - write_meta(d, meta) - return None - - -def cmd_job_follow(cfg, jid=None, offset="0", out=None): - d = job_dir(cfg, jid) - out = out or sys.stdout.buffer - with open(os.path.join(d, "log"), "rb") as f: - f.seek(int(offset)) - while True: - # status first, then read: the last read after the job ends drains the log - done = refresh(cfg, d, read_meta(d))["status"] not in ACTIVE - chunk = f.read() - if chunk: - out.write(chunk) - out.flush() - elif done: - return None - else: - time.sleep(0.5) - - -def cmd_job_kill(cfg, jid=None): - meta = read_meta(job_dir(cfg, jid)) - if meta["status"] != "running": - raise Error(f"job not running: {jid}") - try: - os.killpg(meta["pid"], signal.SIGTERM) - except ProcessLookupError: - pass # ended on its own meanwhile - return {"ok": True} - - -# ── stats ─────────────────────────────────────────────────────────────────── - -def _try(fn): - try: - return fn() - except Exception: - return None - - -def _git(cfg, *args): - return subprocess.run(["git", "-C", cfg["repo_dir"], *args], - capture_output=True, text=True, check=True).stdout - - -def _pkgs(root, pattern): - """Names of the <pkg>/<pkg>.SlackBuild dirs matching pattern under root.""" - out = [] - for d in glob.glob(os.path.join(root, pattern, "")): - name = os.path.basename(os.path.dirname(d)) - if os.path.isfile(os.path.join(d, name + ".SlackBuild")): - out.append(name) - return out - - -def _packages(cfg): - r = cfg["repo_dir"] - mine = {o: _pkgs(r, o + "/*") for o in cfg["overlays"]} - tree = _git(cfg, "ls-tree", "-r", "--name-only", cfg["upstream_branch"]).splitlines() - upstream = {m.group(1) for m in map(PKG.match, tree) if m} - shadowed = upstream & {p for names in mine.values() for p in names} - return {"total": len(_pkgs(r, "*/*")), **{o: len(n) for o, n in mine.items()}, - "shadowed": len(shadowed)} - - -def _slackrepo(cfg): - db = sqlite3.connect(f"file:{cfg['slackrepo_db']}?mode=ro", uri=True) - try: - results = dict(db.execute("select result, count(*) from buildresults group by result")) - built = db.execute("select count(distinct itemid) from revisions where dep = '/'").fetchone()[0] - finally: - db.close() - return {"results": results, "built": built} - - -def _uptime(): - with open("/proc/uptime") as f: - return float(f.read().split()[0]) - - -def cmd_stats(cfg): - dirs = {"repo": cfg["repo_dir"], "state": cfg["state_dir"]} - return { - "load": _try(os.getloadavg), - "uptime": _try(_uptime), - "disk_free": {k: _try(lambda p=p: shutil.disk_usage(p).free) for k, p in dirs.items()}, - "packages": _try(lambda: _packages(cfg)), - "sync": {o: _try(lambda o=o: int(_git(cfg, "log", "-1", "--format=%ct", "--", o + "/"))) - for o in cfg["overlays"]}, - "slackrepo": _try(lambda: _slackrepo(cfg)), - } - - -COMMANDS = { - "version": cmd_version, - "stats": cmd_stats, - "jobs": cmd_jobs, - "job-start": cmd_job_start, - "job-follow": cmd_job_follow, - "job-kill": cmd_job_kill, - "_run": cmd_run, -} - - -def main(argv): - try: - if not argv or argv[0] not in COMMANDS: - names = ",".join(c for c in COMMANDS if not c.startswith("_")) - raise Error(f"usage: sbb-remote {{{names}}} [args]") - cfg = load_config() - os.makedirs(os.path.join(cfg["state_dir"], "jobs"), exist_ok=True) - result = COMMANDS[argv[0]](cfg, *argv[1:]) - except BrokenPipeError: - # the GUI hung up on job-follow; silence the flush at exit too - os.dup2(os.open(os.devnull, os.O_WRONLY), sys.stdout.fileno()) - return 0 - except Exception as e: - print(json.dumps({"error": str(e) if isinstance(e, Error) else f"{type(e).__name__}: {e}"})) - return 1 - if result is not None: - print(json.dumps(result)) - return 0 - - -if __name__ == "__main__": - sys.exit(main(sys.argv[1:])) -```` - -Create `remote/config.example.toml`: - -````toml -# sbb-remote config on the VM: /etc/sb-builder/config.toml -# Every key is optional; these are the defaults. - -state_dir = "/var/lib/sb-builder" -repo_dir = "/var/lib/sbopkg/SBo-danix" -upstream_branch = "current" -slackrepo_db = "/var/lib/slackrepo/SBo-danix/database_SBo-danix_x86_64.sqlite3" -sbpull_script = "/usr/local/share/slackrepo_setup/sbpull.bash" -overlays = ["personal", "pentesting"] -```` - -- [ ] **Step 4: Run the tests** - -Run: `python3 -m unittest tests.test_remote` -Expected: `Ran 27 tests`, `OK`. Two `ResourceWarning: subprocess ... is still running` lines are expected: `job-start` deliberately does not wait for the detached `_run`. - -- [ ] **Step 5: Commit** - -```bash -git add remote/sbb-remote remote/config.example.toml tests/test_remote.py -git commit -F - <<'EOF' -Add sbb-remote, the VM helper - -Single-file stdlib helper the GUI runs over ssh. JSON commands for -version, stats and jobs; job-start runs a fixed table of kinds (sbpull, -sbshadow, each requiring --yes or --no) detached with setsid and no -stdin, one at a time under a flock; job-follow streams a log from a -byte offset so the GUI can drop and resume; job-kill sends SIGTERM to -the job's process group. A job whose runner died is reported lost. - -Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> -EOF -git log --format='%h %G? %s' -1 -``` - ---- - -### Task 3: GUI - -**Files:** -- Create: `sb_builder/__init__.py`, `sb_builder/config.py`, `sb_builder/remote.py`, `sb_builder/fmt.py`, `sb_builder/dashboard.py`, `sb_builder/jobs.py`, `sb_builder/mainwindow.py`, `sb_builder/__main__.py`, `config.example.toml` -- Test: `tests/test_gui.py` - -- [ ] **Step 1: Write the tests** - -Create `tests/test_gui.py`: - -````python -# sb-builder - self-checks for the GUI -# Copyright (C) 2026 Danilo M. <danix@danix.xyz> -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 as -# published by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License along -# with this program; if not, write to the Free Software Foundation, Inc., -# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. - -"""Run: python3 -m unittest discover -s tests (headless, no VM needed)""" - -import os -os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") - -import importlib.machinery # noqa: E402 -import importlib.util # noqa: E402 -import sys # noqa: E402 -import tempfile # noqa: E402 -import unittest # noqa: E402 - -ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..") -sys.path.insert(0, ROOT) - -from PyQt6.QtCore import QObject, pyqtSignal # noqa: E402 -from PyQt6.QtWidgets import QApplication # noqa: E402 - -from sb_builder import VERSION, config # noqa: E402 -from sb_builder.mainwindow import MainWindow # noqa: E402 -from sb_builder.remote import DOWN, MISSING, parse # noqa: E402 - -app = QApplication.instance() or QApplication([]) - - -class ConfigTest(unittest.TestCase): - def load(self, text): - with tempfile.NamedTemporaryFile("w", suffix=".toml", delete=False) as f: - f.write(text) - self.addCleanup(os.unlink, f.name) - return config.load(f.name) - - def test_missing_file(self): - self.assertEqual(config.load("/nonexistent/config.toml"), (config.DEFAULTS, None)) - - def test_example_matches_defaults(self): - self.assertEqual(config.load(os.path.join(ROOT, "config.example.toml")), (config.DEFAULTS, None)) - - def test_override(self): - cfg, err = self.load("[refresh]\njobs = 2\n") - self.assertIsNone(err) - self.assertEqual(cfg["refresh"], {"dashboard": 30, "jobs": 2}) - - def test_unknown_key(self): - cfg, err = self.load("[remote]\nbogus = 1\n") - self.assertIn("remote.bogus", err) - self.assertEqual(cfg, config.DEFAULTS) - - def test_wrong_type(self): - cfg, err = self.load('[vm]\nautostart = "yes"\n') - self.assertIn("vm.autostart: expected bool", err) - self.assertEqual(cfg, config.DEFAULTS) - - -class ParseTest(unittest.TestCase): - def test_cases(self): - self.assertEqual(parse(255, b"", b""), (None, DOWN)) - self.assertEqual(parse(127, b"", b""), (None, MISSING)) - self.assertEqual(parse(0, b'{"a": 1}', b""), ({"a": 1}, None)) - self.assertEqual(parse(1, b'{"error": "boom"}', b""), (None, "boom")) - self.assertEqual(parse(1, b"", b"bash: x\n"), (None, "bash: x")) - - -class VersionTest(unittest.TestCase): - def test_gui_matches_helper(self): - script = os.path.join(ROOT, "remote", "sbb-remote") - loader = importlib.machinery.SourceFileLoader("sbb_remote", script) - sbb = importlib.util.module_from_spec(importlib.util.spec_from_loader("sbb_remote", loader)) - loader.exec_module(sbb) - self.assertEqual(sbb.VERSION, VERSION) - - -class FakeRemote: - def __init__(self, answers): - self.answers = answers - self.calls = [] - self.virsh_answer = (1, "") - - def call(self, cb, *args): - self.calls.append(args) - cb(*self.answers.get(args[0], (None, DOWN))) - - def virsh(self, cb, *args): - self.calls.append(("virsh", *args)) - cb(*self.virsh_answer) - - -class FakeNotifier(QObject): - clicked = pyqtSignal(str) - - def __init__(self): - super().__init__() - self.sent = [] - - def send(self, title, body, jid=""): - self.sent.append((title, body, jid)) - - -STATS = {"load": [0.5, 0.4, 0.3], "uptime": 3600.0, - "disk_free": {"repo": 10 * 2**30, "state": None}, - "packages": {"total": 10, "personal": 2, "pentesting": 1, "shadowed": 1}, - "sync": {"personal": None, "pentesting": 1792351223}, "slackrepo": None} -JOB = {"id": "20260101-000000-sbshadow", "kind": "sbshadow", "args": ["--no"], "argv": [], - "status": "running", "created": 0, "start": 0, "pid": 5} - - -class SmokeTest(unittest.TestCase): - def window(self, **answers): - remote = FakeRemote({"version": ({"version": VERSION, "overlays": ["personal"]}, None), - "stats": (STATS, None), "jobs": ({"jobs": []}, None), **answers}) - w = MainWindow(config.load("/nonexistent")[0], remote, FakeNotifier()) - self.addCleanup(w.deleteLater) - return w, remote - - def test_tabs_and_dashboard(self): - w, _ = self.window() - w.poll_jobs() - self.assertEqual(w.conn.text(), "● connected") - self.assertEqual(w.tabs.tabText(0), "Dashboard") - self.assertIn("total: 10", w.dashboard.labels["Repo"].text()) - self.assertEqual(w.dashboard.labels["slackrepo"].text(), "n/a") - self.assertIn("10.0 GiB", w.dashboard.labels["VM"].text()) - jobs = w.open_jobs() - self.assertIs(w.open_jobs(), jobs) - self.assertEqual(w.tabs.count(), 2) - self.assertEqual(jobs.start_menu.actions()[0].text(), "sbpull personal (preview)") - - def test_end_notification_once(self): - w, remote = self.window(jobs=({"jobs": [JOB]}, None)) - w.poll_jobs() - remote.answers["jobs"] = ({"jobs": [dict(JOB, status="failed", end=5, exit=1)]}, None) - w.poll_jobs() - w.poll_jobs() - self.assertEqual(w.notifier.sent, [("sb-builder: failed", "sbshadow --no", JOB["id"])]) - - def test_down_offers_vm_start(self): - w, remote = self.window(jobs=(None, DOWN)) - remote.virsh_answer = (0, "shut off") - w.poll_jobs() - self.assertIn(("virsh", "domstate", "buildsystem"), remote.calls) - self.assertEqual(w.conn.text(), "● VM off") - self.assertFalse(w.vm_btn.isHidden()) - - def test_log_carriage_return(self): - w, _ = self.window() - jobs = w.open_jobs() - jobs.append_log("10%\r50%\r100%\ndone\r\n") - self.assertEqual(jobs.log.toPlainText(), "100%\ndone\n") - - -if __name__ == "__main__": - unittest.main() -```` - -- [ ] **Step 2: Run them to see them fail** - -Run: `python3 -m unittest tests.test_gui` -Expected: ERROR, `ModuleNotFoundError: No module named 'sb_builder'`. - -- [ ] **Step 3: Package and config** - -Create `sb_builder/__init__.py`: - -````python -# sb-builder - GUI package -# Copyright (C) 2026 Danilo M. <danix@danix.xyz> -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 as -# published by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License along -# with this program; if not, write to the Free Software Foundation, Inc., -# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. - -"""sb-builder: Qt6 GUI driving the SlackBuild tooling on a build VM.""" - -VERSION = "0.1.0" # must match remote/sbb-remote VERSION -```` - -Create `sb_builder/config.py`: - -````python -# sb-builder - GUI configuration -# Copyright (C) 2026 Danilo M. <danix@danix.xyz> -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 as -# published by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License along -# with this program; if not, write to the Free Software Foundation, Inc., -# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. - -"""GUI configuration: built-in defaults overlaid by the user's TOML file.""" - -import copy -import os -import tomllib - -DEFAULTS = { - "remote": {"target": "buildsystem", "ssh_options": []}, - "refresh": {"dashboard": 30, "jobs": 5}, - "notify": {"enabled": True, "events": ["finished", "failed", "killed", "lost", "vm-up"]}, - "vm": {"libvirt_uri": "qemu:///system", "domain": "buildsystem", "autostart": False}, -} - - -def path(): - base = os.environ.get("XDG_CONFIG_HOME") or os.path.expanduser("~/.config") - return os.path.join(base, "sb-builder", "config.toml") - - -def load(p=None): - """Return (config, error). On any problem: the defaults plus the error text.""" - p = p or path() - try: - with open(p, "rb") as f: - data = tomllib.load(f) - except FileNotFoundError: - return copy.deepcopy(DEFAULTS), None - except (OSError, tomllib.TOMLDecodeError) as e: - return copy.deepcopy(DEFAULTS), f"{p}: {e}" - cfg = copy.deepcopy(DEFAULTS) - for sec, keys in data.items(): - if sec not in DEFAULTS or not isinstance(keys, dict): - return copy.deepcopy(DEFAULTS), f"{p}: unknown section [{sec}]" - for k, v in keys.items(): - if k not in DEFAULTS[sec]: - return copy.deepcopy(DEFAULTS), f"{p}: unknown key {sec}.{k}" - want = type(DEFAULTS[sec][k]) - if type(v) is not want: - return copy.deepcopy(DEFAULTS), f"{p}: {sec}.{k}: expected {want.__name__}" - cfg[sec][k] = v - return cfg, None -```` - -Create `config.example.toml`: - -````toml -# sb-builder GUI config: ~/.config/sb-builder/config.toml -# Every key is optional; these are the defaults. - -[remote] -target = "buildsystem" # any ssh destination: host alias, user@host -ssh_options = [] # extra ssh args, e.g. ["-p", "2222"] - -[refresh] -dashboard = 30 # seconds, while the Dashboard tab is visible -jobs = 5 # seconds; doubles up to 30 while the VM is down - -[notify] -enabled = true -events = ["finished", "failed", "killed", "lost", "vm-up"] - -[vm] -libvirt_uri = "qemu:///system" -domain = "buildsystem" # libvirt domain of the VM; "" disables VM control -autostart = false # true: start the VM without asking when it is off -```` - -- [ ] **Step 4: Transport** - -Create `sb_builder/remote.py`: - -````python -# sb-builder - ssh and virsh transport -# Copyright (C) 2026 Danilo M. <danix@danix.xyz> -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 as -# published by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License along -# with this program; if not, write to the Free Software Foundation, Inc., -# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. - -"""The only code that talks to the VM: ssh to sbb-remote, and local virsh.""" - -import json -import os -import shlex - -from PyQt6.QtCore import QObject, QProcess, QStandardPaths - -DOWN = "ssh: connection failed" -MISSING = "sbb-remote is not installed on the VM" - - -def parse(rc, out, err): - """Map one sbb-remote call to (data, error).""" - if rc == 255: - return None, DOWN - if rc == 127: - return None, MISSING - try: - data = json.loads(out) - except ValueError: - return None, err.decode(errors="replace").strip() or f"exit {rc}" - if rc != 0 or "error" in data: - return None, data.get("error", f"exit {rc}") - return data, None - - -class Remote(QObject): - def __init__(self, cfg, parent=None): - super().__init__(parent) - self.cfg = cfg # shared dict, updated in place on config reload - run = QStandardPaths.writableLocation(QStandardPaths.StandardLocation.RuntimeLocation) - self.control = os.path.join(run or "/tmp", "sbb-%C") - - def ssh_args(self, *args): - r = self.cfg["remote"] - # ssh joins its args into a remote shell command: quote every one - return ["-o", "ControlMaster=auto", "-o", f"ControlPath={self.control}", - "-o", "ControlPersist=10m", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5", - *r["ssh_options"], r["target"], "sbb-remote", *map(shlex.quote, args)] - - def call(self, cb, *args): - """Run sbb-remote <args>, then cb(data, error).""" - self._run("ssh", self.ssh_args(*args), lambda rc, out, err: cb(*parse(rc, out, err))) - - def virsh(self, cb, *args): - """Run local virsh against the configured libvirt URI, then cb(rc, text).""" - argv = ["-c", self.cfg["vm"]["libvirt_uri"], *args] - self._run("virsh", argv, - lambda rc, out, err: cb(rc, (out or err).decode(errors="replace").strip())) - - def follow(self, jid, offset): - """Start job-follow; the caller wires readyReadStandardOutput and finished.""" - p = QProcess(self) - p.start("ssh", self.ssh_args("job-follow", jid, str(offset))) - return p - - def _run(self, prog, argv, cb): - p = QProcess(self) - - def done(rc, _status=None): - cb(rc, bytes(p.readAllStandardOutput()), bytes(p.readAllStandardError())) - p.deleteLater() - - p.finished.connect(done) - p.errorOccurred.connect( - lambda e: e == QProcess.ProcessError.FailedToStart and done(-1)) - p.start(prog, argv) -```` - -- [ ] **Step 5: Formatting helpers and Dashboard** - -Create `sb_builder/fmt.py`: - -````python -# sb-builder - formatting helpers -# Copyright (C) 2026 Danilo M. <danix@danix.xyz> -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 as -# published by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License along -# with this program; if not, write to the Free Software Foundation, Inc., -# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. - -"""Text formatting shared by the tabs.""" - -import time - -NA = "n/a" -ICON = {"starting": "…", "running": "▶", "ok": "✔", "failed": "✘", - "killed": "■", "lost": "?"} - - -def fmt_bytes(n): - if n is None: - return NA - for unit in ("B", "KiB", "MiB", "GiB"): - if n < 1024: - return f"{n:.0f} {unit}" if unit == "B" else f"{n:.1f} {unit}" - n /= 1024 - return f"{n:.1f} TiB" - - -def fmt_duration(secs): - if secs is None: - return NA - d, r = divmod(int(secs), 86400) - h, r = divmod(r, 3600) - m, s = divmod(r, 60) - return f"{d}d {h}h" if d else f"{h}h {m}m" if h else f"{m}m {s}s" - - -def fmt_age(ts): - return NA if ts is None else fmt_duration(time.time() - ts) + " ago" - - -def fmt_time(ts): - return time.strftime("%m-%d %H:%M", time.localtime(ts)) - - -def job_title(j): - return " ".join([j["kind"], *j["args"]]) - - -def describe(j): - """One-line job summary: icon, kind and args, start time, duration.""" - dur = None if "start" not in j else j.get("end", time.time()) - j["start"] - return (f"{ICON.get(j['status'], '?')} {job_title(j)} " - f"{fmt_time(j.get('start', j['created']))} {fmt_duration(dur)}") -```` - -Create `sb_builder/dashboard.py`: - -````python -# sb-builder - Dashboard tab -# Copyright (C) 2026 Danilo M. <danix@danix.xyz> -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 as -# published by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License along -# with this program; if not, write to the Free Software Foundation, Inc., -# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. - -"""Dashboard tab: tiles filled from sbb-remote stats and jobs.""" - -import time - -from PyQt6.QtWidgets import (QGridLayout, QGroupBox, QHBoxLayout, QLabel, - QPushButton, QVBoxLayout, QWidget) - -from .fmt import NA, describe, fmt_age, fmt_bytes, fmt_duration - -TILES = ("VM", "Repo", "slackrepo", "Sync", "Jobs", "Orphans") - - -class Dashboard(QWidget): - def __init__(self, parent=None): - super().__init__(parent) - self.labels = {} - grid = QGridLayout() - for i, name in enumerate(TILES): - box = QGroupBox(name) - label = QLabel(NA) - label.setWordWrap(True) - QVBoxLayout(box).addWidget(label) - grid.addWidget(box, i // 3, i % 3) - self.labels[name] = label - self.labels["Orphans"].setText("coming in phase 5") - self.stale = QLabel() - self.refresh_btn = QPushButton("Refresh") - top = QHBoxLayout() - top.addWidget(self.stale) - top.addStretch() - top.addWidget(self.refresh_btn) - layout = QVBoxLayout(self) - layout.addLayout(top) - layout.addLayout(grid) - layout.addStretch() - - def show_stats(self, s): - L = self.labels - load, disk = s.get("load"), s.get("disk_free") or {} - L["VM"].setText( - f"load: {NA if load is None else ' '.join(f'{x:.2f}' for x in load)}\n" - f"uptime: {fmt_duration(s.get('uptime'))}\n" - f"free: repo {fmt_bytes(disk.get('repo'))}, state {fmt_bytes(disk.get('state'))}") - p = s.get("packages") - L["Repo"].setText(NA if p is None else "\n".join(f"{k}: {v}" for k, v in p.items())) - sr = s.get("slackrepo") - if sr is None: - L["slackrepo"].setText(NA) - else: - never = NA if p is None else p["total"] - sr["built"] - L["slackrepo"].setText("\n".join( - [f"built: {sr['built']}", f"never built: {never}", - *(f"{k}: {v}" for k, v in sorted(sr["results"].items()))])) - sync = s.get("sync") or {} - L["Sync"].setText("\n".join(f"{o}: {fmt_age(t)}" for o, t in sync.items()) or NA) - - def show_jobs(self, jobs): - self.labels["Jobs"].setText("\n".join(describe(j) for j in jobs[:10]) or "no jobs yet") - - def set_stale(self, since): - """Grey the tiles when the VM stopped answering; None clears it.""" - for label in self.labels.values(): - label.setEnabled(since is None) - self.stale.setText("" if since is None else - f"stale since {time.strftime('%H:%M', time.localtime(since))}") -```` - -- [ ] **Step 6: Jobs tab** - -Create `sb_builder/jobs.py`: - -````python -# sb-builder - Jobs tab -# Copyright (C) 2026 Danilo M. <danix@danix.xyz> -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 as -# published by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License along -# with this program; if not, write to the Free Software Foundation, Inc., -# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. - -"""Jobs tab: job list, live log through job-follow, start and kill.""" - -import codecs -import re - -from PyQt6.QtCore import Qt, pyqtSignal -from PyQt6.QtGui import QFontDatabase, QTextCursor -from PyQt6.QtWidgets import (QHBoxLayout, QListWidget, QListWidgetItem, QMenu, - QMessageBox, QPlainTextEdit, QPushButton, QSplitter, - QToolButton, QVBoxLayout, QWidget) - -from .fmt import describe - -ANSI = re.compile(rb"\x1b\[[0-9;?]*[A-Za-z]") -SHADOW_PLAN = re.compile(r"^(un)?shadow: ", re.M) -ACTIVE = ("starting", "running") -BAD = ("failed", "killed", "lost") - - -class JobsTab(QWidget): - refresh_requested = pyqtSignal() - - def __init__(self, remote, parent=None): - super().__init__(parent) - self.remote = remote - self.jobs = {} - self.current = None - self.offset = 0 - self.proc = None - self.decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") - - self.start_menu = QMenu(self) - start = QToolButton(text="Start", popupMode=QToolButton.ToolButtonPopupMode.InstantPopup) - start.setMenu(self.start_menu) - self.apply_btn = QPushButton("Apply shadowing") - self.apply_btn.clicked.connect(lambda: self.confirm_start( - "Apply the shadowing plan to the repo (sbshadow --yes)?", "sbshadow", "--yes")) - self.apply_btn.hide() - self.kill_btn = QPushButton("Kill") - self.kill_btn.clicked.connect(self.kill) - bar = QHBoxLayout() - for w in (start, self.apply_btn): - bar.addWidget(w) - bar.addStretch() - bar.addWidget(self.kill_btn) - - self.list = QListWidget() - self.list.currentItemChanged.connect(self._on_select) - self.log = QPlainTextEdit(readOnly=True) - self.log.setFont(QFontDatabase.systemFont(QFontDatabase.SystemFont.FixedFont)) - self.log.setMaximumBlockCount(50000) # ponytail: older lines drop off; the full log stays on the VM - split = QSplitter() - split.addWidget(self.list) - split.addWidget(self.log) - split.setSizes([300, 700]) - layout = QVBoxLayout(self) - layout.addLayout(bar) - layout.addWidget(split) - self.set_overlays([]) - - def set_overlays(self, overlays): - m = self.start_menu - m.clear() - for o in overlays: - m.addAction(f"sbpull {o} (preview)", lambda o=o: self.start("sbpull", o, "--no")) - m.addSeparator() - m.addAction("sbshadow (preview)", lambda: self.start("sbshadow", "--no")) - m.addAction("sbshadow (apply)…", lambda: self.confirm_start( - "Run sbshadow --yes on the repo?", "sbshadow", "--yes")) - - def set_jobs(self, jobs): - self.jobs = {j["id"]: j for j in jobs} - self.list.blockSignals(True) - self.list.clear() - for j in jobs: - item = QListWidgetItem(describe(j)) - item.setData(Qt.ItemDataRole.UserRole, j["id"]) - if j["status"] in BAD: - item.setForeground(Qt.GlobalColor.red) - self.list.addItem(item) - if j["id"] == self.current: - self.list.setCurrentItem(item) - self.list.blockSignals(False) - j = self.jobs.get(self.current) - if j and j["status"] in ACTIVE and self.proc is None: - self._follow() # stream dropped while the job kept running: resume at offset - self._update_buttons() - - def select(self, jid): - for i in range(self.list.count()): - if self.list.item(i).data(Qt.ItemDataRole.UserRole) == jid: - self.list.setCurrentItem(self.list.item(i)) - return - self._show(jid) # not listed yet: just started - - def append_log(self, text): - cur = self.log.textCursor() - cur.movePosition(QTextCursor.MoveOperation.End) - # ponytail: a \r split across two chunks is printed as-is - for i, part in enumerate(text.replace("\r\n", "\n").split("\r")): - if i: # carriage return: the next text replaces the current line - cur.movePosition(QTextCursor.MoveOperation.StartOfBlock, - QTextCursor.MoveMode.KeepAnchor) - cur.removeSelectedText() - cur.insertText(part) - self.log.setTextCursor(cur) - - def stop(self): - if self.proc: - p, self.proc = self.proc, None - p.kill() - - def start(self, *args): - self.remote.call(self._started, "job-start", *args) - - def confirm_start(self, question, *args): - if QMessageBox.question(self, "Confirm", question) == QMessageBox.StandardButton.Yes: - self.start(*args) - - def kill(self): - if QMessageBox.question(self, "Kill job", f"Send SIGTERM to {self.current}?") \ - == QMessageBox.StandardButton.Yes: - self.remote.call(lambda _d, err: err and QMessageBox.warning(self, "Kill failed", err), - "job-kill", self.current) - - def _started(self, data, err): - if err: - QMessageBox.warning(self, "Start failed", err) - return - if "busy" in data: - box = QMessageBox(QMessageBox.Icon.Information, "Busy", - f"busy: {data['busy']}", parent=self) - show = box.addButton("Show job", QMessageBox.ButtonRole.AcceptRole) - box.addButton(QMessageBox.StandardButton.Close) - box.exec() - if box.clickedButton() is show and data["busy"]: - self.select(data["busy"]) - return - self.select(data["id"]) - self.refresh_requested.emit() - - def _on_select(self, item, _prev): - jid = item.data(Qt.ItemDataRole.UserRole) if item else None - if jid != self.current: - self._show(jid) - - def _show(self, jid): - self.stop() - self.current, self.offset = jid, 0 - self.decoder.reset() - self.log.clear() - if jid: - self._follow() - self._update_buttons() - - def _follow(self): - jid = self.current - p = self.proc = self.remote.follow(jid, self.offset) - p.readyReadStandardOutput.connect(lambda: self._chunk(p, jid)) - p.finished.connect(lambda *_: self._ended(p)) - - def _chunk(self, p, jid): - data = bytes(p.readAllStandardOutput()) - if jid != self.current: - return - self.offset += len(data) - self.append_log(self.decoder.decode(ANSI.sub(b"", data))) - - def _ended(self, p): - if p is self.proc: - self.proc = None - p.deleteLater() - self._update_buttons() - - def _update_buttons(self): - j = self.jobs.get(self.current) - self.kill_btn.setEnabled(bool(j) and j["status"] == "running") - self.apply_btn.setVisible( - bool(j) and j["status"] == "ok" and "--no" in j["args"] - and bool(SHADOW_PLAN.search(self.log.toPlainText()))) -```` - -- [ ] **Step 7: Main window and entry point** - -Create `sb_builder/mainwindow.py`: - -````python -# sb-builder - main window -# Copyright (C) 2026 Danilo M. <danix@danix.xyz> -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 as -# published by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License along -# with this program; if not, write to the Free Software Foundation, Inc., -# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. - -"""Main window: tabs, status bar, connection polling, VM control, notifications.""" - -import time - -from PyQt6.QtCore import QMetaType, QObject, QTimer, pyqtSignal, pyqtSlot -from PyQt6.QtDBus import QDBusArgument, QDBusConnection, QDBusInterface, QDBusMessage -from PyQt6.QtWidgets import QLabel, QMainWindow, QMessageBox, QPushButton, QTabBar, QTabWidget - -from . import VERSION, config -from .dashboard import Dashboard -from .fmt import job_title -from .jobs import JobsTab -from .remote import DOWN, MISSING - -NOTIFY = "org.freedesktop.Notifications" -NOTIFY_PATH = "/org/freedesktop/Notifications" -END_EVENT = {"ok": "finished", "failed": "failed", "killed": "killed", "lost": "lost"} -MAX_RETRY_MS = 30000 - - -class Notifier(QObject): - """freedesktop notifications over D-Bus; clicked(job id) on the default action.""" - - clicked = pyqtSignal(str) - - def __init__(self, parent=None): - super().__init__(parent) - bus = QDBusConnection.sessionBus() - self.iface = QDBusInterface(NOTIFY, NOTIFY_PATH, NOTIFY, bus) - bus.connect(NOTIFY, NOTIFY_PATH, NOTIFY, "ActionInvoked", self._action) - self.ids = {} # notification id -> job id - - def send(self, title, body, jid=""): - reply = self.iface.call( - "Notify", "sb-builder", QDBusArgument(0, QMetaType.Type.UInt.value), "", title, body, - QDBusArgument(["default", "Open"], QMetaType.Type.QStringList.value), {}, -1) - if reply.type() == QDBusMessage.MessageType.ReplyMessage: - self.ids[reply.arguments()[0]] = jid - - @pyqtSlot("uint", "QString") - def _action(self, nid, _key): - if nid in self.ids: - self.clicked.emit(self.ids.pop(nid)) - - -class MainWindow(QMainWindow): - def __init__(self, cfg, remote, notifier=None): - super().__init__() - self.cfg, self.remote = cfg, remote - self.notifier = notifier or Notifier(self) - self.notifier.clicked.connect(self.open_job) - self.jobs_tab = None - self.seen = None # job id -> status at the last poll; None before the first - self.online = None # None until the first answer - self.overlays = [] - self.vm_action = None # "start" or "resume" offered on the VM button - self.vm_woken = False # we started the VM: notify once ssh answers - self.running_id = None - self.setWindowTitle("sb-builder") - - self.tabs = QTabWidget(tabsClosable=True) - self.tabs.tabCloseRequested.connect(self._close_tab) - self.tabs.currentChanged.connect(lambda _i: self.poll_stats()) - self.dashboard = Dashboard() - self.dashboard.refresh_btn.clicked.connect(self.poll_stats) - self.tabs.addTab(self.dashboard, "Dashboard") - self.tabs.tabBar().setTabButton(0, QTabBar.ButtonPosition.RightSide, None) - self.setCentralWidget(self.tabs) - - file_menu = self.menuBar().addMenu("&File") - file_menu.addAction("&Reload config", self.reload_config) - file_menu.addAction("&Quit", self.close).setShortcut("Ctrl+Q") - self.menuBar().addMenu("&Open").addAction("&Jobs", self.open_jobs) - - self.conn = QLabel("● connecting") - self.vm_btn = QPushButton() - self.vm_btn.clicked.connect(lambda: self._vm_do(self.vm_action)) - self.vm_btn.hide() - self.job_btn = QPushButton(flat=True) - self.job_btn.clicked.connect(lambda: self.open_job(self.running_id)) - self.job_btn.hide() - for w in (self.job_btn, self.vm_btn, self.conn): - self.statusBar().addPermanentWidget(w) - - self.jobs_timer = QTimer(self, timeout=self.poll_jobs) - self.stats_timer = QTimer(self, timeout=self.poll_stats) - self._start_timers() - QTimer.singleShot(0, self.poll_jobs) - - # ── polling ───────────────────────────────────────────────────────────── - - def _start_timers(self): - self.jobs_timer.start(self.cfg["refresh"]["jobs"] * 1000) - self.stats_timer.start(self.cfg["refresh"]["dashboard"] * 1000) - - def poll_jobs(self): - self.remote.call(self._got_jobs, "jobs") - - def poll_stats(self): - if self.online and self.tabs.currentWidget() is self.dashboard: - self.remote.call(self._got_stats, "stats") - - def _got_jobs(self, data, err): - if err: - return self._failed(err) - self._up() - jobs = data["jobs"] - self._notify_ends(jobs) - self.dashboard.show_jobs(jobs) - if self.jobs_tab: - self.jobs_tab.set_jobs(jobs) - run = next((j for j in jobs if j["status"] == "running"), None) - self.running_id = run and run["id"] - self.job_btn.setVisible(run is not None) - if run: - self.job_btn.setText(f"▶ {job_title(run)}") - - def _got_stats(self, data, err): - if err: - return self._failed(err) - self.dashboard.show_stats(data) - - def _got_version(self, data, err): - if err: - return self._failed(err) - self.overlays = data.get("overlays", []) - if self.jobs_tab: - self.jobs_tab.set_overlays(self.overlays) - if data["version"] != VERSION: - self.conn.setText(f"● helper {data['version']}, GUI {VERSION}: run ./install.sh --remote") - - def _up(self): - if self.online: - return - self.online = True - self.conn.setText("● connected") - self.vm_btn.hide() - self.dashboard.set_stale(None) - self.jobs_timer.setInterval(self.cfg["refresh"]["jobs"] * 1000) - self.remote.call(self._got_version, "version") - self.poll_stats() - if self.vm_woken: - self.vm_woken = False - self.notify("vm-up", f"{self.cfg['vm']['domain']} is up") - - def _failed(self, err): - if err == MISSING: - self.conn.setText("● sbb-remote missing: run ./install.sh --remote") - return - if err != DOWN: - self.statusBar().showMessage(err, 10000) - return - if self.online is not False: - self.online = False - self.dashboard.set_stale(time.time()) - base = self.cfg["refresh"]["jobs"] * 1000 - self.jobs_timer.setInterval(min(self.jobs_timer.interval() * 2, max(MAX_RETRY_MS, base))) - self._check_vm() - - # ── VM control ────────────────────────────────────────────────────────── - - def _check_vm(self): - if not self.cfg["vm"]["domain"]: - self.conn.setText("● down") - return - self.remote.virsh(self._got_domstate, "domstate", self.cfg["vm"]["domain"]) - - def _got_domstate(self, rc, state): - if self.online: - return # ssh came back while virsh ran - self.vm_btn.hide() - if rc != 0: - self.conn.setText("● down") - elif state == "shut off": - self.conn.setText("● VM off") - if self.cfg["vm"]["autostart"]: - self._vm_do("start") - else: - self._offer("Start VM", "start") - elif state == "paused": - self.conn.setText("● VM paused") - self._offer("Resume VM", "resume") - elif state == "running": - self.conn.setText("● VM running, ssh unreachable") - else: - self.conn.setText(f"● VM {state}") - - def _offer(self, label, action): - self.vm_action = action - self.vm_btn.setText(label) - self.vm_btn.show() - - def _vm_do(self, action): - self.vm_btn.hide() - self.vm_woken = True - self.conn.setText(f"● VM {action} requested") - self.remote.virsh( - lambda rc, out: rc and self.statusBar().showMessage(f"virsh {action}: {out}", 10000), - action, self.cfg["vm"]["domain"]) - - # ── notifications ─────────────────────────────────────────────────────── - - def notify(self, event, body, jid=""): - n = self.cfg["notify"] - if n["enabled"] and event in n["events"]: - self.notifier.send(f"sb-builder: {event}", body, jid) - - def _notify_ends(self, jobs): - prev, self.seen = self.seen, {j["id"]: j["status"] for j in jobs} - if prev is None: - return # first poll: old jobs are history, not news - for j in jobs: - event = END_EVENT.get(j["status"]) - if event and prev.get(j["id"], "running") != j["status"]: - self.notify(event, job_title(j), j["id"]) - - # ── tabs ──────────────────────────────────────────────────────────────── - - def open_jobs(self): - if self.jobs_tab is None: - self.jobs_tab = JobsTab(self.remote) - self.jobs_tab.refresh_requested.connect(self.poll_jobs) - self.jobs_tab.set_overlays(self.overlays) - self.tabs.addTab(self.jobs_tab, "Jobs") - self.poll_jobs() - self.tabs.setCurrentWidget(self.jobs_tab) - return self.jobs_tab - - def open_job(self, jid): - self.show() - self.raise_() - self.activateWindow() - if jid: - self.open_jobs().select(jid) - - def _close_tab(self, i): - w = self.tabs.widget(i) - if w is self.dashboard: - return - self.tabs.removeTab(i) - if w is self.jobs_tab: - self.jobs_tab = None - w.stop() - w.deleteLater() - - def reload_config(self): - cfg, err = config.load() - if err: - QMessageBox.warning(self, "Config error", f"{err}\nUsing defaults.") - self.cfg.clear() - self.cfg.update(cfg) # Remote holds the same dict - self._start_timers() -```` - -Create `sb_builder/__main__.py`: - -````python -# sb-builder - entry point -# Copyright (C) 2026 Danilo M. <danix@danix.xyz> -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 as -# published by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License along -# with this program; if not, write to the Free Software Foundation, Inc., -# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. - -"""Entry point: python3 -m sb_builder.""" - -import sys - -from PyQt6.QtWidgets import QApplication, QMessageBox - -from . import config -from .mainwindow import MainWindow -from .remote import Remote - - -def main(): - app = QApplication(sys.argv) - app.setApplicationName("sb-builder") - cfg, err = config.load() - w = MainWindow(cfg, Remote(cfg)) - w.resize(1100, 700) - w.show() - if err: - QMessageBox.warning(w, "Config error", f"{err}\nUsing defaults.") - return app.exec() - - -if __name__ == "__main__": - sys.exit(main()) -```` - -- [ ] **Step 8: Run the whole suite** - -Run: `python3 -m unittest discover -s tests` -Expected: `Ran 38 tests`, `OK` (plus the two expected ResourceWarnings from Task 2). - -- [ ] **Step 9: Commit** - -```bash -git add sb_builder config.example.toml tests/test_gui.py -git commit -F - <<'EOF' -Add the PyQt6 GUI: Dashboard, Jobs, VM control, notifications - -All VM access goes through remote.py (ssh with a shared ControlMaster, -local virsh). The main window polls jobs, backs off while ssh is down, -checks the libvirt domain and offers Start or Resume, and sends -freedesktop notifications when a job ends or the VM comes up. The Jobs -tab streams a job's log with job-follow and resumes at the last offset -after a drop. Config is TOML, validated against built-in defaults. - -Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> -EOF -git log --format='%h %G? %s' -1 -``` - ---- - -### Task 4: Launcher, install script, docs - -**Files:** -- Create: `sb-builder`, `install.sh` -- Modify: `README.md` (Requirements, new Install and Usage), `AGENTS.md` (Testing section) - -- [ ] **Step 1: Launcher** - -Create `sb-builder` and `chmod 755 sb-builder`: - -````sh -#!/bin/sh -# sb-builder - launcher -# Copyright (C) 2026 Danilo M. <danix@danix.xyz> -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 as -# published by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License along -# with this program; if not, write to the Free Software Foundation, Inc., -# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. - -here=$(dirname "$(readlink -f "$0")") -PYTHONPATH="$here${PYTHONPATH:+:$PYTHONPATH}" exec python3 -m sb_builder "$@" -```` - -- [ ] **Step 2: Install script** - -Create `install.sh` and `chmod 755 install.sh`: - -````bash -#!/bin/bash -# sb-builder - install the GUI locally, or the helper on the VM -# Copyright (C) 2026 Danilo M. <danix@danix.xyz> -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 as -# published by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License along -# with this program; if not, write to the Free Software Foundation, Inc., -# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. - -# ./install.sh symlink the launcher into ~/bin (BINDIR overrides), -# create the GUI config from the example if missing -# ./install.sh --remote deploy remote/sbb-remote to the VM (SBB_TARGET, -# default buildsystem) and create its config if missing - -set -euo pipefail -here=$(dirname "$(readlink -f "$0")") - -if [[ ${1:-} == --remote ]]; then - target=${SBB_TARGET:-buildsystem} - read -rp "Deploy sbb-remote to $target:/usr/local/bin (config created only if missing)? [y/N] " a - [[ $a =~ ^[Yy]$ ]] || exit 1 - scp -q "$here/remote/sbb-remote" "$target:/usr/local/bin/sbb-remote" - scp -q "$here/remote/config.example.toml" "$target:/tmp/sbb-config.example.toml" - ssh "$target" 'chmod 755 /usr/local/bin/sbb-remote && - mkdir -p /etc/sb-builder && - { [ -e /etc/sb-builder/config.toml ] || cp /tmp/sbb-config.example.toml /etc/sb-builder/config.toml; } && - rm -f /tmp/sbb-config.example.toml && - sbb-remote version' - exit -fi - -bindir=${BINDIR:-$HOME/bin} -cfgdir=${XDG_CONFIG_HOME:-$HOME/.config}/sb-builder -mkdir -p "$bindir" "$cfgdir" -ln -sfn "$here/sb-builder" "$bindir/sb-builder" -[[ -e $cfgdir/config.toml ]] || cp "$here/config.example.toml" "$cfgdir/config.toml" -echo "installed $bindir/sb-builder -> $here/sb-builder, config in $cfgdir/config.toml" -```` - -- [ ] **Step 3: Syntax and headless launch check** - -```bash -bash -n install.sh && sh -n sb-builder && echo syntax ok -XDG_CONFIG_HOME=$(mktemp -d) QT_QPA_PLATFORM=offscreen timeout 10 ./sb-builder; echo "rc=$?" -``` - -Expected: `syntax ok`, then `rc=124` (timeout killed a running GUI: it started without errors). - -- [ ] **Step 4: README** - -Replace `README.md` with: - -````markdown -# sb-builder - -Desktop GUI for managing a Slackware SlackBuilds build VM from one place. It -runs slackrepo builds, overlay syncs (sbpull, sbshadow), hintfile work -(mkhint) and SBo orphan discovery on the VM over ssh, as background jobs that -keep running when the GUI disconnects and can be reattached later. A -dashboard shows VM and repository state, and the GUI can start the VM through -libvirt when it is off. - -Work in progress. See `docs/superpowers/specs/` for the design. - -## Requirements - -- Desktop: Python 3.11 or newer, PyQt6 (with QtDBus), ssh access to the VM, - optionally `virsh` for starting the VM -- VM: Python 3.11 or newer, root ssh login, the - [slackrepo_setup](https://forge.slackware.nl/danix/slackrepo_setup) tools (`sbpull`, `sbshadow`) - and mkhint - -## Install - - ./install.sh # ~/bin/sb-builder -> this repo, GUI config if missing - ./install.sh --remote # helper to the VM, its config if missing (asks first) - -`BINDIR` changes the launcher directory, `SBB_TARGET` the ssh destination for -`--remote` (default `buildsystem`). Configs: - -- GUI: `~/.config/sb-builder/config.toml`, see `config.example.toml` -- VM: `/etc/sb-builder/config.toml`, see `remote/config.example.toml` - -Every key is optional. `File > Reload config` rereads the GUI config. - -## Usage - - sb-builder - -- **Dashboard**: VM load, uptime and free disk, package counts, slackrepo - build results, last overlay pull, recent jobs. Refreshes while visible. -- **Open > Jobs**: start a job from the Start menu (`sbpull <overlay>` - preview, `sbshadow` preview or apply), follow its log live, kill it. After a - preview that lists `shadow:` or `unshadow:` lines, "Apply shadowing" runs - `sbshadow --yes`. - -Jobs run on the VM detached from the ssh session, one at a time. Closing the -GUI or losing the connection does not stop them; reopening the job resumes -the log. Job state and logs live in `/var/lib/sb-builder/jobs/` on the VM. - -When ssh fails and `vm.domain` is set, the status bar shows the libvirt state -of the VM and offers to start or resume it. Desktop notifications report -finished, failed, killed and lost jobs, and the VM coming up. - -## License - -Copyright (C) 2026 Danilo M. <danix@danix.xyz> - -GPLv2 only. See [LICENSE](LICENSE) for the full text. - -## Development Approach - -This project is developed using AI-assisted tools. Code is generated with the help of AI based on human-provided specifications, design decisions, and iterative feedback. - -All contributions are reviewed, tested, and curated by the maintainer before being included in the codebase. AI is used as a productivity and exploration tool, while human oversight remains central to all decisions. - -The goal is to combine the flexibility of AI-assisted development with standard open-source practices such as transparency, review, and accountability. -```` - -- [ ] **Step 5: AGENTS.md Testing section** - -In `AGENTS.md` replace - -```markdown -No test suite yet. Planned (phase 1 spec): `python3 -m unittest discover -tests`, with the GUI smoke test under `QT_QPA_PLATFORM=offscreen`. -``` - -with - -```markdown - python3 -m unittest discover -s tests - -Headless (`test_gui.py` sets `QT_QPA_PLATFORM=offscreen`), no VM needed. -`test_remote.py` loads `remote/sbb-remote` with `SourceFileLoader` (no `.py` -extension) and points it at a temp state dir through `SBB_CONFIG`. Two -`ResourceWarning: subprocess ... is still running` lines are expected: the -detached `_run` is never waited for. To try the helper on the VM without -installing it, copy it to `/tmp` there and run it with `SBB_CONFIG` pointing -at a config whose `state_dir` is under `/tmp`. -``` - -- [ ] **Step 6: Commit** - -```bash -git add sb-builder install.sh README.md AGENTS.md -git commit -F - <<'EOF' -Add launcher, install script and usage docs - -Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> -EOF -git log --format='%h %G? %s' -1 -git push -``` - ---- - -### Task 5: Deploy and first real run (with the user) - -Ask the user before each VM-touching step. - -- [ ] **Step 1: Local install**: `./install.sh`. Expected: `installed ~/bin/sb-builder -> ...`. -- [ ] **Step 2: Deploy the helper**: `./install.sh --remote`, answer `y`. Expected last line: `{"version": "0.1.0", "overlays": ["personal", "pentesting"]}`. -- [ ] **Step 3: User runs `sb-builder`** and checks: status bar `● connected`; Dashboard tiles filled (about 9957 packages, 48 personal, 3 pentesting, 13 shadowed at planning time); Open > Jobs. -- [ ] **Step 4: Preview job**: Start > `sbshadow (preview)`. The log streams, the job ends ✔, a notification appears; clicking it opens the job. "Apply shadowing" shows only if the plan lists changes. -- [ ] **Step 5: Detach check**: start `sbpull personal (preview)`, quit the GUI while it runs, reopen, select the job: the log continues from the start, the job finishes ✔. `ssh buildsystem ls /var/lib/sb-builder/jobs` shows it. -- [ ] **Step 6: Busy check**: while a job runs, start another: dialog `busy: <id>` with Show job. -- [ ] **Step 7: VM check (optional, user decides)**: with the VM shut off (`virsh shutdown buildsystem` by the user), the status bar shows `● VM off` and Start VM; pressing it boots the VM and a `vm-up` notification arrives when ssh answers. - -Record anything that misbehaves as a follow-up; do not patch it silently. |