Merge pull request #3611 from VWS-Python/cleanup-untyped-runtime-boundaries

Narrow local socket addresses before using their ports

Author: adamtheturtle

tests/mock_vws/test_docker.py

@@ -185,7 +185,11 @@ def _free_port() -> int:
         type=socket.SOCK_STREAM,
     ) as sock:
         sock.bind(("127.0.0.1", 0))
-        return int(sock.getsockname()[1])  # pyrefly: ignore [unknown-argument-type]
+        address = sock.getsockname()
+        assert isinstance(address, tuple)
+        assert isinstance(address[1], int)
+        port: int = address[1]
+        return port
 
 
 @beartype

tests/mock_vws/test_healthcheck.py

@@ -45,9 +45,10 @@ def _unused_port() -> int:
         type=socket.SOCK_STREAM,
     ) as sock:
         sock.bind(("localhost", 0))
-        # The socket stubs cannot specialize this result from the configured
-        # IPv4 address family, although the second tuple item is always a port.
-        port: int = sock.getsockname()[1]  # ty: ignore[unsound-assignment]
+        address = sock.getsockname()
+        assert isinstance(address, tuple)
+        assert isinstance(address[1], int)
+        port: int = address[1]
     return port
 
 

tests/mock_vws/test_httpx2_mock_usage.py

@@ -78,10 +78,12 @@ def _unused_local_url() -> str:
     Returns:
         The URL of a port which was free when this was called.
     """
-    sock = socket.socket()
-    sock.bind(("", 0))
-    port = sock.getsockname()[1]  # pyrefly: ignore [unknown-variable-type]
-    sock.close()
+    with socket.socket() as sock:
+        sock.bind(("", 0))
+        address = sock.getsockname()
+        assert isinstance(address, tuple)
+        assert isinstance(address[1], int)
+        port: int = address[1]
     return f"http://localhost:{port}"
 
 

tests/mock_vws/test_requests_mock_usage.py

@@ -124,10 +124,12 @@ def _bool_matcher(
 @beartype
 def _unused_local_url() -> str:
     """Return a URL for a local address with nothing listening on it."""
-    sock = socket.socket()
-    sock.bind(("", 0))
-    port = sock.getsockname()[1]  # pyrefly: ignore [unknown-variable-type]
-    sock.close()
+    with socket.socket() as sock:
+        sock.bind(("", 0))
+        address = sock.getsockname()
+        assert isinstance(address, tuple)
+        assert isinstance(address[1], int)
+        port: int = address[1]
     return f"http://localhost:{port}"
 
 
@@ -142,11 +144,7 @@ def request_unmocked_address() -> None:
             context of a ``responses`` mock which does not mock local
             addresses.
     """
-    sock = socket.socket()
-    sock.bind(("", 0))
-    port = sock.getsockname()[1]  # pyrefly: ignore [unknown-variable-type]
-    sock.close()
-    _ = requests.get(url=f"http://localhost:{port}", timeout=30)
+    _ = requests.get(url=_unused_local_url(), timeout=30)
 
 
 @beartype
@@ -2075,27 +2073,19 @@ def test_httpx_unmocked_address_blocked() -> None:
         """``MockVWS`` blocks ``httpx`` requests to non-Vuforia
         addresses.
         """
-        sock = socket.socket()
-        sock.bind(("", 0))
-        port = sock.getsockname()[1]  # pyrefly: ignore [unknown-variable-type]
-        sock.close()
         with MockVWS(), pytest.raises(expected_exception=httpx.ConnectError):
-            _ = httpx.get(url=f"http://localhost:{port}", timeout=30)
+            _ = httpx.get(url=_unused_local_url(), timeout=30)
 
     @staticmethod
     def test_httpx_real_http() -> None:
         """When ``real_http=True``, ``httpx`` requests to non-Vuforia
         addresses are not blocked.
         """
-        sock = socket.socket()
-        sock.bind(("", 0))
-        port = sock.getsockname()[1]  # pyrefly: ignore [unknown-variable-type]
-        sock.close()
         with (
             MockVWS(real_http=True),
             pytest.raises(expected_exception=httpx.ConnectError),
         ):
-            _ = httpx.get(url=f"http://localhost:{port}", timeout=30)
+            _ = httpx.get(url=_unused_local_url(), timeout=30)
 
 
 class TestHttpx2AlsoIntercepted:
@@ -2138,27 +2128,19 @@ def test_httpx2_unmocked_address_blocked() -> None:
         """``MockVWS`` blocks ``httpx2`` requests to non-Vuforia
         addresses.
         """
-        sock = socket.socket()
-        sock.bind(("", 0))
-        port = sock.getsockname()[1]  # pyrefly: ignore [unknown-variable-type]
-        sock.close()
         with MockVWS(), pytest.raises(expected_exception=httpx2.ConnectError):
-            _ = httpx2.get(url=f"http://localhost:{port}", timeout=30)
+            _ = httpx2.get(url=_unused_local_url(), timeout=30)
 
     @staticmethod
     def test_httpx2_real_http() -> None:
         """When ``real_http=True``, ``httpx2`` requests to non-Vuforia
         addresses are not blocked.
         """
-        sock = socket.socket()
-        sock.bind(("", 0))
-        port = sock.getsockname()[1]  # pyrefly: ignore [unknown-variable-type]
-        sock.close()
         with (
             MockVWS(real_http=True),
             pytest.raises(expected_exception=httpx2.ConnectError),
         ):
-            _ = httpx2.get(url=f"http://localhost:{port}", timeout=30)
+            _ = httpx2.get(url=_unused_local_url(), timeout=30)
 
 
 class TestModelTargetWebAPI: