From adc447522980471b26fd3d0c6fedf9c42ce39c84 Mon Sep 17 00:00:00 2001 From: Jennifer Linnenberg Date: Wed, 28 Oct 2026 10:29:24 +0100 Subject: [PATCH 1/3] fix: show test while running --- .github/workflows/integration.yml | 6 ++--- CONTRIBUTING.md | 2 +- tests/integration/conftest.py | 38 +++++++++++++++++++++++-------- 3 files changed, 31 insertions(+), 15 deletions(-) diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml index e782b6b..192ca73 100644 --- a/.github/workflows/integration.yml +++ b/.github/workflows/integration.yml @@ -45,7 +45,5 @@ jobs: - name: Run integration tests env: SA_KEY_B64: ${{ secrets.SA_KEY_B64 }} - PROJECT_ID: ${{ secrets.PROJECT_ID }} - TOKEN_URL: ${{ secrets.TOKEN_URL }} - REGION: ${{ secrets.REGION }} - run: uv run pytest tests/integration/ -v --timeout=700 --no-cov + STACKIT_PROJECT_ID: ${{ secrets.STACKIT_PROJECT_ID }} + run: uv run pytest tests/integration/ -v -s --timeout=700 --no-cov diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 4d5614a..45bfe9c 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -74,7 +74,7 @@ We maintain a high bar for reliability and test coverage: ```bash make test ``` - - Integration tests interact with STACKIT PubSub APIs and expect configuration variables (`SA_KEY_B64`, `PROJECT_ID`, `TOKEN_URL`, `REGION`). Run them with `uv run pytest tests/integration/ --no-cov`. + - Integration tests interact with STACKIT PubSub APIs and expect configuration variables (`SA_KEY_B64`, `STACKIT_PROJECT_ID`). Run them with `uv run pytest tests/integration/ --no-cov`. --- diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index fb41f8f..cfaf0a4 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -16,15 +16,15 @@ Requires the following secrets to be set: - SA_KEY_B64: Base64 encoded service account key JSON -- PROJECT_ID: STACKIT project ID -- TOKEN_URL: OAuth token endpoint -- REGION: API region (e.g. eu01) +- STACKIT_PROJECT_ID: STACKIT project ID """ from __future__ import annotations import base64 import contextlib +import hashlib +import json import os import time from pathlib import Path @@ -37,7 +37,15 @@ from pubsub_sdk_python.retry import with_retry SA_KEY_PATH = "sa-key.json" -SA_EMAIL = "break-test-jennifer-qzj3sci8@sa.stackit.cloud" + + +def pytest_report_header() -> list[str]: + """Print the SHA512 sum of STACKIT_PROJECT_ID in the pytest header.""" + pid = os.environ.get("STACKIT_PROJECT_ID", "") + if pid: + digest = hashlib.sha512(pid.encode("utf-8")).hexdigest() + return [f"STACKIT_PROJECT_ID SHA512: {digest}"] + return [] def _load_sa_key() -> str: @@ -340,6 +348,12 @@ def sa_key() -> str: return _load_sa_key() +@pytest.fixture(scope="session") +def sa_email(sa_key: str) -> str: + """Extract the service account email (credentials.iss) from SA_KEY.""" + return json.loads(sa_key)["credentials"]["iss"] + + @pytest.fixture(scope="session") def auth_config(sa_key: str) -> AuthConfig: """Create AuthConfig from SA_KEY.""" @@ -360,9 +374,11 @@ def control_plane_url() -> str: @pytest.fixture(scope="session") def project_id() -> str: """Get the project ID from environment.""" - pid = os.environ.get("PROJECT_ID") + pid = os.environ.get("STACKIT_PROJECT_ID") if not pid: - raise ValueError("PROJECT_ID environment variable is required") + raise ValueError("STACKIT_PROJECT_ID environment variable is required") + digest = hashlib.sha512(pid.encode("utf-8")).hexdigest() + print(f"STACKIT_PROJECT_ID SHA512: {digest}") return pid @@ -378,6 +394,7 @@ def topic_id( project_id: str, region: str, auth_config: AuthConfig, + sa_email: str, ): """Create a topic once for all tests in this run, then delete it.""" auth_header = _create_auth_header(auth_config) @@ -407,9 +424,9 @@ def topic_id( region=region, topic_id=new_topic_id, auth_header=auth_header, - email=SA_EMAIL, + email=sa_email, ) - print(f"Granted publisher access for {SA_EMAIL} on topic {new_topic_id}") + print(f"Granted publisher access for {sa_email} on topic {new_topic_id}") yield new_topic_id @@ -468,6 +485,7 @@ def subscription_id( region: str, topic_id: str, auth_config: AuthConfig, + sa_email: str, ): """Create a subscription once and reuse for all tests that need it.""" auth_header = _create_auth_header(auth_config) @@ -539,9 +557,9 @@ def subscription_id( topic_id=topic_id, subscription_id=sub_id, auth_header=auth_header, - email=SA_EMAIL, + email=sa_email, ) - print(f"Granted subscriber access for {SA_EMAIL} on subscription {sub_id}") + print(f"Granted subscriber access for {sa_email} on subscription {sub_id}") else: sub_id = existing_subs[0]["id"] From be162e6fa982ce28c4f02d0018e12fb8f2d9aebc Mon Sep 17 00:00:00 2001 From: Jennifer Linnenberg Date: Wed, 28 Oct 2026 10:50:22 +0100 Subject: [PATCH 2/3] fix: ruff & ty --- .github/workflows/ci.yml | 3 + .github/workflows/release.yml | 3 + Makefile | 11 +- pyproject.toml | 1 + src/pubsub_sdk_python/auth.py | 5 +- src/pubsub_sdk_python/publisher.py | 8 +- src/pubsub_sdk_python/retry.py | 13 ++- src/pubsub_sdk_python/subscriber.py | 8 +- tests/integration/conftest.py | 161 ++++++++++++++++++---------- tests/test_publisher.py | 41 +++---- tests/test_retry.py | 40 +++---- tests/test_subscriber.py | 41 +++---- uv.lock | 29 ++++- 13 files changed, 220 insertions(+), 144 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a59e7a6..7326f2e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -66,6 +66,9 @@ jobs: - name: Run ruff format check run: uv run ruff format --check src/ tests/ + - name: Run ty type check + run: uv run ty check + test: name: Unit Tests runs-on: ubuntu-latest diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 30b4960..5dcea07 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -46,6 +46,9 @@ jobs: - name: Run ruff format check run: uv run ruff format --check src/ tests/ + - name: Run ty type check + run: uv run ty check + - name: Run pytest with coverage run: uv run pytest --cov=src/pubsub_sdk_python --cov-branch --cov-report=term-missing --cov-fail-under=85 diff --git a/Makefile b/Makefile index cc93962..853c5a1 100644 --- a/Makefile +++ b/Makefile @@ -5,13 +5,16 @@ install: uv pip install -e . lint: - uvx ruff@0.16.4 check + uv run ruff check src/ tests/ format: - uvx ruff@0.16.4 format + uv run ruff format src/ tests/ -check: lint format - uvx ruff@0.16.4 format --check +typecheck: + uv run ty check + +check: lint format typecheck + uv run ruff format --check src/ tests/ test: check install uv run pytest diff --git a/pyproject.toml b/pyproject.toml index fbfcddc..5b0ff2e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,6 +35,7 @@ dev = [ "pytest-cov>=5.0.0", "pytest-timeout>=2.3.0", "ruff>=0.8.0", + "ty>=0.0.93", ] [tool.pytest.ini_options] diff --git a/src/pubsub_sdk_python/auth.py b/src/pubsub_sdk_python/auth.py index ab3fec7..1ccf74b 100644 --- a/src/pubsub_sdk_python/auth.py +++ b/src/pubsub_sdk_python/auth.py @@ -102,8 +102,9 @@ def _patched_fetch(self: KeyAuth) -> None: except requests.RequestException as e: raise requests.RequestException("Initial token fetch failed") from e - KeyAuth._KeyAuth__fetch_token_from_endpoint = _patched_fetch - KeyAuth._KeyAuth__patched = True + key_auth_any: Any = KeyAuth + key_auth_any._KeyAuth__fetch_token_from_endpoint = _patched_fetch + key_auth_any._KeyAuth__patched = True import jwt diff --git a/src/pubsub_sdk_python/publisher.py b/src/pubsub_sdk_python/publisher.py index b7ed233..2cad1e4 100644 --- a/src/pubsub_sdk_python/publisher.py +++ b/src/pubsub_sdk_python/publisher.py @@ -185,8 +185,12 @@ async def _request( # If we've exhausted retries if attempt >= self.retry_config.max_retries: + msg = ( + f"All {self.retry_config.max_retries} retry attempts " + f"exhausted for {path}: {exc}" + ) raise PubSubRetryError( - f"All {self.retry_config.max_retries} retry attempts exhausted for {path}", + msg, causes=[last_exception] if last_exception else None, ) from exc @@ -202,7 +206,7 @@ async def _request( ) await self._safe_sleep(delay) - raise last_exception # type: ignore[misc] + raise PubSubRetryError("Retry loop terminated unexpectedly") @staticmethod async def _safe_sleep(delay: float) -> None: diff --git a/src/pubsub_sdk_python/retry.py b/src/pubsub_sdk_python/retry.py index c9c4226..ee85fae 100644 --- a/src/pubsub_sdk_python/retry.py +++ b/src/pubsub_sdk_python/retry.py @@ -164,7 +164,7 @@ def wrapper(*args: Any, **kwargs: Any) -> T: raise if attempt >= config.max_retries: raise PubSubRetryError( - f"All {config.max_retries} retry attempts exhausted", + f"All {config.max_retries} retry attempts exhausted: {exc}", causes=[last_exception] if last_exception else None, ) from exc delay = _calculate_delay(config, attempt) @@ -172,13 +172,12 @@ def wrapper(*args: Any, **kwargs: Any) -> T: "Retry attempt %d/%d for %s after %.2fs: %s", attempt + 1, config.max_retries, - func.__name__, + getattr(func, "__name__", repr(func)), delay, exc, ) time.sleep(delay) - # Unreachable but mypy needs it - raise last_exception # type: ignore[misc] + raise PubSubRetryError("Retry loop terminated unexpectedly") return wrapper @@ -213,7 +212,7 @@ async def wrapper(*args: Any, **kwargs: Any) -> T: raise if attempt >= config.max_retries: raise PubSubRetryError( - f"All {config.max_retries} retry attempts exhausted", + f"All {config.max_retries} retry attempts exhausted: {exc}", causes=[last_exception] if last_exception else None, ) from exc delay = _calculate_delay(config, attempt) @@ -221,12 +220,12 @@ async def wrapper(*args: Any, **kwargs: Any) -> T: "Retry attempt %d/%d for %s after %.2fs: %s", attempt + 1, config.max_retries, - func.__name__, + getattr(func, "__name__", repr(func)), delay, exc, ) await asyncio.sleep(delay) - raise last_exception # type: ignore[misc] + raise PubSubRetryError("Retry loop terminated unexpectedly") return wrapper diff --git a/src/pubsub_sdk_python/subscriber.py b/src/pubsub_sdk_python/subscriber.py index 8ea47c2..b15da0a 100644 --- a/src/pubsub_sdk_python/subscriber.py +++ b/src/pubsub_sdk_python/subscriber.py @@ -207,8 +207,12 @@ async def _request( raise if attempt >= self.retry_config.max_retries: + msg = ( + f"All {self.retry_config.max_retries} retry attempts " + f"exhausted for {path}: {exc}" + ) raise PubSubRetryError( - f"All {self.retry_config.max_retries} retry attempts exhausted for {path}", + msg, causes=[last_exception] if last_exception else None, ) from exc @@ -224,7 +228,7 @@ async def _request( ) await self._safe_sleep(delay) - raise last_exception # type: ignore[misc] + raise PubSubRetryError("Retry loop terminated unexpectedly") @staticmethod async def _safe_sleep(delay: float) -> None: diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index cfaf0a4..dfcc80b 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -23,7 +23,6 @@ import base64 import contextlib -import hashlib import json import os import time @@ -32,20 +31,28 @@ import httpx import pytest -from pubsub_sdk_python.auth import AuthConfig +from pubsub_sdk_python.auth import AuthConfig, StackitAuthProvider from pubsub_sdk_python.client import PubSubClient from pubsub_sdk_python.retry import with_retry SA_KEY_PATH = "sa-key.json" -def pytest_report_header() -> list[str]: - """Print the SHA512 sum of STACKIT_PROJECT_ID in the pytest header.""" - pid = os.environ.get("STACKIT_PROJECT_ID", "") - if pid: - digest = hashlib.sha512(pid.encode("utf-8")).hexdigest() - return [f"STACKIT_PROJECT_ID SHA512: {digest}"] - return [] +def _log(msg: str) -> None: + """Print an integration test log message on its own line.""" + print(f"\n[integration] {msg}", flush=True) + + +def _format_error(exc: BaseException) -> str: + """Format an exception with its HTTP response body and underlying cause.""" + details = f"{type(exc).__name__}: {exc}" + if isinstance(exc, httpx.HTTPStatusError) and exc.response is not None: + body = exc.response.text.strip() + if body: + details += f" (response body: {body})" + if exc.__cause__ is not None: + details += f" | caused by {_format_error(exc.__cause__)}" + return details def _load_sa_key() -> str: @@ -77,12 +84,25 @@ def _get_control_plane_url() -> str: def _create_auth_header(auth_config: AuthConfig) -> dict[str, str]: """Create an auth header using the StackitAuthProvider.""" - from pubsub_sdk_python.auth import StackitAuthProvider - provider = StackitAuthProvider(auth_config) return provider.get_auth_header() +def _raise_for_status(response: httpx.Response) -> None: + """Raise HTTPStatusError with response body details if request failed.""" + try: + response.raise_for_status() + except httpx.HTTPStatusError as exc: + body = response.text.strip() + if body: + raise httpx.HTTPStatusError( + f"{exc} - response body: {body}", + request=exc.request, + response=exc.response, + ) from None + raise + + @with_retry def _create_topic( control_plane_url: str, @@ -109,7 +129,7 @@ def _create_topic( headers={**auth_header, "Content-Type": "application/json"}, timeout=30.0, ) - response.raise_for_status() + _raise_for_status(response) topic = response.json() return topic["id"] @@ -145,7 +165,7 @@ def _create_subscription( headers={**auth_header, "Content-Type": "application/json"}, timeout=30.0, ) - response.raise_for_status() + _raise_for_status(response) subscription = response.json() return subscription["id"] @@ -158,7 +178,7 @@ def _get_resource(url: str, auth_header: dict[str, str]) -> dict: headers={**auth_header, "Content-Type": "application/json"}, timeout=30.0, ) - response.raise_for_status() + _raise_for_status(response) return response.json() @@ -195,7 +215,7 @@ def _wait_for_resource_active( except httpx.TransportError as exc: # The resource may not be reachable yet right after creation # (e.g. backend routing not provisioned) - keep polling until timeout. - print(f"{resource_type} {resource_id} not reachable yet, retrying: {exc}") + _log(f"{resource_type} {resource_id} not reachable yet, retrying: {exc}") time.sleep(poll_interval) continue @@ -206,6 +226,7 @@ def _wait_for_resource_active( if state == "deleting": raise RuntimeError(f"{resource_type} entered deleting state: {resource_id}") + _log(f"Waiting for {resource_type} {resource_id} to become active (state: {state})") time.sleep(poll_interval) raise TimeoutError(f"{resource_type} {resource_id} did not become active within {timeout}s") @@ -230,7 +251,7 @@ def _delete_topic( timeout=30.0, ) if response.status_code not in (202, 404): - response.raise_for_status() + _raise_for_status(response) @with_retry @@ -254,7 +275,7 @@ def _list_subscriptions( headers={**auth_header, "Content-Type": "application/json"}, timeout=30.0, ) - response.raise_for_status() + _raise_for_status(response) data = response.json() return data.get("subscriptions", []) @@ -282,7 +303,7 @@ def _delete_subscription( timeout=30.0, ) if response.status_code not in (202, 404): - response.raise_for_status() + _raise_for_status(response) @with_retry @@ -310,7 +331,7 @@ def _grant_topic_publisher_access( headers={**auth_header, "Content-Type": "application/json"}, timeout=30.0, ) - response.raise_for_status() + _raise_for_status(response) @with_retry @@ -339,7 +360,7 @@ def _grant_subscription_subscriber_access( headers={**auth_header, "Content-Type": "application/json"}, timeout=30.0, ) - response.raise_for_status() + _raise_for_status(response) @pytest.fixture(scope="session") @@ -365,6 +386,12 @@ def auth_config(sa_key: str) -> AuthConfig: ) +@pytest.fixture(scope="session") +def auth_provider(auth_config: AuthConfig) -> StackitAuthProvider: + """Create a session-scoped StackitAuthProvider for control plane API calls.""" + return StackitAuthProvider(auth_config) + + @pytest.fixture(scope="session") def control_plane_url() -> str: """Get the control plane URL from environment or construct from REGION.""" @@ -377,8 +404,6 @@ def project_id() -> str: pid = os.environ.get("STACKIT_PROJECT_ID") if not pid: raise ValueError("STACKIT_PROJECT_ID environment variable is required") - digest = hashlib.sha512(pid.encode("utf-8")).hexdigest() - print(f"STACKIT_PROJECT_ID SHA512: {digest}") return pid @@ -393,20 +418,18 @@ def topic_id( control_plane_url: str, project_id: str, region: str, - auth_config: AuthConfig, + auth_provider: StackitAuthProvider, sa_email: str, ): """Create a topic once for all tests in this run, then delete it.""" - auth_header = _create_auth_header(auth_config) - new_topic_id = _create_topic( control_plane_url=control_plane_url, project_id=project_id, region=region, - auth_header=auth_header, + auth_header=auth_provider.get_auth_header(), display_name="sdk-integration-test-shared", ) - print(f"Created topic: {new_topic_id}") + _log(f"Created topic: {new_topic_id}") _wait_for_resource_active( control_plane_url=control_plane_url, @@ -414,34 +437,66 @@ def topic_id( region=region, resource_type="topics", resource_id=new_topic_id, - auth_header=auth_header, + auth_header=auth_provider.get_auth_header(), ) - print(f"Topic {new_topic_id} is active") + _log(f"Topic {new_topic_id} is active") _grant_topic_publisher_access( control_plane_url=control_plane_url, project_id=project_id, region=region, topic_id=new_topic_id, - auth_header=auth_header, + auth_header=auth_provider.get_auth_header(), email=sa_email, ) - print(f"Granted publisher access for {sa_email} on topic {new_topic_id}") + _log(f"Granted publisher access for {sa_email} on topic {new_topic_id}") yield new_topic_id - print(f"Deleting shared topic: {new_topic_id}") + _log(f"Deleting shared topic: {new_topic_id}") + try: + subs = _list_subscriptions( + control_plane_url=control_plane_url, + project_id=project_id, + region=region, + topic_id=new_topic_id, + auth_header=auth_provider.get_auth_header(), + ) + for sub in subs: + sub_id = sub.get("id") + if sub_id: + try: + _delete_subscription( + control_plane_url=control_plane_url, + project_id=project_id, + region=region, + topic_id=new_topic_id, + subscription_id=sub_id, + auth_header=auth_provider.get_auth_header(), + ) + _log(f"Deleted subscription {sub_id} on topic {new_topic_id}") + except Exception as sub_exc: + _log( + f"Failed to delete subscription {sub_id} on topic {new_topic_id}: " + f"{_format_error(sub_exc)}" + ) + except Exception as list_exc: + _log( + f"Failed to list subscriptions before deleting topic {new_topic_id}: " + f"{_format_error(list_exc)}" + ) + try: _delete_topic( control_plane_url=control_plane_url, project_id=project_id, region=region, topic_id=new_topic_id, - auth_header=auth_header, + auth_header=auth_provider.get_auth_header(), ) - print(f"Deleted shared topic: {new_topic_id}") + _log(f"Deleted shared topic: {new_topic_id}") except Exception as e: - print(f"Failed to delete shared topic {new_topic_id}: {e}") + _log(f"Failed to delete shared topic {new_topic_id}: {_format_error(e)}") @pytest.fixture(scope="session") @@ -455,9 +510,9 @@ def base_url(topic_id: str) -> str: @pytest.fixture() -def auth_header(auth_config: AuthConfig) -> dict[str, str]: +def auth_header(auth_provider: StackitAuthProvider) -> dict[str, str]: """Create an auth header for API calls.""" - return _create_auth_header(auth_config) + return auth_provider.get_auth_header() @pytest.fixture() @@ -484,23 +539,21 @@ def subscription_id( project_id: str, region: str, topic_id: str, - auth_config: AuthConfig, + auth_provider: StackitAuthProvider, sa_email: str, ): """Create a subscription once and reuse for all tests that need it.""" - auth_header = _create_auth_header(auth_config) - existing_subs = _list_subscriptions( control_plane_url=control_plane_url, project_id=project_id, region=region, topic_id=topic_id, - auth_header=auth_header, + auth_header=auth_provider.get_auth_header(), ) if existing_subs: sub_id = existing_subs[0]["id"] - print(f"Reusing existing subscription: {sub_id}") + _log(f"Reusing existing subscription: {sub_id}") try: _wait_for_resource_active( @@ -509,14 +562,14 @@ def subscription_id( region=region, resource_type="subscriptions", resource_id=sub_id, - auth_header=auth_header, + auth_header=auth_provider.get_auth_header(), poll_interval=10.0, timeout=60.0, topic_id=topic_id, ) - print(f"Subscription {sub_id} is active") + _log(f"Subscription {sub_id} is active") except TimeoutError: - print(f"Subscription {sub_id} not active, creating new subscription") + _log(f"Subscription {sub_id} not active, creating new subscription") with contextlib.suppress(Exception): _delete_subscription( control_plane_url=control_plane_url, @@ -524,7 +577,7 @@ def subscription_id( region=region, topic_id=topic_id, subscription_id=sub_id, - auth_header=auth_header, + auth_header=auth_provider.get_auth_header(), ) existing_subs = [] @@ -534,10 +587,10 @@ def subscription_id( project_id=project_id, region=region, topic_id=topic_id, - auth_header=auth_header, + auth_header=auth_provider.get_auth_header(), display_name="sdk-sub-integration-shared", ) - print(f"Created subscription: {sub_id}") + _log(f"Created subscription: {sub_id}") _wait_for_resource_active( control_plane_url=control_plane_url, @@ -545,10 +598,10 @@ def subscription_id( region=region, resource_type="subscriptions", resource_id=sub_id, - auth_header=auth_header, + auth_header=auth_provider.get_auth_header(), topic_id=topic_id, ) - print(f"Subscription {sub_id} is active") + _log(f"Subscription {sub_id} is active") _grant_subscription_subscriber_access( control_plane_url=control_plane_url, @@ -556,10 +609,10 @@ def subscription_id( region=region, topic_id=topic_id, subscription_id=sub_id, - auth_header=auth_header, + auth_header=auth_provider.get_auth_header(), email=sa_email, ) - print(f"Granted subscriber access for {sa_email} on subscription {sub_id}") + _log(f"Granted subscriber access for {sa_email} on subscription {sub_id}") else: sub_id = existing_subs[0]["id"] @@ -583,8 +636,8 @@ async def purge_between_tests(pubsub_client: PubSubClient): pub = pubsub_client.get_publisher() try: await pub.purge_topic() - print(f"Purged topic: {pubsub_client.topic_id}") + _log(f"Purged topic: {pubsub_client.topic_id}") except Exception as e: - print(f"Failed to purge topic: {e}") + _log(f"Failed to purge topic: {_format_error(e)}") finally: await pub.close() diff --git a/tests/test_publisher.py b/tests/test_publisher.py index ea9892d..f95486d 100644 --- a/tests/test_publisher.py +++ b/tests/test_publisher.py @@ -28,6 +28,7 @@ ) from pubsub_sdk_python.models import PublishMessage from pubsub_sdk_python.publisher import PubSubPublisher +from pubsub_sdk_python.retry import RetryConfig class TestPubSubPublisherInit: @@ -155,18 +156,14 @@ async def test_publish_500_triggers_retry(self, auth_provider): topic_id="test-topic", base_url="https://test.pubsub.eu01.onstackit.cloud", auth_provider=auth_provider, - retry_config=type( - "RetryConfig", - (), - { - "max_retries": 3, - "base_delay": 0.01, - "max_delay": 0.05, - "multiplier": 1.0, - "retryable_exceptions": (PubSubServerError,), - "retryable_status_codes": {503}, - }, - )(), + retry_config=RetryConfig( + max_retries=3, + base_delay=0.01, + max_delay=0.05, + multiplier=1.0, + retryable_exceptions=(PubSubServerError,), + retryable_status_codes={503}, + ), http_client=mock_client, ) @@ -188,18 +185,14 @@ async def test_publish_exhausts_retries(self, auth_provider): topic_id="test-topic", base_url="https://test.pubsub.eu01.onstackit.cloud", auth_provider=auth_provider, - retry_config=type( - "RetryConfig", - (), - { - "max_retries": 2, - "base_delay": 0.01, - "max_delay": 0.05, - "multiplier": 1.0, - "retryable_exceptions": (PubSubServerError,), - "retryable_status_codes": {503}, - }, - )(), + retry_config=RetryConfig( + max_retries=2, + base_delay=0.01, + max_delay=0.05, + multiplier=1.0, + retryable_exceptions=(PubSubServerError,), + retryable_status_codes={503}, + ), http_client=mock_client, ) diff --git a/tests/test_retry.py b/tests/test_retry.py index 4d01d56..d53f9ee 100644 --- a/tests/test_retry.py +++ b/tests/test_retry.py @@ -230,18 +230,14 @@ def _make_publisher(self, auth_provider, side_effect, retry_config=None): ok_response.text = "" if retry_config is None: - retry_config = type( - "RetryConfig", - (), - { - "max_retries": 3, - "base_delay": 0.01, - "max_delay": 0.05, - "multiplier": 1.0, - "retryable_exceptions": (PubSubServerError,), - "retryable_status_codes": {500, 502, 503, 504}, - }, - )() + retry_config = RetryConfig( + max_retries=3, + base_delay=0.01, + max_delay=0.05, + multiplier=1.0, + retryable_exceptions=(PubSubServerError,), + retryable_status_codes={500, 502, 503, 504}, + ) return PubSubPublisher( topic_id="test-topic", @@ -289,18 +285,14 @@ async def test_publisher_retry_exhausted(self, auth_provider): publisher = self._make_publisher( auth_provider, side_effect=[error_response] * 10, - retry_config=type( - "RetryConfig", - (), - { - "max_retries": 2, - "base_delay": 0.01, - "max_delay": 0.05, - "multiplier": 1.0, - "retryable_exceptions": (PubSubServerError,), - "retryable_status_codes": {503}, - }, - )(), + retry_config=RetryConfig( + max_retries=2, + base_delay=0.01, + max_delay=0.05, + multiplier=1.0, + retryable_exceptions=(PubSubServerError,), + retryable_status_codes={503}, + ), ) with pytest.raises(PubSubRetryError): diff --git a/tests/test_subscriber.py b/tests/test_subscriber.py index 15f9d88..d725238 100644 --- a/tests/test_subscriber.py +++ b/tests/test_subscriber.py @@ -28,6 +28,7 @@ PubSubValidationError, ) from pubsub_sdk_python.models import Message +from pubsub_sdk_python.retry import RetryConfig from pubsub_sdk_python.subscriber import PubSubSubscriber @@ -170,18 +171,14 @@ async def test_pull_500_triggers_retry(self, auth_provider): subscription_id="test-sub", base_url="https://test.pubsub.eu01.onstackit.cloud", auth_provider=auth_provider, - retry_config=type( - "RetryConfig", - (), - { - "max_retries": 3, - "base_delay": 0.01, - "max_delay": 0.05, - "multiplier": 1.0, - "retryable_exceptions": (PubSubServerError,), - "retryable_status_codes": {503}, - }, - )(), + retry_config=RetryConfig( + max_retries=3, + base_delay=0.01, + max_delay=0.05, + multiplier=1.0, + retryable_exceptions=(PubSubServerError,), + retryable_status_codes={503}, + ), http_client=mock_client, ) @@ -202,18 +199,14 @@ async def test_pull_exhausts_retries(self, auth_provider): subscription_id="test-sub", base_url="https://test.pubsub.eu01.onstackit.cloud", auth_provider=auth_provider, - retry_config=type( - "RetryConfig", - (), - { - "max_retries": 2, - "base_delay": 0.01, - "max_delay": 0.05, - "multiplier": 1.0, - "retryable_exceptions": (PubSubServerError,), - "retryable_status_codes": {503}, - }, - )(), + retry_config=RetryConfig( + max_retries=2, + base_delay=0.01, + max_delay=0.05, + multiplier=1.0, + retryable_exceptions=(PubSubServerError,), + retryable_status_codes={503}, + ), http_client=mock_client, ) diff --git a/uv.lock b/uv.lock index df45034..48f9a9b 100644 --- a/uv.lock +++ b/uv.lock @@ -538,7 +538,7 @@ wheels = [ [[package]] name = "pubsub-sdk-python" version = "0.1.0" -source = { virtual = "." } +source = { editable = "." } dependencies = [ { name = "httpx" }, { name = "pyjwt" }, @@ -552,6 +552,7 @@ dev = [ { name = "pytest-cov" }, { name = "pytest-timeout" }, { name = "ruff" }, + { name = "ty" }, ] [package.metadata] @@ -568,6 +569,7 @@ dev = [ { name = "pytest-cov", specifier = ">=5.0.0" }, { name = "pytest-timeout", specifier = ">=2.3.0" }, { name = "ruff", specifier = ">=0.8.0" }, + { name = "ty", specifier = ">=0.0.93" }, ] [[package]] @@ -915,6 +917,31 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/7b/61/cceae43728b7de99d9b847560c262873a1f6c98202171fd5ed62640b494b/tomli-2.4.1-py3-none-any.whl", hash = "sha256:0d85819802132122da43cb86656f8d1f8c6587d54ae7dcaf30e90533028b49fe", size = 14583, upload-time = "2026-03-25T20:22:03.012Z" }, ] +[[package]] +name = "ty" +version = "0.0.93" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/78/35/e1cbce56e80f6d911b6b6821d211b4150d0e061d426df038cad78d2800f4/ty-0.0.93.tar.gz", hash = "sha256:18b99060d7f1d601fd4ce1acdcc336d1b3345fea8561307948827e8809c31c25", size = 8105272, upload-time = "2026-10-28T00:52:42.089Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/4e/16/5b0766d3de2b531f47d6f24986f21607546b6ee1861d294ccec69a750b45/ty-0.0.93-py3-none-linux_armv6l.whl", hash = "sha256:74fad0e17d96076364be426b15ed3f052f068ec6e37b9d8dbf9c9c424fcb6f0d", size = 14510907, upload-time = "2026-10-28T00:52:04.298Z" }, + { url = "https://files.pythonhosted.org/packages/14/4a/ba4e9586f22b42f86e95c5f06798d30b288e1d6557b29c137384031c561c/ty-0.0.93-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:83beb545d52c38deb910d977679dcac1c274f5cf8ee0a1868b10d390042c40df", size = 14074052, upload-time = "2026-10-28T00:52:07.324Z" }, + { url = "https://files.pythonhosted.org/packages/ce/3c/00e4195634b61a6ec1f53a2c70777df644a3ee79f6579b48c4e2f6e884d0/ty-0.0.93-py3-none-macosx_11_0_arm64.whl", hash = "sha256:47d6be14343a997c6f828d812513f940929c906f6cf83ca4cdd74db6e16ef261", size = 13132684, upload-time = "2026-10-28T00:52:09.528Z" }, + { url = "https://files.pythonhosted.org/packages/3e/f4/36fb44a79e2d7c8b2d5f2c42cab6d2c1d4d0e8b9fb7e51acb82648fc4d3e/ty-0.0.93-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:d696b82ade77500826d84874d46658020486daa9d39eabc71b94a641247ce484", size = 13955788, upload-time = "2026-10-28T00:52:11.478Z" }, + { url = "https://files.pythonhosted.org/packages/49/21/ae904789846d93e4b01faa33ef9a73f28657dd88b8b99619dbae05f30f1d/ty-0.0.93-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:9882e7b55ee025feceaf8f1210acda368f931c3f597771385f97a460292a9c8b", size = 14252702, upload-time = "2026-10-28T00:52:13.466Z" }, + { url = "https://files.pythonhosted.org/packages/13/8e/fecf0492b20f1e28fafac944f5a6c7a404d8408f2e6fb0fb5666550f994f/ty-0.0.93-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:318871545c28f22564b445cadd022650e0fa892c913c0c1eb4cf6345d72037a2", size = 15192511, upload-time = "2026-10-28T00:52:15.923Z" }, + { url = "https://files.pythonhosted.org/packages/97/1f/67d3284747800456e96892d51df3fefc9f98b844653a86cfc6312902a490/ty-0.0.93-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:1f9c494291a7c83706b380e31606f88b6dcf801499623645861c3099c17edc01", size = 15594668, upload-time = "2026-10-28T00:52:18.133Z" }, + { url = "https://files.pythonhosted.org/packages/ed/1f/2a903b2309c8ea265a0cef1edc00fa9223be42d004171f7f87aded4b1bed/ty-0.0.93-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:43ff1ee239c94260cf2146e23537fb069b7267423723096603051bb7d36ce165", size = 15533756, upload-time = "2026-10-28T00:52:20.271Z" }, + { url = "https://files.pythonhosted.org/packages/dc/75/99bdbbd00220703c51af60160fb1c080f653b45ddb585baf8468588f70ef/ty-0.0.93-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:1c278c86adeb82374e0f5142b44bb628e9dcb663d98e8f274707300d926e0997", size = 16236455, upload-time = "2026-10-28T00:52:22.244Z" }, + { url = "https://files.pythonhosted.org/packages/ca/7d/edbcce5e504275b3574272629c2c7d657c666312fd5a74a5a3202f256e0a/ty-0.0.93-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:02c7cc6f1d9d2fe49eff5b230d243c696ac798ecce034998e3d67872ab2a5236", size = 15065832, upload-time = "2026-10-28T00:52:24.23Z" }, + { url = "https://files.pythonhosted.org/packages/e6/ce/201c42e84b49a8bb6473edfd3e16e311b942178cefd9678f98dda36f25a4/ty-0.0.93-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:50c580e31da3e4dd3f1b7ab1969497996ab55e61ee06d90132ab46a2964249ef", size = 13919356, upload-time = "2026-10-28T00:52:26.893Z" }, + { url = "https://files.pythonhosted.org/packages/68/5c/f323a70974bd57274d4315d1894586498557e21d4d78cd35b550deaf7878/ty-0.0.93-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:7f037c1761a6874cc0072c39950a987cd6d1563d6df3c94bb131b3e727e921c9", size = 14258573, upload-time = "2026-10-28T00:52:29.049Z" }, + { url = "https://files.pythonhosted.org/packages/c8/ba/0d2a2f0892d19c91cef7785ac1bac5de555059f1bad2a0009d83f2df1d0d/ty-0.0.93-py3-none-musllinux_1_2_i686.whl", hash = "sha256:a8bfbcc8a715a5497210cc89eb134476f511b2de3004697c96b5f6eaa0f1602b", size = 14613304, upload-time = "2026-10-28T00:52:31.474Z" }, + { url = "https://files.pythonhosted.org/packages/b8/02/b0f7b83ef79c85b8813e54ed73189995e3d4ec2f1dda791a8d0acd051386/ty-0.0.93-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:19ae7e56c9021d7eafbc305926ca174183b431c63c27e9d2b7db342d45142446", size = 14937750, upload-time = "2026-10-28T00:52:33.528Z" }, + { url = "https://files.pythonhosted.org/packages/06/ad/d2ece9e2ec588394d0ea0c12b683ff68540f5534988b4e4412664bb87a56/ty-0.0.93-py3-none-win32.whl", hash = "sha256:1c677be05fb0ef0864a305fa4a5258190fc0fc9d79f53c1ce93a0342256f21f1", size = 13464721, upload-time = "2026-10-28T00:52:35.916Z" }, + { url = "https://files.pythonhosted.org/packages/a9/17/ef0c1180498564f08331f4be35aea9de09b37c6ca12fa50ad1959d9d1e38/ty-0.0.93-py3-none-win_amd64.whl", hash = "sha256:65072e0a71f9af81a31a4656521c96eaf539452c37287b9e14a81504b22e0864", size = 14091561, upload-time = "2026-10-28T00:52:38.008Z" }, + { url = "https://files.pythonhosted.org/packages/ef/f8/13fb149c7c5a33be119ecea9c56033b32cc1f87c5ac84d577f2236b7c964/ty-0.0.93-py3-none-win_arm64.whl", hash = "sha256:47991915f6a2101cc63da1dd07b2f3bbd226ab4a14ff3c858b20764c38376166", size = 13552165, upload-time = "2026-10-28T00:52:40.111Z" }, +] + [[package]] name = "typing-extensions" version = "4.16.0" From d183735cd7671db0d91e19145fb30da87d3b47b4 Mon Sep 17 00:00:00 2001 From: Jennifer Linnenberg Date: Wed, 28 Oct 2026 11:00:09 +0100 Subject: [PATCH 3/3] feat: better transient error handling --- src/pubsub_sdk_python/retry.py | 169 ++++++++++++++++++++------------- tests/integration/conftest.py | 29 ++++-- tests/test_retry.py | 26 +++++ 3 files changed, 150 insertions(+), 74 deletions(-) diff --git a/src/pubsub_sdk_python/retry.py b/src/pubsub_sdk_python/retry.py index ee85fae..48e5c00 100644 --- a/src/pubsub_sdk_python/retry.py +++ b/src/pubsub_sdk_python/retry.py @@ -21,7 +21,7 @@ import time from collections.abc import Awaitable, Callable from functools import wraps -from typing import Any, TypeVar +from typing import Any, TypeVar, overload import httpx @@ -136,11 +136,24 @@ def _should_retry( return False -def with_retry(func: Callable[..., T]) -> Callable[..., T]: +@overload +def with_retry(func: Callable[..., T]) -> Callable[..., T]: ... + + +@overload +def with_retry( + func: None = None, *, config: RetryConfig | None = None +) -> Callable[[Callable[..., T]], Callable[..., T]]: ... + + +def with_retry( + func: Callable[..., T] | None = None, *, config: RetryConfig | None = None +) -> Callable[..., T] | Callable[[Callable[..., T]], Callable[..., T]]: """Decorator that adds retry logic with exponential backoff to a synchronous function. Args: func: The synchronous function to wrap. + config: Optional RetryConfig to use. Returns: The wrapped function with retry logic. @@ -150,45 +163,66 @@ def with_retry(func: Callable[..., T]) -> Callable[..., T]: def my_api_call(): ... # Will retry on transient errors """ - config = RetryConfig() - - @wraps(func) - def wrapper(*args: Any, **kwargs: Any) -> T: - last_exception: Exception | None = None - for attempt in range(config.max_retries + 1): - try: - return func(*args, **kwargs) - except Exception as exc: - last_exception = exc - if not _should_retry(exc, config=config): - raise - if attempt >= config.max_retries: - raise PubSubRetryError( - f"All {config.max_retries} retry attempts exhausted: {exc}", - causes=[last_exception] if last_exception else None, - ) from exc - delay = _calculate_delay(config, attempt) - logger.warning( - "Retry attempt %d/%d for %s after %.2fs: %s", - attempt + 1, - config.max_retries, - getattr(func, "__name__", repr(func)), - delay, - exc, - ) - time.sleep(delay) - raise PubSubRetryError("Retry loop terminated unexpectedly") - - return wrapper - - -def with_async_retry(func: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]: + retry_config = config or RetryConfig() + + def decorator(fn: Callable[..., T]) -> Callable[..., T]: + @wraps(fn) + def wrapper(*args: Any, **kwargs: Any) -> T: + last_exception: Exception | None = None + for attempt in range(retry_config.max_retries + 1): + try: + return fn(*args, **kwargs) + except Exception as exc: + last_exception = exc + if not _should_retry(exc, config=retry_config): + raise + if attempt >= retry_config.max_retries: + raise PubSubRetryError( + f"All {retry_config.max_retries} retry attempts exhausted: {exc}", + causes=[last_exception] if last_exception else None, + ) from exc + delay = _calculate_delay(retry_config, attempt) + logger.warning( + "Retry attempt %d/%d for %s after %.2fs: %s", + attempt + 1, + retry_config.max_retries, + getattr(fn, "__name__", repr(fn)), + delay, + exc, + ) + time.sleep(delay) + raise PubSubRetryError("Retry loop terminated unexpectedly") + + return wrapper + + if func is not None: + return decorator(func) + return decorator + + +@overload +def with_async_retry(func: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]: ... + + +@overload +def with_async_retry( + func: None = None, *, config: RetryConfig | None = None +) -> Callable[[Callable[..., Awaitable[T]]], Callable[..., Awaitable[T]]]: ... + + +def with_async_retry( + func: Callable[..., Awaitable[T]] | None = None, *, config: RetryConfig | None = None +) -> ( + Callable[..., Awaitable[T]] + | Callable[[Callable[..., Awaitable[T]]], Callable[..., Awaitable[T]]] +): """Async-friendly retry decorator. Adds retry logic with exponential backoff to an async function. Args: func: The async function to wrap. + config: Optional RetryConfig to use. Returns: The wrapped async function with retry logic. @@ -198,36 +232,41 @@ def with_async_retry(func: Callable[..., Awaitable[T]]) -> Callable[..., Awaitab async def my_async_api_call(): ... # Will retry on transient errors """ - config = RetryConfig() - - @wraps(func) - async def wrapper(*args: Any, **kwargs: Any) -> T: - last_exception: Exception | None = None - for attempt in range(config.max_retries + 1): - try: - return await func(*args, **kwargs) - except Exception as exc: - last_exception = exc - if not _should_retry(exc, config=config): - raise - if attempt >= config.max_retries: - raise PubSubRetryError( - f"All {config.max_retries} retry attempts exhausted: {exc}", - causes=[last_exception] if last_exception else None, - ) from exc - delay = _calculate_delay(config, attempt) - logger.warning( - "Retry attempt %d/%d for %s after %.2fs: %s", - attempt + 1, - config.max_retries, - getattr(func, "__name__", repr(func)), - delay, - exc, - ) - await asyncio.sleep(delay) - raise PubSubRetryError("Retry loop terminated unexpectedly") - - return wrapper + retry_config = config or RetryConfig() + + def decorator(fn: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]: + @wraps(fn) + async def wrapper(*args: Any, **kwargs: Any) -> T: + last_exception: Exception | None = None + for attempt in range(retry_config.max_retries + 1): + try: + return await fn(*args, **kwargs) + except Exception as exc: + last_exception = exc + if not _should_retry(exc, config=retry_config): + raise + if attempt >= retry_config.max_retries: + raise PubSubRetryError( + f"All {retry_config.max_retries} retry attempts exhausted: {exc}", + causes=[last_exception] if last_exception else None, + ) from exc + delay = _calculate_delay(retry_config, attempt) + logger.warning( + "Retry attempt %d/%d for %s after %.2fs: %s", + attempt + 1, + retry_config.max_retries, + getattr(fn, "__name__", repr(fn)), + delay, + exc, + ) + await asyncio.sleep(delay) + raise PubSubRetryError("Retry loop terminated unexpectedly") + + return wrapper + + if func is not None: + return decorator(func) + return decorator def get_retry_delay(config: RetryConfig, attempt: int) -> float: diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index dfcc80b..efbe017 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -33,10 +33,18 @@ from pubsub_sdk_python.auth import AuthConfig, StackitAuthProvider from pubsub_sdk_python.client import PubSubClient -from pubsub_sdk_python.retry import with_retry +from pubsub_sdk_python.errors import PubSubServerError +from pubsub_sdk_python.retry import RetryConfig, with_retry SA_KEY_PATH = "sa-key.json" +CONTROL_PLANE_RETRY_CONFIG = RetryConfig( + max_retries=6, + base_delay=2.0, + max_delay=30.0, + multiplier=2.0, +) + def _log(msg: str) -> None: """Print an integration test log message on its own line.""" @@ -90,6 +98,9 @@ def _create_auth_header(auth_config: AuthConfig) -> dict[str, str]: def _raise_for_status(response: httpx.Response) -> None: """Raise HTTPStatusError with response body details if request failed.""" + if response.status_code in {429, 500, 502, 503, 504}: + body = response.text.strip() + raise PubSubServerError(f"Server error {response.status_code}: {body}") try: response.raise_for_status() except httpx.HTTPStatusError as exc: @@ -103,7 +114,7 @@ def _raise_for_status(response: httpx.Response) -> None: raise -@with_retry +@with_retry(config=CONTROL_PLANE_RETRY_CONFIG) def _create_topic( control_plane_url: str, project_id: str, @@ -134,7 +145,7 @@ def _create_topic( return topic["id"] -@with_retry +@with_retry(config=CONTROL_PLANE_RETRY_CONFIG) def _create_subscription( control_plane_url: str, project_id: str, @@ -212,7 +223,7 @@ def _wait_for_resource_active( while time.time() - start_time < timeout: try: resource = _get_resource(url, auth_header) - except httpx.TransportError as exc: + except (httpx.TransportError, PubSubServerError) as exc: # The resource may not be reachable yet right after creation # (e.g. backend routing not provisioned) - keep polling until timeout. _log(f"{resource_type} {resource_id} not reachable yet, retrying: {exc}") @@ -232,7 +243,7 @@ def _wait_for_resource_active( raise TimeoutError(f"{resource_type} {resource_id} did not become active within {timeout}s") -@with_retry +@with_retry(config=CONTROL_PLANE_RETRY_CONFIG) def _delete_topic( control_plane_url: str, project_id: str, @@ -254,7 +265,7 @@ def _delete_topic( _raise_for_status(response) -@with_retry +@with_retry(config=CONTROL_PLANE_RETRY_CONFIG) def _list_subscriptions( control_plane_url: str, project_id: str, @@ -280,7 +291,7 @@ def _list_subscriptions( return data.get("subscriptions", []) -@with_retry +@with_retry(config=CONTROL_PLANE_RETRY_CONFIG) def _delete_subscription( control_plane_url: str, project_id: str, @@ -306,7 +317,7 @@ def _delete_subscription( _raise_for_status(response) -@with_retry +@with_retry(config=CONTROL_PLANE_RETRY_CONFIG) def _grant_topic_publisher_access( control_plane_url: str, project_id: str, @@ -334,7 +345,7 @@ def _grant_topic_publisher_access( _raise_for_status(response) -@with_retry +@with_retry(config=CONTROL_PLANE_RETRY_CONFIG) def _grant_subscription_subscriber_access( control_plane_url: str, project_id: str, diff --git a/tests/test_retry.py b/tests/test_retry.py index d53f9ee..3d3b2e6 100644 --- a/tests/test_retry.py +++ b/tests/test_retry.py @@ -169,6 +169,19 @@ def my_func(): my_func() assert call_count == 4 # 1 initial + 3 retries + def test_sync_with_custom_config(self): + call_count = 0 + + @with_retry(config=RetryConfig(max_retries=1, base_delay=0.01, max_delay=0.02)) + def my_func(): + nonlocal call_count + call_count += 1 + raise PubSubServerError("Server error") + + with pytest.raises(PubSubRetryError, match="All 1 retry attempts exhausted: Server error"): + my_func() + assert call_count == 2 + @pytest.mark.asyncio class TestWithAsyncRetry: @@ -215,6 +228,19 @@ async def my_func(): await my_func() assert call_count == 4 + async def test_async_with_custom_config(self): + call_count = 0 + + @with_async_retry(config=RetryConfig(max_retries=1, base_delay=0.01, max_delay=0.02)) + async def my_func(): + nonlocal call_count + call_count += 1 + raise PubSubServerError("Server error") + + with pytest.raises(PubSubRetryError, match="All 1 retry attempts exhausted: Server error"): + await my_func() + assert call_count == 2 + @pytest.mark.asyncio class TestPublisherRetryIntegration: