1# Copyright 2023 The Pigweed Authors 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); you may not 4# use this file except in compliance with the License. You may obtain a copy of 5# the License at 6# 7# https://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 12# License for the specific language governing permissions and limitations under 13# the License. 14"""Tests for pw_console.socket_client""" 15 16import socket 17import unittest 18 19 20from pw_console import socket_client 21 22 23class TestSocketClient(unittest.TestCase): 24 """Tests for SocketClient.""" 25 26 def test_parse_config_default(self) -> None: 27 config = "default" 28 with unittest.mock.patch.object( 29 socket_client.SocketClient, 'connect', return_value=None 30 ): 31 client = socket_client.SocketClient(config) 32 self.assertEqual( 33 client._socket_init_args, # pylint: disable=protected-access 34 (socket.AF_INET6, socket.SOCK_STREAM), 35 ) 36 self.assertEqual( 37 client._address, # pylint: disable=protected-access 38 ( 39 socket_client.SocketClient.DEFAULT_SOCKET_SERVER, 40 socket_client.SocketClient.DEFAULT_SOCKET_PORT, 41 ), 42 ) 43 44 def test_parse_config_unix_file(self) -> None: 45 # Skip test if UNIX sockets are not supported. 46 if not hasattr(socket, 'AF_UNIX'): 47 return 48 49 config = 'file:fake_file_path' 50 with unittest.mock.patch.object( 51 socket_client.SocketClient, 'connect', return_value=None 52 ): 53 client = socket_client.SocketClient(config) 54 self.assertEqual( 55 client._socket_init_args, # pylint: disable=protected-access 56 ( 57 socket.AF_UNIX, # pylint: disable=no-member 58 socket.SOCK_STREAM, 59 ), 60 ) 61 self.assertEqual( 62 client._address, # pylint: disable=protected-access 63 'fake_file_path', 64 ) 65 66 def _check_config_parsing( 67 self, config: str, expected_address: str, expected_port: int 68 ) -> None: 69 with unittest.mock.patch.object( 70 socket_client.SocketClient, 'connect', return_value=None 71 ): 72 fake_getaddrinfo_return_value = [ 73 (socket.AF_INET6, socket.SOCK_STREAM, 0, None, None) 74 ] 75 with unittest.mock.patch.object( 76 socket, 77 'getaddrinfo', 78 return_value=fake_getaddrinfo_return_value, 79 ) as mock_getaddrinfo: 80 client = socket_client.SocketClient(config) 81 mock_getaddrinfo.assert_called_with( 82 expected_address, expected_port, type=socket.SOCK_STREAM 83 ) 84 # Assert the init args are what is returned by ``getaddrinfo`` 85 # not necessarily the correct ones, since this test should not 86 # perform any network action. 87 self.assertEqual( 88 client._socket_init_args, # pylint: disable=protected-access 89 ( 90 socket.AF_INET6, 91 socket.SOCK_STREAM, 92 ), 93 ) 94 95 def test_parse_config_ipv4_domain(self) -> None: 96 self._check_config_parsing( 97 config='file.com/some_long/path:80', 98 expected_address='file.com/some_long/path', 99 expected_port=80, 100 ) 101 102 def test_parse_config_ipv4_domain_no_port(self) -> None: 103 self._check_config_parsing( 104 config='file.com/some/path', 105 expected_address='file.com/some/path', 106 expected_port=socket_client.SocketClient.DEFAULT_SOCKET_PORT, 107 ) 108 109 def test_parse_config_ipv4_address(self) -> None: 110 self._check_config_parsing( 111 config='8.8.8.8:8080', 112 expected_address='8.8.8.8', 113 expected_port=8080, 114 ) 115 116 def test_parse_config_ipv4_address_no_port(self) -> None: 117 self._check_config_parsing( 118 config='8.8.8.8', 119 expected_address='8.8.8.8', 120 expected_port=socket_client.SocketClient.DEFAULT_SOCKET_PORT, 121 ) 122 123 def test_parse_config_ipv6_domain(self) -> None: 124 self._check_config_parsing( 125 config='[file.com/some_long/path]:80', 126 expected_address='file.com/some_long/path', 127 expected_port=80, 128 ) 129 130 def test_parse_config_ipv6_domain_no_port(self) -> None: 131 self._check_config_parsing( 132 config='[file.com/some/path]', 133 expected_address='file.com/some/path', 134 expected_port=socket_client.SocketClient.DEFAULT_SOCKET_PORT, 135 ) 136 137 def test_parse_config_ipv6_address(self) -> None: 138 self._check_config_parsing( 139 config='[2001:4860:4860::8888:8080]:666', 140 expected_address='2001:4860:4860::8888:8080', 141 expected_port=666, 142 ) 143 144 def test_parse_config_ipv6_address_no_port(self) -> None: 145 self._check_config_parsing( 146 config='[2001:4860:4860::8844]', 147 expected_address='2001:4860:4860::8844', 148 expected_port=socket_client.SocketClient.DEFAULT_SOCKET_PORT, 149 ) 150 151 def test_parse_config_ipv6_local(self) -> None: 152 self._check_config_parsing( 153 config='[fe80::100%eth0]:80', 154 expected_address='fe80::100%eth0', 155 expected_port=80, 156 ) 157 158 def test_parse_config_ipv6_local_no_port(self) -> None: 159 self._check_config_parsing( 160 config='[fe80::100%eth0]', 161 expected_address='fe80::100%eth0', 162 expected_port=socket_client.SocketClient.DEFAULT_SOCKET_PORT, 163 ) 164 165 def test_parse_config_ipv6_local_windows(self) -> None: 166 self._check_config_parsing( 167 config='[fe80::100%4]:80', 168 expected_address='fe80::100%4', 169 expected_port=80, 170 ) 171 172 def test_parse_config_ipv6_local_no_port_windows(self) -> None: 173 self._check_config_parsing( 174 config='[fe80::100%4]', 175 expected_address='fe80::100%4', 176 expected_port=socket_client.SocketClient.DEFAULT_SOCKET_PORT, 177 ) 178 179 180if __name__ == '__main__': 181 unittest.main() 182