From 138370e7e8dee7bbc35b96264c1b11be57d6d45a Mon Sep 17 00:00:00 2001 From: "Danilo M." Date: Tue, 20 Oct 2026 11:15:34 +0200 Subject: sbb-remote 0.5.0: hint jobs replace mkhint's hint editing hint-update and hint-new run the helper's own _hint command: version bumps as mkhint does them (also '-' and dotless URL forms, never inside MD5SUM values), a DELREQUIRES-only hint pinned from the .info first, md5 of the first URL of each DOWNLOAD pair, --no-dl as DOWNLOAD="no", .bak before each write and a final SBB-RESULT line that job-result returns. hint-diff, hint-revert and hint-delete answer at once. neovim and openvino are refused until phase 4b. Job ids accept hyphenated kinds. Co-Authored-By: Claude Opus 5.5 --- remote/config.example.toml | 1 + remote/sbb-remote | 285 ++++++++++++++++++++++++++++++++++++++++++++- sb_builder/__init__.py | 2 +- tests/test_remote.py | 253 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 535 insertions(+), 6 deletions(-) diff --git a/remote/config.example.toml b/remote/config.example.toml index d275401..a77ace9 100644 --- a/remote/config.example.toml +++ b/remote/config.example.toml @@ -12,3 +12,4 @@ hintdir = "/etc/slackrepo/SBo-danix/hintfiles" default_hintdir = "" # searched after hintdir when set genreprc = "/root/.genreprc" # REPOSOWNERGPG names the signing key slackrepo_logdir = "/var/log/slackrepo/SBo-danix" +manifest_packages = ["neovim", "openvino"] # bundled deps: hint-update refuses them (phase 4b) diff --git a/remote/sbb-remote b/remote/sbb-remote index 25427c0..2674ed6 100755 --- a/remote/sbb-remote +++ b/remote/sbb-remote @@ -25,6 +25,7 @@ its arguments, this file validates them and builds the argv itself. import fcntl import glob +import hashlib import json import os import re @@ -37,8 +38,9 @@ import sys import tempfile import time import tomllib +import urllib.request -VERSION = "0.4.1" +VERSION = "0.5.0" DEFAULTS = { "state_dir": "/var/lib/sb-builder", "repo_dir": "/var/lib/sbopkg/SBo-danix", @@ -51,11 +53,13 @@ DEFAULTS = { "default_hintdir": "", "genreprc": "/root/.genreprc", "slackrepo_logdir": "/var/log/slackrepo/SBo-danix", + "manifest_packages": ["neovim", "openvino"], # bundled deps: left to mkhint until phase 4b } FLAGS = ("--yes", "--no") -JOB_ID = re.compile(r"^\d{8}-\d{6}-[a-z]+(-\d+)?$") +JOB_ID = re.compile(r"^\d{8}-\d{6}-[a-z]+(-[a-z]+)*(-\d+)?$") PKG = re.compile(r"^[^/]+/([^/]+)/\1\.SlackBuild$") PKGNAME = re.compile(r"^[A-Za-z0-9._+-]+$") +VERSION_ARG = re.compile(r"^[A-Za-z0-9._+-]{1,64}$") ACTIONS = ("build", "rebuild", "update", "remove") PRESET = "/usr/libexec/gpg-preset-passphrase" ITEM = re.compile(r"^[A-Za-z0-9._+-]+/[A-Za-z0-9._+-]+$") # category/pkg @@ -85,8 +89,9 @@ def load_config(path=None): 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") + for k in ("overlays", "manifest_packages"): + if not all(isinstance(o, str) for o in cfg[k]): + raise Error(f"config: {k}: expected list of strings") return cfg @@ -138,8 +143,67 @@ def kind_publish(cfg, args): return ["sb-publish"], {} +def _pkgmap(cfg): + """{package: category/package} of the repo.""" + return {os.path.basename(d): os.path.relpath(d, cfg["repo_dir"]) + for d in _pkgdirs(cfg["repo_dir"], "*/*")} + + +def _hint_target(cfg, pkg, pkgs=None): + """(category/pkg, /.hint); refuses a hint slackrepo finds elsewhere first.""" + pkgs = pkgs or _pkgmap(cfg) + if not PKGNAME.match(pkg or "") or pkg not in pkgs: + raise Error(f"no such package: {pkg}") + target = os.path.join(cfg["hintdir"], pkg + ".hint") + found = find_hint(cfg, pkgs[pkg]) + if found and found != target: + raise Error(f"hint lives at {found}, not managed") + return pkgs[pkg], target + + +def hint_args(cfg, kind, args): + """Validated ([(pkg, version or None)], no_dl) of a hint-update or hint-new job.""" + no_dl = "--no-dl" in args + for o in (a for a in args if a.startswith("--") and a != "--no-dl"): + raise Error(f"{kind}: unknown option {o}") + rest = [a for a in args if a != "--no-dl"] + if kind == "hint-new": + if not 1 <= len(rest) <= 2: + raise Error("usage: hint-new [] [--no-dl]") + pairs = [(rest[0], rest[1] if len(rest) == 2 else None)] + else: + if not rest: + raise Error("hint-update needs at least one pkg=version") + if not all("=" in a for a in rest): + raise Error("hint-update: expected pkg=version") + pairs = [tuple(a.split("=", 1)) for a in rest] + pkgs = _pkgmap(cfg) + seen = set() + for pkg, ver in pairs: + if ver is not None and not VERSION_ARG.match(ver): + raise Error(f"bad version: {ver}") + _, target = _hint_target(cfg, pkg, pkgs) + if pkg in seen: + raise Error(f"{pkg} named twice") + seen.add(pkg) + if ver is not None and pkg in cfg["manifest_packages"]: + raise Error(f"{pkg}: bundled deps: use mkhint until phase 4b") + if kind == "hint-new" and os.path.exists(target): + raise Error(f"{pkg}: hint exists") + return pairs, no_dl + + +def _self_job(kind): + """A job kind run by this file itself (_hint), argv-free like the others.""" + def build(cfg, args): + hint_args(cfg, kind, args) + return [sys.executable, os.path.abspath(__file__), "_hint", kind, *args], {} + return build + + KINDS = {"sbpull": kind_sbpull, "sbshadow": kind_sbshadow, - "slackrepo": kind_slackrepo, "publish": kind_publish} + "slackrepo": kind_slackrepo, "publish": kind_publish, + **{k: _self_job(k) for k in ("hint-update", "hint-new")}} def signs(kind, args): @@ -549,6 +613,212 @@ def cmd_inventory(cfg): return {"columns": COLUMNS, "rows": rows} +# ── hint text ─────────────────────────────────────────────────────────────── +# Hints are edited as text so comments, order and other keys stay as written. + +PIN_KEYS = ("VERSION", "DOWNLOAD", "MD5SUM", "DOWNLOAD_x86_64", "MD5SUM_x86_64") +DL_PAIRS = (("DOWNLOAD", "MD5SUM"), ("DOWNLOAD_x86_64", "MD5SUM_x86_64")) +SKIP_URLS = ("UNSUPPORTED", "UNTESTED") +DL_TIMEOUT = 60 + + +def _assignment(key): + """KEY=value at line start; a quoted value may span lines. key is a regex.""" + return re.compile(rf'^(?:{key})=(?:"([^"]*)"|(\S*))[^\n]*\n?', re.M) + + +def hint_tokens(text, key): + """Whitespace-separated words of KEY's value, [] when unset or empty.""" + m = _assignment(key).search(text) + if not m: + return [] + value = m.group(1) if m.group(1) is not None else m.group(2) + return value.replace("\\\n", " ").split() + + +def hint_set(text, key, value): + """KEY="value" in place of the first KEY line, else appended.""" + line = f'{key}="{value}"\n' + m = _assignment(key).search(text) + if m: + return text[:m.start()] + line + text[m.end():] + return text + ("\n" if text and not text.endswith("\n") else "") + line + + +def hint_drop(text, key): + return _assignment(key).sub("", text) + + +def swap_version(text, old, new): + """mkhint's bump of old to new, everywhere but in MD5SUM* values. + + Also the '-' form when either has '_' (dated URLs), and the dotless form of + a dotted old version inside URLs, as a whole number (build_4200 for 4.2.00). + """ + pairs = [(old, new)] + if "_" in old or "_" in new: + pairs.append((old.replace("_", "-"), new.replace("_", "-"))) + o, n = old.replace(".", ""), new.replace(".", "") + dotless = "." in old and o != n + out, pos = [], 0 + for m in [*_assignment(r"MD5SUM\w*").finditer(text), None]: + part = text[pos:m.start() if m else len(text)] + for a, b in pairs: + part = part.replace(a, b) + if dotless: + part = re.sub(rf'(://[^\s"]*?)(?"), + ("hint-new", ["x", "1", "2"], "hint-new "), + ("hint-new", ["neovim", "1"], "bundled deps")): + with self.subTest(kind=kind, args=args), self.assertRaisesRegex(sbb.Error, why): + sbb.KINDS[kind](self.cfg, args) + sbb.KINDS["hint-new"](self.cfg, ["neovim"]) # no version: no bundled deps to touch + self.make_pkg("misc/y", self.info, hint="") + with self.assertRaisesRegex(sbb.Error, "exists"): + sbb.KINDS["hint-new"](self.cfg, ["y"]) + + def test_update_backs_up_and_reports(self): + self.make_pkg("misc/x", self.info, hint='DELREQUIRES="rust-opt"\n') + self.make_pkg("misc/y", self.info.replace("x-1.0", "y-1.0"), hint='VERSION="1.0"\n' + f'DOWNLOAD="{self.www.url}/y-1.0.tgz"\nMD5SUM="old"\n') + rc, result, lines = self.run_hint("hint-update", "y=2.0", "x=2.0") + self.assertEqual(rc, 1) + self.assertEqual(result["ok"], ["x"]) + self.assertEqual([p for p, _why in result["failed"]], ["y"]) + self.assertIn("x: downloading", "\n".join(lines)) + self.assertIn(md5(b"two"), self.read(self.hint)) + self.assertEqual(self.read(self.hint + ".bak"), 'DELREQUIRES="rust-opt"\n') + y = os.path.join(self.cfg["hintdir"], "y.hint") + self.assertIn('MD5SUM="old"', self.read(y)) # failed: left as it was + self.assertFalse(os.path.exists(y + ".bak")) + + def test_update_without_hint_creates_one(self): + self.make_pkg("misc/x", self.info) + os.makedirs(self.cfg["hintdir"]) + with open(self.hint + ".bak", "w") as f: + f.write("stale") + rc, result, _ = self.run_hint("hint-update", "x=2.0", "--no-dl") + self.assertEqual((rc, result), (0, {"ok": ["x"], "failed": []})) + self.assertEqual(self.read(self.hint), 'VERSION="2.0"\n#REQUIRES="rust-opt a"\nARCH="x86_64"\n' + 'DOWNLOAD="no"\n') + self.assertFalse(os.path.exists(self.hint + ".bak")) # revert = delete + + def test_new(self): + self.make_pkg("misc/x", self.info) + rc, result, _ = self.run_hint("hint-new", "x") + self.assertEqual((rc, result), (0, {"ok": ["x"], "failed": []})) + self.assertIn('#REQUIRES="rust-opt a"', self.read(self.hint)) + self.assertIn('MD5SUM="old"', self.read(self.hint)) # no version given: no download + + def test_diff_revert_delete(self): + self.make_pkg("misc/x", self.info, hint="A=1\n") + self.assertEqual(sbb.cmd_hint_diff(self.cfg, "x"), {"bak": None, "current": "A=1\n"}) + self.assertEqual(sbb.cmd_hint_delete(self.cfg, "x"), {"ok": True}) + self.assertEqual(sbb.cmd_hint_diff(self.cfg, "x"), {"bak": "A=1\n", "current": None}) + with self.assertRaisesRegex(sbb.Error, "no hint"): + sbb.cmd_hint_delete(self.cfg, "x") + self.assertEqual(sbb.cmd_hint_revert(self.cfg, "x"), {"ok": True}) + self.assertEqual(self.read(self.hint), "A=1\n") + self.assertFalse(os.path.exists(self.hint + ".bak")) + self.assertEqual(sbb.cmd_hint_revert(self.cfg, "x"), {"ok": True}) # no .bak: revert a new hint + self.assertFalse(os.path.exists(self.hint)) + with self.assertRaisesRegex(sbb.Error, "nothing to revert"): + sbb.cmd_hint_revert(self.cfg, "x") + with self.assertRaisesRegex(sbb.Error, "no backup"): + sbb.cmd_hint_diff(self.cfg, "x") + with self.assertRaisesRegex(sbb.Error, "no such package"): + sbb.cmd_hint_diff(self.cfg, "../x") + + def test_job_result(self): + jid, d = self.make_job(["true"]) + with open(os.path.join(d, "log"), "w") as f: + f.write('x: downloading u\nSBB-RESULT {"ok": ["x"], "failed": []}\n{"error": "1 failed"}\n') + self.assertEqual(sbb.cmd_job_result(self.cfg, jid), {"ok": ["x"], "failed": []}) + jid2 = "20260101-000001-test" + os.mkdir(os.path.join(self.tmp, "jobs", jid2)) + open(os.path.join(self.tmp, "jobs", jid2, "log"), "w").close() + with self.assertRaisesRegex(sbb.Error, "no result"): + sbb.cmd_job_result(self.cfg, jid2) + + def test_end_to_end(self): + self.make_pkg("misc/x", self.info, hint='VERSION="1.0"\nDOWNLOAD="u"\nMD5SUM="m"\n') + rc, data = self.run_main("job-start", "hint-update", "x=2.0", "--no-dl") + d = os.path.join(self.tmp, "jobs", data["id"]) + self.assertEqual(self.wait_status(d, ("ok", "failed", "lost"))["status"], "ok") + self.assertEqual(self.run_main("job-result", data["id"]), (0, {"ok": ["x"], "failed": []})) + self.assertEqual(self.read(self.hint), 'VERSION="2.0"\nDOWNLOAD="no"\n') + + class MainTest(Base): def test_unknown_command(self): rc, data = self.run_main("nope") -- cgit v1.2.3