aboutsummaryrefslogtreecommitdiffstats
path: root/tests/test_remote.py
diff options
context:
space:
mode:
authorDanilo M. <danix@danix.xyz>2026-10-20 11:15:34 +0200
committerDanilo M. <danix@danix.xyz>2026-10-20 11:15:34 +0200
commit138370e7e8dee7bbc35b96264c1b11be57d6d45a (patch)
tree99302e99c431138c13799844dedc1ad784086e6d /tests/test_remote.py
parentb1eb7a1847c42dde679f8f0cdac26fc87a7d9571 (diff)
downloadsb-builder-138370e7e8dee7bbc35b96264c1b11be57d6d45a.tar.gz
sb-builder-138370e7e8dee7bbc35b96264c1b11be57d6d45a.zip
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 <noreply@anthropic.com>
Diffstat (limited to 'tests/test_remote.py')
-rw-r--r--tests/test_remote.py253
1 files changed, 253 insertions, 0 deletions
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")