From b8400f38411818aa54a773dffb2c3a0c56b06dba Mon Sep 17 00:00:00 2001 From: "Danilo M." Date: Mon, 19 Oct 2026 20:53:43 +0200 Subject: 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 --- tests/test_gui.py | 199 ++++++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 195 insertions(+), 4 deletions(-) (limited to 'tests') 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("x", 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() -- cgit v1.2.3