From c37c9046ee8578e8db9cedfac3e9eff21f3ea316 Mon Sep 17 00:00:00 2001 From: Albert Callarisa Date: Thu, 16 Jul 2026 08:56:20 +0200 Subject: [PATCH] Pin ports in tests and wait for dapr to exit Signed-off-by: Albert Callarisa --- examples/AGENTS.md | 1 + tests/examples/conftest.py | 176 +++++++++--------- tests/examples/test_dapr_runner.py | 164 +++++++++------- tests/examples/test_demo_actor_grpc.py | 2 +- tests/examples/test_distributed_lock.py | 1 - tests/examples/test_error_handling.py | 1 - tests/examples/test_metadata.py | 1 - tests/integration/AGENTS.md | 12 +- tests/integration/apps/invoke_receiver.py | 2 +- tests/integration/apps/pubsub_subscriber.py | 2 +- tests/integration/conftest.py | 31 ++- tests/integration/test_actor_grpc.py | 2 +- tests/integration/test_configuration_async.py | 2 +- tests/integration/test_conversation_async.py | 2 +- tests/integration/test_crypto_async.py | 2 +- .../test_distributed_lock_async.py | 2 +- tests/integration/test_invoke.py | 4 +- tests/integration/test_invoke_async.py | 4 +- .../integration/test_invoke_binding_async.py | 2 +- tests/integration/test_jobs_async.py | 2 +- tests/integration/test_metadata_async.py | 2 +- tests/integration/test_pubsub.py | 4 +- tests/integration/test_pubsub_async.py | 4 +- tests/integration/test_secret_store_async.py | 2 +- tests/integration/test_state_store_async.py | 2 +- tests/port_utils.py | 68 +++++++ tests/test_port_utils.py | 37 ++++ 27 files changed, 349 insertions(+), 185 deletions(-) create mode 100644 tests/port_utils.py create mode 100644 tests/test_port_utils.py diff --git a/examples/AGENTS.md b/examples/AGENTS.md index b5df9aa04..f098a2b10 100644 --- a/examples/AGENTS.md +++ b/examples/AGENTS.md @@ -8,6 +8,7 @@ The `examples/` directory serves as the **user-facing documentation**. Each exam 2. Tests use a `DaprRunner` helper (defined in `tests/examples/conftest.py`) that wraps `dapr run` commands 3. `DaprRunner.run()` executes a command and captures stdout; `DaprRunner.start()`/`stop()` manage background services 4. Tests assert that expected output lines appear in the captured output +5. The runner pins every sidecar listener port the test didn't set itself and blocks until those ports are bindable before launching. Test-pinned ports must stay below 32768 — see the "Port allocation" section in `tests/integration/AGENTS.md` for the full rationale. Run examples locally (requires a running Dapr runtime via `dapr init`): diff --git a/tests/examples/conftest.py b/tests/examples/conftest.py index aec84838c..b1e8bf579 100644 --- a/tests/examples/conftest.py +++ b/tests/examples/conftest.py @@ -8,16 +8,26 @@ import pytest +from tests.port_utils import SidecarPorts, wait_for_ports_free from tests.process_utils import get_kwargs_for_process_group, terminate_process_group REPO_ROOT = Path(__file__).resolve().parent.parent.parent EXAMPLES_DIR = REPO_ROOT / 'examples' -DAPR_PORT_BIND_FAILURE_MARKERS = ( - 'bind: address already in use', - 'failed to start internal gRPC server: could not listen on any endpoint', -) DAPR_SIDECAR_READY_MARKER = "You're up and running!" +# A DaprRunner has at most one background (start/stop) and one foreground +# (run) sidecar alive at a time, so two fixed port blocks cover every test. +FOREGROUND_PORTS = SidecarPorts(http=13601, grpc=13602, internal_grpc=13603, metrics=13604) +BACKGROUND_PORTS = SidecarPorts(http=13611, grpc=13612, internal_grpc=13613, metrics=13614) + +PORT_FLAGS = ( + '--dapr-http-port', + '--dapr-grpc-port', + '--dapr-internal-grpc-port', + '--metrics-port', + '--app-port', +) + def pytest_configure(config: pytest.Config) -> None: config.addinivalue_line('markers', 'example_dir(name): set the example directory for a test') @@ -33,19 +43,65 @@ def __init__(self, cwd: Path) -> None: @staticmethod def _terminate(proc: subprocess.Popen[str]) -> None: - if proc.poll() is not None: - return + """Terminates a ``dapr run`` invocation and everything it spawned. - terminate_process_group(proc) - try: - proc.wait(timeout=10) - except subprocess.TimeoutExpired: - terminate_process_group(proc, force=True) - proc.wait() + The group is force-killed even after a clean CLI exit: daprd shuts + down gracefully after the CLI and there is no value in waiting for + it in tests — the port gate in ``_prepare_launch`` protects the next + sidecar against anything the sweep misses. + """ + if proc.poll() is None: + terminate_process_group(proc) + try: + proc.wait(timeout=10) + except subprocess.TimeoutExpired: + pass + terminate_process_group(proc, force=True) + proc.wait() + + @staticmethod + def _with_pinned_ports(args: str, ports: SidecarPorts) -> list[str]: + """Builds the ``dapr run`` argv, pinning any listener port the caller left unset. + + Random CLI ports are never used: the CLI's picker can hand the same + port to two listeners of one daprd (see "Port allocation" in + tests/integration/AGENTS.md). + """ + tokens = shlex.split(args) + separator = tokens.index('--') if '--' in tokens else len(tokens) + dapr_tokens = tokens[:separator] + + def is_set(flag: str) -> bool: + return any(token == flag or token.startswith(f'{flag}=') for token in dapr_tokens) + + additions = [ + token + for flag, port in ports.as_flags().items() + if not is_set(flag) + for token in (flag, str(port)) + ] + return [*dapr_tokens, *additions, *tokens[separator:]] @staticmethod - def _is_dapr_port_bind_failure(output: str) -> bool: - return all(marker in output for marker in DAPR_PORT_BIND_FAILURE_MARKERS) + def _listener_ports(tokens: list[str]) -> list[int]: + """Extracts every port the sidecar and its app will bind from the argv.""" + separator = tokens.index('--') if '--' in tokens else len(tokens) + dapr_tokens = tokens[:separator] + + ports: list[int] = [] + for index, token in enumerate(dapr_tokens): + for flag in PORT_FLAGS: + if token == flag: + ports.append(int(dapr_tokens[index + 1])) + elif token.startswith(f'{flag}='): + ports.append(int(token.split('=', 1)[1])) + return ports + + def _prepare_launch(self, args: str, ports: SidecarPorts) -> list[str]: + """Pins the sidecar's ports and blocks until all of them are bindable.""" + tokens = self._with_pinned_ports(args, ports) + wait_for_ports_free(self._listener_ports(tokens)) + return tokens def run( self, @@ -53,7 +109,6 @@ def run( *, timeout: int = 30, until: list[str] | None = None, - port_bind_retries: int = 3, ) -> str: """Run a foreground command, block until it finishes, and return output. @@ -61,30 +116,15 @@ def run( own). For long-lived background services, use ``start()``/``stop()``. Args: - args: Arguments passed to ``dapr run``. + args: Arguments passed to ``dapr run``. Listener ports not set + here are pinned from ``FOREGROUND_PORTS``. timeout: Maximum seconds to wait before killing the process. until: If provided, the process is terminated as soon as every string in this list has appeared in the accumulated output. - port_bind_retries: Retry count for Dapr sidecar startup failures - caused by a transient random-port collision. """ - attempts = max(1, port_bind_retries + 1) - for attempt in range(attempts): - output = self._run_once(args, timeout=timeout, until=until) - if attempt < attempts - 1 and self._is_dapr_port_bind_failure(output): - print( - 'Dapr sidecar failed to bind a random port; ' - f'retrying startup after {2**attempt}s ' - f'(attempt {attempt + 1}/{attempts})', - flush=True, - ) - time.sleep(2**attempt) - continue - return output - - def _run_once(self, args: str, *, timeout: int, until: list[str] | None) -> str: + tokens = self._prepare_launch(args, FOREGROUND_PORTS) proc = subprocess.Popen( - args=('dapr', 'run', *shlex.split(args)), + args=('dapr', 'run', *tokens), cwd=self._cwd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, @@ -114,7 +154,7 @@ def _run_once(self, args: str, *, timeout: int, until: list[str] | None) -> str: return ''.join(lines) - def start(self, args: str, *, wait: int = 30, port_bind_retries: int = 3) -> None: + def start(self, args: str, *, wait: int = 30) -> None: """Start a long-lived background service. Use this for servers/subscribers that must stay alive while a second @@ -122,41 +162,25 @@ def start(self, args: str, *, wait: int = 30, port_bind_retries: int = 3) -> Non output. Stdout is written to a temp file to avoid pipe-buffer deadlocks. Args: - args: Arguments passed to ``dapr run``. + args: Arguments passed to ``dapr run``. Listener ports not set + here are pinned from ``BACKGROUND_PORTS``. wait: Maximum seconds to poll for sidecar readiness before proceeding anyway. - port_bind_retries: Retry count for Dapr sidecar startup failures - caused by a transient random-port collision. """ - attempts = max(1, port_bind_retries + 1) - for attempt in range(attempts): - output_file = tempfile.NamedTemporaryFile(mode='w+', suffix='.log') - proc = subprocess.Popen( - args=('dapr', 'run', *shlex.split(args)), - cwd=self._cwd, - stdout=output_file, - stderr=subprocess.STDOUT, - text=True, - **get_kwargs_for_process_group(), - ) - self._wait_until_ready(proc, output_file, timeout=wait) - - can_retry = attempt < attempts - 1 - if can_retry and self._started_with_port_bind_failure(proc, output_file): - self._terminate(proc) - output_file.close() - print( - 'Dapr background sidecar failed to bind a random port; ' - f'retrying startup after {2**attempt}s ' - f'(attempt {attempt + 1}/{attempts})', - flush=True, - ) - time.sleep(2**attempt) - continue - - self._bg_process = proc - self._bg_output_file = output_file - return + tokens = self._prepare_launch(args, BACKGROUND_PORTS) + output_file = tempfile.NamedTemporaryFile(mode='w+', suffix='.log') + proc = subprocess.Popen( + args=('dapr', 'run', *tokens), + cwd=self._cwd, + stdout=output_file, + stderr=subprocess.STDOUT, + text=True, + **get_kwargs_for_process_group(), + ) + self._wait_until_ready(proc, output_file, timeout=wait) + + self._bg_process = proc + self._bg_output_file = output_file @staticmethod def _wait_until_ready( @@ -164,9 +188,8 @@ def _wait_until_ready( ) -> None: """Polls the sidecar log every second until `dapr run` reports readiness. - Returns early when the process exits (``start`` then inspects the - output for a port-bind failure) and gives up after ``timeout`` seconds, - at which point the caller proceeds as if ready. + Returns early when the process exits and gives up after ``timeout`` + seconds, at which point the caller proceeds as if ready. The log is re-opened by name for each read: the ``output_file`` handle shares its offset with the child process, so seeking it directly would @@ -181,19 +204,6 @@ def _wait_until_ready( return time.sleep(1) - def _started_with_port_bind_failure( - self, proc: subprocess.Popen[str], output_file: IO[str] - ) -> bool: - """Whether a background sidecar already exited from a random-port collision. - - Reads the log only once the process has exited; seeking the temp file - while daprd is still writing would corrupt its shared append offset. - """ - if proc.poll() is None: - return False - output_file.seek(0) - return self._is_dapr_port_bind_failure(output_file.read()) - def stop(self) -> str: """Stop the background service and return its captured output.""" if self._bg_process is None: diff --git a/tests/examples/test_dapr_runner.py b/tests/examples/test_dapr_runner.py index d89736053..dc5dd8be2 100644 --- a/tests/examples/test_dapr_runner.py +++ b/tests/examples/test_dapr_runner.py @@ -5,7 +5,33 @@ import pytest -from tests.examples.conftest import DaprRunner +from tests.examples import conftest as examples_conftest +from tests.examples.conftest import BACKGROUND_PORTS, FOREGROUND_PORTS, DaprRunner + + +@pytest.fixture(autouse=True) +def no_real_process_groups(monkeypatch: pytest.MonkeyPatch) -> None: + """Fake processes have no real process group to signal.""" + monkeypatch.setattr( + examples_conftest, + 'terminate_process_group', + lambda proc, *, force=False: None, + ) + + +@pytest.fixture(autouse=True) +def ports_always_free(monkeypatch: pytest.MonkeyPatch) -> list[list[int]]: + """Stubs the port-free gate; returns the port lists it was called with.""" + gated: list[list[int]] = [] + monkeypatch.setattr( + examples_conftest, + 'wait_for_ports_free', + lambda ports, *, timeout=30.0: gated.append(list(ports)), + ) + return gated + + +SIDECAR_READY_OUTPUT = "✅ You're up and running! Both Dapr and your app logs will appear here.\n" class FakeProcess: @@ -23,9 +49,8 @@ def wait(self, timeout: int | None = None) -> int: class FakeBackgroundProcess: """Stand-in for a ``dapr run`` background process started via ``start()``. - A real background sidecar writes to the file object passed as ``stdout`` and - keeps running (``poll()`` returns ``None``); a sidecar that died on a port - bind has exited (``poll()`` returns a non-``None`` code). + A real background sidecar writes to the file object passed as ``stdout`` + and keeps running (``poll()`` returns ``None``). """ def __init__(self, output: str, returncode: int | None, stdout: IO[str]) -> None: @@ -40,105 +65,100 @@ def wait(self, timeout: int | None = None) -> int: return self._returncode or 0 -def test_run_retries_transient_dapr_port_bind_failure( - monkeypatch, tmp_path: Path, capsys: pytest.CaptureFixture[str] -) -> None: - outputs = [ - ( - 'level=error msg="Failed to listen for gRPC server on TCP address :33223 ' - 'with error: listen tcp :33223: bind: address already in use"\n' - 'level=fatal msg="Fatal error from runtime: failed to start internal gRPC ' - 'server: could not listen on any endpoint"\n' - ), - "{'secretKey': 'secretValue'}\n", - ] - popen_calls = [] +def test_run_pins_ports_the_caller_did_not_set(monkeypatch, tmp_path: Path) -> None: + argv_seen: list[tuple[str, ...]] = [] def fake_popen(*args, **kwargs) -> FakeProcess: - popen_calls.append((args, kwargs)) - return FakeProcess(outputs.pop(0)) + argv_seen.append(kwargs['args']) + return FakeProcess('done\n') monkeypatch.setattr(subprocess, 'Popen', fake_popen) - sleeps: list[int] = [] - monkeypatch.setattr(time, 'sleep', sleeps.append) - output = DaprRunner(tmp_path).run('--app-id=secretsapp -- python3 example.py', timeout=1) + DaprRunner(tmp_path).run('--app-id=metadata -- python3 app.py', timeout=1) - assert output == "{'secretKey': 'secretValue'}\n" - assert len(popen_calls) == 2 - assert sleeps == [1] - assert ( - 'Dapr sidecar failed to bind a random port; retrying startup after 1s' - in capsys.readouterr().out - ) + argv = argv_seen[0] + separator = argv.index('--') + dapr_argv = argv[:separator] + for flag, port in FOREGROUND_PORTS.as_flags().items(): + flag_position = dapr_argv.index(flag) + assert dapr_argv[flag_position + 1] == str(port) + assert list(argv[separator:]) == ['--', 'python3', 'app.py'] -def test_run_does_not_retry_non_port_bind_failure(monkeypatch, tmp_path: Path) -> None: - popen_calls = [] +def test_run_keeps_port_flags_the_caller_set(monkeypatch, tmp_path: Path) -> None: + argv_seen: list[tuple[str, ...]] = [] def fake_popen(*args, **kwargs) -> FakeProcess: - popen_calls.append((args, kwargs)) - return FakeProcess('application failed before printing expected output\n') + argv_seen.append(kwargs['args']) + return FakeProcess('done\n') monkeypatch.setattr(subprocess, 'Popen', fake_popen) - output = DaprRunner(tmp_path).run('--app-id=secretsapp -- python3 example.py', timeout=1) - - assert output == 'application failed before printing expected output\n' - assert len(popen_calls) == 1 - - -PORT_BIND_FAILURE_OUTPUT = ( - 'level=error msg="Failed to listen for gRPC server on TCP address :38779 ' - 'with error: listen tcp :38779: bind: address already in use"\n' - 'level=fatal msg="Fatal error from runtime: failed to start internal gRPC ' - 'server: could not listen on any endpoint"\n' -) + DaprRunner(tmp_path).run( + '--app-id pub --dapr-grpc-port=3500 --dapr-http-port 3501 -- python3 pub.py', + timeout=1, + ) + argv = argv_seen[0] + assert '--dapr-grpc-port=3500' in argv + assert argv.count('--dapr-http-port') == 1 + assert str(FOREGROUND_PORTS.grpc) not in argv + assert str(FOREGROUND_PORTS.http) not in argv + assert str(FOREGROUND_PORTS.internal_grpc) in argv + assert str(FOREGROUND_PORTS.metrics) in argv -SIDECAR_READY_OUTPUT = "✅ You're up and running! Both Dapr and your app logs will appear here.\n" +def test_run_waits_for_every_listener_port_before_launching(monkeypatch, tmp_path: Path) -> None: + events: list[str] = [] + gated_ports: list[int] = [] -def test_start_retries_transient_dapr_port_bind_failure( - monkeypatch, tmp_path: Path, capsys: pytest.CaptureFixture[str] -) -> None: - attempts = [ - (PORT_BIND_FAILURE_OUTPUT, 1), - ('INFO: Application startup complete.\n', None), - ] - popen_calls = [] + def fake_gate(ports, *, timeout=30.0) -> None: + events.append('gate') + gated_ports.extend(ports) - def fake_popen(*args, **kwargs) -> FakeBackgroundProcess: - popen_calls.append((args, kwargs)) - output, returncode = attempts.pop(0) - return FakeBackgroundProcess(output, returncode, kwargs['stdout']) + def fake_popen(*args, **kwargs) -> FakeProcess: + events.append('launch') + return FakeProcess('done\n') + monkeypatch.setattr(examples_conftest, 'wait_for_ports_free', fake_gate) monkeypatch.setattr(subprocess, 'Popen', fake_popen) - sleeps: list[int] = [] - monkeypatch.setattr(time, 'sleep', sleeps.append) - DaprRunner(tmp_path).start('--app-id demo-actor -- uvicorn demo:app', wait=0) + DaprRunner(tmp_path).run('--app-id recv --app-port 8088 -- python3 recv.py', timeout=1) - assert len(popen_calls) == 2 - assert sleeps == [1] - assert ( - 'Dapr background sidecar failed to bind a random port; retrying startup after 1s' - in capsys.readouterr().out - ) + assert events == ['gate', 'launch'] + assert sorted(gated_ports) == sorted([8088, *FOREGROUND_PORTS]) -def test_start_does_not_retry_non_port_bind_failure(monkeypatch, tmp_path: Path) -> None: - popen_calls = [] +def test_start_pins_the_background_port_block(monkeypatch, tmp_path: Path) -> None: + argv_seen: list[tuple[str, ...]] = [] def fake_popen(*args, **kwargs) -> FakeBackgroundProcess: - popen_calls.append((args, kwargs)) - return FakeBackgroundProcess('app crashed for an unrelated reason\n', 1, kwargs['stdout']) + argv_seen.append(kwargs['args']) + return FakeBackgroundProcess(SIDECAR_READY_OUTPUT, None, kwargs['stdout']) monkeypatch.setattr(subprocess, 'Popen', fake_popen) DaprRunner(tmp_path).start('--app-id demo-actor -- uvicorn demo:app', wait=0) - assert len(popen_calls) == 1 + argv = argv_seen[0] + for flag, port in BACKGROUND_PORTS.as_flags().items(): + flag_position = argv.index(flag) + assert argv[flag_position + 1] == str(port) + + +def test_terminate_sweeps_the_group_even_after_a_clean_exit( + monkeypatch: pytest.MonkeyPatch, +) -> None: + group_kills: list[bool] = [] + monkeypatch.setattr( + examples_conftest, + 'terminate_process_group', + lambda proc, *, force=False: group_kills.append(force), + ) + + DaprRunner._terminate(FakeProcess('dapr CLI already exited\n', returncode=0)) + + assert group_kills == [True] def test_start_returns_without_sleeping_when_sidecar_is_already_ready( diff --git a/tests/examples/test_demo_actor_grpc.py b/tests/examples/test_demo_actor_grpc.py index cfb02c6ed..e2a2bb8a8 100644 --- a/tests/examples/test_demo_actor_grpc.py +++ b/tests/examples/test_demo_actor_grpc.py @@ -17,7 +17,7 @@ from tests.actor_grpc_utils import actor_stream_supported -DAPR_GRPC_PORT = 50061 +DAPR_GRPC_PORT = 13561 EXPECTED_SERVICE = [ 'DemoActor is hosted over the Dapr gRPC actor stream', diff --git a/tests/examples/test_distributed_lock.py b/tests/examples/test_distributed_lock.py index 47e243c90..74fe6ba58 100644 --- a/tests/examples/test_distributed_lock.py +++ b/tests/examples/test_distributed_lock.py @@ -15,7 +15,6 @@ def test_distributed_lock(dapr): output = dapr.run( '--app-id=locksapp --app-protocol grpc --resources-path components/ -- python3 lock.py', - timeout=10, ) for line in EXPECTED_LINES: assert line in output, f'Missing in output: {line}' diff --git a/tests/examples/test_error_handling.py b/tests/examples/test_error_handling.py index 68c46af0e..77d4152e1 100644 --- a/tests/examples/test_error_handling.py +++ b/tests/examples/test_error_handling.py @@ -16,7 +16,6 @@ def test_error_handling(dapr): output = dapr.run( '--resources-path components -- python3 error_handling.py', - timeout=10, ) for line in EXPECTED_LINES: assert line in output, f'Missing in output: {line}' diff --git a/tests/examples/test_metadata.py b/tests/examples/test_metadata.py index fb9641342..badb143af 100644 --- a/tests/examples/test_metadata.py +++ b/tests/examples/test_metadata.py @@ -17,7 +17,6 @@ def test_metadata(dapr): output = dapr.run( '--app-id=my-metadata-app --app-protocol grpc --resources-path components/ ' '-- python3 app.py', - timeout=10, ) for line in EXPECTED_LINES: assert line in output, f'Missing in output: {line}' diff --git a/tests/integration/AGENTS.md b/tests/integration/AGENTS.md index 19a5b7330..2e4e77b4a 100644 --- a/tests/integration/AGENTS.md +++ b/tests/integration/AGENTS.md @@ -109,11 +109,17 @@ Async counterparts exercise `dapr.aio.clients.DaprClient` (the gRPC async client | `test_crypto_async.py` | `encrypt`, `decrypt` | | `test_conversation_async.py` | `converse_alpha1`, `converse_alpha2` | -Async tests use `pytest-asyncio` in auto mode (configured in `pyproject.toml`). Any `async def test_*` is run as a coroutine — no decorator required. The sidecar fixture stays sync (it just starts `dapr run`); each test creates a short-lived `async with AsyncDaprClient(address='127.0.0.1:50001') as d:` block. +Async tests use `pytest-asyncio` in auto mode (configured in `pyproject.toml`). Any `async def test_*` is run as a coroutine — no decorator required. The sidecar fixture stays sync (it just starts `dapr run`); each test creates a short-lived `async with AsyncDaprClient(address='127.0.0.1:13501') as d:` block. ## Port allocation -All sidecars default to gRPC port 50001 and HTTP port 3500. Since fixtures are module-scoped and tests run sequentially, only one sidecar is active at a time. If parallel execution is needed in the future, sidecars will need dynamic port allocation. +All sidecars default to gRPC port 13501, HTTP port 3500, internal gRPC port 13502, and metrics port 9091. Since fixtures are module-scoped and tests run sequentially, only one sidecar is active at a time. If parallel execution is needed in the future, sidecars will need dynamic port allocation. + +Every listener port is pinned deliberately: when one is left unset, the Dapr CLI picks a random free port, and its picker can assign the same port to two listeners of the same daprd — observed in CI across several pairs (metrics vs. internal gRPC, HTTP vs. gRPC, metrics vs. HTTP). Usually that kills the sidecar at startup with `bind: address already in use`; in the metrics-vs-HTTP case the sidecar even passes the CLI readiness check while serving Prometheus text on the API port, so the failure is silent. + +All pinned ports (sidecar and app alike) must sit **below the OS ephemeral source-port range** — Linux hands out 32768–60999 for outbound connections, macOS and Windows 49152+. Any process's outbound localhost socket can land on a port in that range and block daprd from binding it (observed in CI when the internal gRPC port was pinned to 50002: an unrelated established socket held it for over a minute). This is why the suite uses the 135xx block instead of Dapr's conventional 500xx ports. (The examples suite pins the same way via `DaprRunner`, using 136xx blocks.) + +`start_sidecar()` additionally waits until every pinned port is bindable (`tests/port_utils.py`) before launching `dapr run`, so a previous sidecar draining past its teardown costs a short wait instead of a startup bind failure. ## Helper apps @@ -140,6 +146,6 @@ Some building blocks (invoke, pubsub) require an app process running alongside t - **`localsecretstore.yaml` uses a relative path** (`secrets.json`) resolved against `cwd=INTEGRATION_DIR`. Same pattern applies to `localbinding.yaml` (`./.binding-data`) and `cryptostore.yaml` (`./keys`). - **`bindings.localstorage` refuses to initialize if `rootPath` does not exist** — `conftest.py` creates `.binding-data/` at import time so every sidecar can load the component. - **`statestore.yaml` has `actorStateStore: "true"`** because workflow uses the actor runtime. The flag is additive — regular state tests are unaffected. -- **Workflow tests run the `WorkflowRuntime` in-process** and connect to the sidecar's gRPC port (default 50001). No external app is needed. +- **Workflow tests run the `WorkflowRuntime` in-process** and connect to the sidecar's gRPC port (default 13501). No external app is needed. - **Dapr may normalize response fields** — e.g., `content_type` may lose charset parameters when proxied through gRPC. Assert on the media type prefix, not the full string. - **Error shapes vary** — `invoke_binding` surfaces sidecar errors as raw `grpc.RpcError`, while other APIs (jobs, state) wrap them in `DaprGrpcError`. Match what the method actually raises. diff --git a/tests/integration/apps/invoke_receiver.py b/tests/integration/apps/invoke_receiver.py index 41592eb0e..2f2b9b4e3 100644 --- a/tests/integration/apps/invoke_receiver.py +++ b/tests/integration/apps/invoke_receiver.py @@ -10,4 +10,4 @@ def my_method(request: InvokeMethodRequest) -> InvokeMethodResponse: return InvokeMethodResponse(b'INVOKE_RECEIVED', 'text/plain; charset=UTF-8') -app.run(50051) +app.run(13503) diff --git a/tests/integration/apps/pubsub_subscriber.py b/tests/integration/apps/pubsub_subscriber.py index 23defc9c2..38e6fdb4f 100644 --- a/tests/integration/apps/pubsub_subscriber.py +++ b/tests/integration/apps/pubsub_subscriber.py @@ -19,4 +19,4 @@ def handle_topic_a(event: SubscriptionMessage) -> TopicEventResponse: return TopicEventResponse('success') -app.run(50051) +app.run(13503) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 1b683dc83..3ea3c7eb1 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -10,6 +10,7 @@ from dapr.clients import DaprClient from dapr.conf import settings from tests.crypto_utils import remove_test_keys, write_test_keys +from tests.port_utils import wait_for_ports_free from tests.process_utils import get_kwargs_for_process_group, terminate_process_group from tests.wait_utils import wait_until @@ -37,18 +38,27 @@ def start_sidecar( self, app_id: str, *, - grpc_port: int = 50001, + grpc_port: int = 13501, http_port: int = 3500, + internal_grpc_port: int = 13502, + metrics_port: int = 9091, app_port: int | None = None, app_cmd: str | None = None, resources: Path | None = None, ) -> DaprClient: """Start a Dapr sidecar and return a connected DaprClient. + Every listener port is pinned and startup blocks until each one is + bindable. See the "Port allocation" section in AGENTS.md for the full + rationale (CLI random-port picker bug, ephemeral-range port theft, + teardown races). + Args: app_id: Dapr application ID. grpc_port: Sidecar gRPC port. http_port: Sidecar HTTP port (also used for the SDK health check). + internal_grpc_port: Sidecar-to-sidecar gRPC port. + metrics_port: Sidecar metrics port. app_port: Port the app listens on (implies ``--app-protocol grpc``). app_cmd: Shell command to start alongside the sidecar. resources: Path to resources YAML directory. Defaults to @@ -67,12 +77,19 @@ def start_sidecar( str(grpc_port), '--dapr-http-port', str(http_port), + '--dapr-internal-grpc-port', + str(internal_grpc_port), + '--metrics-port', + str(metrics_port), ] + gated_ports = [grpc_port, http_port, internal_grpc_port, metrics_port] if app_port is not None: cmd.extend(['--app-port', str(app_port), '--app-protocol', 'grpc']) + gated_ports.append(app_port) if app_cmd is not None: cmd.extend(['--', *shlex.split(app_cmd)]) + wait_for_ports_free(gated_ports) proc = subprocess.Popen( cmd, cwd=INTEGRATION_DIR, @@ -100,6 +117,13 @@ def start_sidecar( return client def cleanup(self) -> None: + """Closes clients and takes down every sidecar process group. + + Each group is force-killed even after a clean CLI exit: daprd shuts + down gracefully after the CLI and there is no value in waiting for + it in tests — the port gate in ``start_sidecar`` protects the next + sidecar against anything the sweep misses. + """ for client in self.clients: client.close() self.clients.clear() @@ -110,8 +134,9 @@ def cleanup(self) -> None: try: proc.wait(timeout=10) except subprocess.TimeoutExpired: - terminate_process_group(proc, force=True) - proc.wait() + pass + terminate_process_group(proc, force=True) + proc.wait() self.processes.clear() diff --git a/tests/integration/test_actor_grpc.py b/tests/integration/test_actor_grpc.py index 23d9964d7..ac3d86a09 100644 --- a/tests/integration/test_actor_grpc.py +++ b/tests/integration/test_actor_grpc.py @@ -33,7 +33,7 @@ from dapr.actor import ActorId, ActorProxy from tests.actor_grpc_utils import actor_stream_supported -GRPC_PORT = 50031 +GRPC_PORT = 13531 HTTP_PORT = 3531 APP_PORT = 9131 diff --git a/tests/integration/test_configuration_async.py b/tests/integration/test_configuration_async.py index d520fb3d2..ca194fea6 100644 --- a/tests/integration/test_configuration_async.py +++ b/tests/integration/test_configuration_async.py @@ -18,7 +18,7 @@ from dapr.aio.clients import DaprClient as AsyncDaprClient STORE = 'configurationstore' -GRPC_ADDRESS = '127.0.0.1:50001' +GRPC_ADDRESS = '127.0.0.1:13501' @pytest.fixture(scope='module') diff --git a/tests/integration/test_conversation_async.py b/tests/integration/test_conversation_async.py index 32fbe8b46..7d6141d13 100644 --- a/tests/integration/test_conversation_async.py +++ b/tests/integration/test_conversation_async.py @@ -23,7 +23,7 @@ ) COMPONENT = 'echo' -GRPC_ADDRESS = '127.0.0.1:50001' +GRPC_ADDRESS = '127.0.0.1:13501' @pytest.fixture(scope='module') diff --git a/tests/integration/test_crypto_async.py b/tests/integration/test_crypto_async.py index c20fa754d..7486454f8 100644 --- a/tests/integration/test_crypto_async.py +++ b/tests/integration/test_crypto_async.py @@ -18,7 +18,7 @@ from dapr.aio.clients import DaprClient as AsyncDaprClient from dapr.clients.grpc._crypto import DecryptOptions, EncryptOptions -GRPC_ADDRESS = '127.0.0.1:50001' +GRPC_ADDRESS = '127.0.0.1:13501' CRYPTO_COMPONENT = 'cryptostore' RSA_KEY = 'rsa-private-key.pem' SYMMETRIC_KEY = 'symmetric-key-256' diff --git a/tests/integration/test_distributed_lock_async.py b/tests/integration/test_distributed_lock_async.py index cbb511d05..059c8039b 100644 --- a/tests/integration/test_distributed_lock_async.py +++ b/tests/integration/test_distributed_lock_async.py @@ -19,7 +19,7 @@ from dapr.clients.grpc._response import UnlockResponseStatus STORE = 'lockstore' -GRPC_ADDRESS = '127.0.0.1:50001' +GRPC_ADDRESS = '127.0.0.1:13501' # The distributed lock API re-emits the alpha warnings on every test run. pytestmark = pytest.mark.filterwarnings('ignore::UserWarning') diff --git a/tests/integration/test_invoke.py b/tests/integration/test_invoke.py index 45abdcdcb..e2fdd4c68 100644 --- a/tests/integration/test_invoke.py +++ b/tests/integration/test_invoke.py @@ -5,8 +5,8 @@ def client(dapr_env, apps_dir): return dapr_env.start_sidecar( app_id='invoke-receiver', - grpc_port=50001, - app_port=50051, + grpc_port=13501, + app_port=13503, app_cmd=f'python3 {apps_dir / "invoke_receiver.py"}', ) diff --git a/tests/integration/test_invoke_async.py b/tests/integration/test_invoke_async.py index c9e8fc690..241df704e 100644 --- a/tests/integration/test_invoke_async.py +++ b/tests/integration/test_invoke_async.py @@ -17,14 +17,14 @@ from dapr.aio.clients import DaprClient as AsyncDaprClient -GRPC_ADDRESS = '127.0.0.1:50001' +GRPC_ADDRESS = '127.0.0.1:13501' @pytest.fixture(scope='module') def sidecar(dapr_env, apps_dir): dapr_env.start_sidecar( app_id='invoke-receiver-async', - app_port=50051, + app_port=13503, app_cmd=f'python3 {apps_dir / "invoke_receiver.py"}', ) diff --git a/tests/integration/test_invoke_binding_async.py b/tests/integration/test_invoke_binding_async.py index 90fdaf1ce..ad27fe462 100644 --- a/tests/integration/test_invoke_binding_async.py +++ b/tests/integration/test_invoke_binding_async.py @@ -22,7 +22,7 @@ BINDING = 'localbinding' BINDING_ROOT = Path(__file__).resolve().parent / '.binding-data' -GRPC_ADDRESS = '127.0.0.1:50001' +GRPC_ADDRESS = '127.0.0.1:13501' @pytest.fixture(scope='module') diff --git a/tests/integration/test_jobs_async.py b/tests/integration/test_jobs_async.py index 1433345c5..10cf15b8a 100644 --- a/tests/integration/test_jobs_async.py +++ b/tests/integration/test_jobs_async.py @@ -22,7 +22,7 @@ from dapr.clients.exceptions import DaprGrpcError from tests.naming_utils import unique_name -GRPC_ADDRESS = '127.0.0.1:50001' +GRPC_ADDRESS = '127.0.0.1:13501' # The jobs API re-emits the alpha warnings on every test run. pytestmark = pytest.mark.filterwarnings('ignore::UserWarning') diff --git a/tests/integration/test_metadata_async.py b/tests/integration/test_metadata_async.py index 24f95c2f0..24c7978ed 100644 --- a/tests/integration/test_metadata_async.py +++ b/tests/integration/test_metadata_async.py @@ -17,7 +17,7 @@ from dapr.aio.clients import DaprClient as AsyncDaprClient -GRPC_ADDRESS = '127.0.0.1:50001' +GRPC_ADDRESS = '127.0.0.1:13501' @pytest.fixture(scope='module') diff --git a/tests/integration/test_pubsub.py b/tests/integration/test_pubsub.py index 8c07d5ae8..9985b62b2 100644 --- a/tests/integration/test_pubsub.py +++ b/tests/integration/test_pubsub.py @@ -24,8 +24,8 @@ def _fetch_received(client: DaprClient, key: str) -> bytes | None: def client(dapr_env, apps_dir, flush_redis): return dapr_env.start_sidecar( app_id='test-subscriber', - grpc_port=50001, - app_port=50051, + grpc_port=13501, + app_port=13503, app_cmd=f'python3 {apps_dir / "pubsub_subscriber.py"}', ) diff --git a/tests/integration/test_pubsub_async.py b/tests/integration/test_pubsub_async.py index 119814f89..79c988d34 100644 --- a/tests/integration/test_pubsub_async.py +++ b/tests/integration/test_pubsub_async.py @@ -24,7 +24,7 @@ STORE = 'statestore' PUBSUB = 'pubsub' TOPIC = 'TOPIC_A' -GRPC_ADDRESS = '127.0.0.1:50001' +GRPC_ADDRESS = '127.0.0.1:13501' async def _fetch_received(d: AsyncDaprClient, key: str) -> bytes | None: @@ -36,7 +36,7 @@ async def _fetch_received(d: AsyncDaprClient, key: str) -> bytes | None: def sidecar(dapr_env, apps_dir, flush_redis): dapr_env.start_sidecar( app_id='test-subscriber-async', - app_port=50051, + app_port=13503, app_cmd=f'python3 {apps_dir / "pubsub_subscriber.py"}', ) diff --git a/tests/integration/test_secret_store_async.py b/tests/integration/test_secret_store_async.py index 8f6b7537b..acea0bd63 100644 --- a/tests/integration/test_secret_store_async.py +++ b/tests/integration/test_secret_store_async.py @@ -18,7 +18,7 @@ from dapr.aio.clients import DaprClient as AsyncDaprClient STORE = 'localsecretstore' -GRPC_ADDRESS = '127.0.0.1:50001' +GRPC_ADDRESS = '127.0.0.1:13501' @pytest.fixture(scope='module') diff --git a/tests/integration/test_state_store_async.py b/tests/integration/test_state_store_async.py index 37f714dd4..5814b5fff 100644 --- a/tests/integration/test_state_store_async.py +++ b/tests/integration/test_state_store_async.py @@ -20,7 +20,7 @@ from tests.naming_utils import unique_name STORE = 'statestore' -GRPC_ADDRESS = '127.0.0.1:50001' +GRPC_ADDRESS = '127.0.0.1:13501' @pytest.fixture(scope='module') diff --git a/tests/port_utils.py b/tests/port_utils.py new file mode 100644 index 000000000..0e2b64867 --- /dev/null +++ b/tests/port_utils.py @@ -0,0 +1,68 @@ +""" +Sidecar ports in tests are always pinned, never left to the Dapr CLI: the +CLI's random free-port picker can hand the same port to two listeners of one +daprd, and pinned values must sit below the OS ephemeral source-port range +(32768+ on Linux, 49152+ on macOS/Windows) or any process's outbound localhost +connection can steal them from daprd. See tests/integration/AGENTS.md. +""" + +from __future__ import annotations + +import socket +import sys +from typing import Iterable, NamedTuple + +from tests.wait_utils import wait_until + + +class SidecarPorts(NamedTuple): + """One sidecar's pinned listener ports, keyed by their ``dapr run`` flag.""" + + http: int + grpc: int + internal_grpc: int + metrics: int + + def as_flags(self) -> dict[str, int]: + return { + '--dapr-http-port': self.http, + '--dapr-grpc-port': self.grpc, + '--dapr-internal-grpc-port': self.internal_grpc, + '--metrics-port': self.metrics, + } + + +def _can_bind(port: int) -> bool: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + if sys.platform != 'win32': + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + sock.bind(('', port)) + except OSError: + return False + return True + + +def wait_for_ports_free(ports: Iterable[int], *, timeout: float = 30.0) -> None: + """Blocks until every port accepts a fresh bind. + + Gates sidecar startup on the one invariant that matters: the ports it is + about to bind are actually free. A previous test's daprd or app draining + past its teardown then costs a short wait here instead of a fatal + "bind: address already in use" inside the new sidecar. SO_REUSEADDR + mirrors daprd's own (Go) bind semantics, so TIME_WAIT does not count as + busy. + + Raises: + TimeoutError: listing the ports still held after ``timeout`` seconds. + """ + pending = list(dict.fromkeys(ports)) + + def _all_bindable() -> bool: + return all(_can_bind(port) for port in pending) + + try: + wait_until(_all_bindable, timeout=timeout, interval=0.2) + except TimeoutError: + ports_busy = [port for port in pending if not _can_bind(port)] + raise TimeoutError(f'Ports still in use after {timeout}s: {ports_busy}') from None diff --git a/tests/test_port_utils.py b/tests/test_port_utils.py new file mode 100644 index 000000000..55ff5ff5c --- /dev/null +++ b/tests/test_port_utils.py @@ -0,0 +1,37 @@ +import socket +from contextlib import closing + +import pytest + +from tests.port_utils import SidecarPorts, wait_for_ports_free + + +def _listener() -> socket.socket: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.bind(('', 0)) + sock.listen(1) + return sock + + +def test_wait_returns_once_ports_are_free() -> None: + with closing(_listener()) as sock: + port = sock.getsockname()[1] + wait_for_ports_free([port], timeout=5) + + +def test_wait_times_out_while_a_listener_holds_the_port() -> None: + with closing(_listener()) as sock: + port = sock.getsockname()[1] + with pytest.raises(TimeoutError, match=str(port)): + wait_for_ports_free([port], timeout=0.3) + + +def test_sidecar_ports_map_to_dapr_run_flags() -> None: + ports = SidecarPorts(http=13601, grpc=13602, internal_grpc=13603, metrics=13604) + + assert ports.as_flags() == { + '--dapr-http-port': 13601, + '--dapr-grpc-port': 13602, + '--dapr-internal-grpc-port': 13603, + '--metrics-port': 13604, + }