headquarter/apps/api/tests/unit/test_terminal_session.py

"""Unit tests for TerminalSession."""

import asyncio
from unittest.mock import MagicMock, patch

import pytest

from src.services.terminal_session import TerminalSession


@pytest.fixture
def mock_pty():
    """Mock pty.openpty to return predictable fds."""
    master_fd = 10
    slave_fd = 11
    with (
        patch(
            "src.services.terminal_session.pty.openpty",
            return_value=(master_fd, slave_fd),
        ),
        patch("src.services.terminal_session.os.close") as mock_close,
    ):
        yield master_fd, slave_fd, mock_close


class TestTerminalSessionStart:
    def test_init_state(self, mock_pty):
        session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")

        assert session.session_id == "sess-1"
        assert session.container_id == "container-abc"
        assert session._echo_enabled is True
        assert session._exit_reason is None


class TestTerminalSessionEchoDetection:
    @patch("src.services.terminal_session.termios.tcgetattr")
    def test_detect_echo_state_enabled(self, mock_tcgetattr):
        session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
        session._master_fd = 10

        # termios.ECHO flag set
        attrs = [[], [], [], __import__("termios").ECHO, [], [], []]
        mock_tcgetattr.return_value = attrs

        result = session._detect_echo_state()
        assert result is True

    @patch("src.services.terminal_session.termios.tcgetattr")
    def test_detect_echo_state_disabled(self, mock_tcgetattr):
        session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
        session._master_fd = 10

        # termios.ECHO flag NOT set
        attrs = [[], [], [], 0, [], [], []]
        mock_tcgetattr.return_value = attrs

        result = session._detect_echo_state()
        assert result is False

    def test_detect_echo_state_no_master_fd(self):
        session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
        session._master_fd = None

        result = session._detect_echo_state()
        assert result is True  # default


class TestTerminalSessionResize:
    @patch("src.services.terminal_session.fcntl.ioctl")
    def test_resize_sets_size(self, mock_ioctl):
        session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
        session._master_fd = 10

        # Should not raise
        asyncio.run(session.resize(120, 40))
        mock_ioctl.assert_called_once()

    def test_resize_when_closed(self):
        session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
        session._closed = True

        # Should not raise
        asyncio.run(session.resize(120, 40))


class TestTerminalSessionWriteInput:
    @patch("src.services.terminal_session.os.write")
    def test_write_input(self, mock_write):
        session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
        session._master_fd = 10

        asyncio.run(session.write_input(b"hello"))
        mock_write.assert_called_once_with(10, b"hello")

    def test_write_input_when_closed(self):
        session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
        session._closed = True

        # Should not raise
        asyncio.run(session.write_input(b"hello"))


class TestTerminalSessionReadOutput:
    @patch("src.services.terminal_session.select.select")
    @patch("src.services.terminal_session.os.read")
    def test_read_output_with_data(self, mock_read, mock_select):
        session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
        session._master_fd = 10

        mock_select.return_value = ([10], [], [])
        mock_read.return_value = b"output"

        result = asyncio.run(session.read_output())
        assert result == b"output"

    @patch("src.services.terminal_session.select.select")
    def test_read_output_no_data(self, mock_select):
        session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
        session._master_fd = 10

        mock_select.return_value = ([], [], [])

        result = asyncio.run(session.read_output())
        assert result == b""


class TestTerminalSessionClose:
    @patch("src.services.terminal_session.os.close")
    @patch("src.services.terminal_session.asyncio.wait_for")
    async def test_close_sets_exit_reason(self, mock_wait_for, mock_close):
        session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
        session._master_fd = 10
        session.process = MagicMock()
        session.process.returncode = 0

        await session.close()
        assert session._exit_reason == "process_exit"
        assert session._closed is True

    async def test_close_idempotent(self):
        session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
        session._closed = True

        # Should not raise
        await session.close()


class TestTerminalSessionIsAlive:
    def test_is_alive_with_running_process(self):
        session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
        session.process = MagicMock()
        session.process.returncode = None

        assert session.is_alive() is True

    def test_is_alive_with_exited_process(self):
        session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
        session.process = MagicMock()
        session.process.returncode = 0

        assert session.is_alive() is False

    def test_is_alive_no_process(self):
        session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
        session.process = None

        assert session.is_alive() is False