"""bin/stack/{start,start-assistant,stop,status} — offline contract + fake PATH.""" from __future__ import annotations import os import stat import subprocess import tempfile import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[2] METHODS = ("start", "start-assistant", "start-mail-sync", "stop", "status") class StackLayoutTest(unittest.TestCase): def test_methods_are_bash_with_usage_comment(self) -> None: lib = ROOT / "bin" / "stack" / "lib.sh" self.assertTrue(lib.is_file(), "missing bin/stack/lib.sh") for name in METHODS: p = ROOT / "bin" / "stack" / name self.assertTrue(p.is_file(), f"missing bin/stack/{name}") self.assertTrue(os.access(p, os.X_OK), f"bin/stack/{name} must be executable") lines = p.read_text().splitlines() self.assertEqual(lines[0], "#!/usr/bin/env bash", name) self.assertTrue(lines[1].startswith("# bin/stack/"), name) text = "\n".join(lines) self.assertIn("lib.sh", text, name) self.assertNotIn("/mnt/", text, name) self.assertNotIn("/home/", text, name) def test_lib_has_no_host_paths_or_secrets(self) -> None: lib = (ROOT / "bin" / "stack" / "lib.sh").read_text() self.assertIn("stack_start", lib) self.assertIn("stack_start_assistant", lib) self.assertIn("stack_start_mail_sync", lib) self.assertIn("stack_stop", lib) self.assertIn("stack_status", lib) self.assertIn("mail-sync", lib) self.assertIn("qwen3.5:9b", lib) self.assertIn("picoclaw agent", lib) self.assertIn("--no-deps", lib) self.assertIn("tools/list", lib) self.assertNotIn("/mnt/", lib) self.assertNotIn("/home/", lib) self.assertNotIn("password", lib.lower()) self.assertNotIn("GITEA_TOKEN", lib) def test_start_does_not_launch_picoclaw(self) -> None: start = (ROOT / "bin" / "stack" / "start").read_text() self.assertIn("stack_start", start) self.assertNotIn("stack_start_assistant", start) self.assertNotIn("picoclaw agent", start) def test_start_mail_sync_is_etl_only(self) -> None: src = (ROOT / "bin" / "stack" / "start-mail-sync").read_text() self.assertIn("stack_start_mail_sync", src) self.assertNotIn("stack_start_assistant", src) ep = (ROOT / "bin" / "docker-entrypoint").read_text() self.assertIn('MAIL_SYNC_SRC:=onlyoffice,gmail', ep) self.assertIn('MAIL_SYNC_INDEX:=0', ep) self.assertIn('interval="${1:-300}"', ep) self.assertIn('MAIL_SYNC_INDEX" = "1"', ep) def test_start_assistant_attaches_agent(self) -> None: src = (ROOT / "bin" / "stack" / "start-assistant").read_text() self.assertIn("stack_start_assistant", src) self.assertIn("--no-attach", src) def test_stop_does_not_down_volumes(self) -> None: lib = (ROOT / "bin" / "stack" / "lib.sh").read_text() self.assertIn(" compose ", lib) self.assertRegex(lib, r"\bstop\b") self.assertNotIn(" compose down", lib) self.assertNotIn("compose down", lib) def test_docs_name_stack_commands(self) -> None: runbook = (ROOT / "docs" / "runbook.md").read_text() pico = (ROOT / "docs" / "picoclaw.md").read_text() agents = (ROOT / "AGENTS.md").read_text() for text in (runbook, pico, agents): self.assertIn("bin/stack/start", text) self.assertIn("bin/stack/start-assistant", text) self.assertIn("bin/stack/status", text) self.assertIn("bin/stack/stop", text) def test_help_prints_comments_not_source(self) -> None: r = subprocess.run( [str(ROOT / "bin" / "stack" / "start"), "--help"], cwd=str(ROOT), capture_output=True, text=True, check=False, ) self.assertEqual(r.returncode, 0, r.stderr) self.assertIn("bin/stack/start", r.stdout) self.assertNotIn("set -euo pipefail", r.stdout) self.assertNotIn("source ", r.stdout) class StackFakePathTest(unittest.TestCase): def _fake_bin(self, tmp: Path, *, health_ok: bool) -> Path: bindir = tmp / "bin" bindir.mkdir() curl = bindir / "curl" docker = bindir / "docker" log = tmp / "docker.log" curl.write_text( f"""#!/usr/bin/env bash url="" for a in "$@"; do case "$a" in http*) url=$a ;; esac done if [ "{int(health_ok)}" = "0" ] && [[ "$url" == */health ]]; then echo '{{"status":"down"}}' exit 7 fi case "$url" in */mcp) echo '{{"jsonrpc":"2.0","id":1,"result":{{"tools":[{{"name":"search"}},{{"name":"get"}},{{"name":"audit"}}]}}}}' ;; */api/tags) echo '{{"models":[{{"name":"qwen3.5:9b"}}]}}' ;; */api/pull) echo '{{"status":"success"}}' ;; *) echo '{{"status":"ok"}}' ;; esac """ ) docker.write_text( f"""#!/usr/bin/env bash echo "$*" >> "{log}" exit 0 """ ) curl.chmod(curl.stat().st_mode | stat.S_IEXEC) docker.chmod(docker.stat().st_mode | stat.S_IEXEC) return bindir def _env(self, bindir: Path) -> dict[str, str]: env = os.environ.copy() env["PATH"] = f"{bindir}:{env.get('PATH', '')}" env["STACK_WAIT_SECS"] = "1" env["STACK_WAIT_INTERVAL"] = "0" return env def test_start_skips_compose_when_brain_healthy(self) -> None: with tempfile.TemporaryDirectory() as raw: tmp = Path(raw) bindir = self._fake_bin(tmp, health_ok=True) log = tmp / "docker.log" r = subprocess.run( [str(ROOT / "bin" / "stack" / "start")], cwd=str(ROOT), env=self._env(bindir), capture_output=True, text=True, check=False, ) self.assertEqual(r.returncode, 0, r.stderr) if log.exists(): logged = log.read_text() self.assertNotIn("up -d", logged, "healthy brain must not docker compose up") self.assertIn("ps", logged) # status probes mail-sync via compose ps def test_start_ups_brain_when_unhealthy(self) -> None: with tempfile.TemporaryDirectory() as raw: tmp = Path(raw) bindir = self._fake_bin(tmp, health_ok=False) log = tmp / "docker.log" r = subprocess.run( [str(ROOT / "bin" / "stack" / "start")], cwd=str(ROOT), env=self._env(bindir), capture_output=True, text=True, check=False, ) self.assertNotEqual(r.returncode, 0, "unhealthy brain without recovering compose must fail") self.assertTrue(log.exists(), r.stderr) logged = log.read_text() self.assertIn("up -d", logged) self.assertIn("brain", logged) self.assertNotIn("picoclaw", logged) def test_stop_stops_named_services(self) -> None: with tempfile.TemporaryDirectory() as raw: tmp = Path(raw) bindir = self._fake_bin(tmp, health_ok=True) log = tmp / "docker.log" r = subprocess.run( [str(ROOT / "bin" / "stack" / "stop")], cwd=str(ROOT), env=self._env(bindir), capture_output=True, text=True, check=False, ) self.assertEqual(r.returncode, 0, r.stderr) logged = log.read_text() self.assertIn("stop", logged) self.assertNotIn(" down", logged) for svc in ("brain", "brain-mcp", "reasoner", "picoclaw", "mail-sync"): self.assertIn(svc, logged) def test_start_assistant_no_attach_starts_picoclaw(self) -> None: with tempfile.TemporaryDirectory() as raw: tmp = Path(raw) bindir = self._fake_bin(tmp, health_ok=True) log = tmp / "docker.log" r = subprocess.run( [str(ROOT / "bin" / "stack" / "start-assistant"), "--no-attach"], cwd=str(ROOT), env=self._env(bindir), capture_output=True, text=True, check=False, ) self.assertEqual(r.returncode, 0, r.stderr + r.stdout) logged = log.read_text() if log.exists() else "" self.assertIn("picoclaw", logged) self.assertIn("--no-deps", logged) self.assertNotIn("picoclaw agent", logged)