"""Tests for read-only box lookups over HTTPS (no SSH). Covers the agent-facing read path: box-relay.sh (agent client) -> exec-constrained.py named ops -> box-ctl.py backend verbs -> super-cli.py lookups. Live-socket round-trips are intentionally NOT covered here (loopback TCP is unavailable in some sandboxes); instead we assert the exact argv each op builds and execute the fast argv directly. """ import argparse import importlib.util import io import json import subprocess import sys import unittest from contextlib import redirect_stdout from pathlib import Path from unittest import mock REPO_ROOT = Path(__file__).resolve().parent.parent BOX_CTL = REPO_ROOT / "bin" / "box-ctl.py" RELAY = REPO_ROOT / "bin" / "box-relay.sh" def _load(name, relpath): spec = importlib.util.spec_from_file_location(name, REPO_ROOT / relpath) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod exec_constrained = _load("exec_constrained_read", "bin/exec-constrained.py") super_cli = _load("super_cli_read", "bin/super-cli.py") def _box_ctl(*args): return subprocess.run( [sys.executable, str(BOX_CTL), *args], capture_output=True, text=True, timeout=60) class ExecReadOpsTests(unittest.TestCase): def test_ops_registered_and_read_only(self): self.assertIn("fleet.unread", exec_constrained.OPS) self.assertIn("dm.log", exec_constrained.OPS) self.assertFalse(exec_constrained.OPS["fleet.unread"]["side_effecting"]) self.assertFalse(exec_constrained.OPS["dm.log"]["side_effecting"]) def test_default_perms_include_read_ops(self): # Any valid fleet signer can read; canary stays ping-only. self.assertTrue(exec_constrained.permitted("some-unknown-identity", "fleet.unread")) self.assertTrue(exec_constrained.permitted("some-unknown-identity", "dm.log")) self.assertFalse(exec_constrained.permitted("exec-canary", "fleet.unread")) self.assertFalse(exec_constrained.permitted("exec-canary", "dm.log")) def test_fleet_unread_validate(self): v = exec_constrained.OPS["fleet.unread"]["validate"] self.assertEqual(v({}), {"agent": None}) self.assertEqual(v({"agent": "pip"}), {"agent": "pip"}) with self.assertRaises(exec_constrained.OpError): v({"agent": "nope"}) with self.assertRaises(exec_constrained.OpError): v({"bogus": 1}) def test_dm_log_validate(self): v = exec_constrained.OPS["dm.log"]["validate"] self.assertEqual(v({}), {"limit": 20, "agent": None}) self.assertEqual(v({"limit": 5, "agent": "opm"}), {"limit": 5, "agent": "opm"}) with self.assertRaises(exec_constrained.OpError): v({"limit": 0}) with self.assertRaises(exec_constrained.OpError): v({"limit": 101}) with self.assertRaises(exec_constrained.OpError): v({"agent": "nope"}) with self.assertRaises(exec_constrained.OpError): v({"bogus": 1}) def test_build_argv_shapes(self): unread = exec_constrained.OPS["fleet.unread"] argv = unread["build"]({"agent": None}) self.assertEqual(argv[-1], "unread") self.assertNotIn("--agent", argv) argv = unread["build"]({"agent": "muse"}) self.assertEqual(argv[-3:], ["unread", "--agent", "muse"]) dmlog = exec_constrained.OPS["dm.log"] argv = dmlog["build"]({"limit": 5, "agent": "opm"}) self.assertEqual(argv[-4:], ["dm-log", "5", "--agent", "opm"]) argv = dmlog["build"]({"limit": 20, "agent": None}) self.assertEqual(argv[-2:], ["dm-log", "20"]) # argv only, never a shell string. self.assertIsInstance(argv, list) def test_dm_log_built_argv_executes(self): spec = exec_constrained.OPS["dm.log"] clean = spec["validate"]({"limit": 2}) argv = spec["build"](clean) argv[0] = sys.executable # hermetic interpreter, same script + args r = subprocess.run(argv, capture_output=True, text=True, timeout=60) self.assertEqual(r.returncode, 0, r.stderr) data = json.loads(r.stdout) self.assertTrue(data["ok"]) self.assertEqual(len(data["entries"]), 2) class BoxCtlReadVerbsTests(unittest.TestCase): def test_unread_rejects_unknown_agent(self): r = _box_ctl("unread", "--agent", "nope") self.assertNotEqual(r.returncode, 0) self.assertEqual(json.loads(r.stdout)["code"], "BAD_NODE") def test_unread_rejects_positional_and_missing_value(self): r = _box_ctl("unread", "pip") self.assertNotEqual(r.returncode, 0) self.assertEqual(json.loads(r.stdout)["code"], "BAD_ARGS") r = _box_ctl("unread", "--agent") self.assertNotEqual(r.returncode, 0) self.assertEqual(json.loads(r.stdout)["code"], "BAD_ARGS") def test_dm_log_rejects_bad_limit_and_agent(self): r = _box_ctl("dm-log", "abc") self.assertNotEqual(r.returncode, 0) self.assertEqual(json.loads(r.stdout)["code"], "BAD_LIMIT") r = _box_ctl("dm-log", "5", "--agent", "nope") self.assertNotEqual(r.returncode, 0) self.assertEqual(json.loads(r.stdout)["code"], "BAD_NODE") r = _box_ctl("dm-log", "1", "2") self.assertNotEqual(r.returncode, 0) self.assertEqual(json.loads(r.stdout)["code"], "BAD_ARGS") def test_dm_log_back_compat_limit_only(self): r = _box_ctl("dm-log", "2") self.assertEqual(r.returncode, 0, r.stderr) data = json.loads(r.stdout) self.assertTrue(data["ok"]) self.assertEqual(len(data["entries"]), 2) def test_quality_validate_new_verbs(self): r = _box_ctl("quality-validate", "unread", "--agent", "pip") data = json.loads(r.stdout) self.assertTrue(data["valid"], r.stdout) r = _box_ctl("quality-validate", "dm-log", "5", "--agent", "opm") self.assertTrue(json.loads(r.stdout)["valid"], r.stdout) r = _box_ctl("quality-validate", "unread", "--agent", "nope") self.assertFalse(json.loads(r.stdout)["valid"], r.stdout) r = _box_ctl("quality-validate", "unread", "extra-positional") self.assertFalse(json.loads(r.stdout)["valid"], r.stdout) class SuperCliUnreadTests(unittest.TestCase): def test_lookup_dispatches_unread(self): args = argparse.Namespace(target="unread", lookup_args=[], json=False) with mock.patch.object(super_cli, "_lookup_unreads") as m: super_cli.cmd_lookup(args) m.assert_called_once_with(args) def test_lookup_unreads_json_shape(self): fleet = [ {"node": "muse", "title": "muse (2)", "url": "https://muse.ai/thread/abc123", "approval_pending": False}, {"node": "pip", "title": "muse", "url": "https://muse.ai/", "approval_pending": True}, ] args = argparse.Namespace(json=True) buf = io.StringIO() with mock.patch.object(super_cli, "collect_fleet_data", return_value=fleet): with redirect_stdout(buf): super_cli._lookup_unreads(args) data = json.loads(buf.getvalue()) self.assertTrue(data["ok"]) by_node = {n["node"]: n for n in data["nodes"]} self.assertEqual(by_node["muse"]["unread"], 2) self.assertEqual(by_node["muse"]["thread"], "abc123") self.assertFalse(by_node["muse"]["approval_pending"]) self.assertEqual(by_node["pip"]["unread"], 0) self.assertTrue(by_node["pip"]["approval_pending"]) self.assertEqual(by_node["pip"]["thread"], "home") class BoxRelayClientTests(unittest.TestCase): def test_relay_syntax_valid(self): r = subprocess.run(["bash", "-n", str(RELAY)], capture_output=True, text=True, timeout=30) self.assertEqual(r.returncode, 0, r.stderr) def test_relay_help_lists_read_commands(self): r = subprocess.run(["bash", str(RELAY), "help"], capture_output=True, text=True, timeout=30) self.assertEqual(r.returncode, 0, r.stderr) self.assertIn("box unread", r.stdout) self.assertIn("box dm log", r.stdout) def test_relay_maps_read_commands_to_ops(self): text = RELAY.read_text() self.assertIn("fleet.unread", text) self.assertIn('"dm.log"', text) if __name__ == "__main__": unittest.main()