box/tests/test_tmux_auto_approver.py

#!/usr/bin/env python3
"""test_tmux_auto_approver.py — Unit tests for Tmux worker management and regex auto-approver."""

import json
import os
import sys
import tempfile
import time
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch

REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT / "bin"))

import tmux_auto_approver
from tmux_auto_approver import (
    DEFAULT_RULES,
    GUARDRAIL_PATTERNS,
    AutoApproverRunner,
    AutoApproverState,
    MatchRule,
    MatchVerdict,
    RegexApproverEngine,
    TmuxPaneInfo,
    TmuxWorkerTally,
    gather_tmux_tally,
    infer_agent_for_session,
)


class TestRegexApproverEngine(unittest.TestCase):
    def setUp(self):
        self.engine = RegexApproverEngine(DEFAULT_RULES)

    def test_muse_code_run_numbered_match(self):
        text = """
Terminal output before...
Would you like to run the following bash command?
  echo "hello"
› 1. Yes, proceed (y)
  2. No, skip (n)
"""
        verdict = self.engine.evaluate(text)
        self.assertTrue(verdict.matched)
        self.assertEqual(verdict.rule_id, "muse_code_run_numbered")
        self.assertEqual(verdict.key, "1")
        self.assertFalse(verdict.press_enter)

    def test_muse_code_run_yn_match(self):
        text = "Some actions\n› 1. Yes, proceed (y)\n"
        verdict = self.engine.evaluate(text)
        self.assertTrue(verdict.matched)
        self.assertIn("muse_code_run", verdict.rule_id)
        self.assertEqual(verdict.key, "1")

    def test_muse_code_allow_execution_match(self):
        text = "Allow execution of script /tmp/test.sh? [y/N]"
        verdict = self.engine.evaluate(text)
        self.assertTrue(verdict.matched)
        self.assertEqual(verdict.rule_id, "muse_code_allow_execution")
        self.assertEqual(verdict.key, "y")
        self.assertTrue(verdict.press_enter)

    def test_choice_abc_match(self):
        text = "Select option from the choices below:\nA) Deploy to production\nB) Staging"
        verdict = self.engine.evaluate(text)
        self.assertTrue(verdict.matched)
        self.assertEqual(verdict.rule_id, "choice_abc")
        self.assertEqual(verdict.key, "A")
        self.assertTrue(verdict.press_enter)

    def test_menu_numbered_match(self):
        text = "Option: Please pick a number:\n(1) Accept terms\n(2) Decline"
        verdict = self.engine.evaluate(text)
        self.assertTrue(verdict.matched)
        self.assertEqual(verdict.rule_id, "menu_numbered")
        self.assertEqual(verdict.key, "1")
        self.assertTrue(verdict.press_enter)

    def test_confirm_yn_match(self):
        text = "Do you want to continue? [y/n]: "
        verdict = self.engine.evaluate(text)
        self.assertTrue(verdict.matched)
        self.assertEqual(verdict.rule_id, "confirm_yn")
        self.assertEqual(verdict.key, "y")
        self.assertTrue(verdict.press_enter)

    def test_enter_to_continue_match(self):
        text = "Task completed successfully. Press [Enter] to continue..."
        verdict = self.engine.evaluate(text)
        self.assertTrue(verdict.matched)
        self.assertEqual(verdict.rule_id, "enter_to_continue")
        self.assertEqual(verdict.key, "Enter")

    def test_guardrail_sudo_password_blocked(self):
        text = "[sudo] password for super: "
        verdict = self.engine.evaluate(text)
        self.assertFalse(verdict.matched)
        self.assertTrue(verdict.is_blocked)
        self.assertIn("guardrail", verdict.blocked_reason.lower())

    def test_guardrail_passkey_blocked(self):
        text = "Insert security key or enter passkey PIN:"
        verdict = self.engine.evaluate(text)
        self.assertFalse(verdict.matched)
        self.assertTrue(verdict.is_blocked)
        self.assertIn("guardrail", verdict.blocked_reason.lower())

    def test_guardrail_destructive_command_blocked(self):
        text = "Running dangerous cleanup: rm -rf /"
        verdict = self.engine.evaluate(text)
        self.assertFalse(verdict.matched)
        self.assertTrue(verdict.is_blocked)
        self.assertIn("guardrail", verdict.blocked_reason.lower())

    def test_empty_text_returns_unmatched(self):
        verdict = self.engine.evaluate("")
        self.assertFalse(verdict.matched)
        self.assertFalse(verdict.is_blocked)


class TestAgentInference(unittest.TestCase):
    def test_infer_agent_names(self):
        self.assertEqual(infer_agent_for_session("/tmp/tmux-muse.sock", "worker-1"), "muse")
        self.assertEqual(infer_agent_for_session("/tmp/tmux-pip.sock", "agent-job"), "pip")
        self.assertEqual(infer_agent_for_session("/tmp/tmux-1000/default", "646-retention"), "646")
        self.assertEqual(infer_agent_for_session("/tmp/tmux-1000/default", "opm"), "opm")
        self.assertEqual(infer_agent_for_session("/tmp/tmux-1000/default", "dev-test"), "dev")
        self.assertEqual(infer_agent_for_session("/tmp/tmux-1000/default", "def"), "def")
        self.assertEqual(infer_agent_for_session("/tmp/tmux-1000/lte", "main"), "host")


class TestAutoApproverRunner(unittest.TestCase):
    def setUp(self):
        self.temp_dir = tempfile.TemporaryDirectory()
        self.orig_state = tmux_auto_approver.STATE_FILE
        self.orig_audit = tmux_auto_approver.AUDIT_LOG_FILE
        tmux_auto_approver.STATE_FILE = Path(self.temp_dir.name) / "test_state.json"
        tmux_auto_approver.AUDIT_LOG_FILE = Path(self.temp_dir.name) / "test_audit.jsonl"

    def tearDown(self):
        tmux_auto_approver.STATE_FILE = self.orig_state
        tmux_auto_approver.AUDIT_LOG_FILE = self.orig_audit
        self.temp_dir.cleanup()

    @patch("tmux_auto_approver.run_tmux_cmd")
    @patch("tmux_auto_approver.capture_pane_text")
    @patch("tmux_auto_approver.gather_tmux_tally")
    def test_run_once_dry_run_dispatches_match(self, mock_tally, mock_capture, mock_tmux_cmd):
        mock_pane = TmuxPaneInfo(
            socket="/tmp/tmux-1000/default",
            session="muse",
            window_idx=0,
            pane_id="%37",
            pane_pid=12345,
            current_command="muse-bin",
            active=True,
            attached=True,
            title="muse terminal",
            agent_node="muse",
            auto_approve=True,
        )

        mock_tally.return_value = TmuxWorkerTally(
            total_sockets=1,
            total_sessions=1,
            total_panes=1,
            active_workers=1,
            by_agent={"muse": {"sessions": 1, "panes": 1, "active_commands": ["muse-bin"], "auto_approve": True}},
            panes=[mock_pane],
        )

        mock_capture.return_value = "Would you like to run the following?\n› 1. Yes, proceed (y)"

        runner = AutoApproverRunner(dry_run=True)
        results = runner.run_once()

        self.assertEqual(len(results), 1)
        self.assertEqual(results[0]["action"], "DRY_RUN_MATCH")
        self.assertEqual(results[0]["key_sent"], "1")
        self.assertEqual(results[0]["rule_name"], "Muse Code Run (Numbered)")

        # Verify deduplication within 15 seconds
        results2 = runner.run_once()
        self.assertEqual(len(results2), 0)

    @patch("tmux_auto_approver.gather_tmux_tally")
    def test_run_once_global_disabled(self, mock_tally):
        runner = AutoApproverRunner(dry_run=True)
        runner.state.global_enabled = False
        runner.state.save()
        results = runner.run_once()
        self.assertEqual(len(results), 1)
        self.assertEqual(results[0]["status"], "disabled")


class TestTallyGathering(unittest.TestCase):
    @patch("tmux_auto_approver.get_existing_sockets")
    @patch("tmux_auto_approver.run_tmux_cmd")
    def test_gather_tmux_tally_parsing(self, mock_run_cmd, mock_sockets):
        mock_sockets.return_value = ["/tmp/tmux-1000/default"]
        # Format: #{session_name}___#{window_index}___#{pane_id}___#{pane_pid}___#{pane_current_command}___#{pane_active}___#{session_attached}___#{pane_title}
        mock_run_cmd.return_value = (
            0,
            "muse___0___%1___1001___muse-bin___1___1___muse worker\n"
            "dev___0___%2___1002___bash___0___0___dev terminal\n",
            "",
        )

        state = AutoApproverState()
        tally = gather_tmux_tally(state)

        self.assertEqual(tally.total_sockets, 1)
        self.assertEqual(tally.total_sessions, 2)
        self.assertEqual(tally.total_panes, 2)
        self.assertEqual(tally.active_workers, 1)  # only muse-bin is active, bash is shell
        self.assertEqual(len(tally.panes), 2)
        self.assertEqual(tally.panes[0].agent_node, "muse")
        self.assertEqual(tally.panes[1].agent_node, "dev")


if __name__ == "__main__":
    unittest.main()