diff options
| -rw-r--r-- | remote/config.example.toml | 1 | ||||
| -rwxr-xr-x | remote/sbb-remote | 285 | ||||
| -rw-r--r-- | sb_builder/__init__.py | 2 | ||||
| -rw-r--r-- | 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, <hintdir>/<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 <pkg> [<version>] [--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"]*?)(?<!\d){re.escape(o)}(?!\d)', + lambda x: x.group(1) + n, part) + out.append(part) + if m: + out.append(m.group(0)) + pos = m.end() + return "".join(out) + + +def _pin(info): + """The version pin of an .info: its VERSION, DOWNLOAD* and MD5SUM* lines.""" + lines = [m.group(0) for k in PIN_KEYS if (m := _assignment(k).search(info))] + if not lines or not lines[0].startswith("VERSION="): + raise Error("no VERSION in the .info") + return "".join(ln.rstrip("\n") + "\n" for ln in lines) + + +def new_hint(info): + """mkhint's hint from an .info: the pin, REQUIRES commented, ARCH. + + No phantom deps: a package that has them already has a hint, the + DELREQUIRES one sbshadow's mkhint -F writes after every sync. + """ + text = _pin(info) + m = _assignment("REQUIRES").search(info) + if m: + text += "#" + m.group(0).rstrip("\n") + "\n" + return text + 'ARCH="x86_64"\n' + + +def _md5_url(url): + req = urllib.request.Request(url, headers={"User-Agent": "sbb-remote"}) + h = hashlib.md5() + with urllib.request.urlopen(req, timeout=DL_TIMEOUT) as r: + for chunk in iter(lambda: r.read(1 << 16), b""): + h.update(chunk) + return h.hexdigest() + + +def apply_version(text, info, new, no_dl, log): + """The hint text at version new (None: keep the version), log(str) per download. + + An unpinned hint (DELREQUIRES only) gets the .info pin first. Downloads the + first URL of each DOWNLOAD pair for its md5; continuation URLs keep theirs, + as in mkhint. no_dl: no download, DOWNLOAD="no" in place of every + DOWNLOAD*/MD5SUM* line, since DOWNLOAD_$ARCH would override it. + """ + if not hint_tokens(text, "VERSION"): + text = _pin(info) + ("" if hint_tokens(text, "ARCH") else 'ARCH="x86_64"\n') + text + if new is not None: + text = swap_version(text, hint_tokens(text, "VERSION")[0], new) + if no_dl: + return hint_drop(text, r"(?:DOWNLOAD|MD5SUM)\w*") + 'DOWNLOAD="no"\n' + if new is None: + return text + for dl, sums in DL_PAIRS: + urls = hint_tokens(text, dl) + if not urls or urls[0] in SKIP_URLS: + continue + log(f"downloading {urls[0]}") + text = hint_set(text, sums, " \\\n ".join([_md5_url(urls[0]), *hint_tokens(text, sums)[1:]])) + return text + + +# ── hint jobs and commands ────────────────────────────────────────────────── + +def _read(path): + try: + with open(path, errors="replace") as f: + return f.read() + except FileNotFoundError: + return None + + +def _write(path, text): + os.makedirs(os.path.dirname(path), exist_ok=True) + tmp = path + ".tmp" + with open(tmp, "w") as f: + f.write(text) + os.replace(tmp, path) + + +def _save(target, text): + """Write a hint, its previous text to .bak; a new hint drops a stale .bak (revert = delete).""" + if os.path.exists(target): + shutil.copy2(target, target + ".bak") + elif os.path.exists(target + ".bak"): + os.remove(target + ".bak") + _write(target, text) + + +def cmd_hint(cfg, kind=None, *args): + """Body of the hint jobs: log lines, then SBB-RESULT {"ok": [...], "failed": [[pkg, why]]}.""" + if kind not in ("hint-update", "hint-new"): + raise Error(f"unknown hint job: {kind}") + ok, failed = [], [] + pkgs = _pkgmap(cfg) + pairs, no_dl = hint_args(cfg, kind, list(args)) + for pkg, ver in pairs: + try: + path, target = _hint_target(cfg, pkg, pkgs) + info = _read(os.path.join(cfg["repo_dir"], path, pkg + ".info")) or "" + text = _read(target) + if text is None: + text = new_hint(info) + _save(target, apply_version(text, info, ver, no_dl, + lambda s, p=pkg: print(f"{p}: {s}", flush=True))) + print(f"{pkg}: ok", flush=True) + ok.append(pkg) + except (Error, OSError) as e: + print(f"{pkg}: failed: {e}", flush=True) + failed.append([pkg, str(e)]) + print("SBB-RESULT " + json.dumps({"ok": ok, "failed": failed}), flush=True) + if failed: + raise Error(f"{len(failed)} failed") + + +def cmd_hint_diff(cfg, pkg=None): + _, target = _hint_target(cfg, pkg) + bak, current = _read(target + ".bak"), _read(target) + if bak is None and current is None: + raise Error("no backup") + return {"bak": bak, "current": current} + + +def cmd_hint_revert(cfg, pkg=None): + _, target = _hint_target(cfg, pkg) + if os.path.exists(target + ".bak"): + os.replace(target + ".bak", target) + elif os.path.exists(target): + os.remove(target) # a new hint: reverting it is deleting it + else: + raise Error("nothing to revert") + return {"ok": True} + + +def cmd_hint_delete(cfg, pkg=None): + _, target = _hint_target(cfg, pkg) + if not os.path.exists(target): + raise Error(f"no hint for {pkg}") + os.replace(target, target + ".bak") + return {"ok": True} + + +def cmd_job_result(cfg, jid=None): + """The SBB-RESULT of a hint job's log.""" + text, _ = _tail(os.path.join(job_dir(cfg, jid), "log"), LOG_TAIL) + for line in reversed(text.splitlines()): + if line.startswith("SBB-RESULT "): + return json.loads(line.split(" ", 1)[1]) + raise Error(f"no result in {jid}") + + # ── signing key ───────────────────────────────────────────────────────────── def _signing_key(cfg): @@ -626,7 +896,12 @@ COMMANDS = { "build-log": cmd_build_log, "gpg-status": cmd_gpg_status, "gpg-unlock": cmd_gpg_unlock, + "hint-diff": cmd_hint_diff, + "hint-revert": cmd_hint_revert, + "hint-delete": cmd_hint_delete, + "job-result": cmd_job_result, "_run": cmd_run, + "_hint": cmd_hint, } diff --git a/sb_builder/__init__.py b/sb_builder/__init__.py index e0f577e..21a93ab 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.4.1" # must match remote/sbb-remote VERSION +VERSION = "0.5.0" # must match remote/sbb-remote VERSION diff --git a/tests/test_remote.py b/tests/test_remote.py index db3d0a1..14622b8 100644 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -18,6 +18,9 @@ import contextlib import fcntl +import functools +import hashlib +import http.server import importlib.machinery import importlib.util import io @@ -25,6 +28,7 @@ import json import os import shutil import sqlite3 +import sys import tempfile import threading import time @@ -179,6 +183,255 @@ class KindsTest(Base): sbb.cmd_job_start(self.cfg, "rm", "-rf", "/") +FLATPAK_INFO = ('PRGNAM="flatpak"\nVERSION="1.16.0"\nHOMEPAGE="https://flatpak.org/"\n' + 'DOWNLOAD="https://example.org/flatpak-1.16.0.tar.xz"\n' + 'MD5SUM="bdf866646fb312a341f84669f625488b"\n' + 'DOWNLOAD_x86_64=""\nMD5SUM_x86_64=""\n' + 'REQUIRES="bubblewrap ostree"\nMAINTAINER="Some One"\nEMAIL="one@example.org"\n') + + +class HintTextTest(unittest.TestCase): + def test_swap_forms(self): + text = ('VERSION="1.2.00"\nDOWNLOAD="https://example.org/build_1200/x-1.2.00.tar.gz"\n' + 'MD5SUM="00001200aa"\nDELREQUIRES="lib1200"\n') + self.assertEqual(sbb.swap_version(text, "1.2.00", "1.2.15"), + 'VERSION="1.2.15"\nDOWNLOAD="https://example.org/build_1215/x-1.2.15.tar.gz"\n' + 'MD5SUM="00001200aa"\nDELREQUIRES="lib1200"\n') + + def test_swap_underscore_as_dash(self): + text = 'VERSION="2026_07_07"\nDOWNLOAD="https://example.org/2026-07-07.tgz"\n' + self.assertEqual(sbb.swap_version(text, "2026_07_07", "2026_08_01"), + 'VERSION="2026_08_01"\nDOWNLOAD="https://example.org/2026-08-01.tgz"\n') + + def test_md5_values_untouched(self): + text = 'VERSION="4215"\nMD5SUM="aa4215bb \\\n cc4215dd"\nMD5SUM_x86_64="4215"\n' + self.assertEqual(sbb.swap_version(text, "4215", "4216"), + 'VERSION="4216"\nMD5SUM="aa4215bb \\\n cc4215dd"\nMD5SUM_x86_64="4215"\n') + + def test_tokens_set_drop(self): + text = 'A="x \\\n y"\nB=plain\nDOWNLOAD_x86_64=""\n' + self.assertEqual(sbb.hint_tokens(text, "A"), ["x", "y"]) + self.assertEqual(sbb.hint_tokens(text, "B"), ["plain"]) + self.assertEqual(sbb.hint_tokens(text, "C"), []) + self.assertEqual(sbb.hint_set(text, "B", "1 2"), 'A="x \\\n y"\nB="1 2"\nDOWNLOAD_x86_64=""\n') + self.assertEqual(sbb.hint_set("Z=1", "B", "2"), 'Z=1\nB="2"\n') + self.assertEqual(sbb.hint_drop(text, r"(?:A|DOWNLOAD\w*)"), "B=plain\n") + + def test_new_hint(self): + self.assertEqual(sbb.new_hint(FLATPAK_INFO), + 'VERSION="1.16.0"\nDOWNLOAD="https://example.org/flatpak-1.16.0.tar.xz"\n' + 'MD5SUM="bdf866646fb312a341f84669f625488b"\n' + 'DOWNLOAD_x86_64=""\nMD5SUM_x86_64=""\n' + '#REQUIRES="bubblewrap ostree"\nARCH="x86_64"\n') + with self.assertRaisesRegex(sbb.Error, "no VERSION"): + sbb.new_hint('PRGNAM="x"\n') + + +class Served: + """Files of a temp dir over http on localhost, for md5 downloads.""" + + def __init__(self, test, files): + root = os.path.join(test.tmp, "www") + os.makedirs(root) + for name, data in files.items(): + with open(os.path.join(root, name), "wb") as f: + f.write(data) + handler = functools.partial(http.server.SimpleHTTPRequestHandler, directory=root) + handler.log_message = lambda *a: None + self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler) + threading.Thread(target=self.server.serve_forever, daemon=True).start() + test.addCleanup(self.server.server_close) + test.addCleanup(self.server.shutdown) + self.url = f"http://127.0.0.1:{self.server.server_port}" + + +def md5(data): + return hashlib.md5(data).hexdigest() + + +class ApplyVersionTest(Base): + def setUp(self): + super().setUp() + self.www = Served(self, {"x-2.0.tgz": b"two", "dep.tgz": b"dep"}) + self.log = [] + + def apply(self, text, new, no_dl=False, info=""): + return sbb.apply_version(text, info, new, no_dl, self.log.append) + + def test_download_refreshes_first_md5_only(self): + u = self.www.url + text = (f'VERSION="1.0"\nDOWNLOAD="{u}/x-1.0.tgz \\\n {u}/dep.tgz"\n' + 'MD5SUM="old1 \\\n keep2"\nDOWNLOAD_x86_64=""\nMD5SUM_x86_64=""\nARCH="x86_64"\n') + self.assertEqual(self.apply(text, "2.0"), + f'VERSION="2.0"\nDOWNLOAD="{u}/x-2.0.tgz \\\n {u}/dep.tgz"\n' + f'MD5SUM="{md5(b"two")} \\\n keep2"\nDOWNLOAD_x86_64=""\nMD5SUM_x86_64=""\n' + 'ARCH="x86_64"\n') + self.assertEqual(self.log, [f"downloading {u}/x-2.0.tgz"]) + + def test_arch_pair_and_unsupported(self): + u = self.www.url + text = (f'VERSION="1.0"\nDOWNLOAD="UNSUPPORTED"\nMD5SUM=""\n' + f'DOWNLOAD_x86_64="{u}/x-1.0.tgz"\nMD5SUM_x86_64="old"\n') + self.assertIn(f'MD5SUM_x86_64="{md5(b"two")}"', self.apply(text, "2.0")) + self.assertIn('DOWNLOAD="UNSUPPORTED"\nMD5SUM=""\n', self.apply(text, "2.0")) + + def test_unpinned_hint_pins_from_info(self): + u = self.www.url + info = (f'PRGNAM="x"\nVERSION="1.0"\nDOWNLOAD="{u}/x-1.0.tgz"\nMD5SUM="old"\n' + 'DOWNLOAD_x86_64=""\nMD5SUM_x86_64=""\nREQUIRES="a"\n') + self.assertEqual(self.apply('DELREQUIRES="rust-opt"\n', "2.0", info=info), + f'VERSION="2.0"\nDOWNLOAD="{u}/x-2.0.tgz"\nMD5SUM="{md5(b"two")}"\n' + 'DOWNLOAD_x86_64=""\nMD5SUM_x86_64=""\nARCH="x86_64"\nDELREQUIRES="rust-opt"\n') + self.assertEqual(self.apply('DELREQUIRES="rust-opt"\n', "2.0", no_dl=True, info=info), + 'VERSION="2.0"\nARCH="x86_64"\nDELREQUIRES="rust-opt"\nDOWNLOAD="no"\n') + + def test_no_dl(self): + text = ('VERSION="1.0"\nDOWNLOAD="https://example.org/x-1.0.tgz \\\n https://example.org/d"\n' + 'MD5SUM="a \\\n b"\nDOWNLOAD_x86_64="https://example.org/y"\nMD5SUM_x86_64="c"\n' + 'ARCH="x86_64"\n') + self.assertEqual(self.apply(text, "2.0", no_dl=True), + 'VERSION="2.0"\nARCH="x86_64"\nDOWNLOAD="no"\n') + self.assertEqual(self.log, []) + + def test_no_version_no_download(self): + self.assertEqual(self.apply('VERSION="1.0"\nDOWNLOAD="x"\nMD5SUM="m"\n', None), + 'VERSION="1.0"\nDOWNLOAD="x"\nMD5SUM="m"\n') + + def test_download_failure_raises(self): + with self.assertRaises(OSError): + self.apply(f'VERSION="1.0"\nDOWNLOAD="{self.www.url}/x-1.0.tgz"\nMD5SUM="m"\n', "3.0") + + +class HintJobsTest(Base): + def setUp(self): + super().setUp() + self.www = Served(self, {"x-2.0.tgz": b"two"}) + self.info = (f'PRGNAM="x"\nVERSION="1.0"\nDOWNLOAD="{self.www.url}/x-1.0.tgz"\nMD5SUM="old"\n' + 'DOWNLOAD_x86_64=""\nMD5SUM_x86_64=""\nREQUIRES="rust-opt a"\n') + self.hint = os.path.join(self.cfg["hintdir"], "x.hint") + + def read(self, path): + with open(path) as f: + return f.read() + + def run_hint(self, *args): + out = io.StringIO() + with contextlib.redirect_stdout(out): + try: + sbb.cmd_hint(self.cfg, *args) + rc = 0 + except sbb.Error: + rc = 1 + lines = out.getvalue().splitlines() + self.assertTrue(lines[-1].startswith("SBB-RESULT ")) + return rc, json.loads(lines[-1].split(" ", 1)[1]), lines + + def test_kinds_argv(self): + self.make_pkg("misc/x", self.info) + me = [sys.executable, os.path.abspath(sbb.__file__), "_hint"] + self.assertEqual(sbb.KINDS["hint-update"](self.cfg, ["x=2.0", "--no-dl"]), + (me + ["hint-update", "x=2.0", "--no-dl"], {})) + self.assertEqual(sbb.KINDS["hint-new"](self.cfg, ["x", "2.0"]), (me + ["hint-new", "x", "2.0"], {})) + self.assertFalse(sbb.signs("hint-update", ["x=2.0"])) + + def test_kinds_reject(self): + self.make_pkg("misc/x", self.info) + self.make_pkg("misc/neovim", self.info) + self.make_pkg("misc/shadowed", self.info) + os.makedirs(os.path.join(self.cfg["repo_dir"], "misc/shadowed"), exist_ok=True) + with open(os.path.join(self.cfg["repo_dir"], "misc/shadowed/shadowed.hint"), "w") as f: + f.write("") + for kind, args, why in (("hint-update", [], "at least one"), + ("hint-update", ["x"], "pkg=version"), + ("hint-update", ["x=1;rm"], "bad version"), + ("hint-update", ["x=" + "1" * 65], "bad version"), + ("hint-update", ["nope=1"], "no such package"), + ("hint-update", ["x=1", "x=2"], "twice"), + ("hint-update", ["x=1", "--force"], "unknown option"), + ("hint-update", ["neovim=1"], "bundled deps"), + ("hint-update", ["shadowed=1"], "not managed"), + ("hint-new", [], "hint-new <pkg>"), + ("hint-new", ["x", "1", "2"], "hint-new <pkg>"), + ("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") |