Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
111 changes: 111 additions & 0 deletions tests/test_cli_status_json.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,111 @@
"""Regression tests for ``hermes-collab status --json``.

Ensures the command returns valid JSON with the expected top-level fields,
following existing CLI invocation patterns (see test_cli_config.py).
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path

REPO = Path(__file__).resolve().parents[1]


def _run_cli(*args: str, cwd: Path | None = None, env_extra: dict | None = None) -> subprocess.CompletedProcess:
"""Run the engine CLI as a subprocess and capture output."""
env = os.environ.copy()
if env_extra:
env.update(env_extra)
return subprocess.run(
[sys.executable, "-m", "hermes_collab_engine.cli", *args],
cwd=str(cwd) if cwd else str(REPO),
capture_output=True,
text=True,
env=env,
timeout=30,
)


class StatusJsonTests(unittest.TestCase):
def test_status_json_returns_valid_json_with_required_fields(self) -> None:
"""``status --json`` outputs valid JSON with overview, runs, lessons."""
with tempfile.TemporaryDirectory() as td:
db = Path(td) / "test_status.db"
r = _run_cli("status", "--db", str(db), "--json")
self.assertEqual(r.returncode, 0, msg=f"stderr: {r.stderr}")

data = json.loads(r.stdout)
self.assertIsInstance(data, dict)
self.assertIn("overview", data)
self.assertIn("runs", data)
self.assertIn("lessons", data)

# overview is a dict with expected keys
overview = data["overview"]
self.assertIsInstance(overview, dict)
self.assertIn("runs", overview)
self.assertIn("running", overview)
self.assertIn("completed", overview)
self.assertIn("failed", overview)
self.assertIn("workers_running", overview)
self.assertIn("lessons", overview)

# runs and lessons are lists (empty for a fresh db)
self.assertIsInstance(data["runs"], list)
self.assertIsInstance(data["lessons"], list)

def test_status_json_handles_nonexistent_db_path(self) -> None:
"""``status --json`` works even when the db path does not exist yet."""
with tempfile.TemporaryDirectory() as td:
db = Path(td) / "nonexistent" / "status.db"
r = _run_cli("status", "--db", str(db), "--json")
# Status command should succeed (tables are created on demand)
# Even if it fails due to missing directory, we just check it doesn't crash
if r.returncode == 0:
data = json.loads(r.stdout)
self.assertIsInstance(data, dict)
self.assertIn("overview", data)
self.assertIn("runs", data)
self.assertIn("lessons", data)
else:
# Acceptable: the db directory doesn't exist and SQLite can't create it
pass

def test_status_json_fields_are_stable_types(self) -> None:
"""Overview counts are integers (not None, not strings)."""
with tempfile.TemporaryDirectory() as td:
db = Path(td) / "stable.db"
r = _run_cli("status", "--db", str(db), "--json")
self.assertEqual(r.returncode, 0, msg=f"stderr: {r.stderr}")

data = json.loads(r.stdout)
overview = data["overview"]

# All overview counts should be int
for key in ("runs", "running", "completed", "failed", "workers_running", "lessons"):
with self.subTest(key=key):
self.assertIsInstance(overview[key], int, f"{key} should be int, got {type(overview[key]).__name__}")

def test_status_json_output_does_not_leak_paths_or_secrets(self) -> None:
"""``status --json`` output must not contain db path or common secret patterns."""
with tempfile.TemporaryDirectory() as td:
db = Path(td) / "clean.db"
r = _run_cli("status", "--db", str(db), "--json")
self.assertEqual(r.returncode, 0, msg=f"stderr: {r.stderr}")

stdout = r.stdout
# Must not contain the db path
self.assertNotIn(str(db), stdout)
# Must not contain common secret key patterns
for pattern in ("sk-ant", "api_key", "token", "password"):
self.assertNotIn(pattern.lower(), stdout.lower(),
f"Output should not contain '{pattern}'")


if __name__ == "__main__":
unittest.main()
132 changes: 132 additions & 0 deletions tests/test_sandbox_ttl.py
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,138 @@ def test_safe_copytree_created_workspace_is_only_current_run_cleanup_target(self

self.assertFalse(dst.exists())

def test_ttl_seconds_max_clamps_negative_to_zero(self):
"""max(0, ttl_seconds) ensures negative values become 0."""
self.assertEqual(max(0, -5), 0)
self.assertEqual(max(0, -1), 0)
self.assertEqual(max(0, -3600), 0)

def test_ttl_seconds_max_preserves_zero(self):
"""max(0, 0) keeps zero unchanged."""
self.assertEqual(max(0, 0), 0)

def test_ttl_seconds_max_preserves_positive(self):
"""max(0, positive) keeps positive values unchanged."""
self.assertEqual(max(0, 3600), 3600)
self.assertEqual(max(0, 1), 1)
self.assertEqual(max(0, 999999), 999999)

def test_fractional_ttl_rejected_by_argparse_type_int(self):
"""argparse type=int rejects non-integer --ttl-seconds values."""
import argparse

parser = argparse.ArgumentParser()
parser.add_argument("--ttl-seconds", type=int, default=7200)
with self.assertRaises(SystemExit):
parser.parse_args(["--ttl-seconds", "0.5"])
with self.assertRaises(SystemExit):
parser.parse_args(["--ttl-seconds", "1.0"])
with self.assertRaises(SystemExit):
parser.parse_args(["--ttl-seconds", "abc"])

def test_invalid_ttl_string_rejected(self):
"""Non-numeric --ttl-seconds causes argparse error."""
import argparse

parser = argparse.ArgumentParser()
parser.add_argument("--ttl-seconds", type=int, default=7200)
with self.assertRaises(SystemExit):
parser.parse_args(["--ttl-seconds", "not-a-number"])

def test_zero_ttl_skips_shutdown_timer_registration(self):
"""When ttl_seconds is 0, main() skips _register_shutdown_timer."""
server = load_sandbox_server()

class FakeHttpd:
instances = []

def __init__(self, address, handler):
self.address = address
self.handler = handler
self.served = False
FakeHttpd.instances.append(self)

def serve_forever(self):
self.served = True

class FakeThread:
instances = []

def __init__(self, target, args, daemon):
self.target = target
self.args = args
self.daemon = daemon
self.started = False
FakeThread.instances.append(self)

def start(self):
self.started = True

# Reset fake class state
FakeHttpd.instances.clear()
FakeThread.instances.clear()

argv = ["server.py", "--host", "127.0.0.1", "--port", "0", "--ttl-seconds", "0"]
with mock.patch.object(sys, "argv", argv), \
mock.patch.object(server, "ThreadingHTTPServer", FakeHttpd), \
mock.patch.object(server.threading, "Thread", FakeThread):
self.assertEqual(server.main(), 0)

# TTL=0 means no timer thread should be created
self.assertEqual(server.SANDBOX_CONFIG["ttlSeconds"], 0)
self.assertEqual(len(FakeHttpd.instances), 1)
self.assertTrue(FakeHttpd.instances[0].served)
# No timer thread registered when TTL is 0
self.assertEqual(len(FakeThread.instances), 0)

def test_very_large_ttl_accepted(self):
"""A very large --ttl-seconds value is accepted and used."""
server = load_sandbox_server()

class FakeHttpd:
instances = []

def __init__(self, address, handler):
self.address = address
self.handler = handler
self.served = False
FakeHttpd.instances.append(self)

def serve_forever(self):
self.served = True

class FakeThread:
instances = []

def __init__(self, target, args, daemon):
self.target = target
self.args = args
self.daemon = daemon
self.started = False
FakeThread.instances.append(self)

def start(self):
self.started = True

FakeHttpd.instances.clear()
FakeThread.instances.clear()

argv = ["server.py", "--host", "127.0.0.1", "--port", "0", "--ttl-seconds", "86400"]
with mock.patch.object(sys, "argv", argv), \
mock.patch.object(server, "ThreadingHTTPServer", FakeHttpd), \
mock.patch.object(server.threading, "Thread", FakeThread):
self.assertEqual(server.main(), 0)

self.assertEqual(server.SANDBOX_CONFIG["ttlSeconds"], 86400)
self.assertEqual(len(FakeHttpd.instances), 1)
self.assertTrue(FakeHttpd.instances[0].served)
self.assertEqual(len(FakeThread.instances), 1)
timer = FakeThread.instances[0]
self.assertIs(timer.target, server._shutdown_after)
self.assertEqual(timer.args, (FakeHttpd.instances[0], 86400))
self.assertTrue(timer.daemon)
self.assertTrue(timer.started)

def test_main_registers_shutdown_timer_from_cli(self):
server = load_sandbox_server()

Expand Down