Replace fixed test ports with free port helper

Author: andystaples

tests/durabletask/_port_utils.py

@@ -0,0 +1,27 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Shared test helpers for picking network ports.
+
+Tests start an in-memory backend that binds a TCP port. Hard-coding
+fixed ports causes intermittent failures on Windows because Hyper-V (and
+other components) reserve large, dynamic ranges of TCP ports. Asking the
+OS for a free port avoids those collisions.
+"""
+
+from __future__ import annotations
+
+import socket
+
+
+def find_free_port() -> int:
+    """Return a TCP port number that is currently free.
+
+    Binds a socket to port 0 so the OS assigns an available port, then
+    releases it and returns the chosen number. There is an inherent race
+    between releasing the socket and the caller binding the port, but in
+    practice it is reliable for tests and far safer than fixed ports.
+    """
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("localhost", 0))
+        return s.getsockname()[1]

tests/durabletask/entities/test_class_based_entities_e2e.py

@@ -12,13 +12,16 @@
 from durabletask import client, entities, task, worker
 from durabletask.testing import create_test_backend
 
-HOST = "localhost:50059"
+from _port_utils import find_free_port
+
+PORT = find_free_port()
+HOST = f"localhost:{PORT}"
 
 
 @pytest.fixture(autouse=True)
 def backend():
     """Create an in-memory backend for entity testing."""
-    b = create_test_backend(port=50059)
+    b = create_test_backend(port=PORT)
     yield b
     b.stop()
     b.reset()

tests/durabletask/entities/test_entity_failure_handling.py

@@ -12,13 +12,16 @@
 from durabletask import client, entities, task, worker
 from durabletask.testing import create_test_backend
 
-HOST = "localhost:50057"
+from _port_utils import find_free_port
+
+PORT = find_free_port()
+HOST = f"localhost:{PORT}"
 
 
 @pytest.fixture(autouse=True)
 def backend():
     """Create an in-memory backend for entity testing."""
-    b = create_test_backend(port=50057)
+    b = create_test_backend(port=PORT)
     yield b
     b.stop()
     b.reset()

tests/durabletask/entities/test_function_based_entities_e2e.py

@@ -12,13 +12,16 @@
 from durabletask import client, entities, task, worker
 from durabletask.testing import create_test_backend
 
-HOST = "localhost:50056"
+from _port_utils import find_free_port
+
+PORT = find_free_port()
+HOST = f"localhost:{PORT}"
 
 
 @pytest.fixture(autouse=True)
 def backend():
     """Create an in-memory backend for entity testing."""
-    b = create_test_backend(port=50056)
+    b = create_test_backend(port=PORT)
     yield b
     b.stop()
     b.reset()

tests/durabletask/extensions/history_export/test_activities.py

@@ -36,8 +36,9 @@
 )
 from durabletask.testing import create_test_backend
 
+from _port_utils import find_free_port
 
-PORT = 50261
+PORT = find_free_port()
 HOST = f"localhost:{PORT}"
 
 

tests/durabletask/extensions/history_export/test_client.py

@@ -35,9 +35,9 @@
 from durabletask.testing import create_test_backend
 
 from ._test_helpers import wait_until
+from _port_utils import find_free_port
 
-
-PORT = 50263
+PORT = find_free_port()
 HOST = f"localhost:{PORT}"
 
 

tests/durabletask/extensions/history_export/test_entity.py

@@ -31,9 +31,9 @@
 from durabletask.testing import create_test_backend
 
 from ._test_helpers import wait_until
+from _port_utils import find_free_port
 
-
-PORT = 50260
+PORT = find_free_port()
 HOST = f"localhost:{PORT}"
 
 _WINDOW_START = datetime(2025, 1, 1, tzinfo=timezone.utc)

tests/durabletask/extensions/history_export/test_orchestrator.py

@@ -33,9 +33,9 @@
 from durabletask.testing import create_test_backend
 
 from ._test_helpers import wait_until
+from _port_utils import find_free_port
 
-
-PORT = 50262
+PORT = find_free_port()
 HOST = f"localhost:{PORT}"
 
 

tests/durabletask/test_batch_actions.py

@@ -15,7 +15,9 @@
 from durabletask.testing import create_test_backend
 from durabletask.worker import TaskHubGrpcWorker
 
-BATCH_TEST_PORT = 50058
+from _port_utils import find_free_port
+
+BATCH_TEST_PORT = find_free_port()
 HOST = f"localhost:{BATCH_TEST_PORT}"
 
 

tests/durabletask/test_large_payload_e2e.py

@@ -22,6 +22,8 @@
 from durabletask import client, task, worker
 from durabletask.testing import create_test_backend
 
+from _port_utils import find_free_port
+
 # Skip the entire module if azure-storage-blob is not installed.
 azure_blob = pytest.importorskip("azure.storage.blob")
 
@@ -30,8 +32,8 @@
 # Azurite well-known connection string
 AZURITE_CONN_STR = "UseDevelopmentStorage=true"
 
-HOST = "localhost:50070"
-BACKEND_PORT = 50070
+BACKEND_PORT = find_free_port()
+HOST = f"localhost:{BACKEND_PORT}"
 
 # Use a unique container per test run to avoid collisions.
 TEST_CONTAINER = f"e2e-payloads-{uuid.uuid4().hex[:8]}"