From fb64c88bb8672ee9bdb8172eec0252966d67404a Mon Sep 17 00:00:00 2001 From: "Danilo M." Date: Mon, 19 Oct 2026 19:11:44 +0200 Subject: 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 --- remote/config.example.toml | 9 ++ remote/sbb-remote | 355 +++++++++++++++++++++++++++++++++++++++++++++ tests/test_remote.py | 281 +++++++++++++++++++++++++++++++++++ 3 files changed, 645 insertions(+) create mode 100644 remote/config.example.toml create mode 100755 remote/sbb-remote create mode 100644 tests/test_remote.py diff --git a/remote/config.example.toml b/remote/config.example.toml new file mode 100644 index 0000000..d7fcd7f --- /dev/null +++ b/remote/config.example.toml @@ -0,0 +1,9 @@ +# sbb-remote config on the VM: /etc/sb-builder/config.toml +# Every key is optional; these are the defaults. + +state_dir = "/var/lib/sb-builder" +repo_dir = "/var/lib/sbopkg/SBo-danix" +upstream_branch = "current" +slackrepo_db = "/var/lib/slackrepo/SBo-danix/database_SBo-danix_x86_64.sqlite3" +sbpull_script = "/usr/local/share/slackrepo_setup/sbpull.bash" +overlays = ["personal", "pentesting"] diff --git a/remote/sbb-remote b/remote/sbb-remote new file mode 100755 index 0000000..2d877d6 --- /dev/null +++ b/remote/sbb-remote @@ -0,0 +1,355 @@ +#!/usr/bin/env python3 +# sbb-remote - sb-builder helper on the build VM +# Copyright (C) 2026 Danilo M. +# +# 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. + +"""Helper run over ssh by the sb-builder GUI. + +Every command prints one JSON object on stdout, except job-follow, which +streams raw log bytes. Jobs run detached (setsid, stdin=/dev/null), one at a +time under a flock, from a fixed table of kinds: the GUI sends a kind and +its arguments, this file validates them and builds the argv itself. +""" + +import fcntl +import glob +import json +import os +import re +import shutil +import signal +import sqlite3 +import subprocess +import sys +import time +import tomllib + +VERSION = "0.1.0" +DEFAULTS = { + "state_dir": "/var/lib/sb-builder", + "repo_dir": "/var/lib/sbopkg/SBo-danix", + "upstream_branch": "current", + "slackrepo_db": "/var/lib/slackrepo/SBo-danix/database_SBo-danix_x86_64.sqlite3", + "sbpull_script": "/usr/local/share/slackrepo_setup/sbpull.bash", + "overlays": ["personal", "pentesting"], +} +FLAGS = ("--yes", "--no") +JOB_ID = re.compile(r"^\d{8}-\d{6}-[a-z]+(-\d+)?$") +PKG = re.compile(r"^[^/]+/([^/]+)/\1\.SlackBuild$") +ACTIVE = ("starting", "running") +START_GRACE = 10 # seconds a job may sit in "starting" before it counts as lost + + +class Error(Exception): + pass + + +def load_config(path=None): + path = path or os.environ.get("SBB_CONFIG", "/etc/sb-builder/config.toml") + cfg = dict(DEFAULTS) + try: + with open(path, "rb") as f: + data = tomllib.load(f) + except FileNotFoundError: + return cfg + except tomllib.TOMLDecodeError as e: + raise Error(f"config: {e}") + for k, v in data.items(): + if k not in DEFAULTS: + raise Error(f"config: {k}: unknown key") + if type(v) is not type(DEFAULTS[k]): + raise Error(f"config: {k}: expected {type(DEFAULTS[k]).__name__}") + cfg[k] = v + if not all(isinstance(o, str) for o in cfg["overlays"]): + raise Error("config: overlays: expected list of strings") + return cfg + + +# ── job kinds: (cfg, args) -> argv, raising Error on bad args ─────────────── + +def _flag(args): + if len(args) != 1 or args[0] not in FLAGS: + raise Error("expected exactly one of --yes, --no") + return args[0] + + +def kind_sbpull(cfg, args): + if not args or args[0] not in cfg["overlays"]: + raise Error(f"sbpull: overlay must be one of {cfg['overlays']}") + return ["bash", "-c", '. "$1" && sbpull "$2" "$3"', "_", + cfg["sbpull_script"], args[0], _flag(args[1:])] + + +def kind_sbshadow(cfg, args): + return ["sbshadow", _flag(args)] + + +KINDS = {"sbpull": kind_sbpull, "sbshadow": kind_sbshadow} + + +# ── job state ─────────────────────────────────────────────────────────────── + +def job_dir(cfg, jid): + if not JOB_ID.match(jid or ""): + raise Error(f"bad job id: {jid}") + d = os.path.join(cfg["state_dir"], "jobs", jid) + if not os.path.isdir(d): + raise Error(f"no such job: {jid}") + return d + + +def read_meta(d): + with open(os.path.join(d, "meta.json")) as f: + return json.load(f) + + +def write_meta(d, meta): + tmp = os.path.join(d, "meta.json.tmp") + with open(tmp, "w") as f: + json.dump(meta, f) + os.replace(tmp, os.path.join(d, "meta.json")) + + +def lock_file(cfg): + return open(os.path.join(cfg["state_dir"], "lock"), "a") + + +def refresh(cfg, d, meta): + """Mark an active job lost when no runner holds the lock. + + _run writes the final status before it releases the lock, so a free lock + plus a status that is still active (re-read under the lock) means the + runner died: killed, crashed, or the VM rebooted. + """ + if meta["status"] not in ACTIVE: + return meta + if meta["status"] == "starting" and time.time() - meta["created"] < START_GRACE: + return meta + with lock_file(cfg) as lk: + try: + fcntl.flock(lk, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + return meta + meta = read_meta(d) + if meta["status"] in ACTIVE: + meta["status"] = "lost" + write_meta(d, meta) + return meta + + +def all_jobs(cfg): + out = [] + for d in sorted(glob.glob(os.path.join(cfg["state_dir"], "jobs", "*")), reverse=True): + try: + out.append(refresh(cfg, d, read_meta(d))) + except (OSError, ValueError): + pass # a dir being created or removed right now + return out + + +# ── commands ──────────────────────────────────────────────────────────────── + +def cmd_version(cfg): + return {"version": VERSION, "overlays": cfg["overlays"]} + + +def cmd_jobs(cfg): + # ponytail: logs and job dirs are kept forever; prune once disk usage shows on the dashboard + return {"jobs": all_jobs(cfg)} + + +def cmd_job_start(cfg, kind=None, *args): + if kind not in KINDS: + raise Error(f"unknown kind: {kind}") + argv = KINDS[kind](cfg, list(args)) + base = time.strftime("%Y%m%d-%H%M%S-") + kind + jid, n = base, 1 + while True: + d = os.path.join(cfg["state_dir"], "jobs", jid) + try: + os.mkdir(d) + break + except FileExistsError: + n += 1 + jid = f"{base}-{n}" + write_meta(d, {"id": jid, "kind": kind, "args": list(args), "argv": argv, + "status": "starting", "created": time.time()}) + open(os.path.join(d, "log"), "wb").close() + subprocess.Popen([sys.executable, os.path.abspath(__file__), "_run", jid], + stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, start_new_session=True) + for _ in range(60): + status = read_meta(d)["status"] + if status == "busy": + shutil.rmtree(d) + return {"busy": next((m["id"] for m in all_jobs(cfg) if m["status"] == "running"), None)} + if status != "starting": + break + time.sleep(0.05) + return {"id": jid} + + +def cmd_run(cfg, jid=None): + d = job_dir(cfg, jid) + meta = read_meta(d) + with lock_file(cfg) as lk: + try: + fcntl.flock(lk, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + meta["status"] = "busy" + write_meta(d, meta) + return None + with open(os.path.join(d, "log"), "ab") as log: + meta["start"] = time.time() + try: + p = subprocess.Popen(meta["argv"], stdin=subprocess.DEVNULL, stdout=log, + stderr=subprocess.STDOUT, start_new_session=True) + except OSError as e: + log.write(f"sbb-remote: {e}\n".encode()) + rc = 127 + else: + meta.update(pid=p.pid, status="running") + write_meta(d, meta) + rc = p.wait() + meta.update(end=time.time(), exit=rc, + status="ok" if rc == 0 else "killed" if rc == -signal.SIGTERM else "failed") + write_meta(d, meta) + return None + + +def cmd_job_follow(cfg, jid=None, offset="0", out=None): + d = job_dir(cfg, jid) + out = out or sys.stdout.buffer + with open(os.path.join(d, "log"), "rb") as f: + f.seek(int(offset)) + while True: + # status first, then read: the last read after the job ends drains the log + done = refresh(cfg, d, read_meta(d))["status"] not in ACTIVE + chunk = f.read() + if chunk: + out.write(chunk) + out.flush() + elif done: + return None + else: + time.sleep(0.5) + + +def cmd_job_kill(cfg, jid=None): + meta = read_meta(job_dir(cfg, jid)) + if meta["status"] != "running": + raise Error(f"job not running: {jid}") + try: + os.killpg(meta["pid"], signal.SIGTERM) + except ProcessLookupError: + pass # ended on its own meanwhile + return {"ok": True} + + +# ── stats ─────────────────────────────────────────────────────────────────── + +def _try(fn): + try: + return fn() + except Exception: + return None + + +def _git(cfg, *args): + return subprocess.run(["git", "-C", cfg["repo_dir"], *args], + capture_output=True, text=True, check=True).stdout + + +def _pkgs(root, pattern): + """Names of the /.SlackBuild dirs matching pattern under root.""" + out = [] + for d in glob.glob(os.path.join(root, pattern, "")): + name = os.path.basename(os.path.dirname(d)) + if os.path.isfile(os.path.join(d, name + ".SlackBuild")): + out.append(name) + return out + + +def _packages(cfg): + r = cfg["repo_dir"] + mine = {o: _pkgs(r, o + "/*") for o in cfg["overlays"]} + tree = _git(cfg, "ls-tree", "-r", "--name-only", cfg["upstream_branch"]).splitlines() + upstream = {m.group(1) for m in map(PKG.match, tree) if m} + shadowed = upstream & {p for names in mine.values() for p in names} + return {"total": len(_pkgs(r, "*/*")), **{o: len(n) for o, n in mine.items()}, + "shadowed": len(shadowed)} + + +def _slackrepo(cfg): + db = sqlite3.connect(f"file:{cfg['slackrepo_db']}?mode=ro", uri=True) + try: + results = dict(db.execute("select result, count(*) from buildresults group by result")) + built = db.execute("select count(distinct itemid) from revisions where dep = '/'").fetchone()[0] + finally: + db.close() + return {"results": results, "built": built} + + +def _uptime(): + with open("/proc/uptime") as f: + return float(f.read().split()[0]) + + +def cmd_stats(cfg): + dirs = {"repo": cfg["repo_dir"], "state": cfg["state_dir"]} + return { + "load": _try(os.getloadavg), + "uptime": _try(_uptime), + "disk_free": {k: _try(lambda p=p: shutil.disk_usage(p).free) for k, p in dirs.items()}, + "packages": _try(lambda: _packages(cfg)), + "sync": {o: _try(lambda o=o: int(_git(cfg, "log", "-1", "--format=%ct", "--", o + "/"))) + for o in cfg["overlays"]}, + "slackrepo": _try(lambda: _slackrepo(cfg)), + } + + +COMMANDS = { + "version": cmd_version, + "stats": cmd_stats, + "jobs": cmd_jobs, + "job-start": cmd_job_start, + "job-follow": cmd_job_follow, + "job-kill": cmd_job_kill, + "_run": cmd_run, +} + + +def main(argv): + try: + if not argv or argv[0] not in COMMANDS: + names = ",".join(c for c in COMMANDS if not c.startswith("_")) + raise Error(f"usage: sbb-remote {{{names}}} [args]") + cfg = load_config() + os.makedirs(os.path.join(cfg["state_dir"], "jobs"), exist_ok=True) + result = COMMANDS[argv[0]](cfg, *argv[1:]) + except BrokenPipeError: + # the GUI hung up on job-follow; silence the flush at exit too + os.dup2(os.open(os.devnull, os.O_WRONLY), sys.stdout.fileno()) + return 0 + except Exception as e: + print(json.dumps({"error": str(e) if isinstance(e, Error) else f"{type(e).__name__}: {e}"})) + return 1 + if result is not None: + print(json.dumps(result)) + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) 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. +# +# 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() -- cgit v1.2.3