fenrir/tests/unit/test_unix_remote_driver.py

from unittest.mock import Mock, mock_open, patch

from fenrirscreenreader.remoteDriver import unixDriver


class FakeClientSocket:
    def __init__(self, data):
        self.data = data
        self.sent = b""

    def recv(self, _size):
        return self.data

    def sendall(self, data):
        self.sent += data


def test_main_socket_routes_to_ancestor_private_socket(mock_environment):
    driver = unixDriver.driver()
    driver.env = mock_environment
    driver.bound_sockets = [
        (Mock(), "/tmp/fenrirscreenreader-999.sock"),
        (Mock(), unixDriver.MAIN_SOCKET_FILE),
    ]
    client_sock = FakeClientSocket(b"command say routed")
    event_queue = Mock()

    with patch.object(driver, "_get_peer_pid", return_value=1234), patch.object(
        driver,
        "_find_ancestor_private_socket",
        return_value="/tmp/fenrirscreenreader-222.sock",
    ), patch.object(
        driver,
        "_find_available_private_socket",
        return_value="/tmp/fenrirscreenreader-222.sock",
    ), patch.object(
        driver, "_forward_remote_to_socket", return_value=True
    ) as forward:
        driver._handle_client(
            client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE
        )

    forward.assert_called_once_with(
        "command say routed", "/tmp/fenrirscreenreader-222.sock"
    )
    event_queue.put.assert_not_called()


def test_main_socket_routes_to_first_available_private_socket(
    mock_environment,
):
    driver = unixDriver.driver()
    driver.env = mock_environment
    driver.bound_sockets = [
        (Mock(), "/tmp/fenrirscreenreader-999.sock"),
        (Mock(), unixDriver.MAIN_SOCKET_FILE),
    ]
    client_sock = FakeClientSocket(b"command say fallback")
    event_queue = Mock()

    with patch.object(driver, "_get_peer_pid", return_value=1234), patch.object(
        driver,
        "_find_ancestor_private_socket",
        return_value="",
    ), patch.object(
        driver,
        "_find_available_private_socket",
        return_value="/tmp/fenrirscreenreader-333.sock",
    ), patch.object(
        driver, "_forward_remote_to_socket", return_value=True
    ) as forward:
        driver._handle_client(
            client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE
        )

    forward.assert_called_once_with(
        "command say fallback", "/tmp/fenrirscreenreader-333.sock"
    )
    event_queue.put.assert_not_called()


def test_main_socket_handles_command_locally_without_available_target(
    mock_environment,
):
    driver = unixDriver.driver()
    driver.env = mock_environment
    client_sock = FakeClientSocket(b"command say local")
    event_queue = Mock()

    with patch.object(driver, "_get_peer_pid", return_value=1234), patch.object(
        driver,
        "_find_available_private_socket",
        return_value="",
    ):
        driver._handle_client(
            client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE
        )

    event_queue.put.assert_called_once_with(
        {
            "Type": unixDriver.FenrirEventType.remote_incomming,
            "data": "command say local",
        }
    )


def test_vcsa_main_socket_owner_handles_command_locally(mock_environment):
    driver = unixDriver.driver()
    driver.env = mock_environment
    driver.bound_sockets = [(Mock(), unixDriver.MAIN_SOCKET_FILE)]
    client_sock = FakeClientSocket(b"command say root")
    event_queue = Mock()

    with patch.object(driver, "_find_available_private_socket") as find_available:
        driver._handle_client(
            client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE
        )

    find_available.assert_not_called()
    event_queue.put.assert_called_once_with(
        {
            "Type": unixDriver.FenrirEventType.remote_incomming,
            "data": "command say root",
        }
    )


def test_private_socket_handles_command_locally(mock_environment):
    driver = unixDriver.driver()
    driver.env = mock_environment
    client_sock = FakeClientSocket(b"command say private")
    event_queue = Mock()

    driver._handle_client(
        client_sock, event_queue, "/tmp/fenrirscreenreader-222.sock"
    )

    event_queue.put.assert_called_once_with(
        {
            "Type": unixDriver.FenrirEventType.remote_incomming,
            "data": "command say private",
        }
    )


def test_list_command_still_returns_response_from_main_socket(mock_environment):
    driver = unixDriver.driver()
    driver.env = mock_environment
    mock_environment["runtime"]["RemoteManager"] = Mock()
    client_sock = FakeClientSocket(b"command list")
    event_queue = Mock()
    mock_environment["runtime"][
        "RemoteManager"
    ].handle_remote_incomming_with_response.return_value = {
        "success": True,
        "message": "pid=1",
    }

    driver._handle_client(client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE)

    assert client_sock.sent == b"pid=1\n"
    event_queue.put.assert_not_called()


def test_get_parent_pid_parses_process_names_with_spaces():
    driver = unixDriver.driver()

    stat_file = mock_open(
        read_data="1234 (name with spaces) S 567 1 1 0 -1 0\n"
    )
    with patch("builtins.open", stat_file):
        assert driver._get_parent_pid(1234) == 567


def test_find_available_private_socket_prefers_ancestor_socket(
    mock_environment,
):
    driver = unixDriver.driver()
    driver.env = mock_environment

    with patch(
        "fenrirscreenreader.remoteDriver.unixDriver.remoteInstanceRegistry.list_instances",
        return_value=[
            {
                "pid": 111,
                "socket_files": ["/tmp/fenrirscreenreader-111.sock"],
            },
            {
                "pid": 222,
                "socket_files": ["/tmp/fenrirscreenreader-222.sock"],
            },
        ],
    ), patch.object(driver, "_is_socket_active", return_value=True):
        assert (
            driver._find_available_private_socket(
                "/tmp/fenrirscreenreader-222.sock"
            )
            == "/tmp/fenrirscreenreader-222.sock"
        )


def test_find_available_private_socket_skips_main_socket(
    mock_environment,
):
    driver = unixDriver.driver()
    driver.env = mock_environment

    with patch(
        "fenrirscreenreader.remoteDriver.unixDriver.remoteInstanceRegistry.list_instances",
        return_value=[
            {
                "pid": 111,
                "socket_files": [
                    unixDriver.MAIN_SOCKET_FILE,
                    "/tmp/fenrirscreenreader-111.sock",
                ],
            }
        ],
    ), patch.object(driver, "_is_socket_active", return_value=True):
        assert (
            driver._find_available_private_socket()
            == "/tmp/fenrirscreenreader-111.sock"
        )