diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_remote.py | 159 |
1 files changed, 155 insertions, 4 deletions
diff --git a/tests/test_remote.py b/tests/test_remote.py index 31d0597..c2b9857 100644 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -45,7 +45,9 @@ class Base(unittest.TestCase): with open(conf, "w") as f: f.write(f'state_dir = "{self.tmp}"\n' f'repo_dir = "{self.tmp}/repo"\n' - f'slackrepo_db = "{self.tmp}/slackrepo.sqlite3"\n') + f'slackrepo_db = "{self.tmp}/slackrepo.sqlite3"\n' + f'hintdir = "{self.tmp}/hints"\n' + f'genreprc = "{self.tmp}/genreprc"\n') patcher = mock.patch.dict(os.environ, {"SBB_CONFIG": conf}) patcher.start() self.addCleanup(patcher.stop) @@ -61,6 +63,34 @@ class Base(unittest.TestCase): open(os.path.join(d, "log"), "wb").close() return jid, d + def make_pkg(self, path, info="", hint=None, desc=None): + d = os.path.join(self.cfg["repo_dir"], path) + pkg = os.path.basename(path) + os.makedirs(d, exist_ok=True) + open(os.path.join(d, pkg + ".SlackBuild"), "w").close() + with open(os.path.join(d, pkg + ".info"), "w") as f: + f.write(info) + if desc: + with open(os.path.join(d, "slack-desc"), "w") as f: + f.write(f"{pkg}: {pkg} ({desc})\n{pkg}:\n") + if hint is not None: + os.makedirs(self.cfg["hintdir"], exist_ok=True) + with open(os.path.join(self.cfg["hintdir"], pkg + ".hint"), "w") as f: + f.write(hint) + + def fake_bin(self, name, script): + bindir = os.path.join(self.tmp, "bin") + os.makedirs(bindir, exist_ok=True) + path = os.path.join(bindir, name) + with open(path, "w") as f: + f.write("#!/bin/sh\n" + script) + os.chmod(path, 0o755) + if bindir not in os.environ["PATH"]: + patcher = mock.patch.dict(os.environ, {"PATH": bindir + os.pathsep + os.environ["PATH"]}) + patcher.start() + self.addCleanup(patcher.stop) + return path + def hold_lock(self): f = open(os.path.join(self.tmp, "lock"), "a") fcntl.flock(f, fcntl.LOCK_EX) @@ -109,11 +139,33 @@ class ConfigTest(Base): class KindsTest(Base): def test_sbpull_argv(self): self.assertEqual(sbb.KINDS["sbpull"](self.cfg, ["personal", "--no"]), - ["bash", "-c", '. "$1" && sbpull "$2" "$3"', "_", - self.cfg["sbpull_script"], "personal", "--no"]) + (["bash", "-c", '. "$1" && sbpull "$2" "$3"', "_", + self.cfg["sbpull_script"], "personal", "--no"], {})) def test_sbshadow_argv(self): - self.assertEqual(sbb.KINDS["sbshadow"](self.cfg, ["--yes"]), ["sbshadow", "--yes"]) + self.assertEqual(sbb.KINDS["sbshadow"](self.cfg, ["--yes"]), (["sbshadow", "--yes"], {})) + + def test_slackrepo_argv(self): + self.make_pkg("misc/nuclei") + self.make_pkg("personal/foo") + kind = sbb.KINDS["slackrepo"] + self.assertEqual(kind(self.cfg, ["update", "--preview", "nuclei", "foo", "nuclei"]), + (["slackrepo", "--repo=SBo-danix", "update", "--preview", "nuclei", "foo"], {})) + self.assertEqual(kind(self.cfg, ["build", "--no-publish", "foo"]), + (["slackrepo", "--repo=SBo-danix", "build", "foo"], {"SBB_PUBLISH": "n"})) + self.assertEqual(kind(self.cfg, ["update"]), (["slackrepo", "--repo=SBo-danix", "update"], {})) + + def test_slackrepo_rejects(self): + self.make_pkg("misc/nuclei") + for args in (["install", "nuclei"], ["build"], ["remove"], ["build", "nope"], + ["build", "../etc"], ["build", "--force", "nuclei"]): + with self.subTest(args=args), self.assertRaises(sbb.Error): + sbb.KINDS["slackrepo"](self.cfg, args) + + def test_publish(self): + self.assertEqual(sbb.KINDS["publish"](self.cfg, []), (["sb-publish"], {})) + with self.assertRaises(sbb.Error): + sbb.KINDS["publish"](self.cfg, ["x"]) def test_rejects_bad_args(self): for kind, args in (("sbpull", ["evil", "--no"]), ("sbpull", ["personal"]), @@ -138,6 +190,100 @@ class MainTest(Base): (0, {"version": sbb.VERSION, "overlays": ["personal", "pentesting"]})) +class InventoryTest(Base): + def test_rows(self): + self.make_pkg("misc/nuclei", 'VERSION="3.11.0"\nMAINTAINER="Some One"\n' + 'REQUIRES="go \\\n git"\n', desc="Fast scanner") + self.make_pkg("personal/foo", 'VERSION="1.0"\n', + hint='VERSION="1.1" # bumped\nDELREQUIRES="a b"\nOPTIONS="X=$(uname -m)"\n') + os.makedirs(os.path.join(self.cfg["repo_dir"], "misc", "notapkg")) + db = sqlite3.connect(self.cfg["slackrepo_db"]) + db.executescript(""" + create table buildresults (itemid text primary key, time text, result text); + create table revisions (itemid text, dep text, deplist text, version text, + built text, rev text, os text, hintcksum text, primary key (itemid, dep)); + insert into buildresults values ('misc/nuclei', '5', 'ok'), ('nuclei', '1', 'bad'); + insert into revisions values ('misc/nuclei', '/', '/', '3.11.0', '1788978222', 'r', 'o', '/'); + """) + db.close() + inv = sbb.cmd_inventory(self.cfg) + rows = [dict(zip(inv["columns"], r)) for r in inv["rows"]] + self.assertEqual([r["path"] for r in rows], ["misc/nuclei", "personal/foo"]) + nuclei, foo = rows + self.assertEqual((nuclei["version"], nuclei["built_version"], nuclei["built_time"], + nuclei["result"], nuclei["maintainer"], nuclei["requires"], + nuclei["summary"], nuclei["hint_path"]), + ("3.11.0", "3.11.0", 1788978222, "ok", "Some One", ["go", "git"], + "Fast scanner", None)) + self.assertEqual((foo["version"], foo["info_version"], foo["hint_version"], foo["result"]), + ("1.1", "1.0", "1.1", None)) + self.assertEqual(foo["hints"]["DELREQUIRES"], "a b") + self.assertEqual(foo["hints"]["OPTIONS"], {"value": '"X=$(uname -m)"', "shell": True}) + json.dumps(inv) + + def test_hint_search_order(self): + self.make_pkg("misc/foo") + sub = os.path.join(self.cfg["hintdir"], "misc", "foo") + os.makedirs(sub) + with open(os.path.join(sub, "foo.hint"), "w") as f: + f.write('VERSION="2"\n') + self.assertEqual(sbb.find_hint(self.cfg, "misc/foo"), os.path.join(sub, "foo.hint")) + with open(os.path.join(self.cfg["hintdir"], "foo.hint"), "w") as f: + f.write('VERSION="3"\n') + self.assertEqual(sbb.find_hint(self.cfg, "misc/foo"), + os.path.join(self.cfg["hintdir"], "foo.hint")) + in_repo = os.path.join(self.cfg["repo_dir"], "misc", "foo", "foo.hint") + open(in_repo, "w").close() + self.assertEqual(sbb.find_hint(self.cfg, "misc/foo"), in_repo) + + def test_missing_db(self): + self.make_pkg("misc/foo", 'VERSION="1"\n') + row = sbb.cmd_inventory(self.cfg)["rows"][0] + self.assertEqual(row[4:7], [None, None, None]) + + +class GpgTest(Base): + def setup_key(self, cached): + with open(self.cfg["genreprc"], "w") as f: + f.write('REPOSOWNERGPG="ABCD1234"\nUSE_GPGAGENT=1\n') + self.fake_bin("gpg", 'case "$*" in\n' + ' *--list-secret-keys*) printf "sec:u:255:22:K:1::::::scESC:\\n' + 'grp:::::::::GRIPSIGN:\\nssb:u:255:18:S:1::::::e:\\ngrp:::::::::GRIPENC:\\n" ;;\n' + f' *-bas*) [ -f "{self.tmp}/good" ] ;;\n' + 'esac\n') + flag = "1" if cached else "-" + self.fake_bin("gpg-connect-agent", f'echo "S KEYINFO GRIPSIGN D - - {flag} P - - -"; echo OK\n') + + def test_status(self): + self.assertEqual(sbb.cmd_gpg_status(self.cfg), {"unlocked": None, "key": None}) + self.setup_key(cached=False) + self.assertEqual(sbb.cmd_gpg_status(self.cfg), {"unlocked": False, "key": "ABCD1234"}) + self.setup_key(cached=True) + self.assertEqual(sbb.cmd_gpg_status(self.cfg)["unlocked"], True) + + def test_locked_refuses_real_run_only(self): + self.setup_key(cached=False) + self.make_pkg("misc/foo") + with self.assertRaisesRegex(sbb.Error, "signing key locked"): + sbb.cmd_job_start(self.cfg, "slackrepo", "build", "foo") + self.fake_bin("slackrepo", "exit 0\n") + self.assertIn("id", sbb.cmd_job_start(self.cfg, "slackrepo", "build", "--preview", "foo")) + + def test_unlock(self): + self.setup_key(cached=False) + log = os.path.join(self.tmp, "preset.log") + preset = self.fake_bin("preset", f'echo "$* $(cat)" >> {log}\n') + with mock.patch.object(sbb, "PRESET", preset): + with self.assertRaisesRegex(sbb.Error, "bad passphrase"): + sbb.cmd_gpg_unlock(self.cfg, stdin=io.BytesIO(b"wrong\n")) + open(os.path.join(self.tmp, "good"), "w").close() + self.assertEqual(sbb.cmd_gpg_unlock(self.cfg, stdin=io.BytesIO(b"right\n")), + {"unlocked": True}) + with open(log) as f: + self.assertEqual(f.read().splitlines(), + ["--preset GRIPSIGN wrong", "--forget GRIPSIGN ", "--preset GRIPSIGN right"]) + + class JobsTest(Base): def log(self, d): with open(os.path.join(d, "log"), "rb") as f: @@ -220,6 +366,11 @@ class JobsTest(Base): patcher.start() self.addCleanup(patcher.stop) + def test_env_reaches_child(self): + jid, d = self.make_job(["sh", "-c", 'echo "publish=$SBB_PUBLISH"'], env={"SBB_PUBLISH": "n"}) + sbb.cmd_run(self.cfg, jid) + self.assertEqual(self.log(d), b"publish=n\n") + def test_start_end_to_end(self): self.fake_sbshadow() rc, data = self.run_main("job-start", "sbshadow", "--no") |