Replace sleep() with polls when possible

Signed-off-by: Sergio Herrera <627709+seherv@users.noreply.github.com>

Author: seherv

CLAUDE.md

@@ -1,6 +1,7 @@
 @AGENTS.md
 
 Use pathlib instead of os.path.
+Use httpx instead of urllib.
 Use modern Python (3.10+) features.
 Make all code strongly typed.
 Keep conditional nesting to a minimum, and use guard clauses when possible.

tests/integration/conftest.py

@@ -4,13 +4,16 @@
 import time
 from contextlib import contextmanager
 from pathlib import Path
-from typing import Any, Generator, Iterator
+from typing import Any, Callable, Generator, Iterator, TypeVar
 
+import httpx
 import pytest
 
 from dapr.clients import DaprClient
 from dapr.conf import settings
 
+T = TypeVar('T')
+
 INTEGRATION_DIR = Path(__file__).resolve().parent
 COMPONENTS_DIR = INTEGRATION_DIR / 'components'
 APPS_DIR = INTEGRATION_DIR / 'apps'
@@ -38,7 +41,6 @@ def start_sidecar(
         app_port: int | None = None,
         app_cmd: str | None = None,
         components: Path | None = None,
-        wait: int = 5,
     ) -> DaprClient:
         """Start a Dapr sidecar and return a connected DaprClient.
 
@@ -50,7 +52,6 @@ def start_sidecar(
             app_cmd: Shell command to start alongside the sidecar.
             components: Path to component YAML directory.  Defaults to
                 ``tests/integration/components/``.
-            wait: Seconds to sleep after launching (before the SDK health check).
         """
         resources = components or self._default_components
 
@@ -82,18 +83,22 @@ def start_sidecar(
             )
         self._processes.append(proc)
 
-        # Give the sidecar a moment to bind its ports before the SDK health
-        # check starts hitting the HTTP endpoint.
-        time.sleep(wait)
-
         # Point the SDK health check at the actual sidecar HTTP port.
         # DaprHealth.wait_for_sidecar() reads settings.DAPR_HTTP_PORT, which
         # is initialized once at import time and won't reflect a non-default
-        # http_port unless we update it here.
+        # http_port unless we update it here. The DaprClient constructor
+        # polls /healthz/outbound on this port, so we don't need to sleep first.
         settings.DAPR_HTTP_PORT = http_port
 
         client = DaprClient(address=f'127.0.0.1:{grpc_port}')
         self._clients.append(client)
+
+        # /healthz/outbound (polled by DaprClient) only checks sidecar-side
+        # readiness. When we launched an app alongside the sidecar, also wait
+        # for /v1.0/healthz so invoke_method et al. don't race the app's server.
+        if app_cmd is not None:
+            _wait_for_app_health(http_port)
+
         return client
 
     def cleanup(self) -> None:
@@ -114,15 +119,68 @@ def cleanup(self) -> None:
         self._log_files.clear()
 
 
+def _wait_until(
+    predicate: Callable[[], T | None],
+    timeout: float = 10.0,
+    interval: float = 0.1,
+) -> T:
+    """Poll `predicate` until it returns a truthy value. eaises `TimeoutError` if it never does."""
+    deadline = time.monotonic() + timeout
+    while True:
+        result = predicate()
+        if result:
+            return result
+        if time.monotonic() >= deadline:
+            raise TimeoutError(f'wait_until timed out after {timeout}s')
+        time.sleep(interval)
+
+
+def _wait_for_app_health(http_port: int, timeout: float = 30.0) -> None:
+    """Poll Dapr's app-facing /v1.0/healthz endpoint until it returns 2xx.
+
+    ``/v1.0/healthz`` requires the app behind the sidecar to be reachable,
+    unlike ``/v1.0/healthz/outbound`` which only checks sidecar readiness.
+    """
+    url = f'http://127.0.0.1:{http_port}/v1.0/healthz'
+
+    def _check() -> bool:
+        try:
+            response = httpx.get(url, timeout=2.0)
+        except httpx.HTTPError:
+            return False
+        return response.is_success
+
+    _wait_until(_check, timeout=timeout, interval=0.2)
+
+
 @contextmanager
-def _preserve_http_port() -> Iterator[None]:
-    # start_sidecar() mutates settings.DAPR_HTTP_PORT.
-    # This restores the original value so it does not leak across test modules.
-    original = settings.DAPR_HTTP_PORT
+def _isolate_dapr_settings() -> Iterator[None]:
+    """Pin SDK HTTP settings to the local test sidecar for the duration.
+
+    ``DaprHealth.get_api_url()`` consults three settings (see
+    ``dapr/clients/http/helpers.py``):
+
+    - ``DAPR_HTTP_ENDPOINT``, if set, wins and bypasses host/port entirely.
+    - ``DAPR_RUNTIME_HOST`` is the host component of the fallback URL.
+    - ``DAPR_HTTP_PORT`` is the port component of the fallback URL.
+
+    Any of these may be populated from the developer's environment (the Dapr
+    CLI sets them); without an override the SDK health check could target the
+    wrong sidecar. All three are snapshotted and restored so the test's
+    mutations don't leak across modules either.
+    """
+    originals = {
+        'DAPR_HTTP_ENDPOINT': settings.DAPR_HTTP_ENDPOINT,
+        'DAPR_RUNTIME_HOST': settings.DAPR_RUNTIME_HOST,
+        'DAPR_HTTP_PORT': settings.DAPR_HTTP_PORT,
+    }
+    settings.DAPR_HTTP_ENDPOINT = None
+    settings.DAPR_RUNTIME_HOST = '127.0.0.1'
     try:
         yield
     finally:
-        settings.DAPR_HTTP_PORT = original
+        for name, value in originals.items():
+            setattr(settings, name, value)
 
 
 @pytest.fixture(scope='module')
@@ -133,14 +191,20 @@ def dapr_env() -> Generator[DaprTestEnvironment, Any, None]:
     avoiding port conflicts from rapid start/stop cycles and cutting total
     test time significantly.
     """
-    with _preserve_http_port():
+    with _isolate_dapr_settings():
         env = DaprTestEnvironment()
         try:
             yield env
         finally:
             env.cleanup()
 
 
+@pytest.fixture
+def wait_until() -> Callable[..., Any]:
+    """Returns the ``_wait_until(predicate, timeout=10, interval=0.1)`` helper."""
+    return _wait_until
+
+
 @pytest.fixture(scope='module')
 def apps_dir() -> Path:
     return APPS_DIR

tests/integration/test_configuration.py

@@ -86,6 +86,5 @@ def test_unsubscribe_returns_true(self, client):
             keys=['cfg-unsub-key'],
             handler=lambda _id, _resp: None,
         )
-        time.sleep(1)
         ok = client.unsubscribe_configuration(store_name=STORE, id=sub_id)
         assert ok

tests/integration/test_pubsub.py

@@ -1,6 +1,5 @@
 import json
 import subprocess
-import time
 import uuid
 
 import pytest
@@ -35,11 +34,10 @@ def client(dapr_env, apps_dir):
         grpc_port=50001,
         app_port=50051,
         app_cmd=f'python3 {apps_dir / "pubsub_subscriber.py"}',
-        wait=10,
     )
 
 
-def test_published_messages_are_received_by_subscriber(client):
+def test_published_messages_are_received_by_subscriber(client, wait_until):
     run_id = uuid.uuid4().hex
     for n in range(1, 4):
         client.publish_event(
@@ -48,14 +46,14 @@ def test_published_messages_are_received_by_subscriber(client):
             data=json.dumps({'run_id': run_id, 'id': n, 'message': 'hello world'}),
             data_content_type='application/json',
         )
-        time.sleep(1)
-
-    time.sleep(3)
 
     for n in range(1, 4):
-        state = client.get_state(store_name=STORE, key=f'received-{run_id}-{n}')
-        assert state.data != b'', f'Subscriber did not receive message {n}'
-        msg = json.loads(state.data)
+        key = f'received-{run_id}-{n}'
+        data = wait_until(
+            lambda k=key: client.get_state(store_name=STORE, key=k).data or None,
+            timeout=10,
+        )
+        msg = json.loads(data)
         assert msg['id'] == n
         assert msg['message'] == 'hello world'