aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorDanilo M. <danix@danix.xyz>2026-10-19 20:53:43 +0200
committerDanilo M. <danix@danix.xyz>2026-10-19 20:53:43 +0200
commitb8400f38411818aa54a773dffb2c3a0c56b06dba (patch)
treee6390c860efb005e424b91594c6c619c120d9b75
parentd29fc1cd128b7d6a18e3e50d8ff4d6262f7e88ae (diff)
downloadsb-builder-b8400f38411818aa54a773dffb2c3a0c56b06dba.tar.gz
sb-builder-b8400f38411818aa54a773dffb2c3a0c56b06dba.zip
Add Packages tab, hint view, Run it and signing key unlock
Packages tab over the helper inventory: query bar with cat:/maint:/ req:/status: prefixes, overlay and status chips, check column, and a detail pane with versions, dependencies after hints and a structured read-only view of the hint (hints.py mirrors slackrepo's parser). Actions start slackrepo previews; the Jobs tab offers Run it with a publish box (default from [builds] publish), asking for the signing key passphrase first when the key is locked. Status bar shows the key state; Dashboard tiles open their tabs; inventory reloads after slackrepo and publish jobs. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
-rw-r--r--config.example.toml3
-rw-r--r--sb_builder/__init__.py2
-rw-r--r--sb_builder/config.py1
-rw-r--r--sb_builder/dashboard.py14
-rw-r--r--sb_builder/hints.py195
-rw-r--r--sb_builder/jobs.py63
-rw-r--r--sb_builder/mainwindow.py43
-rw-r--r--sb_builder/packages.py332
-rw-r--r--sb_builder/remote.py12
-rw-r--r--tests/test_gui.py199
10 files changed, 844 insertions, 20 deletions
diff --git a/config.example.toml b/config.example.toml
index 22afaec..d5fd2c3 100644
--- a/config.example.toml
+++ b/config.example.toml
@@ -17,3 +17,6 @@ events = ["finished", "failed", "killed", "lost", "vm-up"]
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
+
+[builds]
+publish = true # initial state of the "publish" box next to Run it
diff --git a/sb_builder/__init__.py b/sb_builder/__init__.py
index 6428da7..7c58f9a 100644
--- a/sb_builder/__init__.py
+++ b/sb_builder/__init__.py
@@ -16,4 +16,4 @@
"""sb-builder: Qt6 GUI driving the SlackBuild tooling on a build VM."""
-VERSION = "0.1.0" # must match remote/sbb-remote VERSION
+VERSION = "0.2.0" # must match remote/sbb-remote VERSION
diff --git a/sb_builder/config.py b/sb_builder/config.py
index 4549b66..da1d388 100644
--- a/sb_builder/config.py
+++ b/sb_builder/config.py
@@ -25,6 +25,7 @@ DEFAULTS = {
"refresh": {"dashboard": 30, "jobs": 5},
"notify": {"enabled": True, "events": ["finished", "failed", "killed", "lost", "vm-up"]},
"vm": {"libvirt_uri": "qemu:///system", "domain": "buildsystem", "autostart": False},
+ "builds": {"publish": True},
}
diff --git a/sb_builder/dashboard.py b/sb_builder/dashboard.py
index d4152ae..85c19f4 100644
--- a/sb_builder/dashboard.py
+++ b/sb_builder/dashboard.py
@@ -19,7 +19,7 @@
import html
import time
-from PyQt6.QtCore import QEvent, QRectF, Qt
+from PyQt6.QtCore import QEvent, QRectF, Qt, pyqtSignal
from PyQt6.QtGui import QColor, QFont, QPainter, QPalette
from PyQt6.QtWidgets import (QFrame, QGridLayout, QHBoxLayout, QLabel, QPushButton,
QToolTip, QVBoxLayout, QWidget)
@@ -44,12 +44,24 @@ def muted(label):
class Card(QFrame):
"""Rounded panel with a muted title; content goes into self.body."""
+ clicked = pyqtSignal()
+
def __init__(self, title):
super().__init__(objectName="card")
self.body = QVBoxLayout(self)
self.body.setContentsMargins(16, 12, 16, 14)
self.body.addWidget(muted(QLabel(title)))
+ def link(self, slot):
+ """Make the whole card open something."""
+ self.clicked.connect(slot)
+ self.setCursor(Qt.CursorShape.PointingHandCursor)
+
+ def mouseReleaseEvent(self, event):
+ if event.button() == Qt.MouseButton.LeftButton:
+ self.clicked.emit()
+ super().mouseReleaseEvent(event)
+
class Tile(Card):
"""Stat tile: title, one large value, a muted detail line."""
diff --git a/sb_builder/hints.py b/sb_builder/hints.py
new file mode 100644
index 0000000..34ef97d
--- /dev/null
+++ b/sb_builder/hints.py
@@ -0,0 +1,195 @@
+# sb-builder - slackrepo hint interpretation
+# 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.
+
+"""What a slackrepo hint file does, the way slackrepo reads it.
+
+Rules from slackrepo's libexec/functions.d/parsefunctions.sh (hint and fixup
+departments) and slackrepo.hint(5). Pure functions, no Qt.
+"""
+
+import os
+import re
+
+KNOWN = {"SKIP", "VERSION", "OPTIONS", "GROUPADD", "USERADD", "NOWARNING", "NOWARNINGS",
+ "ADDREQUIRES", "DELREQUIRES", "BUILDTIME", "CONFLICTS", "INSTALL", "NUMJOBS",
+ "ANSWER", "CLEANUP", "PRAGMA", "SPECIAL", "ARCH", "DOWNLOAD", "MD5SUM", "SHA256SUM",
+ "NODOWNLOAD"} # NODOWNLOAD: written by mkhint, shown as DOWNLOAD="no" by user decision
+PER_ARCH = re.compile(r"^(DOWNLOAD|MD5SUM|SHA256SUM)_\w+$")
+SHELLS = {"sh", "bash", "zsh", "ksh", "csh", "tcsh", "false", "nologin", "true"}
+
+PRAGMAS = {
+ "python3": "force python3 support: adds python3 to the deps, sets PYTHON3=yes",
+ "kernelmodule": "rebuild when the host kernel changes; depmod on uninstall",
+ "kernel": "rebuild when the host kernel changes",
+ "kernelversion": "uname reports $KERNEL as the kernel version",
+ "multilib_ldflags": "patch the SlackBuild to set LDFLAGS on a multilib host",
+ "wget": "download with a fake wget user agent",
+ "curl": "download with a fake curl user agent",
+ "download_basename": "symlink downloads to the URL's basename",
+ "download_urldecode": "URL-decode DOWNLOAD values before use",
+ "need_net": "network not blocked during the build",
+ "need_X": "X display not blocked during the build",
+ "noexport_ARCH": "set ARCH by patching the SlackBuild, not via the environment",
+ "noexport_BUILD": "set BUILD by patching the SlackBuild, not via the environment",
+ "noexport_TAG": "set TAG by patching the SlackBuild, not via the environment",
+ "noremove": "do not uninstall packages before building",
+ "nofakeroot": "build as root with sudo instead of fakeroot",
+ "no_make_test": "comment out 'make test' in the SlackBuild",
+ "stubs-32": "provide /usr/include/gnu/stubs-32.h during the build",
+ "abstar": "patch the SlackBuild to use tar --absolute-names",
+}
+
+
+def _raw(v):
+ return v["value"] if isinstance(v, dict) else v
+
+
+def pragma_meaning(p):
+ if p in PRAGMAS:
+ return PRAGMAS[p]
+ if p.startswith("x86arch="):
+ return f"on 32 bit x86, build with ARCH={p[8:]}"
+ if p.startswith("unset_"):
+ return f"unset {p[6:]} for the build"
+ return "unknown pragma"
+
+
+def parse_groupadd(value, pkg):
+ """[(gid, name)] like slackrepo: numeric field is the gid, other the name."""
+ out = []
+ for spec in value.split():
+ gid, name = None, pkg
+ for field in spec.split(":"):
+ if field[:1].isdigit():
+ gid = field
+ elif field:
+ name = field
+ out.append((gid, name))
+ return out
+
+
+def parse_useradd(value, pkg):
+ """[dict(uid, name, group, home, shell, extra)] with slackrepo's defaults."""
+ out = []
+ for spec in value.split():
+ u = {"uid": None, "name": pkg, "group": "", "home": "/dev/null",
+ "shell": "/bin/false", "extra": []}
+ for field in spec.split(":"):
+ if field.startswith("-g"):
+ u["group"] = field[2:]
+ elif field.startswith("-d"):
+ u["home"] = field[2:]
+ elif field.startswith("-s"):
+ u["shell"] = field[2:]
+ elif field.startswith("-"):
+ u["extra"].append(f"{field[:2]} {field[2:]}".strip())
+ elif field.startswith("/"):
+ # ponytail: slackrepo tests -x on the build host; guess by name instead
+ key = "shell" if os.path.basename(field) in SHELLS else "home"
+ u[key] = field
+ elif field[:1].isdigit():
+ u["uid"] = field
+ elif field:
+ u["name"] = field
+ u["group"] = u["group"] or u["name"]
+ out.append(u)
+ return out
+
+
+def dependencies(hints, requires):
+ """[(dep, mark)] as slackrepo computes them; mark in '', del, pragma, add, build."""
+ deleted = set(hints.get("DELREQUIRES", "").split()) | {"%README%"}
+ out = [(r, "del" if r in deleted else "") for r in requires]
+ if "python3" in hints.get("PRAGMA", hints.get("SPECIAL", "")).split():
+ out.append(("python3", "pragma"))
+ out += [(r, "add") for r in hints.get("ADDREQUIRES", "").split()]
+ out += [(r, "build") for r in hints.get("BUILDTIME", "").split()]
+ return out
+
+
+def interpret(hints, pkg, info_version=None, requires=()):
+ """Structured view of one package's hints.
+
+ hints: {key: value or {"value": raw, "shell": True}} from the helper, or None.
+ Returns {"skip", "sections": [(title, [(label, value)])], "deps",
+ "conflicts", "ignored", "shell"}.
+ """
+ raw = hints or {}
+ shell = sorted(k for k, v in raw.items() if isinstance(v, dict))
+ h = {k: _raw(v) for k, v in raw.items()}
+ view = {"skip": None, "sections": [], "deps": dependencies(h, list(requires)),
+ "conflicts": h.get("CONFLICTS", "").split(), "shell": shell,
+ "ignored": sorted(k for k in h if k not in KNOWN and not PER_ARCH.match(k))}
+
+ skip = h.get("SKIP")
+ if skip is not None and skip.lower() != "no":
+ view["skip"] = "always skipped" if skip.lower() == "yes" else skip
+
+ # version, downloads and checksums, resolved together as slackrepo does
+ arch = h.get("ARCH")
+ resolved = {}
+ for var in ("DOWNLOAD", "MD5SUM", "SHA256SUM"):
+ resolved[var] = h.get(f"{var}_{arch}") if arch and h.get(f"{var}_{arch}") else h.get(var)
+ if h.get("VERSION"):
+ resolved["MD5SUM"] = resolved["MD5SUM"] or "no"
+ resolved["SHA256SUM"] = resolved["SHA256SUM"] or "no"
+ rows = []
+ if h.get("VERSION"):
+ rows.append(("version", f"{info_version or '?'} β†’ {h['VERSION']}"))
+ if arch is not None:
+ rows.append(("arch", arch or "(empty)"))
+ if resolved["DOWNLOAD"] == "no" or h.get("NODOWNLOAD", "").lower() == "yes":
+ rows.append(("download", "no download, the source must already be in the cache"))
+ elif resolved["DOWNLOAD"]:
+ rows += [("download", url) for url in resolved["DOWNLOAD"].split()]
+ if resolved["MD5SUM"] == "no":
+ rows.append(("md5sum", "not checked"))
+ elif resolved["MD5SUM"]:
+ rows.append(("md5sum", resolved["MD5SUM"]))
+ if any(k.startswith("SHA256SUM") for k in h):
+ rows.append(("sha256sum", "not checked" if resolved["SHA256SUM"] == "no"
+ else resolved["SHA256SUM"]))
+ if rows:
+ view["sections"].append(("Version override", rows))
+
+ rows = [("option", o) for o in h.get("OPTIONS", "").split()]
+ if "NUMJOBS" in h:
+ rows.append(("make jobs", h["NUMJOBS"]))
+ if h.get("INSTALL"):
+ rows.append(("install after build", "no" if h["INSTALL"][:1] in "Nn0" else "yes"))
+ if "ANSWER" in h:
+ rows.append(("answers to the build", h["ANSWER"].replace("\\n", " ⏎ ")))
+ nowarning = h.get("NOWARNING") or h.get("NOWARNINGS")
+ if nowarning:
+ rows.append(("hide warnings matching", nowarning))
+ if rows:
+ view["sections"].append(("Build settings", rows))
+
+ pragmas = h.get("PRAGMA") or h.get("SPECIAL") or ""
+ if pragmas:
+ view["sections"].append(("Pragmas", [(p, pragma_meaning(p)) for p in pragmas.split()]))
+
+ rows = [("group", f"{name} (gid {gid or '?'})") for gid, name in parse_groupadd(h.get("GROUPADD", ""), pkg)]
+ for u in parse_useradd(h.get("USERADD", ""), pkg):
+ extra = f", {' '.join(u['extra'])}" if u["extra"] else ""
+ rows.append(("user", f"{u['name']} (uid {u['uid'] or '?'}), group {u['group']}, "
+ f"home {u['home']}, shell {u['shell']}{extra}"))
+ if rows:
+ view["sections"].append(("Host changes", rows))
+
+ if h.get("CLEANUP"):
+ view["sections"].append(("On uninstall", [("run", h["CLEANUP"])]))
+ return view
diff --git a/sb_builder/jobs.py b/sb_builder/jobs.py
index bcbd812..b413597 100644
--- a/sb_builder/jobs.py
+++ b/sb_builder/jobs.py
@@ -21,9 +21,9 @@ 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 PyQt6.QtWidgets import (QCheckBox, QHBoxLayout, QInputDialog, QLineEdit, QListWidget,
+ QListWidgetItem, QMenu, QMessageBox, QPlainTextEdit, QPushButton,
+ QSplitter, QToolButton, QVBoxLayout, QWidget)
from .fmt import describe
@@ -36,9 +36,10 @@ BAD = ("failed", "killed", "lost")
class JobsTab(QWidget):
refresh_requested = pyqtSignal()
- def __init__(self, remote, parent=None):
+ def __init__(self, remote, publish=True, parent=None):
super().__init__(parent)
self.remote = remote
+ self.pending = ()
self.jobs = {}
self.current = None
self.offset = 0
@@ -52,10 +53,15 @@ class JobsTab(QWidget):
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.publish_box = QCheckBox("publish", checked=publish)
+ self.publish_box.setToolTip("Push to the mirror when the build ends; "
+ "unticked, use Publish later")
+ self.run_btn = QPushButton("Run it")
+ self.run_btn.clicked.connect(self.run_preview)
self.kill_btn = QPushButton("Kill")
self.kill_btn.clicked.connect(self.kill)
bar = QHBoxLayout()
- for w in (start, self.apply_btn):
+ for w in (start, self.apply_btn, self.run_btn, self.publish_box):
bar.addWidget(w)
bar.addStretch()
bar.addWidget(self.kill_btn)
@@ -131,6 +137,7 @@ class JobsTab(QWidget):
p.deleteLater()
def start(self, *args):
+ self.pending = args
self.remote.call(self._started, "job-start", *args)
def confirm_start(self, question, *args):
@@ -143,7 +150,49 @@ class JobsTab(QWidget):
self.remote.call(lambda _d, err: err and QMessageBox.warning(self, "Kill failed", err),
"job-kill", self.current)
+ def run_preview(self):
+ """Start the previewed slackrepo run for real."""
+ args = [a for a in self.jobs[self.current]["args"] if a != "--preview"]
+ if not self.publish_box.isChecked():
+ args.insert(1, "--no-publish")
+ self.start_signed("slackrepo", *args)
+
+ def ask_passphrase(self, retry):
+ text, ok = QInputDialog.getText(
+ self, "Signing key", ("Wrong passphrase, try again." if retry else
+ "Passphrase of the package signing key:"),
+ QLineEdit.EchoMode.Password)
+ return text if ok else None
+
+ def start_signed(self, *args, retry=False):
+ """Unlock the signing key first when it is locked, then start the job."""
+ def status(data, err):
+ if err:
+ QMessageBox.warning(self, "Signing key", err)
+ elif data["unlocked"] is False:
+ self._unlock(args, retry)
+ else:
+ self.start(*args)
+ self.remote.call(status, "gpg-status")
+
+ def _unlock(self, args, retry):
+ passphrase = self.ask_passphrase(retry)
+ if passphrase is None:
+ return
+
+ def unlocked(_data, err):
+ if err == "bad passphrase":
+ self._unlock(args, True)
+ elif err:
+ QMessageBox.warning(self, "Signing key", err)
+ else:
+ self.start(*args)
+ self.remote.call(unlocked, "gpg-unlock", stdin=passphrase.encode() + b"\n")
+
def _started(self, data, err):
+ if err == "signing key locked":
+ self.start_signed(*self.pending) # locked since the last check: ask again
+ return
if err:
QMessageBox.warning(self, "Start failed", err)
return
@@ -198,3 +247,7 @@ class JobsTab(QWidget):
self.apply_btn.setVisible(
bool(j) and j["status"] == "ok" and "--no" in j["args"]
and bool(SHADOW_PLAN.search(self.log.toPlainText())))
+ preview = bool(j) and j["kind"] == "slackrepo" and j["status"] == "ok" \
+ and "--preview" in j["args"]
+ self.run_btn.setVisible(preview)
+ self.publish_box.setVisible(preview)
diff --git a/sb_builder/mainwindow.py b/sb_builder/mainwindow.py
index 734811b..e84389c 100644
--- a/sb_builder/mainwindow.py
+++ b/sb_builder/mainwindow.py
@@ -26,6 +26,7 @@ from . import VERSION, config
from .dashboard import Dashboard
from .fmt import job_title
from .jobs import JobsTab
+from .packages import PackagesTab
from .remote import DOWN, MISSING
NOTIFY = "org.freedesktop.Notifications"
@@ -66,6 +67,7 @@ class MainWindow(QMainWindow):
self.notifier = notifier or Notifier(self)
self.notifier.clicked.connect(self.open_job)
self.jobs_tab = None
+ self.packages_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 = []
@@ -79,6 +81,8 @@ class MainWindow(QMainWindow):
self.tabs.currentChanged.connect(lambda _i: self.poll_stats())
self.dashboard = Dashboard()
self.dashboard.refresh_btn.clicked.connect(self.poll_stats)
+ self.dashboard.tiles["Packages"].link(self.open_packages)
+ self.dashboard.jobs.link(self.open_jobs)
self.tabs.addTab(self.dashboard, "Dashboard")
self.tabs.tabBar().setTabButton(0, QTabBar.ButtonPosition.RightSide, None)
self.setCentralWidget(self.tabs)
@@ -86,20 +90,24 @@ class MainWindow(QMainWindow):
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)
+ open_menu = self.menuBar().addMenu("&Open")
+ open_menu.addAction("&Packages", self.open_packages)
+ open_menu.addAction("&Jobs", self.open_jobs)
self.conn = QLabel("● connecting")
+ self.key = QLabel()
+ self.key.hide()
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):
+ for w in (self.job_btn, self.key, 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.stats_timer = QTimer(self, timeout=self.poll_slow)
self._start_timers()
QTimer.singleShot(0, self.poll_jobs)
@@ -112,6 +120,18 @@ class MainWindow(QMainWindow):
def poll_jobs(self):
self.remote.call(self._got_jobs, "jobs")
+ def poll_slow(self):
+ self.poll_stats()
+ if self.online:
+ self.remote.call(self._got_key, "gpg-status")
+
+ def _got_key(self, data, err):
+ if err:
+ return self._failed(err)
+ self.key.setVisible(data["unlocked"] is not None)
+ self.key.setText("πŸ”“ key unlocked" if data["unlocked"] else "πŸ”’ key locked")
+ self.key.setToolTip(f"package signing key {data['key']}")
+
def poll_stats(self):
if self.online and self.tabs.currentWidget() is self.dashboard:
self.remote.call(self._got_stats, "stats")
@@ -154,7 +174,7 @@ class MainWindow(QMainWindow):
self.dashboard.set_stale(None)
self.jobs_timer.setInterval(self.cfg["refresh"]["jobs"] * 1000)
self.remote.call(self._got_version, "version")
- self.poll_stats()
+ self.poll_slow()
if self.vm_woken:
self.vm_woken = False
self.notify("vm-up", f"{self.cfg['vm']['domain']} is up")
@@ -229,12 +249,14 @@ class MainWindow(QMainWindow):
event = END_EVENT.get(j["status"])
if event and prev.get(j["id"], "running") != j["status"]:
self.notify(event, job_title(j), j["id"])
+ if j["kind"] in ("slackrepo", "publish") and self.packages_tab:
+ self.packages_tab.reload() # build state changed
# ── tabs ────────────────────────────────────────────────────────────────
def open_jobs(self):
if self.jobs_tab is None:
- self.jobs_tab = JobsTab(self.remote)
+ self.jobs_tab = JobsTab(self.remote, self.cfg["builds"]["publish"])
self.jobs_tab.refresh_requested.connect(self.poll_jobs)
self.jobs_tab.set_overlays(self.overlays)
self.tabs.addTab(self.jobs_tab, "Jobs")
@@ -242,6 +264,15 @@ class MainWindow(QMainWindow):
self.tabs.setCurrentWidget(self.jobs_tab)
return self.jobs_tab
+ def open_packages(self):
+ if self.packages_tab is None:
+ self.packages_tab = PackagesTab(self.remote)
+ self.packages_tab.start_requested.connect(lambda args: self.open_jobs().start(*args))
+ self.tabs.addTab(self.packages_tab, "Packages")
+ self.packages_tab.reload()
+ self.tabs.setCurrentWidget(self.packages_tab)
+ return self.packages_tab
+
def open_job(self, jid):
self.show()
self.raise_()
@@ -257,6 +288,8 @@ class MainWindow(QMainWindow):
if w is self.jobs_tab:
self.jobs_tab = None
w.stop()
+ if w is self.packages_tab:
+ self.packages_tab = None
w.deleteLater()
def reload_config(self):
diff --git a/sb_builder/packages.py b/sb_builder/packages.py
new file mode 100644
index 0000000..34ce7f5
--- /dev/null
+++ b/sb_builder/packages.py
@@ -0,0 +1,332 @@
+# sb-builder - Packages 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.
+
+"""Packages tab: search the inventory, read a package's hints, start slackrepo previews."""
+
+import html
+
+from PyQt6.QtCore import QAbstractTableModel, QModelIndex, QSortFilterProxyModel, Qt, pyqtSignal
+from PyQt6.QtGui import QColor
+from PyQt6.QtWidgets import (QHBoxLayout, QLabel, QLineEdit, QMessageBox, QPushButton,
+ QScrollArea, QSplitter, QTableView, QVBoxLayout, QWidget)
+
+from . import hints as hintlib
+from .dashboard import STATUS
+from .fmt import NA, fmt_time
+
+OVERLAYS = ("personal", "pentesting")
+STATUSES = ("ok", "failed", "never", "outdated")
+COLUMNS = ("", "name", "category", "version", "built", "status", "maintainer")
+STATUS_COLOR = {"ok": STATUS["ok"], "failed": STATUS["failed"], "outdated": STATUS["aborted"]}
+FIELDS = {"cat": "category", "maint": "maintainer", "req": "requires", "status": "status"}
+
+
+def status(row):
+ # ponytail: outdated compares versions only; slackrepo also rebuilds on
+ # SlackBuild or dependency changes, which its preview shows
+ if not row["built_version"] and not row["result"]:
+ return "never"
+ if row["result"] in ("failed", "bad", "aborted"):
+ return "failed"
+ if row["built_version"] != row["version"]:
+ return "outdated"
+ return "ok"
+
+
+def overlay(row):
+ cat = row["path"].split("/", 1)[0]
+ return cat if cat in OVERLAYS else "upstream"
+
+
+def parse_query(text):
+ """[(field or None, term)], lower-cased; unknown prefixes count as plain text."""
+ terms = []
+ for word in text.lower().split():
+ field, sep, term = word.partition(":")
+ if sep and field in FIELDS and term:
+ terms.append((FIELDS[field], term))
+ else:
+ terms.append((None, word))
+ return terms
+
+
+def matches(row, terms, overlays=(), statuses=()):
+ """All terms match, and the row is in a toggled chip of each non-empty group."""
+ if overlays and overlay(row) not in overlays:
+ return False
+ if statuses and row["status"] not in statuses:
+ return False
+ for field, term in terms:
+ if field is None:
+ hay = f"{row['name']} {row['summary'] or ''}"
+ elif field == "requires":
+ hay = " ".join(row["requires"])
+ else:
+ hay = row[field] or ""
+ if term not in hay.lower():
+ return False
+ return True
+
+
+def load_rows(inventory):
+ rows = []
+ for values in inventory["rows"]:
+ row = dict(zip(inventory["columns"], values))
+ row["category"], row["name"] = row["path"].split("/", 1)
+ row["status"] = status(row)
+ rows.append(row)
+ return rows
+
+
+class Model(QAbstractTableModel):
+ def __init__(self):
+ super().__init__()
+ self.rows = []
+ self.checked = set() # package names
+
+ def set_rows(self, rows):
+ self.beginResetModel()
+ self.rows = rows
+ names = {r["name"] for r in rows}
+ self.checked &= names
+ self.endResetModel()
+
+ def rowCount(self, parent=QModelIndex()):
+ return 0 if parent.isValid() else len(self.rows)
+
+ def columnCount(self, parent=QModelIndex()):
+ return len(COLUMNS)
+
+ def headerData(self, section, orientation, role=Qt.ItemDataRole.DisplayRole):
+ if orientation == Qt.Orientation.Horizontal and role == Qt.ItemDataRole.DisplayRole:
+ return COLUMNS[section]
+ return None
+
+ def flags(self, index):
+ f = Qt.ItemFlag.ItemIsEnabled | Qt.ItemFlag.ItemIsSelectable
+ return f | Qt.ItemFlag.ItemIsUserCheckable if index.column() == 0 else f
+
+ def data(self, index, role=Qt.ItemDataRole.DisplayRole):
+ row, col = self.rows[index.row()], index.column()
+ if role == Qt.ItemDataRole.CheckStateRole and col == 0:
+ return Qt.CheckState.Checked if row["name"] in self.checked else Qt.CheckState.Unchecked
+ if role == Qt.ItemDataRole.DisplayRole:
+ return (None, row["name"], row["category"], row["version"],
+ row["built_version"] or "", row["status"], row["maintainer"] or "")[col]
+ if role == Qt.ItemDataRole.ForegroundRole and col == 5 and row["status"] in STATUS_COLOR:
+ return QColor(STATUS_COLOR[row["status"]])
+ if role == Qt.ItemDataRole.UserRole:
+ return row
+ return None
+
+ def setData(self, index, value, role=Qt.ItemDataRole.EditRole):
+ if role != Qt.ItemDataRole.CheckStateRole or index.column() != 0:
+ return False
+ name = self.rows[index.row()]["name"]
+ if Qt.CheckState(value) == Qt.CheckState.Checked:
+ self.checked.add(name)
+ else:
+ self.checked.discard(name)
+ self.dataChanged.emit(index, index)
+ return True
+
+
+class Filter(QSortFilterProxyModel):
+ def __init__(self):
+ super().__init__()
+ self.terms, self.overlays, self.statuses = [], set(), set()
+
+ def update(self, text, overlays, statuses):
+ self.terms, self.overlays, self.statuses = parse_query(text), overlays, statuses
+ self.invalidateFilter()
+
+ def filterAcceptsRow(self, source_row, _parent):
+ row = self.sourceModel().rows[source_row]
+ return matches(row, self.terms, self.overlays, self.statuses)
+
+
+def chip(text):
+ b = QPushButton(text, checkable=True)
+ b.setStyleSheet("QPushButton { border: 1px solid palette(mid); border-radius: 10px;"
+ " padding: 2px 10px; }"
+ "QPushButton:checked { background: palette(highlight);"
+ " color: palette(highlighted-text); }")
+ return b
+
+
+def detail_html(row):
+ """The detail pane for one package: facts, dependencies, then its hints."""
+ view = hintlib.interpret(row["hints"], row["name"], row["info_version"], row["requires"])
+ e = html.escape
+ out = [f"<h3>{e(row['name'])}</h3><p>{e(row['path'])}"
+ f"{' Β· ' + e(row['summary']) if row['summary'] else ''}</p>"]
+ if view["skip"]:
+ out.append(f'<p style="color:{STATUS["aborted"]}"><b>&#9888; Skipped by slackrepo:</b> '
+ f"{e(view['skip'])}</p>")
+ facts = [(".info version", row["info_version"]), ("hint version", row["hint_version"]),
+ ("built version", row["built_version"]),
+ ("built", fmt_time(row["built_time"]) if row["built_time"] else None),
+ ("last result", row["result"]), ("maintainer", row["maintainer"])]
+ out.append(table([(k, v or NA) for k, v in facts]))
+ marks = {"": "", "del": "<s>{}</s> <small>(DELREQUIRES)</small>",
+ "add": "+{} <small>(ADDREQUIRES)</small>", "build": "+{} <small>(build time only)</small>",
+ "pragma": "+{} <small>(pragma)</small>"}
+ deps = [(marks[m] or "{}").format(e(d)) for d, m in view["deps"]]
+ out.append(f"<h4>Dependencies</h4><p>{' &nbsp; '.join(deps) or 'none'}</p>")
+ if view["conflicts"]:
+ out.append(f"<p>Conflicts: {e(' '.join(view['conflicts']))}</p>")
+ for title, rows in view["sections"]:
+ out.append(f"<h4>{e(title)}</h4>" + table(rows))
+ if view["shell"]:
+ out.append(f"<p><small>Expanded by the shell at build time, shown as written: "
+ f"{e(', '.join(view['shell']))}</small></p>")
+ if view["ignored"]:
+ out.append(f'<p style="color:{STATUS["aborted"]}">Not read by slackrepo: '
+ f"{e(', '.join(view['ignored']))}</p>")
+ if row["hint_path"]:
+ out.append(f"<p><small>hint: {e(row['hint_path'])}</small></p>")
+ return "".join(out)
+
+
+def table(rows):
+ cells = "".join(f"<tr><td style='padding-right:12px'><i>{html.escape(str(k))}</i></td>"
+ f"<td>{html.escape(str(v))}</td></tr>" for k, v in rows)
+ return f"<table>{cells}</table>"
+
+
+class PackagesTab(QWidget):
+ start_requested = pyqtSignal(list) # job-start arguments
+
+ def __init__(self, remote, parent=None):
+ super().__init__(parent)
+ self.remote = remote
+ self.model = Model()
+ self.proxy = Filter()
+ self.proxy.setSourceModel(self.model)
+
+ self.query = QLineEdit(placeholderText="search (cat: maint: req: status:)")
+ self.query.textChanged.connect(self.refilter)
+ self.chips = {}
+ chips = QHBoxLayout()
+ for group in (("personal", "pentesting", "upstream"), STATUSES):
+ for name in group:
+ self.chips[name] = chip(name)
+ self.chips[name].toggled.connect(self.refilter)
+ chips.addWidget(self.chips[name])
+ chips.addSpacing(16)
+ chips.addStretch()
+ self.refresh_btn = QPushButton("Refresh")
+ self.refresh_btn.clicked.connect(self.reload)
+ chips.addWidget(self.refresh_btn)
+
+ self.table = QTableView(sortingEnabled=True)
+ self.table.setModel(self.proxy)
+ self.table.setSelectionBehavior(QTableView.SelectionBehavior.SelectRows)
+ self.table.setSelectionMode(QTableView.SelectionMode.SingleSelection)
+ self.table.verticalHeader().hide()
+ self.table.horizontalHeader().setStretchLastSection(True)
+ for col, width in enumerate((28, 220, 110, 120, 120, 80)):
+ self.table.setColumnWidth(col, width)
+ self.table.selectionModel().currentRowChanged.connect(self.show_detail)
+ self.detail = QLabel(textFormat=Qt.TextFormat.RichText, wordWrap=True,
+ alignment=Qt.AlignmentFlag.AlignTop)
+ self.detail.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
+ self.detail.setContentsMargins(12, 8, 12, 8)
+ scroll = QScrollArea(widgetResizable=True)
+ scroll.setWidget(self.detail)
+ split = QSplitter()
+ split.addWidget(self.table)
+ split.addWidget(scroll)
+ split.setSizes([700, 400])
+
+ self.count = QLabel()
+ bar = QHBoxLayout()
+ bar.addWidget(self.count)
+ for label, slot in (("Select visible", self.select_visible), ("Clear", self.clear)):
+ b = QPushButton(label)
+ b.clicked.connect(slot)
+ bar.addWidget(b)
+ bar.addStretch()
+ for action in ("build", "update", "rebuild", "remove"):
+ b = QPushButton(action.capitalize())
+ b.clicked.connect(lambda _c, a=action: self.act(a))
+ bar.addWidget(b)
+ for label, slot in (("Update all", self.update_all), ("Publish", self.publish)):
+ b = QPushButton(label)
+ b.clicked.connect(slot)
+ bar.addWidget(b)
+
+ layout = QVBoxLayout(self)
+ layout.addWidget(self.query)
+ layout.addLayout(chips)
+ layout.addWidget(split)
+ layout.addLayout(bar)
+ self.model.dataChanged.connect(lambda *_: self.update_count())
+ self.update_count()
+
+ def reload(self):
+ self.count.setText("loading…")
+ self.remote.call(self._loaded, "inventory")
+
+ def _loaded(self, data, err):
+ if err:
+ self.count.setText(f"inventory failed: {err}")
+ return
+ self.model.set_rows(load_rows(data))
+ self.update_count()
+
+ def refilter(self):
+ on = {n for n, b in self.chips.items() if b.isChecked()}
+ self.proxy.update(self.query.text(), on - set(STATUSES), on & set(STATUSES))
+ self.update_count()
+
+ def update_count(self):
+ self.count.setText(f"{self.proxy.rowCount():,} shown Β· {len(self.model.checked)} selected")
+
+ def select_visible(self):
+ for i in range(self.proxy.rowCount()):
+ self.model.checked.add(self.proxy.index(i, 0).data(Qt.ItemDataRole.UserRole)["name"])
+ self.model.layoutChanged.emit()
+ self.update_count()
+
+ def clear(self):
+ self.model.checked.clear()
+ self.model.layoutChanged.emit()
+ self.update_count()
+
+ def targets(self):
+ """Checked packages, else the current row; sorted names."""
+ if self.model.checked:
+ return sorted(self.model.checked)
+ current = self.table.currentIndex()
+ return [current.data(Qt.ItemDataRole.UserRole)["name"]] if current.isValid() else []
+
+ def act(self, action):
+ names = self.targets()
+ if names:
+ self.start_requested.emit(["slackrepo", action, "--preview", *names])
+
+ def update_all(self):
+ self.start_requested.emit(["slackrepo", "update", "--preview"])
+
+ def publish(self):
+ if QMessageBox.question(self, "Publish", "Push the package repository to the mirror now?") \
+ == QMessageBox.StandardButton.Yes:
+ self.start_requested.emit(["publish"])
+
+ def show_detail(self, current, _previous):
+ row = current.data(Qt.ItemDataRole.UserRole) if current.isValid() else None
+ self.detail.setText(detail_html(row) if row else "")
diff --git a/sb_builder/remote.py b/sb_builder/remote.py
index 5340f6c..f87ef66 100644
--- a/sb_builder/remote.py
+++ b/sb_builder/remote.py
@@ -55,9 +55,10 @@ class Remote(QObject):
"-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 call(self, cb, *args, stdin=None):
+ """Run sbb-remote <args>, then cb(data, error). stdin: bytes for the helper."""
+ self._run("ssh", self.ssh_args(*args), lambda rc, out, err: cb(*parse(rc, out, err)),
+ stdin)
def virsh(self, cb, *args):
"""Run local virsh against the configured libvirt URI, then cb(rc, text)."""
@@ -71,7 +72,7 @@ class Remote(QObject):
p.start("ssh", self.ssh_args("job-follow", jid, str(offset)))
return p
- def _run(self, prog, argv, cb):
+ def _run(self, prog, argv, cb, stdin=None):
p = QProcess(self)
def done(rc, _status=None):
@@ -82,3 +83,6 @@ class Remote(QObject):
p.errorOccurred.connect(
lambda e: e == QProcess.ProcessError.FailedToStart and done(-1))
p.start(prog, argv)
+ if stdin is not None:
+ p.write(stdin)
+ p.closeWriteChannel()
diff --git a/tests/test_gui.py b/tests/test_gui.py
index b7110f0..482d8b4 100644
--- a/tests/test_gui.py
+++ b/tests/test_gui.py
@@ -33,6 +33,8 @@ from PyQt6.QtCore import QCoreApplication, QEvent, QObject, QProcess, pyqtSignal
from PyQt6.QtWidgets import QApplication # noqa: E402
from sb_builder import VERSION, config # noqa: E402
+from sb_builder import hints as hintlib # noqa: E402
+from sb_builder import packages as pk # noqa: E402
from sb_builder.mainwindow import MainWindow # noqa: E402
from sb_builder.remote import DOWN, MISSING, parse # noqa: E402
@@ -92,14 +94,20 @@ class FakeRemote:
self.calls = []
self.virsh_answer = (1, "")
- def call(self, cb, *args):
- self.calls.append(args)
- cb(*self.answers.get(args[0], (None, DOWN)))
+ def call(self, cb, *args, stdin=None):
+ self.calls.append(args if stdin is None else (*args, stdin))
+ answer = self.answers.get(args[0], (None, DOWN))
+ cb(*(answer(args, stdin) if callable(answer) else answer))
def virsh(self, cb, *args):
self.calls.append(("virsh", *args))
cb(*self.virsh_answer)
+ def follow(self, jid, offset):
+ p = QProcess(app) # parented, like Remote's, so it outlives this call
+ p.start("true", [])
+ return p
+
class FakeNotifier(QObject):
clicked = pyqtSignal(str)
@@ -120,12 +128,136 @@ JOB = {"id": "20260101-000000-sbshadow", "kind": "sbshadow", "args": ["--no"], "
"status": "running", "created": 0, "start": 0, "pid": 5}
+def inv_row(path, version="1.0", built=None, result=None, hints=None, requires=(),
+ summary=None, maintainer="Some One", info_version=None):
+ return [path, version, info_version or version, None, built, 1788978222 if built else None,
+ result, maintainer, list(requires), summary, "/h/x.hint" if hints else None, hints]
+
+
+COLS = ["path", "version", "info_version", "hint_version", "built_version", "built_time",
+ "result", "maintainer", "requires", "summary", "hint_path", "hints"]
+INVENTORY = {"columns": COLS, "rows": [
+ inv_row("misc/nuclei", "3.11.1", built="3.11.0", result="ok", summary="Fast scanner",
+ requires=["go"]),
+ inv_row("personal/foo", built="1.0", result="ok", hints={"DELREQUIRES": "x"}, requires=["x", "y"]),
+ inv_row("pentesting/nmap", built="7.9", result="bad"),
+ inv_row("network/curl"),
+]}
+PREVIEW = {"id": "20260101-000000-slackrepo", "kind": "slackrepo", "status": "ok",
+ "args": ["update", "--preview", "nuclei"], "argv": [], "created": 0, "start": 0, "end": 1}
+
+
+class HintsTest(unittest.TestCase):
+ def view(self, hints, requires=(), info_version="1.0"):
+ return hintlib.interpret(hints, "pkg", info_version, requires)
+
+ def section(self, view, title):
+ return dict(view["sections"]).get(title)
+
+ def test_no_hint(self):
+ v = self.view(None, requires=["a"])
+ self.assertEqual((v["skip"], v["sections"], v["deps"], v["ignored"]),
+ (None, [], [("a", "")], []))
+
+ def test_version_arch_and_forced_checksum_skip(self):
+ v = self.view({"VERSION": "2.0", "ARCH": "x86_64", "DOWNLOAD": "http://a/src.tgz",
+ "DOWNLOAD_x86_64": "http://a/bin-x64.tgz http://a/extra.tgz"})
+ self.assertEqual(self.section(v, "Version override"), [
+ ("version", "1.0 \u2192 2.0"), ("arch", "x86_64"), ("download", "http://a/bin-x64.tgz"),
+ ("download", "http://a/extra.tgz"), ("md5sum", "not checked")])
+
+ def test_arch_variant_empty_falls_back(self):
+ v = self.view({"ARCH": "x86_64", "DOWNLOAD": "http://a/s", "MD5SUM": "abc", "MD5SUM_x86_64": ""})
+ self.assertEqual(self.section(v, "Version override")[1:],
+ [("download", "http://a/s"), ("md5sum", "abc")])
+
+ def test_no_download_forms(self):
+ for hints in ({"DOWNLOAD": "no"}, {"NODOWNLOAD": "yes"}):
+ with self.subTest(hints=hints):
+ self.assertEqual(self.section(self.view(hints), "Version override"),
+ [("download", "no download, the source must already be in the cache")])
+ self.assertEqual(self.view(hints)["ignored"], [])
+
+ def test_dependencies(self):
+ v = self.view({"DELREQUIRES": "b %README%", "ADDREQUIRES": "z", "BUILDTIME": "cmake",
+ "PRAGMA": "python3 need_net", "CONFLICTS": "old"},
+ requires=["a", "b", "%README%"])
+ self.assertEqual(v["deps"], [("a", ""), ("b", "del"), ("%README%", "del"),
+ ("python3", "pragma"), ("z", "add"), ("cmake", "build")])
+ self.assertEqual(v["conflicts"], ["old"])
+ self.assertEqual(self.section(v, "Pragmas")[1], ("need_net", "network not blocked during the build"))
+
+ def test_skip_forms(self):
+ self.assertIsNone(self.view({"SKIP": "no"})["skip"])
+ self.assertEqual(self.view({"SKIP": "yes"})["skip"], "always skipped")
+ self.assertEqual(self.view({"SKIP": "remove nc first"})["skip"], "remove nc first")
+
+ def test_build_settings(self):
+ v = self.view({"OPTIONS": "A=yes B=no", "NUMJOBS": "-j1", "INSTALL": "N",
+ "ANSWER": "ok\\nyes", "NOWARNINGS": "Nonstandard"})
+ self.assertEqual(self.section(v, "Build settings"), [
+ ("option", "A=yes"), ("option", "B=no"), ("make jobs", "-j1"),
+ ("install after build", "no"), ("answers to the build", "ok \u23ce yes"),
+ ("hide warnings matching", "Nonstandard")])
+ self.assertEqual(dict(self.section(self.view({"INSTALL": "1"}), "Build settings")),
+ {"install after build": "yes"})
+
+ def test_users_and_groups(self):
+ v = self.view({"GROUPADD": "209:postgres 300", "USERADD": "209:postgres:-d/var/lib/pgsql:-r"})
+ self.assertEqual(self.section(v, "Host changes"), [
+ ("group", "postgres (gid 209)"), ("group", "pkg (gid 300)"),
+ ("user", "postgres (uid 209), group postgres, home /var/lib/pgsql, shell /bin/false, -r")])
+
+ def test_pragma_forms(self):
+ self.assertEqual(hintlib.pragma_meaning("x86arch=i686"), "on 32 bit x86, build with ARCH=i686")
+ self.assertEqual(hintlib.pragma_meaning("unset_DISTCC_HOSTS"), "unset DISTCC_HOSTS for the build")
+ self.assertEqual(hintlib.pragma_meaning("bogus"), "unknown pragma")
+
+ def test_ignored_and_shell(self):
+ v = self.view({"ANSWERS": "x", "FOO": "1", "DOWNLOAD_arm": "u",
+ "OPTIONS": {"value": '"X=$(uname -m)"', "shell": True}})
+ self.assertEqual(v["ignored"], ["ANSWERS", "FOO"])
+ self.assertEqual(v["shell"], ["OPTIONS"])
+
+ def test_cleanup(self):
+ self.assertEqual(self.section(self.view({"CLEANUP": "depmod -a"}), "On uninstall"),
+ [("run", "depmod -a")])
+
+
+class FilterTest(unittest.TestCase):
+ def setUp(self):
+ self.rows = {r["name"]: r for r in pk.load_rows(INVENTORY)}
+
+ def test_status(self):
+ self.assertEqual({n: r["status"] for n, r in self.rows.items()},
+ {"nuclei": "outdated", "foo": "ok", "nmap": "failed", "curl": "never"})
+
+ def test_query(self):
+ self.assertEqual(pk.parse_query("Nuc cat:misc bogus:x status:"),
+ [(None, "nuc"), ("category", "misc"), (None, "bogus:x"), (None, "status:")])
+
+ def names(self, text="", overlays=(), statuses=()):
+ terms = pk.parse_query(text)
+ return sorted(n for n, r in self.rows.items() if pk.matches(r, terms, overlays, statuses))
+
+ def test_matching(self):
+ self.assertEqual(self.names("scanner"), ["nuclei"])
+ self.assertEqual(self.names("req:go"), ["nuclei"])
+ self.assertEqual(self.names("maint:some status:o"), ["foo", "nuclei"])
+ self.assertEqual(self.names(overlays={"personal", "pentesting"}), ["foo", "nmap"])
+ self.assertEqual(self.names(overlays={"upstream"}, statuses={"never", "failed"}), ["curl"])
+ self.assertEqual(self.names("cat:misc", statuses={"ok"}), [])
+
+
class SmokeTest(unittest.TestCase):
def window(self, **answers):
remote = FakeRemote({"version": ({"version": VERSION, "overlays": ["personal"]}, None),
- "stats": (STATS, None), "jobs": ({"jobs": []}, None), **answers})
+ "stats": (STATS, None), "jobs": ({"jobs": []}, None),
+ "gpg-status": ({"unlocked": True, "key": "K"}, None),
+ "inventory": (INVENTORY, None), **answers})
w = MainWindow(config.load("/nonexistent")[0], remote, FakeNotifier())
self.addCleanup(w.deleteLater)
+ self.addCleanup(lambda: w.jobs_tab and w.jobs_tab.stop())
return w, remote
def test_tabs_and_dashboard(self):
@@ -192,6 +324,65 @@ class SmokeTest(unittest.TestCase):
app.processEvents()
self.assertIsNone(w.jobs_tab)
+ def test_packages_tab(self):
+ w, remote = self.window(**{"job-start": ({"id": "20260101-000001-slackrepo"}, None)})
+ w.poll_jobs()
+ tab = w.open_packages()
+ self.assertEqual(tab.proxy.rowCount(), 4)
+ tab.chips["failed"].setChecked(True)
+ self.assertEqual(tab.proxy.rowCount(), 1)
+ tab.chips["failed"].setChecked(False)
+ started = []
+ tab.start_requested.connect(started.append)
+ tab.act("update")
+ self.assertEqual(started, []) # nothing checked, no current row
+ tab.table.setCurrentIndex(tab.proxy.index(1, 1))
+ self.assertIn("<s>x</s>", tab.detail.text())
+ tab.act("update")
+ tab.model.checked.update({"nuclei", "curl"})
+ tab.act("build")
+ tab.update_all()
+ self.assertEqual(started, [["slackrepo", "update", "--preview", "foo"],
+ ["slackrepo", "build", "--preview", "curl", "nuclei"],
+ ["slackrepo", "update", "--preview"]])
+
+ def test_run_preview_with_signing(self):
+ w, remote = self.window(jobs=({"jobs": [PREVIEW]}, None),
+ **{"gpg-status": ({"unlocked": False, "key": "K"}, None),
+ "gpg-unlock": lambda args, stdin: (
+ (None, "bad passphrase") if stdin == b"wrong\n" else ({}, None)),
+ "job-start": ({"id": "20260101-000001-slackrepo"}, None)})
+ w.poll_jobs()
+ jobs = w.open_jobs()
+ jobs.select(PREVIEW["id"])
+ self.assertTrue(jobs.run_btn.isVisibleTo(jobs))
+ answers = iter(["wrong", "right"])
+ jobs.ask_passphrase = lambda retry: next(answers)
+ jobs.publish_box.setChecked(False)
+ jobs.run_preview()
+ self.assertIn(("gpg-unlock", b"wrong\n"), remote.calls)
+ self.assertIn(("gpg-unlock", b"right\n"), remote.calls)
+ self.assertEqual(remote.calls[-2], ("job-start", "slackrepo", "update", "--no-publish", "nuclei"))
+
+ def test_signing_cancel_starts_nothing(self):
+ w, remote = self.window(jobs=({"jobs": [PREVIEW]}, None),
+ **{"gpg-status": ({"unlocked": False, "key": "K"}, None)})
+ w.poll_jobs()
+ jobs = w.open_jobs()
+ jobs.select(PREVIEW["id"])
+ jobs.ask_passphrase = lambda retry: None
+ jobs.run_preview()
+ self.assertFalse(any(c[0] == "job-start" for c in remote.calls))
+
+ def test_key_indicator_and_tile_links(self):
+ w, _ = self.window(**{"gpg-status": ({"unlocked": False, "key": "K"}, None)})
+ w.poll_jobs()
+ self.assertEqual(w.key.text(), "πŸ”’ key locked")
+ w.dashboard.tiles["Packages"].clicked.emit()
+ self.assertIs(w.tabs.currentWidget(), w.packages_tab)
+ w.dashboard.jobs.clicked.emit()
+ self.assertIs(w.tabs.currentWidget(), w.jobs_tab)
+
def test_log_carriage_return(self):
w, _ = self.window()
jobs = w.open_jobs()