box/tests/test_main_nav.py

#!/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"), "1dfb3199-2f99-446c-83e3-848ae2da0a12")
        self.assertEqual(dm.resolve_sidechat_target("heartbeat"), "757198c3-c1b2-48b8-ba2b-062c84f71b02")

    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)