pw_console/py/socket_client_test.py

# Copyright 2023 The Pigweed Authors
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy of
# the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.
"""Tests for pw_console.socket_client"""

import socket
import unittest


from pw_console import socket_client


class TestSocketClient(unittest.TestCase):
    """Tests for SocketClient."""

    def test_parse_config_default(self) -> None:
        config = "default"
        with unittest.mock.patch.object(
            socket_client.SocketClient, 'connect', return_value=None
        ):
            client = socket_client.SocketClient(config)
            self.assertEqual(
                client._socket_init_args,  # pylint: disable=protected-access
                (socket.AF_INET6, socket.SOCK_STREAM),
            )
            self.assertEqual(
                client._address,  # pylint: disable=protected-access
                (
                    socket_client.SocketClient.DEFAULT_SOCKET_SERVER,
                    socket_client.SocketClient.DEFAULT_SOCKET_PORT,
                ),
            )

    def test_parse_config_unix_file(self) -> None:
        # Skip test if UNIX sockets are not supported.
        if not hasattr(socket, 'AF_UNIX'):
            return

        config = 'file:fake_file_path'
        with unittest.mock.patch.object(
            socket_client.SocketClient, 'connect', return_value=None
        ):
            client = socket_client.SocketClient(config)
            self.assertEqual(
                client._socket_init_args,  # pylint: disable=protected-access
                (
                    socket.AF_UNIX,  # pylint: disable=no-member
                    socket.SOCK_STREAM,
                ),
            )
            self.assertEqual(
                client._address,  # pylint: disable=protected-access
                'fake_file_path',
            )

    def _check_config_parsing(
        self, config: str, expected_address: str, expected_port: int
    ) -> None:
        with unittest.mock.patch.object(
            socket_client.SocketClient, 'connect', return_value=None
        ):
            fake_getaddrinfo_return_value = [
                (socket.AF_INET6, socket.SOCK_STREAM, 0, None, None)
            ]
            with unittest.mock.patch.object(
                socket,
                'getaddrinfo',
                return_value=fake_getaddrinfo_return_value,
            ) as mock_getaddrinfo:
                client = socket_client.SocketClient(config)
                mock_getaddrinfo.assert_called_with(
                    expected_address, expected_port, type=socket.SOCK_STREAM
                )
                # Assert the init args are what is returned by ``getaddrinfo``
                # not necessarily the correct ones, since this test should not
                # perform any network action.
                self.assertEqual(
                    client._socket_init_args,  # pylint: disable=protected-access
                    (
                        socket.AF_INET6,
                        socket.SOCK_STREAM,
                    ),
                )

    def test_parse_config_ipv4_domain(self) -> None:
        self._check_config_parsing(
            config='file.com/some_long/path:80',
            expected_address='file.com/some_long/path',
            expected_port=80,
        )

    def test_parse_config_ipv4_domain_no_port(self) -> None:
        self._check_config_parsing(
            config='file.com/some/path',
            expected_address='file.com/some/path',
            expected_port=socket_client.SocketClient.DEFAULT_SOCKET_PORT,
        )

    def test_parse_config_ipv4_address(self) -> None:
        self._check_config_parsing(
            config='8.8.8.8:8080',
            expected_address='8.8.8.8',
            expected_port=8080,
        )

    def test_parse_config_ipv4_address_no_port(self) -> None:
        self._check_config_parsing(
            config='8.8.8.8',
            expected_address='8.8.8.8',
            expected_port=socket_client.SocketClient.DEFAULT_SOCKET_PORT,
        )

    def test_parse_config_ipv6_domain(self) -> None:
        self._check_config_parsing(
            config='[file.com/some_long/path]:80',
            expected_address='file.com/some_long/path',
            expected_port=80,
        )

    def test_parse_config_ipv6_domain_no_port(self) -> None:
        self._check_config_parsing(
            config='[file.com/some/path]',
            expected_address='file.com/some/path',
            expected_port=socket_client.SocketClient.DEFAULT_SOCKET_PORT,
        )

    def test_parse_config_ipv6_address(self) -> None:
        self._check_config_parsing(
            config='[2001:4860:4860::8888:8080]:666',
            expected_address='2001:4860:4860::8888:8080',
            expected_port=666,
        )

    def test_parse_config_ipv6_address_no_port(self) -> None:
        self._check_config_parsing(
            config='[2001:4860:4860::8844]',
            expected_address='2001:4860:4860::8844',
            expected_port=socket_client.SocketClient.DEFAULT_SOCKET_PORT,
        )

    def test_parse_config_ipv6_local(self) -> None:
        self._check_config_parsing(
            config='[fe80::100%eth0]:80',
            expected_address='fe80::100%eth0',
            expected_port=80,
        )

    def test_parse_config_ipv6_local_no_port(self) -> None:
        self._check_config_parsing(
            config='[fe80::100%eth0]',
            expected_address='fe80::100%eth0',
            expected_port=socket_client.SocketClient.DEFAULT_SOCKET_PORT,
        )

    def test_parse_config_ipv6_local_windows(self) -> None:
        self._check_config_parsing(
            config='[fe80::100%4]:80',
            expected_address='fe80::100%4',
            expected_port=80,
        )

    def test_parse_config_ipv6_local_no_port_windows(self) -> None:
        self._check_config_parsing(
            config='[fe80::100%4]',
            expected_address='fe80::100%4',
            expected_port=socket_client.SocketClient.DEFAULT_SOCKET_PORT,
        )


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