fix: improve flaky test resilience for channel type and transient errors

- Broaden _is_transient_error to catch 429, rate limits, and resource
  limit errors (e.g. "maximum number of custom channel types")
- Use str(exc) instead of http_response.text for reliable phrase matching
- Add @cleanup_channel_types decorator that deletes stale (>2min)
  non-builtin channel types before tests to free slots in shared CI env
- Apply @retry_on_transient_error and @cleanup_channel_types to channel
  type CRUD tests and event_hooks test

Author: aliev

tests/base.py

@@ -1,4 +1,6 @@
+import datetime
 import functools
+import inspect
 import time
 from abc import ABC
 
@@ -52,14 +54,19 @@ def _is_transient_error(exc: Exception) -> bool:
     if isinstance(exc, (httpx.ReadTimeout, httpx.ConnectTimeout, httpx.ConnectError)):
         return True
     if isinstance(exc, StreamAPIException):
-        body = ""
-        try:
-            body = exc.http_response.text or ""
-        except Exception:
-            pass
-        if "upstream connect error" in body or "disconnect" in body:
+        if exc.status_code in (429, 502, 503, 504):
             return True
-        if exc.status_code in (502, 503, 504):
+        msg = str(exc).lower()
+        if any(
+            phrase in msg
+            for phrase in (
+                "upstream connect error",
+                "disconnect",
+                "maximum number of",
+                "rate limit",
+                "too many",
+            )
+        ):
             return True
     return False
 
@@ -85,3 +92,44 @@ def wrapper(*args, **kwargs):
         return wrapper
 
     return decorator
+
+
+_BUILTIN_CHANNEL_TYPES = frozenset(
+    {"messaging", "livestream", "team", "gaming", "commerce"}
+)
+_STALE_THRESHOLD = datetime.timedelta(minutes=2)
+
+
+def cleanup_channel_types(func):
+    """Decorator that deletes stale test channel types before the test runs.
+
+    Frees slots toward the 50-type limit without interfering with parallel
+    runners (only removes types older than 2 minutes).  Expects a ``client``
+    pytest fixture parameter.
+    """
+
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs):
+        # Resolve 'client' from pytest kwargs or positional args
+        client = kwargs.get("client")
+        if client is None:
+            sig = inspect.signature(func)
+            params = list(sig.parameters)
+            idx = params.index("client") if "client" in params else 0
+            client = args[idx]
+
+        now = datetime.datetime.now(datetime.timezone.utc)
+        resp = client.chat.list_channel_types()
+        for name, config in resp.data.channel_types.items():
+            if name in _BUILTIN_CHANNEL_TYPES:
+                continue
+            if config and (now - config.created_at) > _STALE_THRESHOLD:
+                try:
+                    client.chat.delete_channel_type(name=name)
+                except Exception:
+                    pass
+        time.sleep(2)
+
+        return func(*args, **kwargs)
+
+    return wrapper

tests/test_chat_misc.py

@@ -17,6 +17,7 @@
     QueryFutureChannelBansPayload,
     SortParamRequest,
 )
+from tests.base import cleanup_channel_types, retry_on_transient_error
 
 
 def test_get_app_settings(client: Stream):
@@ -389,6 +390,8 @@ def test_query_future_channel_bans(client: Stream, random_users):
             pass
 
 
+@retry_on_transient_error()
+@cleanup_channel_types
 def test_create_channel_type(client: Stream):
     """Create a channel type with custom settings."""
     type_name = f"testtype{uuid.uuid4().hex[:8]}"
@@ -413,6 +416,8 @@ def test_create_channel_type(client: Stream):
             pass
 
 
+@retry_on_transient_error()
+@cleanup_channel_types
 def test_update_channel_type_mark_messages_pending(client: Stream):
     """Update a channel type with mark_messages_pending=True."""
     type_name = f"testtype{uuid.uuid4().hex[:8]}"
@@ -445,6 +450,8 @@ def test_update_channel_type_mark_messages_pending(client: Stream):
             pass
 
 
+@retry_on_transient_error()
+@cleanup_channel_types
 def test_update_channel_type_push_notifications(client: Stream):
     """Update a channel type with push_notifications=False."""
     type_name = f"testtype{uuid.uuid4().hex[:8]}"
@@ -477,6 +484,8 @@ def test_update_channel_type_push_notifications(client: Stream):
             pass
 
 
+@retry_on_transient_error()
+@cleanup_channel_types
 def test_delete_channel_type(client: Stream):
     """Create and delete a channel type with retry."""
     type_name = f"testdeltype{uuid.uuid4().hex[:8]}"
@@ -540,6 +549,7 @@ def test_get_rate_limits_specific_endpoints(client: Stream):
         assert info.remaining >= 0
 
 
+@retry_on_transient_error()
 def test_event_hooks_sqs_sns(client: Stream):
     """Test setting SQS, SNS, and pending_message event hooks."""
     # Save original hooks to restore later