|
6 | 6 |
|
7 | 7 | from __future__ import annotations |
8 | 8 |
|
| 9 | +from ipaddress import ip_address |
9 | 10 | import socket |
10 | 11 | from unittest.mock import MagicMock, patch |
11 | 12 |
|
@@ -51,14 +52,39 @@ def block_all_network(request: pytest.FixtureRequest): |
51 | 52 | return |
52 | 53 |
|
53 | 54 | orig_connect = socket.socket.connect |
| 55 | + orig_getaddrinfo = socket.getaddrinfo |
| 56 | + |
| 57 | + def is_allowed_host(host) -> bool: |
| 58 | + if host is None: |
| 59 | + return True |
| 60 | + if isinstance(host, bytes): |
| 61 | + try: |
| 62 | + host = host.decode("ascii") |
| 63 | + except UnicodeDecodeError: |
| 64 | + return False |
| 65 | + if host == "localhost": |
| 66 | + return True |
| 67 | + try: |
| 68 | + address = ip_address(host) |
| 69 | + except (TypeError, ValueError): |
| 70 | + return False |
| 71 | + return address.is_loopback or address.is_unspecified |
54 | 72 |
|
55 | 73 | def guarded_connect(self, address, *args, **kwargs): |
56 | 74 | host = address[0] if isinstance(address, (tuple, list)) else address |
57 | | - if host in ("127.0.0.1", "localhost", "::1"): |
| 75 | + if is_allowed_host(host): |
58 | 76 | return orig_connect(self, address, *args, **kwargs) |
59 | 77 | raise RuntimeError(f"Blocked unmocked external network connection to {address} during test execution.") |
60 | 78 |
|
61 | | - with patch.object(socket.socket, "connect", guarded_connect): |
| 79 | + def guarded_getaddrinfo(host, port, *args, **kwargs): |
| 80 | + if is_allowed_host(host): |
| 81 | + return orig_getaddrinfo(host, port, *args, **kwargs) |
| 82 | + raise RuntimeError(f"Blocked unmocked external network connection to {(host, port)} during test execution.") |
| 83 | + |
| 84 | + with ( |
| 85 | + patch.object(socket, "getaddrinfo", guarded_getaddrinfo), |
| 86 | + patch.object(socket.socket, "connect", guarded_connect), |
| 87 | + ): |
62 | 88 | yield |
63 | 89 |
|
64 | 90 |
|
|
0 commit comments