path: root/tests/test_remote.py

# sb-builder - self-checks for the remote helper
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

"""Run: python3 -m unittest discover -s tests"""

import contextlib
import fcntl
import importlib.machinery
import importlib.util
import io
import json
import os
import shutil
import sqlite3
import tempfile
import threading
import time
import unittest
from unittest import mock

SCRIPT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "remote", "sbb-remote")
_loader = importlib.machinery.SourceFileLoader("sbb_remote", SCRIPT)
sbb = importlib.util.module_from_spec(importlib.util.spec_from_loader("sbb_remote", _loader))
_loader.exec_module(sbb)


class Base(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, self.tmp)
        conf = os.path.join(self.tmp, "config.toml")
        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')
        patcher = mock.patch.dict(os.environ, {"SBB_CONFIG": conf})
        patcher.start()
        self.addCleanup(patcher.stop)
        self.cfg = sbb.load_config()
        os.makedirs(os.path.join(self.tmp, "jobs"))

    def make_job(self, argv, **meta):
        jid = "20260101-000000-test"
        d = os.path.join(self.tmp, "jobs", jid)
        os.mkdir(d)
        sbb.write_meta(d, {"id": jid, "kind": "test", "args": [], "argv": argv,
                           "status": "starting", "created": time.time(), **meta})
        open(os.path.join(d, "log"), "wb").close()
        return jid, d

    def hold_lock(self):
        f = open(os.path.join(self.tmp, "lock"), "a")
        fcntl.flock(f, fcntl.LOCK_EX)
        self.addCleanup(f.close)

    def run_main(self, *argv):
        out = io.StringIO()
        with contextlib.redirect_stdout(out):
            rc = sbb.main(list(argv))
        return rc, json.loads(out.getvalue())

    def wait_status(self, d, statuses):
        for _ in range(100):
            if sbb.read_meta(d)["status"] in statuses:
                break
            time.sleep(0.05)
        return sbb.read_meta(d)


class ConfigTest(Base):
    def write(self, text):
        p = os.path.join(self.tmp, "c.toml")
        with open(p, "w") as f:
            f.write(text)
        return p

    def test_missing_file_gives_defaults(self):
        self.assertEqual(sbb.load_config(os.path.join(self.tmp, "nope.toml")), sbb.DEFAULTS)

    def test_override(self):
        self.assertEqual(sbb.load_config(self.write('overlays = ["a"]\n'))["overlays"], ["a"])

    def test_example_matches_defaults(self):
        example = os.path.join(os.path.dirname(SCRIPT), "config.example.toml")
        self.assertEqual(sbb.load_config(example), sbb.DEFAULTS)

    def test_unknown_key(self):
        with self.assertRaisesRegex(sbb.Error, "bogus: unknown key"):
            sbb.load_config(self.write("bogus = 1\n"))

    def test_wrong_type(self):
        with self.assertRaisesRegex(sbb.Error, "state_dir: expected str"):
            sbb.load_config(self.write("state_dir = 1\n"))


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"])

    def test_sbshadow_argv(self):
        self.assertEqual(sbb.KINDS["sbshadow"](self.cfg, ["--yes"]), ["sbshadow", "--yes"])

    def test_rejects_bad_args(self):
        for kind, args in (("sbpull", ["evil", "--no"]), ("sbpull", ["personal"]),
                           ("sbpull", ["personal", "--force"]), ("sbshadow", []),
                           ("sbshadow", ["--yes", "--no"])):
            with self.subTest(kind=kind, args=args), self.assertRaises(sbb.Error):
                sbb.KINDS[kind](self.cfg, args)

    def test_unknown_kind(self):
        with self.assertRaisesRegex(sbb.Error, "unknown kind"):
            sbb.cmd_job_start(self.cfg, "rm", "-rf", "/")


class MainTest(Base):
    def test_unknown_command(self):
        rc, data = self.run_main("nope")
        self.assertEqual(rc, 1)
        self.assertIn("usage", data["error"])

    def test_version(self):
        self.assertEqual(self.run_main("version"),
                         (0, {"version": sbb.VERSION, "overlays": ["personal", "pentesting"]}))


class JobsTest(Base):
    def log(self, d):
        with open(os.path.join(d, "log"), "rb") as f:
            return f.read()

    def test_run_ok(self):
        jid, d = self.make_job(["sh", "-c", "echo hello"])
        sbb.cmd_run(self.cfg, jid)
        m = sbb.read_meta(d)
        self.assertEqual((m["status"], m["exit"]), ("ok", 0))
        self.assertEqual(self.log(d), b"hello\n")

    def test_run_failed(self):
        jid, d = self.make_job(["sh", "-c", "exit 3"])
        sbb.cmd_run(self.cfg, jid)
        m = sbb.read_meta(d)
        self.assertEqual((m["status"], m["exit"]), ("failed", 3))

    def test_run_missing_program(self):
        jid, d = self.make_job(["/nonexistent/prog"])
        sbb.cmd_run(self.cfg, jid)
        self.assertEqual(sbb.read_meta(d)["status"], "failed")
        self.assertIn(b"sbb-remote:", self.log(d))

    def test_kill(self):
        jid, d = self.make_job(["sleep", "30"])
        t = threading.Thread(target=sbb.cmd_run, args=(self.cfg, jid))
        t.start()
        self.wait_status(d, ("running",))
        self.assertEqual(sbb.cmd_job_kill(self.cfg, jid), {"ok": True})
        t.join(5)
        self.assertEqual(sbb.read_meta(d)["status"], "killed")

    def test_kill_not_running(self):
        jid, _ = self.make_job(["true"], status="ok")
        with self.assertRaisesRegex(sbb.Error, "not running"):
            sbb.cmd_job_kill(self.cfg, jid)

    def test_busy(self):
        self.hold_lock()
        jid, d = self.make_job(["true"])
        sbb.cmd_run(self.cfg, jid)
        self.assertEqual(sbb.read_meta(d)["status"], "busy")

    def test_lost_when_lock_free(self):
        self.make_job(["true"], status="running", pid=1)
        self.assertEqual(sbb.cmd_jobs(self.cfg)["jobs"][0]["status"], "lost")

    def test_running_kept_while_lock_held(self):
        self.hold_lock()
        self.make_job(["true"], status="running", pid=1)
        self.assertEqual(sbb.cmd_jobs(self.cfg)["jobs"][0]["status"], "running")

    def test_starting_grace(self):
        _, d = self.make_job(["true"])
        self.assertEqual(sbb.cmd_jobs(self.cfg)["jobs"][0]["status"], "starting")
        sbb.write_meta(d, dict(sbb.read_meta(d), created=time.time() - 60))
        self.assertEqual(sbb.cmd_jobs(self.cfg)["jobs"][0]["status"], "lost")

    def test_follow_from_offset(self):
        jid, d = self.make_job(["true"], status="ok")
        with open(os.path.join(d, "log"), "wb") as f:
            f.write(b"hello world")
        out = io.BytesIO()
        sbb.cmd_job_follow(self.cfg, jid, "6", out=out)
        self.assertEqual(out.getvalue(), b"world")

    def test_bad_job_id(self):
        with self.assertRaisesRegex(sbb.Error, "bad job id"):
            sbb.cmd_job_follow(self.cfg, "../../etc", "0")

    def fake_sbshadow(self):
        bindir = os.path.join(self.tmp, "bin")
        os.mkdir(bindir)
        fake = os.path.join(bindir, "sbshadow")
        with open(fake, "w") as f:
            f.write('#!/bin/sh\necho "shadow:   cat/$1"\n')
        os.chmod(fake, 0o755)
        patcher = mock.patch.dict(os.environ, {"PATH": bindir + os.pathsep + os.environ["PATH"]})
        patcher.start()
        self.addCleanup(patcher.stop)

    def test_start_end_to_end(self):
        self.fake_sbshadow()
        rc, data = self.run_main("job-start", "sbshadow", "--no")
        self.assertEqual(rc, 0)
        d = os.path.join(self.tmp, "jobs", data["id"])
        self.assertEqual(self.wait_status(d, ("ok", "failed", "lost"))["status"], "ok")
        self.assertEqual(self.log(d), b"shadow:   cat/--no\n")

    def test_start_busy(self):
        self.fake_sbshadow()
        self.hold_lock()
        jid, _ = self.make_job(["true"], status="running", pid=1)
        self.assertEqual(self.run_main("job-start", "sbshadow", "--no"), (0, {"busy": jid}))
        self.assertEqual(os.listdir(os.path.join(self.tmp, "jobs")), [jid])


class StatsTest(Base):
    def test_missing_sources_are_null(self):
        s = sbb.cmd_stats(self.cfg)
        self.assertIsNotNone(s["load"])
        self.assertIsNone(s["packages"])
        self.assertIsNone(s["slackrepo"])
        self.assertEqual(s["sync"], {"personal": None, "pentesting": None})
        json.dumps(s)

    def test_counts(self):
        r = self.cfg["repo_dir"]
        for p in ("cat/foo", "cat/bar", "personal/foo", "personal/baz", "pentesting/nmap"):
            os.makedirs(os.path.join(r, p))
            open(os.path.join(r, p, os.path.basename(p) + ".SlackBuild"), "w").close()
        os.makedirs(os.path.join(r, "cat", "notapkg"))

        def fake_git(cfg, *args):
            if args[0] == "ls-tree":
                return "cat/foo/foo.SlackBuild\ncat/foo/README\ncat/old/old.SlackBuild\n"
            return "1792351223\n"

        with mock.patch.object(sbb, "_git", fake_git):
            s = sbb.cmd_stats(self.cfg)
        self.assertEqual(s["packages"], {"total": 5, "personal": 2, "pentesting": 1, "shadowed": 1})
        self.assertEqual(s["sync"]["personal"], 1792351223)

    def test_slackrepo(self):
        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 ('a', '1', 'ok'), ('b', '1', 'ok'), ('c', '1', 'failed');
            insert into revisions values ('cat/a', '/', '/', '1', '1', 'r', 'o', '/'),
                                         ('cat/a', 'dep', '/', '1', '1', 'r', 'o', '/');
        """)
        db.close()
        self.assertEqual(sbb.cmd_stats(self.cfg)["slackrepo"],
                         {"results": {"ok": 2, "failed": 1}, "built": 1})


if __name__ == "__main__":
    unittest.main()