diff options
Diffstat (limited to 'tests/test_remote.py')
| -rw-r--r-- | tests/test_remote.py | 253 |
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") |