Replace fixed test ports with free port helper

tests/durabletask/__init__.py

@@ -0,0 +1,2 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.

tests/durabletask/_port_utils.py

@@ -0,0 +1,33 @@
+# 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 free TCP port by binding to port 0 and reading the assignment.
+
+    Probes IPv6 loopback first to match the backend's ``[::]`` bind (so an
+    IPv6-occupied port isn't wrongly reported free), falling back to IPv4.
+    """
+    if socket.has_ipv6:
+        try:
+            with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s:
+                s.bind(("::1", 0))
+                return s.getsockname()[1]
+        except OSError:
+            pass  # IPv6 not usable on this host; fall back to IPv4.
+
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("127.0.0.1", 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 tests.durabletask._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 tests.durabletask._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 tests.durabletask._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 tests.durabletask._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 tests.durabletask._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 tests.durabletask._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 tests.durabletask._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 tests.durabletask._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 tests.durabletask._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]}"

tests/durabletask/test_orchestration_async_e2e.py

@@ -9,13 +9,16 @@
 from durabletask import client, task, worker
 from durabletask.testing import create_test_backend
 
-HOST = "localhost:50060"
+from tests.durabletask._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 testing."""
-    b = create_test_backend(port=50060)
+    b = create_test_backend(port=PORT)
     yield b
     b.stop()
     b.reset()

tests/durabletask/test_orchestration_e2e.py

@@ -13,13 +13,16 @@
 import durabletask.history as history
 from durabletask.testing import create_test_backend
 
-HOST = "localhost:50054"
+from tests.durabletask._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 testing."""
-    b = create_test_backend(port=50054)
+    b = create_test_backend(port=PORT)
     yield b
     b.stop()
     b.reset()

tests/durabletask/test_orchestration_versioning_e2e.py

@@ -8,13 +8,16 @@
 from durabletask import client, task, worker
 from durabletask.testing import create_test_backend
 
-HOST = "localhost:50055"
+from tests.durabletask._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 testing."""
-    b = create_test_backend(port=50055)
+    b = create_test_backend(port=PORT)
     yield b
     b.stop()
     b.reset()

tests/durabletask/test_work_item_filters_e2e.py

@@ -18,13 +18,16 @@
 )
 from durabletask.testing import create_test_backend
 
-HOST = "localhost:50060"
+from tests.durabletask._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 testing."""
-    b = create_test_backend(port=50060)
+    b = create_test_backend(port=PORT)
     yield b
     b.stop()
     b.reset()