Add sbb-remote, the VM helper

Single-file stdlib helper the GUI runs over ssh. JSON commands for
version, stats and jobs; job-start runs a fixed table of kinds (sbpull,
sbshadow, each requiring --yes or --no) detached with setsid and no
stdin, one at a time under a flock; job-follow streams a log from a
byte offset so the GUI can drop and resume; job-kill sends SIGTERM to
the job's process group. A job whose runner died is reported lost.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 281 insertions, 0 deletions

diff --git a/tests/test_remote.py b/tests/test_remote.py
new file mode 100644
index 0000000..31d0597
--- /dev/null
+++ b/tests/test_remote.py
@@ -0,0 +1,281 @@
+# 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()