fix: never replay the webhook test call after a 5xx or a lost connection

src/controlresell/retry.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import random
+import re
 import time
 import uuid
 from dataclasses import dataclass
@@ -59,36 +60,47 @@ def _should_retry(response: httpx.Response) -> bool:
     return response.status_code in _RETRIED_STATUSES or _in_progress(response)
 
 
+_WEBHOOK_TEST_PATH = re.compile(r"/webhook-endpoints/[^/]+/test/?$")
+
+
+def _is_webhook_test(request: httpx.Request) -> bool:
+    """``POST /webhook-endpoints/{id}/test`` sends a real event and the API ignores its key: a replay would send a
+    second event. It is retried on a 429 (answered before the event leaves) and on a failed connect, nothing else."""
+    return request.method == "POST" and _WEBHOOK_TEST_PATH.search(request.url.path) is not None
+
+
 def _with_key(request: httpx.Request) -> None:
     if request.method == "POST" and IDEMPOTENCY_KEY_HEADER not in request.headers:
         request.headers[IDEMPOTENCY_KEY_HEADER] = str(uuid.uuid4())
 
 
 class RetryTransport(httpx.BaseTransport):
     """Gives each ``POST`` an ``Idempotency-Key`` (kept across its retries), retries the statuses above and network
-    errors, and waits for ``Retry-After`` when the API sends it."""
+    errors, and waits for ``Retry-After`` when the API sends it. The webhook test call is retried on a 429 and on a
+    failed connect only."""
 
     def __init__(self, base: httpx.BaseTransport, policy: RetryPolicy | None = None) -> None:
         self._base = base
         self._policy = policy or RetryPolicy()
 
     def handle_request(self, request: httpx.Request) -> httpx.Response:
         _with_key(request)
+        replayable = not _is_webhook_test(request)
         attempt = 0
         while True:
             last = attempt >= self._policy.max_retries
             try:
                 response = self._base.handle_request(request)
             except httpx.TransportError as cause:
-                if last:
+                if last or (not replayable and not isinstance(cause, (httpx.ConnectError, httpx.ConnectTimeout))):
                     raise ControlResellConnectionError(cause) from cause
                 time.sleep(self._policy.backoff(attempt))
                 attempt += 1
                 continue
             if last:
                 return response
             response.read()
-            if not _should_retry(response):
+            if (not replayable and response.status_code != 429) or not _should_retry(response):
                 return response
             time.sleep(_delay(self._policy, response, attempt))
             response.close()

tests/test_client.py

@@ -6,6 +6,7 @@
 from controlresell import (
     IDEMPOTENCY_KEY_HEADER,
     ConflictError,
+    ControlResellConnectionError,
     ControlResell,
     NotFoundError,
     RateLimitError,
@@ -527,3 +528,50 @@ def handler(request):
     assert sdk.webhooks.rotate_secret(ENDPOINT_ID, expires_in_hours=0).secret == "whsec_new"
     assert json.loads(requests[0].content) == {"expiresInHours": 0}
     assert sdk.webhooks.delete_endpoint(ENDPOINT_ID) is None
+
+
+@pytest.mark.parametrize("status", [500, 502, 503, 504])
+def test_webhooks_test_is_never_replayed_after_a_5xx(status):
+    requests, sdk = client(lambda request: reply(envelope("unavailable"), status))
+    with pytest.raises(ServerError):
+        sdk.webhooks.test(ENDPOINT_ID)
+    assert len(requests) == 1
+
+
+def test_webhooks_test_is_never_replayed_after_a_lost_connection():
+    def handler(request):
+        raise httpx.ReadError("connection lost", request=request)
+
+    requests, sdk = client(handler)
+    with pytest.raises(ControlResellConnectionError):
+        sdk.webhooks.test(ENDPOINT_ID)
+    assert len(requests) == 1
+
+
+def test_webhooks_test_is_replayed_after_a_failed_connect_or_a_429_but_other_calls_keep_the_retries():
+    def refused_once(request):
+        if len(requests) == 1:
+            raise httpx.ConnectError("refused", request=request)
+        return reply({"success": True, "statusCode": 200, "durationMs": 12, "error": None})
+
+    requests, sdk = client(refused_once)
+    assert sdk.webhooks.test(ENDPOINT_ID).success is True
+    assert len(requests) == 2
+
+    def busy_once(request):
+        if len(requests) == 1:
+            return reply(envelope("webhook_endpoint_busy"), 429, {"Retry-After": "0"})
+        return reply({"success": True, "statusCode": 200, "durationMs": 12, "error": None})
+
+    requests, sdk = client(busy_once)
+    assert sdk.webhooks.test(ENDPOINT_ID).success is True
+    assert len(requests) == 2
+
+    def unavailable_once(request):
+        if len(requests) == 1:
+            return reply(envelope("unavailable"), 503)
+        return reply({"id": 1, "username": "u", "currency": "EUR"})
+
+    requests, sdk = client(unavailable_once)
+    sdk.me.get()
+    assert len(requests) == 2