Add missing sync and new async integration tests

.gitignore

@@ -136,3 +136,9 @@ coverage.lcov
 
 # macOS specific files
 .DS_Store
+
+# Integration test scratch dirs
+tests/integration/.binding-data/
+
+# Crypto test material generated at test time (see tests/crypto_utils.py)
+tests/integration/keys/

CLAUDE.md

@@ -1,14 +1,20 @@
 @AGENTS.md
 
-Use pathlib instead of os.path.
-Use httpx instead of urllib.
-subprocess(`shell=True`) is used only when it makes the code more readable. Use either shlex or args lists.
-subprocess calls should have a reasonable timeout.
-Use modern Python (3.10+) features.
 Make all code strongly typed.
 Keep conditional nesting to a minimum, and use guard clauses when possible.
-Aim for medium "visual complexity": use intermediate variables to store results of nested/complex function calls, but don't create a new variable for everything.
-Avoid comments unless there is a gotcha, a complex algorithm or anything an experienced code reviewer needs to be aware of. Focus on making better Google-style docstrings instead.
+Aim for medium visual complexity: use intermediate variables to store results of nested/complex function calls. A complex function call could be:
+- `f(Object(a=1, b=2, c=3))`, the inner object has more than 2 meaningful args
+- `f(Object((a, b)))`, 2 levels of nesting or anything with a long chain of closing parens
+- `small_transformation(ImportantObject())`, the object itself is the main subject of the function but the transformation steals the focus
+Use descriptive, self-documenting names for these intermediate variables.
+Closely related variable names should share a root and use different suffixes. For example, `request_original` and `request_clean`, but not `clean_request`.
+Avoid comments unless there is a gotcha, a complex algorithm or anything an experienced code reviewer needs to be aware of. Focus on making short but descriptive Google-style docstrings instead.
 
-The user is not always right. Be skeptical and do not blindly comply if something doesn't make sense.
+Use modern Python (3.10+) features.
+Use pathlib instead of os.path.
+Use httpx instead of urllib.
+`subprocess(shell=True)` is used only when it makes the code more readable. Use either shlex or args lists.
+Anything that can have an explicit timeout should have one.
 Code should be cross-platform and production ready.
+
+The user is not always right. Be skeptical and do not blindly comply if something doesn't make sense.

dev-requirements.txt

@@ -4,12 +4,16 @@ mypy-protobuf>=2.9
 tox>=4.3.0
 coverage>=5.3
 pytest>=7.0
+pytest-asyncio>=0.23
 wheel
 # used in unit test only
 opentelemetry-sdk
 opentelemetry-instrumentation-grpc
 httpx>=0.28.1
 pyOpenSSL>=26.0.0
+# used by tests to generate crypto keys at runtime
+cryptography>=42.0.0
+redis>=7.4.0
 # needed for type checking
 Flask>=1.1
 # needed for auto fix
@@ -20,3 +24,4 @@ python-dotenv>=1.2.2
 pydantic>=2.13.3
 # needed for yaml file generation in examples
 PyYAML>=6.0.3
+# needed for direct Redis access in integration tests

examples/pubsub-streaming-async/publisher.py

@@ -44,7 +44,7 @@ async def publish_events():
             )
 
             # Print the request
-            print(req_data, flush=True)
+            print(req_data)
 
             await asyncio.sleep(1)
 

examples/pubsub-streaming-async/subscriber.py

@@ -19,7 +19,7 @@ def process_message(message):
     global counter
     counter += 1
     # Process the message here
-    print(f'Processing message: {message.data()} from {message.topic()}...', flush=True)
+    print(f'Processing message: {message.data()} from {message.topic()}...')
     return 'success'
 
 

pyproject.toml

@@ -28,3 +28,4 @@ markers = [
     'example_dir(name): set the example directory for the dapr fixture',
 ]
 pythonpath = ["."]
+asyncio_mode = "auto"

tests/conftest.py

@@ -0,0 +1,40 @@
+import subprocess
+from typing import Callable
+
+import pytest
+
+REDIS_CONTAINER = 'dapr_redis'
+
+
+@pytest.fixture(scope='session')
+def flush_redis() -> None:
+    """Flush the ``dapr_redis`` container once per session."""
+    subprocess.run(
+        args=('docker', 'exec', REDIS_CONTAINER, 'redis-cli', 'FLUSHDB'),
+        check=True,
+        capture_output=True,
+        timeout=10,
+    )
+
+
+@pytest.fixture(scope='session')
+def redis_set_config() -> Callable[..., None]:
+    """Dapr encodes values in the config store as ``value||version``"""
+
+    def _set(key: str, value: str, version: int = 1) -> None:
+        subprocess.run(
+            args=(
+                'docker',
+                'exec',
+                REDIS_CONTAINER,
+                'redis-cli',
+                'SET',
+                key,
+                f'{value}||{version}',
+            ),
+            check=True,
+            capture_output=True,
+            timeout=10,
+        )
+
+    return _set

tests/crypto_utils.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+import secrets
+from pathlib import Path
+
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+
+RSA_KEY_FILENAME = 'rsa-private-key.pem'
+SYMMETRIC_KEY_FILENAME = 'symmetric-key-256'
+
+_RSA_KEY_SIZE = 4096
+_SYMMETRIC_KEY_BYTES = 32
+
+
+def write_test_keys(target_dir: Path) -> None:
+    """Write a fresh 4096-bit RSA private key (PKCS8 PEM) and a 256-bit AES key.
+
+    File names match those expected by ``examples/crypto/crypto.py`` and the
+    ``cryptostore.yaml`` component used by the integration tests.
+    """
+    target_dir.mkdir(parents=True, exist_ok=True)
+
+    rsa_key = rsa.generate_private_key(public_exponent=65537, key_size=_RSA_KEY_SIZE)
+    rsa_pem = rsa_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.PKCS8,
+        encryption_algorithm=serialization.NoEncryption(),
+    )
+    (target_dir / RSA_KEY_FILENAME).write_bytes(rsa_pem)
+    (target_dir / SYMMETRIC_KEY_FILENAME).write_bytes(secrets.token_bytes(_SYMMETRIC_KEY_BYTES))
+
+
+def remove_test_keys(target_dir: Path) -> None:
+    for name in (RSA_KEY_FILENAME, SYMMETRIC_KEY_FILENAME):
+        (target_dir / name).unlink(missing_ok=True)

tests/examples/conftest.py

@@ -8,7 +8,7 @@
 
 import pytest
 
-from tests._process_utils import get_kwargs_for_process_group, terminate_process_group
+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'

tests/examples/test_configuration.py

@@ -1,10 +1,7 @@
-import subprocess
 import time
 
 import pytest
 
-REDIS_CONTAINER = 'dapr_redis'
-
 EXPECTED_LINES = [
     'Got key=orderId1 value=100 version=1 metadata={}',
     'Got key=orderId2 value=200 version=1 metadata={}',
@@ -13,33 +10,17 @@
 ]
 
 
-@pytest.fixture()
-def redis_config():
-    """Seed configuration values in Redis before the test."""
-    subprocess.run(
-        ('docker', 'exec', 'dapr_redis', 'redis-cli', 'SET', 'orderId1', '100||1'),
-        check=True,
-        capture_output=True,
-    )
-    subprocess.run(
-        ('docker', 'exec', 'dapr_redis', 'redis-cli', 'SET', 'orderId2', '200||1'),
-        check=True,
-        capture_output=True,
-    )
-
-
 @pytest.mark.example_dir('configuration')
-def test_configuration(dapr, redis_config):
+def test_configuration(dapr, redis_set_config):
+    redis_set_config('orderId1', '100')
+    redis_set_config('orderId2', '200')
+
     dapr.start(
         '--app-id configexample --resources-path components/ -- python3 configuration.py',
         wait=5,
     )
     # Update Redis to trigger the subscription notification
-    subprocess.run(
-        ('docker', 'exec', 'dapr_redis', 'redis-cli', 'SET', 'orderId2', '210||2'),
-        check=True,
-        capture_output=True,
-    )
+    redis_set_config('orderId2', '210', version=2)
     # configuration.py sleeps 10s after subscribing before it unsubscribes.
     # Wait long enough for the full script to finish.
     time.sleep(10)

tests/examples/test_crypto.py

@@ -16,7 +16,7 @@
 
 
 @pytest.fixture()
-def crypto_artifacts():
+def cleanup_crypto_outputs():
     """Clean up output files written by the crypto example on teardown.
 
     Example RSA and AES keys are in ``examples/crypto/keys/``.
@@ -27,7 +27,7 @@ def crypto_artifacts():
 
 
 @pytest.mark.example_dir('crypto')
-def test_crypto(dapr, crypto_artifacts):
+def test_crypto(dapr, cleanup_crypto_outputs):
     output = dapr.run(
         '--app-id crypto --resources-path ./components/ -- python3 crypto.py',
         timeout=30,
@@ -38,7 +38,7 @@ def test_crypto(dapr, crypto_artifacts):
 
 
 @pytest.mark.example_dir('crypto')
-def test_crypto_async(dapr, crypto_artifacts):
+def test_crypto_async(dapr, cleanup_crypto_outputs):
     output = dapr.run(
         '--app-id crypto-async --resources-path ./components/ -- python3 crypto-async.py',
         timeout=30,

tests/examples/test_langgraph_checkpointer.py

@@ -1,9 +1,10 @@
 import subprocess
-import time
 
 import httpx
 import pytest
 
+from tests.wait_utils import wait_until
+
 OLLAMA_URL = 'http://localhost:11434'
 MODEL = 'llama3.2:latest'
 
@@ -31,15 +32,6 @@ def _model_available() -> bool:
     return any(m['name'] == MODEL for m in resp.json().get('models', []))
 
 
-def _wait_for_ollama(timeout: float = 30.0, interval: float = 0.5) -> None:
-    deadline = time.monotonic() + timeout
-    while time.monotonic() < deadline:
-        if _ollama_ready():
-            return
-        time.sleep(interval)
-    raise TimeoutError(f'ollama serve did not become ready within {timeout}s')
-
-
 @pytest.fixture()
 def ollama():
     """Ensure Ollama is running and the required model is pulled.
@@ -57,7 +49,7 @@ def ollama():
             )
         except FileNotFoundError:
             pytest.skip('ollama is not installed')
-        _wait_for_ollama()
+        wait_until(_ollama_ready, timeout=30.0, interval=0.5)
 
     if not _model_available():
         subprocess.run(['ollama', 'pull', MODEL], check=True, capture_output=True)
@@ -69,17 +61,6 @@ def ollama():
         started.wait(timeout=10)
 
 
-@pytest.fixture()
-def flush_redis():
-    """This test is not replayable if the checkpointer state store is not clean."""
-    subprocess.run(
-        ['docker', 'exec', 'dapr_redis', 'redis-cli', 'FLUSHDB'],
-        capture_output=True,
-        check=True,
-        timeout=10,
-    )
-
-
 @pytest.mark.example_dir('langgraph-checkpointer')
 def test_langgraph_checkpointer(dapr, ollama, flush_redis):
     output = dapr.run(

tests/examples/test_pubsub_streaming_async.py

@@ -6,7 +6,7 @@
     "Processing message: {'id': 3, 'message': 'hello world'} from TOPIC_B1...",
     "Processing message: {'id': 4, 'message': 'hello world'} from TOPIC_B1...",
     "Processing message: {'id': 5, 'message': 'hello world'} from TOPIC_B1...",
-    'Closing subscription...',
+    "Closing subscription...",
 ]
 
 EXPECTED_HANDLER_SUBSCRIBER = [
@@ -15,7 +15,7 @@
     "Processing message: {'id': 3, 'message': 'hello world'} from TOPIC_B2...",
     "Processing message: {'id': 4, 'message': 'hello world'} from TOPIC_B2...",
     "Processing message: {'id': 5, 'message': 'hello world'} from TOPIC_B2...",
-    'Closing subscription...',
+    "Closing subscription...",
 ]
 
 EXPECTED_PUBLISHER = [
@@ -30,12 +30,12 @@
 @pytest.mark.example_dir('pubsub-streaming-async')
 def test_pubsub_streaming_async(dapr):
     dapr.start(
-        '--app-id python-subscriber --app-protocol grpc -- python3 subscriber.py --topic=TOPIC_B1',
+        '--app-id python-subscriber --app-protocol grpc -- python3 -u subscriber.py --topic=TOPIC_B1',
         wait=5,
     )
     publisher_output = dapr.run(
         '--app-id python-publisher --app-protocol grpc --dapr-grpc-port=3500 '
-        '--enable-app-health-check -- python3 publisher.py --topic=TOPIC_B1',
+        '--enable-app-health-check -- python3 -u publisher.py --topic=TOPIC_B1',
         timeout=30,
     )
     for line in EXPECTED_PUBLISHER:
@@ -49,12 +49,12 @@ def test_pubsub_streaming_async(dapr):
 @pytest.mark.example_dir('pubsub-streaming-async')
 def test_pubsub_streaming_async_handler(dapr):
     dapr.start(
-        '--app-id python-subscriber --app-protocol grpc -- python3 subscriber-handler.py --topic=TOPIC_B2',
+        '--app-id python-subscriber --app-protocol grpc -- python3 -u subscriber-handler.py --topic=TOPIC_B2',
         wait=5,
     )
     publisher_output = dapr.run(
         '--app-id python-publisher --app-protocol grpc --dapr-grpc-port=3500 '
-        '--enable-app-health-check -- python3 publisher.py --topic=TOPIC_B2',
+        '--enable-app-health-check -- python3 -u publisher.py --topic=TOPIC_B2',
         timeout=30,
     )
     for line in EXPECTED_PUBLISHER: