Add missing sync and new async integration tests

.github/workflows/build-push-to-main.yaml

@@ -52,10 +52,7 @@ jobs:
       run: uv run mypy
     - name: Run unit-tests
       run: |
-        uv run coverage run -m unittest discover -v ./tests
-        for dir in ext/*/tests; do
-            uv run coverage run -a -m unittest discover -v "./$dir"
-        done
+        uv run coverage run -m unittest discover -v .
         uv run coverage run -a -m pytest -m "not e2e" ./ext/dapr-ext-workflow/tests/durabletask/
         uv run coverage xml
     - name: Upload test coverage

.github/workflows/build-tag.yaml

@@ -52,10 +52,7 @@ jobs:
       run: uv run mypy
     - name: Run unit-tests
       run: |
-        uv run coverage run -m unittest discover -v ./tests
-        for dir in ext/*/tests; do
-            uv run coverage run -a -m unittest discover -v "./$dir"
-        done
+        uv run coverage run -m unittest discover -v .
         uv run coverage run -a -m pytest -m "not e2e" ./ext/dapr-ext-workflow/tests/durabletask/
         uv run coverage xml
     - name: Upload test coverage

.github/workflows/run-tests.yaml

@@ -1,4 +1,4 @@
-name: validate-examples
+name: run-tests
 
 on:
   push:
@@ -86,7 +86,7 @@ jobs:
       - name: Install uv
         uses: astral-sh/setup-uv@v5
       - name: Install dependencies
-        run: uv sync --frozen --all-packages --group examples
+        run: uv sync --frozen --all-packages --group tests
       - name: Set up Dapr CLI
         uses: dapr/.github/.github/actions/setup-dapr-cli@main
         with:
@@ -103,9 +103,9 @@ jobs:
           nohup ollama serve &
           sleep 10
           ollama pull llama3.2:latest
-      - name: Check examples
-        run: |
-          uv run pytest tests/examples/
       - name: Run integration tests
         run: |
           uv run pytest tests/integration/
+      - name: Validate examples
+        run: |
+          uv run pytest tests/examples/

.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.

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

@@ -71,14 +71,17 @@ dev = [
     "opentelemetry-exporter-zipkin~=1.11.1",
     "httpx~=0.28.1",
     "pyOpenSSL~=26.0.0",
+    "cryptography>=42.0.0",
+    "redis>=7.4.0",
     "Flask~=3.1.3",
     "pytest~=9.0.2",
+    "pytest-asyncio>=0.23",
     "ruff==0.14.1",
     "python-dotenv~=1.2.2",
     "pydantic~=2.13.3",
     "PyYAML~=6.0.3",
 ]
-examples = [
+tests = [
     {include-group = "dev"},
     "mechanical-markdown~=0.8.0",
     "langchain-ollama~=1.0.1",
@@ -142,3 +145,4 @@ markers = [
     'example_dir(name): set the example directory for the dapr fixture',
 ]
 pythonpath = ["."]
+asyncio_mode = "auto"

tests/conftest.py

@@ -0,0 +1,76 @@
+import subprocess
+from typing import Callable, Iterator
+
+import pytest
+
+from tests.ollama_utils import DEFAULT_MODEL, model_available, ollama_ready
+from tests.process_utils import get_kwargs_for_process_group, terminate_process_group
+from tests.wait_utils import wait_until
+
+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[[str, str, int], 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
+
+
+@pytest.fixture(scope='session')
+def ollama() -> Iterator[None]:
+    """Ensure an Ollama server with the default model is running for the session."""
+    started: subprocess.Popen[str] | None = None
+    try:
+        if not ollama_ready():
+            try:
+                started = subprocess.Popen(
+                    ['ollama', 'serve'],
+                    stdout=subprocess.DEVNULL,
+                    stderr=subprocess.DEVNULL,
+                    text=True,
+                    **get_kwargs_for_process_group(),
+                )
+            except FileNotFoundError as exc:
+                pytest.fail(f'ollama CLI is not installed: {exc}')
+            wait_until(ollama_ready, timeout=30.0, interval=0.5)
+
+        if not model_available(DEFAULT_MODEL):
+            subprocess.run(['ollama', 'pull', DEFAULT_MODEL], check=True, capture_output=True)
+
+        yield
+    finally:
+        if started and started.poll() is None:
+            terminate_process_group(started)
+            try:
+                started.wait(timeout=10)
+            except subprocess.TimeoutExpired:
+                terminate_process_group(started, force=True)
+                started.wait()

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 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,