|
19 | 19 | # limitations under the License. |
20 | 20 |
|
21 | 21 |
|
22 | | -from unittest import TestCase |
| 22 | +import pytest |
| 23 | +import unittest.mock as mock |
| 24 | +from socket import ( |
| 25 | + AF_INET, |
| 26 | + AF_INET6, |
| 27 | +) |
23 | 28 |
|
| 29 | +from neo4j.addressing import ( |
| 30 | + Address, |
| 31 | + IPv4Address, |
| 32 | + IPv6Address, |
| 33 | +) |
24 | 34 | from neo4j import GraphDatabase |
25 | | -from neo4j.addressing import Address |
26 | 35 |
|
| 36 | +mock_socket_ipv4 = mock.Mock() |
| 37 | +mock_socket_ipv4.getpeername = lambda: ("127.0.0.1", 7687) # (address, port) |
27 | 38 |
|
28 | | -class RoutingTableParseAddressTestCase(TestCase): |
| 39 | +mock_socket_ipv6 = mock.Mock() |
| 40 | +mock_socket_ipv6.getpeername = lambda: ("[::1]", 7687, 0, 0) # (address, port, flow info, scope id) |
29 | 41 |
|
30 | | - def test_should_parse_ipv4_address_and_port(self): |
31 | | - parsed = Address.parse("127.0.0.1:7687") |
32 | | - self.assertEqual(parsed, ("127.0.0.1", 7687)) |
| 42 | +# python -m pytest tests/unit/test_addressing.py -s |
33 | 43 |
|
34 | | - def test_should_parse_ipv6_address_and_port(self): |
35 | | - parsed = Address.parse("[::1]:7687") |
36 | | - self.assertEqual(parsed, ("::1", 7687, 0, 0)) |
37 | 44 |
|
38 | | - def test_should_parse_host_name_and_port(self): |
39 | | - parsed = Address.parse("localhost:7687") |
40 | | - self.assertEqual(parsed, ("localhost", 7687)) |
| 45 | +@pytest.mark.parametrize( |
| 46 | + "test_input, expected", |
| 47 | + [ |
| 48 | + (("127.0.0.1", 7687), {"family": AF_INET, "host": "127.0.0.1", "port": 7687, "str": "127.0.0.1:7687", "repr": "IPv4Address(('127.0.0.1', 7687))"}), |
| 49 | + (("localhost", 7687), {"family": AF_INET, "host": "localhost", "port": 7687, "str": "localhost:7687", "repr": "IPv4Address(('localhost', 7687))"}), |
| 50 | + ((None, None), {"family": AF_INET, "host": None, "port": None, "str": "None:None", "repr": "IPv4Address((None, None))"}), |
| 51 | + (("::1", 7687), {"family": AF_INET, "host": "::1", "port": 7687, "str": "::1:7687", "repr": "IPv4Address(('::1', 7687))"}), |
| 52 | + (("::1", 7687, 0, 0), {"family": AF_INET6, "host": "::1", "port": 7687, "str": "[::1]:7687", "repr": "IPv6Address(('::1', 7687, 0, 0))"}), |
| 53 | + (("::1", 7687, 1, 2), {"family": AF_INET6, "host": "::1", "port": 7687, "str": "[::1]:7687", "repr": "IPv6Address(('::1', 7687, 1, 2))"}), |
| 54 | + ((None, None, None, None), {"family": AF_INET6, "host": None, "port": None, "str": "[None]:None", "repr": "IPv6Address((None, None, None, None))"}), |
| 55 | + (Address(("127.0.0.1", 7687)), {"family": AF_INET, "host": "127.0.0.1", "port": 7687, "str": "127.0.0.1:7687", "repr": "IPv4Address(('127.0.0.1', 7687))"}), |
| 56 | + (Address(("::1", 7687, 1, 2)), {"family": AF_INET6, "host": "::1", "port": 7687, "str": "[::1]:7687", "repr": "IPv6Address(('::1', 7687, 1, 2))"}), |
| 57 | + ] |
| 58 | +) |
| 59 | +def test_address_initialization(test_input, expected): |
| 60 | + # python -m pytest tests/unit/test_addressing.py -s -k test_address_initialization |
| 61 | + address = Address(test_input) |
| 62 | + assert address.family == expected["family"] |
| 63 | + assert address.host == expected["host"] |
| 64 | + assert address.port == expected["port"] |
| 65 | + assert str(address) == expected["str"] |
| 66 | + assert repr(address) == expected["repr"] |
41 | 67 |
|
42 | | - def verify_routing_context(self, expected, query): |
43 | | - context = GraphDatabase._parse_routing_context(query) |
44 | | - self.assertEqual(context, expected) |
45 | 68 |
|
46 | | - def test_parse_routing_context(self): |
47 | | - self.verify_routing_context({"name": "molly", "color": "white"}, "name=molly&color=white") |
48 | | - self.verify_routing_context({"name": "molly", "color": "white"}, "name=molly&color=white") |
49 | | - self.verify_routing_context({"name": "molly", "color": "white"}, "name=molly&color=white") |
| 69 | +@pytest.mark.parametrize( |
| 70 | + "test_input", |
| 71 | + [ |
| 72 | + Address(("127.0.0.1", 7687)), |
| 73 | + Address(("127.0.0.1", 7687, 1, 2)), |
| 74 | + ] |
| 75 | +) |
| 76 | +def test_address_init_with_address_object_returns_same_instance(test_input): |
| 77 | + # python -m pytest tests/unit/test_addressing.py -s -k test_address_init_with_address_object_returns_same_instance |
| 78 | + address = Address(test_input) |
| 79 | + assert address is test_input |
| 80 | + assert id(address) == id(test_input) |
50 | 81 |
|
51 | | - def test_should_error_when_value_missing(self): |
52 | | - with self.assertRaises(ValueError): |
53 | | - GraphDatabase._parse_routing_context("name=&color=white") |
54 | 82 |
|
55 | | - def test_should_error_when_key_duplicate(self): |
56 | | - with self.assertRaises(ValueError): |
57 | | - GraphDatabase._parse_routing_context("name=molly&name=white") |
| 83 | +@pytest.mark.parametrize( |
| 84 | + "test_input, expected", |
| 85 | + [ |
| 86 | + (("127.0.0.1",), ValueError), |
| 87 | + (("127.0.0.1", 7687, 0), ValueError), |
| 88 | + (("[::1]", 7687, 0), ValueError), |
| 89 | + (("[::1]", 7687, 0, 0, 0), ValueError), |
| 90 | + ] |
| 91 | +) |
| 92 | +def test_address_initialization_with_incorrect_input(test_input, expected): |
| 93 | + # python -m pytest tests/unit/test_addressing.py -s -k test_address_initialization_with_incorrect_input |
| 94 | + with pytest.raises(expected): |
| 95 | + address = Address(test_input) |
58 | 96 |
|
59 | 97 |
|
60 | | -def test_address_init_with_address_object_returns_same_instance(): |
61 | | - a = Address(("localhost", 7687)) |
62 | | - b = Address(a) |
63 | | - assert id(a) == id(b) |
| 98 | +@pytest.mark.parametrize( |
| 99 | + "test_input, expected", |
| 100 | + [ |
| 101 | + (mock_socket_ipv4, ("127.0.0.1", 7687)), |
| 102 | + (mock_socket_ipv6, ("[::1]", 7687, 0, 0)) |
| 103 | + ] |
| 104 | +) |
| 105 | +def test_address_from_socket(test_input, expected): |
| 106 | + # python -m pytest tests/unit/test_addressing.py -s -k test_address_from_socket |
| 107 | + |
| 108 | + address = Address.from_socket(test_input) |
| 109 | + assert address == expected |
| 110 | + |
| 111 | + |
| 112 | +def test_address_from_socket_with_none(): |
| 113 | + # python -m pytest tests/unit/test_addressing.py -s -k test_address_from_socket_with_none |
| 114 | + with pytest.raises(AttributeError): |
| 115 | + address = Address.from_socket(None) |
| 116 | + |
| 117 | + |
| 118 | +@pytest.mark.parametrize( |
| 119 | + "test_input, expected", |
| 120 | + [ |
| 121 | + ("127.0.0.1:7687", ("127.0.0.1", 7687)), |
| 122 | + ("localhost:7687", ("localhost", 7687)), |
| 123 | + (":7687", ("localhost", 7687)), |
| 124 | + (":", ("localhost", 0)), |
| 125 | + ("", ("localhost", 0)), |
| 126 | + (":abcd", ("localhost", "abcd")), |
| 127 | + (" ", (" ", 0)), |
| 128 | + ] |
| 129 | +) |
| 130 | +def test_address_parse_with_ipv4(test_input, expected): |
| 131 | + # python -m pytest tests/unit/test_addressing.py -s -k test_address_parse_with_ipv4 |
| 132 | + parsed = Address.parse(test_input) |
| 133 | + assert parsed == expected |
| 134 | + |
| 135 | + |
| 136 | +@pytest.mark.parametrize( |
| 137 | + "test_input, expected", |
| 138 | + [ |
| 139 | + ("[::1]:7687", ("::1", 7687, 0, 0)), |
| 140 | + ("[::1]:abcd", ("::1", "abcd", 0, 0)), |
| 141 | + ("[::1]:", ("::1", 0, 0, 0)), |
| 142 | + ("[::1]", ("::1", 0, 0, 0)), |
| 143 | + ] |
| 144 | +) |
| 145 | +def test_address_should_parse_ipv6(test_input, expected): |
| 146 | + # python -m pytest tests/unit/test_addressing.py -s -k test_address_should_parse_ipv6 |
| 147 | + parsed = Address.parse(test_input) |
| 148 | + assert parsed == expected |
| 149 | + |
| 150 | + |
| 151 | +@pytest.mark.parametrize( |
| 152 | + "test_input, expected", |
| 153 | + [ |
| 154 | + (None, TypeError), |
| 155 | + (123, TypeError), |
| 156 | + (("127.0.0.1", 7687), TypeError), |
| 157 | + (("[::1]", 7687, 1, 2), TypeError), |
| 158 | + (Address(("127.0.0.1", 7687)), TypeError), |
| 159 | + ] |
| 160 | +) |
| 161 | +def test_address_parse_with_invalid_input(test_input, expected): |
| 162 | + # python -m pytest tests/unit/test_addressing.py -s -k test_address_parse_with_invalid_input |
| 163 | + with pytest.raises(expected): |
| 164 | + parsed = Address.parse(test_input) |
| 165 | + |
| 166 | + |
| 167 | +@pytest.mark.parametrize( |
| 168 | + "test_input, expected", |
| 169 | + [ |
| 170 | + (("localhost:7687 [::1]:7687",), 2), |
| 171 | + (("localhost:7687", "[::1]:7687"), 2), |
| 172 | + (("localhost:7687 localhost:7688", "[::1]:7687"), 3), |
| 173 | + (("localhost:7687 localhost:7687", "[::1]:7687"), 3), |
| 174 | + ] |
| 175 | +) |
| 176 | +def test_address_parse_list(test_input, expected): |
| 177 | + # python -m pytest tests/unit/test_addressing.py -s -k test_address_parse_list |
| 178 | + addresses = Address.parse_list(*test_input) |
| 179 | + assert len(addresses) == expected |
| 180 | + |
| 181 | + |
| 182 | +@pytest.mark.parametrize( |
| 183 | + "test_input, expected", |
| 184 | + [ |
| 185 | + (("localhost:7687", None), TypeError), |
| 186 | + (("localhost:7687", 123), TypeError), |
| 187 | + (("localhost:7687", ("127.0.0.1", 7687)), TypeError), |
| 188 | + (("localhost:7687", ("[::1]", 7687, 1, 2)), TypeError), |
| 189 | + (("localhost:7687", Address(("127.0.0.1", 7687))), TypeError), |
| 190 | + ] |
| 191 | +) |
| 192 | +def test_address_parse_list_with_invalid_input(test_input, expected): |
| 193 | + # python -m pytest tests/unit/test_addressing.py -s -k test_address_parse_list_with_invalid_input |
| 194 | + with pytest.raises(TypeError): |
| 195 | + addresses = Address.parse_list(*test_input) |
| 196 | + |
| 197 | + |
| 198 | +def test_address_resolve(): |
| 199 | + # python -m pytest tests/unit/test_addressing.py -s -k test_address_resolve |
| 200 | + address = Address(("127.0.0.1", 7687)) |
| 201 | + resolved = address.resolve() |
| 202 | + assert isinstance(resolved, Address) is False |
| 203 | + assert isinstance(resolved, list) is True |
| 204 | + assert len(resolved) == 1 |
| 205 | + assert resolved[0] == IPv4Address(('127.0.0.1', 7687)) |
| 206 | + |
| 207 | + |
| 208 | +def test_address_resolve_with_custom_resolver_none(): |
| 209 | + # python -m pytest tests/unit/test_addressing.py -s -k test_address_resolve_with_custom_resolver_none |
| 210 | + address = Address(("127.0.0.1", 7687)) |
| 211 | + resolved = address.resolve(resolver=None) |
| 212 | + assert isinstance(resolved, Address) is False |
| 213 | + assert isinstance(resolved, list) is True |
| 214 | + assert len(resolved) == 1 |
| 215 | + assert resolved[0] == IPv4Address(('127.0.0.1', 7687)) |
| 216 | + |
| 217 | + |
| 218 | +@pytest.mark.parametrize( |
| 219 | + "test_input, expected", |
| 220 | + [ |
| 221 | + (Address(("127.0.0.1", "abcd")), ValueError), |
| 222 | + (Address((None, None)), ValueError), |
| 223 | + ] |
| 224 | +) |
| 225 | +def test_address_resolve_with_unresolvable_address(test_input, expected): |
| 226 | + # python -m pytest tests/unit/test_addressing.py -s -k test_address_resolve_with_unresolvable_address |
| 227 | + with pytest.raises(expected): |
| 228 | + test_input.resolve(resolver=None) |
| 229 | + |
| 230 | + |
| 231 | +def test_address_resolve_with_custom_resolver(): |
| 232 | + # python -m pytest tests/unit/test_addressing.py -s -k test_address_resolve_with_custom_resolver |
| 233 | + custom_resolver = lambda a: [("127.0.0.1", 7687), ("localhost", 1234)] |
| 234 | + |
| 235 | + address = Address(("127.0.0.1", 7687)) |
| 236 | + resolved = address.resolve(resolver=custom_resolver) |
| 237 | + assert isinstance(resolved, Address) is False |
| 238 | + assert isinstance(resolved, list) is True |
| 239 | + assert len(resolved) == 3 |
| 240 | + assert resolved[0] == IPv4Address(('127.0.0.1', 7687)) |
| 241 | + assert resolved[1] == IPv6Address(('::1', 1234, 0, 0)) |
| 242 | + assert resolved[2] == IPv4Address(('127.0.0.1', 1234)) |
| 243 | + |
| 244 | + |
| 245 | +def test_graphdatabase_parse_routing_context(): |
| 246 | + # python -m pytest tests/unit/test_addressing.py -s -k test_graphdatabase_parse_routing_context |
| 247 | + context = GraphDatabase._parse_routing_context(query="name=molly&color=white") |
| 248 | + assert context == {"name": "molly", "color": "white"} |
| 249 | + |
| 250 | + |
| 251 | +def test_graphdatabase_parse_routing_context_should_error_when_value_missing(): |
| 252 | + # python -m pytest tests/unit/test_addressing.py -s -k test_graphdatabase_parse_routing_context_should_error_when_value_missing |
| 253 | + with pytest.raises(ValueError): |
| 254 | + GraphDatabase._parse_routing_context("name=&color=white") |
| 255 | + |
| 256 | + |
| 257 | +def test_graphdatabase_parse_routing_context_should_error_when_key_duplicate(): |
| 258 | + # python -m pytest tests/unit/test_addressing.py -s -k test_graphdatabase_parse_routing_context_should_error_when_key_duplicate |
| 259 | + with pytest.raises(ValueError): |
| 260 | + GraphDatabase._parse_routing_context("name=molly&name=white") |
0 commit comments