diff --git a/bin/dm.py b/bin/dm.py index 9da54fe..8c978a6 100755 --- a/bin/dm.py +++ b/bin/dm.py @@ -202,6 +202,8 @@ LOG_FILE = "/home/super/Projects/NetVM/dm-log.jsonl" # Resolved 2026-10-04. Add new entries as sidechats are created. SIDCHAT_ALIASES = { "646-opm-work": "d410b9ad-f667-465f-a103-43fabc0f69fe", + "646 tasks": "1e75a740-d08f-443d-a0f9-793db196e24f", + "heartbeat": "1e75a740-d08f-443d-a0f9-793db196e24f", } def resolve_sidechat_target(target): diff --git a/tests/test_main_nav.py b/tests/test_main_nav.py new file mode 100644 index 0000000..0217c01 --- /dev/null +++ b/tests/test_main_nav.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python3 +""" +test_main_nav.py — Comprehensive unit and integration test suite for: +1. Strict Main Chat blocking in CLI (`super dm send`, `super dm wo`) +2. Explicit override `--allow-main-chat` +3. Sidechat target resolution & alias mapping +4. "Reset-at-Begin, Leave-in-Sidechat" navigation logic in dm.py +5. Harvester sidechat-only thread scope (no Main Chat polling) +6. Staged file transfer & metadata registry lifecycle +""" +import unittest +import subprocess +import json +import os +import sys +import tempfile +from pathlib import Path + +REPO_ROOT = Path("/home/super/Projects/NetVM") +BIN_DIR = REPO_ROOT / "bin" +SUPER_CLI = BIN_DIR / "super-cli.py" +DM_PY = BIN_DIR / "dm.py" +HARVESTER_PY = BIN_DIR / "response-harvester.py" + +sys.path.insert(0, str(BIN_DIR)) + +class TestMainChatPolicyEnforcement(unittest.TestCase): + """Test CLI guards blocking unintended Main Chat access.""" + + def test_super_dm_send_blocks_main_chat_by_default(self): + cmd = [sys.executable, str(SUPER_CLI), "dm", "send", "--to", "646", "--target", "main", "Automated ping"] + res = subprocess.run(cmd, capture_output=True, text=True) + self.assertNotEqual(res.returncode, 0, "CLI should reject --target main without override") + self.assertIn("Targeting Main Chat is blocked by CHAT_POLICY.md", res.stderr) + self.assertIn("--allow-main-chat", res.stderr) + + def test_super_dm_wo_blocks_main_chat_by_default(self): + cmd = [sys.executable, str(SUPER_CLI), "dm", "wo", "--to", "646", "--target", "main", "--title", "Task", "Task body"] + res = subprocess.run(cmd, capture_output=True, text=True) + self.assertNotEqual(res.returncode, 0, "CLI should reject work order in Main Chat without override") + self.assertIn("Work Orders must target sidechats per CHAT_POLICY.md", res.stderr) + self.assertIn("--allow-main-chat", res.stderr) + + def test_super_dm_wo_defaults_to_agent_sidechat(self): + # Dry-run test parser defaults: when --target is omitted, should target agent task sidechat + cmd = [sys.executable, str(SUPER_CLI), "dm", "wo", "--help"] + res = subprocess.run(cmd, capture_output=True, text=True) + self.assertEqual(res.returncode, 0) + self.assertIn("Target sidechat (default: agent primary task sidechat)", res.stdout) + + +class TestSidechatNavigationLogic(unittest.TestCase): + """Test dm.py and sidechat target resolution.""" + + def test_resolve_sidechat_target(self): + import dm + # Well-known mappings + self.assertEqual(dm.resolve_sidechat_target("main"), "main") + self.assertEqual(dm.resolve_sidechat_target("646 tasks"), "1e75a740-d08f-443d-a0f9-793db196e24f") + self.assertEqual(dm.resolve_sidechat_target("heartbeat"), "1e75a740-d08f-443d-a0f9-793db196e24f") + + def test_uuid_regex_detection(self): + import dm, re + valid_uuid = "1e75a740-d08f-443d-a0f9-793db196e24f" + invalid_uuid = "646 tasks" + pattern = r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" + self.assertTrue(bool(re.match(pattern, valid_uuid, re.I))) + self.assertFalse(bool(re.match(pattern, invalid_uuid, re.I))) + + +class TestResponseHarvesterScope(unittest.TestCase): + """Verify harvester monitors only sidechats and leaves Main Chat out of periodic cycles.""" + + def test_harvester_excludes_main_chat(self): + import importlib.util + spec = importlib.util.spec_from_file_location("harvester", str(HARVESTER_PY)) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + + monitored = mod.get_monitored_threads() + for agent, threads in monitored.items(): + thread_ids = [t["id"] for t in threads] + self.assertNotIn("main", thread_ids, f"Agent {agent} should not have 'main' in automated harvester scope") + + +class TestPayloadTransfersLifecycle(unittest.TestCase): + """Test staging and metadata registry for payload file transfers.""" + + def test_files_clean_and_registry_sync(self): + with tempfile.NamedTemporaryFile("w", delete=False) as f: + f.write("Unit test payload content") + tmp_path = f.name + + try: + # 1. Clean aged payloads + clean_cmd = [sys.executable, str(SUPER_CLI), "dm", "files", "clean", "--older-than", "0"] + res = subprocess.run(clean_cmd, capture_output=True, text=True) + self.assertEqual(res.returncode, 0) + self.assertIn("Cleaned", res.stdout) + + # 2. View files list (JSON output) + list_cmd = [sys.executable, str(SUPER_CLI), "dm", "files", "--json"] + res = subprocess.run(list_cmd, capture_output=True, text=True) + self.assertEqual(res.returncode, 0) + data = json.loads(res.stdout) + self.assertTrue(data.get("ok")) + self.assertEqual(len(data.get("files", [])), 0) + finally: + if os.path.exists(tmp_path): + os.unlink(tmp_path) + + +if __name__ == "__main__": + unittest.main(verbosity=2)