rp/tests/test_multiplexer_commands.py

from unittest.mock import Mock, patch
from pr.commands.multiplexer_commands import (
    show_sessions,
    attach_session,
    detach_session,
    kill_session,
    send_command,
    show_session_log,
    show_session_status,
    list_waiting_sessions,
)


class TestShowSessions:
    @patch("pr.commands.multiplexer_commands.list_active_sessions")
    @patch("pr.commands.multiplexer_commands.get_session_status")
    def test_show_sessions_no_sessions(self, mock_status, mock_list):
        mock_list.return_value = {}
        show_sessions()
        mock_list.assert_called_once()

    @patch("pr.commands.multiplexer_commands.list_active_sessions")
    @patch("pr.commands.multiplexer_commands.get_session_status")
    def test_show_sessions_with_sessions(self, mock_status, mock_list):
        mock_list.return_value = {
            "session1": {
                "metadata": {
                    "process_type": "test",
                    "start_time": 123.0,
                    "interaction_count": 5,
                    "state": "running",
                },
                "output_summary": {"stdout_lines": 10, "stderr_lines": 2},
            }
        }
        mock_status.return_value = {"is_active": True, "pid": 123}
        show_sessions()
        mock_list.assert_called_once()
        mock_status.assert_called_once_with("session1")


class TestAttachSession:
    def test_attach_session_no_args(self):
        attach_session([])

    @patch("pr.commands.multiplexer_commands.get_session_status")
    def test_attach_session_not_found(self, mock_status):
        mock_status.return_value = None
        attach_session(["session1"])

    @patch("pr.commands.multiplexer_commands.get_session_status")
    @patch("pr.commands.multiplexer_commands.read_session_output")
    def test_attach_session_success(self, mock_read, mock_status):
        mock_status.return_value = {"is_active": True, "metadata": {"process_type": "test"}}
        mock_read.return_value = {"stdout": "line1\nline2", "stderr": ""}
        attach_session(["session1"])

    @patch("pr.commands.multiplexer_commands.get_session_status")
    @patch("pr.commands.multiplexer_commands.read_session_output")
    def test_attach_session_with_stderr(self, mock_read, mock_status):
        mock_status.return_value = {"is_active": False, "metadata": {"process_type": "test"}}
        mock_read.return_value = {"stdout": "", "stderr": "error1\nerror2"}
        attach_session(["session1"])

    @patch("pr.commands.multiplexer_commands.get_session_status")
    @patch("pr.commands.multiplexer_commands.read_session_output")
    def test_attach_session_read_error(self, mock_read, mock_status):
        mock_status.return_value = {"is_active": True, "metadata": {"process_type": "test"}}
        mock_read.side_effect = Exception("test error")
        attach_session(["session1"])


class TestDetachSession:
    def test_detach_session_no_args(self):
        detach_session([])

    @patch("pr.commands.multiplexer_commands.get_multiplexer")
    def test_detach_session_not_found(self, mock_get):
        mock_get.return_value = None
        detach_session(["session1"])

    @patch("pr.commands.multiplexer_commands.get_multiplexer")
    def test_detach_session_success(self, mock_get):
        mock_mux = Mock()
        mock_get.return_value = mock_mux
        detach_session(["session1"])
        assert mock_mux.show_output is False


class TestKillSession:
    def test_kill_session_no_args(self):
        kill_session([])

    @patch("pr.commands.multiplexer_commands.close_interactive_session")
    def test_kill_session_success(self, mock_close):
        kill_session(["session1"])
        mock_close.assert_called_once_with("session1")

    @patch("pr.commands.multiplexer_commands.close_interactive_session")
    def test_kill_session_error(self, mock_close):
        mock_close.side_effect = Exception("test error")
        kill_session(["session1"])


class TestSendCommand:
    def test_send_command_no_args(self):
        send_command([])

    def test_send_command_insufficient_args(self):
        send_command(["session1"])

    @patch("pr.commands.multiplexer_commands.send_input_to_session")
    def test_send_command_success(self, mock_send):
        send_command(["session1", "ls", "-la"])
        mock_send.assert_called_once_with("session1", "ls -la")

    @patch("pr.commands.multiplexer_commands.send_input_to_session")
    def test_send_command_error(self, mock_send):
        mock_send.side_effect = Exception("test error")
        send_command(["session1", "ls"])


class TestShowSessionLog:
    def test_show_session_log_no_args(self):
        show_session_log([])

    @patch("pr.commands.multiplexer_commands.read_session_output")
    def test_show_session_log_success(self, mock_read):
        mock_read.return_value = {"stdout": "stdout content", "stderr": "stderr content"}
        show_session_log(["session1"])

    @patch("pr.commands.multiplexer_commands.read_session_output")
    def test_show_session_log_no_stderr(self, mock_read):
        mock_read.return_value = {"stdout": "stdout content", "stderr": ""}
        show_session_log(["session1"])

    @patch("pr.commands.multiplexer_commands.read_session_output")
    def test_show_session_log_error(self, mock_read):
        mock_read.side_effect = Exception("test error")
        show_session_log(["session1"])


class TestShowSessionStatus:
    def test_show_session_status_no_args(self):
        show_session_status([])

    @patch("pr.commands.multiplexer_commands.get_session_status")
    def test_show_session_status_not_found(self, mock_status):
        mock_status.return_value = None
        show_session_status(["session1"])

    @patch("pr.commands.multiplexer_commands.get_session_status")
    @patch("pr.commands.multiplexer_commands.get_global_detector")
    def test_show_session_status_success(self, mock_detector, mock_status):
        mock_status.return_value = {
            "is_active": True,
            "pid": 123,
            "metadata": {
                "process_type": "test",
                "start_time": 123.0,
                "last_activity": 456.0,
                "interaction_count": 5,
                "state": "running",
            },
            "output_summary": {"stdout_lines": 10, "stderr_lines": 2},
        }
        mock_detector_instance = Mock()
        mock_detector_instance.get_session_info.return_value = {
            "current_state": "waiting",
            "is_waiting": True,
        }
        mock_detector.return_value = mock_detector_instance
        show_session_status(["session1"])

    @patch("pr.commands.multiplexer_commands.get_session_status")
    @patch("pr.commands.multiplexer_commands.get_global_detector")
    def test_show_session_status_no_detector_info(self, mock_detector, mock_status):
        mock_status.return_value = {
            "is_active": False,
            "metadata": {
                "process_type": "test",
                "start_time": 123.0,
                "last_activity": 456.0,
                "interaction_count": 5,
                "state": "running",
            },
            "output_summary": {"stdout_lines": 10, "stderr_lines": 2},
        }
        mock_detector_instance = Mock()
        mock_detector_instance.get_session_info.return_value = None
        mock_detector.return_value = mock_detector_instance
        show_session_status(["session1"])


class TestListWaitingSessions:
    @patch("pr.commands.multiplexer_commands.list_active_sessions")
    @patch("pr.commands.multiplexer_commands.get_global_detector")
    def test_list_waiting_sessions_no_sessions(self, mock_detector, mock_list):
        mock_list.return_value = {}
        mock_detector_instance = Mock()
        mock_detector_instance.is_waiting_for_input.return_value = False
        mock_detector.return_value = mock_detector_instance
        list_waiting_sessions()

    @patch("pr.commands.multiplexer_commands.list_active_sessions")
    @patch("pr.commands.multiplexer_commands.get_session_status")
    @patch("pr.commands.multiplexer_commands.get_global_detector")
    def test_list_waiting_sessions_with_waiting(self, mock_detector, mock_status, mock_list):
        mock_list.return_value = ["session1"]
        mock_detector_instance = Mock()
        mock_detector_instance.is_waiting_for_input.return_value = True
        mock_detector_instance.get_session_info.return_value = {
            "current_state": "waiting",
            "is_waiting": True,
        }
        mock_detector_instance.get_response_suggestions.return_value = ["yes", "no", "quit"]
        mock_detector.return_value = mock_detector_instance
        mock_status.return_value = {"metadata": {"process_type": "test"}}
        list_waiting_sessions()

    @patch("pr.commands.multiplexer_commands.list_active_sessions")
    @patch("pr.commands.multiplexer_commands.get_global_detector")
    def test_list_waiting_sessions_no_waiting(self, mock_detector, mock_list):
        mock_list.return_value = ["session1"]
        mock_detector_instance = Mock()
        mock_detector_instance.is_waiting_for_input.return_value = False
        mock_detector.return_value = mock_detector_instance
        list_waiting_sessions()