chore(tests): stop leaking channels on cleanup failure

The CI test suite creates ~94 channels per run (~50 sync + ~44 async)
across fixtures and explicit cases, but cleanup historically ran
through three escape hatches that silently leaked channels into the
shared test app:

1. channel fixture used client.delete_channels([cid], hard_delete=True),
   which returns a task_id immediately. When the task queue is backed
   up (the same backend slowness that breaks test_delete_channels) the
   fixture says 'done' before the delete actually fires.
2. Every cleanup path swallowed exceptions: except Exception: pass.
   Failures invisible.
3. test_query_drafts (sync + async) created a second channel,
   channel2, and did not delete it on the failure path.

Stale channels drift across runs and eventually break unrelated tests
that query the shared app expecting clean state.

Three fixes:

* Swap client.delete_channels for the synchronous channel.delete(hard=True)
  HTTP DELETE in every fixture. No task_id race; the channel is gone
  before the fixture returns.
* Log cleanup exceptions to stderr ('[cleanup] channel <cid> failed:
  ...'). Visible in CI logs, doesn't fail the test.
* Add a session-scoped, autouse=True sweep that queries channels tagged
  {'test': True, 'language': 'python'} at start and end of the run and
  hard-deletes whatever is still hanging around. Self-healing across
  runs; tag test_query_drafts's channel2 so the sweep catches it too.

Author: nijeesh-stream

stream_chat/tests/async_chat/conftest.py

@@ -1,5 +1,6 @@
 import asyncio
 import os
+import sys
 import uuid
 from typing import Dict, List
 
@@ -26,13 +27,66 @@ def pytest_configure(config):
     config.addinivalue_line("markers", "incremental: mark test incremental")
 
 
+def _warn_cleanup_failure(label: str, identifier: str, exc: BaseException) -> None:
+    """See sync conftest for rationale; mirrored to keep both suites consistent."""
+    print(
+        f"[cleanup] {label} {identifier} failed: {exc.__class__.__name__}: {exc}",
+        file=sys.stderr,
+    )
+
+
 @pytest.fixture(scope="module")
 def event_loop():
     loop = asyncio.get_event_loop_policy().new_event_loop()
     yield loop
     loop.close()
 
 
+@pytest.fixture(scope="session", autouse=True)
+def _purge_stale_test_channels_async(event_loop):
+    """Async mirror of the sync session sweep; queries channels tagged
+    ``{"test": True, "language": "python"}`` and hard-deletes them at
+    session start and end so leaks don't compound across runs."""
+
+    async def sweep() -> None:
+        base_url = os.environ.get("STREAM_HOST")
+        options = {"base_url": base_url} if base_url else {}
+        async with StreamChatAsync(
+            api_key=os.environ["STREAM_KEY"],
+            api_secret=os.environ["STREAM_SECRET"],
+            timeout=10,
+            **options,
+        ) as client:
+            try:
+                response = await client.query_channels(
+                    {"test": True, "language": "python"},
+                    sort=[{"field": "created_at", "direction": -1}],
+                    limit=30,
+                )
+            except Exception as exc:
+                print(
+                    f"[cleanup] sweep query_channels failed: "
+                    f"{exc.__class__.__name__}: {exc}",
+                    file=sys.stderr,
+                )
+                return
+            cids = [c["channel"]["cid"] for c in response.get("channels", [])]
+            if not cids:
+                return
+            try:
+                await client.delete_channels(cids, hard_delete=True)
+                print(
+                    f"[cleanup] swept {len(cids)} leaked test channels",
+                    file=sys.stderr,
+                )
+            except Exception as exc:
+                _warn_cleanup_failure("sweep delete_channels", str(len(cids)), exc)
+
+    event_loop.run_until_complete(sweep())
+    yield
+    event_loop.run_until_complete(sweep())
+
+
 @pytest.fixture(scope="function", autouse=True)
 @pytest.mark.asyncio
 async def client():
@@ -85,10 +139,12 @@ async def channel(client: StreamChatAsync, random_user: Dict):
     await channel.create(random_user["id"])
     yield channel
 
+    # Synchronous channel.delete (HTTP DELETE), not the async-task
+    # delete_channels — see sync conftest for the leak rationale.
     try:
-        await client.delete_channels([channel.cid], hard_delete=True)
-    except Exception:
-        pass
+        await channel.delete(hard=True)
+    except Exception as exc:
+        _warn_cleanup_failure("channel", channel.cid, exc)
 
 
 @pytest.fixture(scope="function")
@@ -107,10 +163,10 @@ async def command(client: StreamChatAsync):
             ):
                 try:
                     await client.delete_command(cmd["name"])
-                except Exception:
-                    pass
-    except Exception:
-        pass
+                except Exception as exc:
+                    _warn_cleanup_failure("stale command", cmd["name"], exc)
+    except Exception as exc:
+        _warn_cleanup_failure("list_commands", "<sweep>", exc)
 
     response = await client.create_command(
         dict(name=str(uuid.uuid4()), description="My command")
@@ -120,8 +176,8 @@ async def command(client: StreamChatAsync):
 
     try:
         await client.delete_command(response["command"]["name"])
-    except Exception:
-        pass
+    except Exception as exc:
+        _warn_cleanup_failure("command", response["command"]["name"], exc)
 
 
 @pytest.fixture(scope="function")
@@ -145,8 +201,8 @@ async def fellowship_of_the_ring(client: StreamChatAsync):
     ]
     try:
         await client.restore_users([m["id"] for m in members])
-    except Exception:
-        pass
+    except Exception as exc:
+        _warn_cleanup_failure("restore_users", "fellowship", exc)
     await client.upsert_users(members)
     channel = client.channel(
         "team", "fellowship-of-the-ring", {"members": [m["id"] for m in members]}
@@ -155,15 +211,15 @@ async def fellowship_of_the_ring(client: StreamChatAsync):
     yield
     try:
         await channel.delete(hard=True)
-        await hard_delete_users(client, [m["id"] for m in members])
-    except Exception:
-        pass
+    except Exception as exc:
+        _warn_cleanup_failure("channel", channel.cid, exc)
+    await hard_delete_users(client, [m["id"] for m in members])
 
 
 async def hard_delete_users(client: StreamChatAsync, user_ids: List[str]):
     try:
         await client.delete_users(
             user_ids, "hard", conversations="hard", messages="hard"
         )
-    except Exception:
-        pass
+    except Exception as exc:
+        _warn_cleanup_failure("delete_users", ",".join(user_ids), exc)

stream_chat/tests/async_chat/test_draft.py

@@ -84,65 +84,80 @@ async def test_query_drafts(
         draft1 = {"text": "Draft in channel 1"}
         await channel.create_draft(draft1, random_user["id"])
 
-        # Create another channel with a draft
-        channel2 = client.channel("messaging", str(uuid.uuid4()))
-        await channel2.create(random_user["id"])
-
-        draft2 = {"text": "Draft in channel 2"}
-        await channel2.create_draft(draft2, random_user["id"])
-
-        # Query all drafts for the user
-        response = await client.query_drafts(random_user["id"])
-
-        assert "drafts" in response
-        assert len(response["drafts"]) == 2
-
-        # Query drafts for a specific channel
-        response = await client.query_drafts(
-            random_user["id"], filter={"channel_cid": channel2.cid}
+        # Create another channel with a draft. Tag with {"test": True,
+        # "language": "python"} so the session-level GC sweep in conftest
+        # catches it if cleanup at the end of this test doesn't run.
+        channel2 = client.channel(
+            "messaging",
+            str(uuid.uuid4()),
+            {"test": True, "language": "python"},
         )
+        await channel2.create(random_user["id"])
 
-        assert "drafts" in response
-        assert len(response["drafts"]) == 1
-        draft = response["drafts"][0]
-        assert draft["channel_cid"] == channel2.cid
-        assert draft["message"]["text"] == "Draft in channel 2"
-
-        # Query drafts with sort
-        response = await client.query_drafts(
-            random_user["id"],
-            sort=[{"field": "created_at", "direction": SortOrder.ASC}],
-        )
-
-        assert "drafts" in response
-        assert len(response["drafts"]) == 2
-        assert response["drafts"][0]["channel_cid"] == channel.cid
-        assert response["drafts"][1]["channel_cid"] == channel2.cid
-
-        # Query drafts with pagination
-        response = await client.query_drafts(
-            random_user["id"],
-            options={"limit": 1},
-        )
-
-        assert "drafts" in response
-        assert len(response["drafts"]) == 1
-        assert response["drafts"][0]["channel_cid"] == channel2.cid
-
-        assert response["next"] is not None
-
-        # Query drafts with pagination
-        response = await client.query_drafts(
-            random_user["id"],
-            options={"limit": 1, "next": response["next"]},
-        )
-
-        assert "drafts" in response
-        assert len(response["drafts"]) == 1
-        assert response["drafts"][0]["channel_cid"] == channel.cid
-
-        # Cleanup
         try:
-            await channel2.delete()
-        except Exception:
-            pass
+            draft2 = {"text": "Draft in channel 2"}
+            await channel2.create_draft(draft2, random_user["id"])
+
+            # Query all drafts for the user
+            response = await client.query_drafts(random_user["id"])
+
+            assert "drafts" in response
+            assert len(response["drafts"]) == 2
+
+            # Query drafts for a specific channel
+            response = await client.query_drafts(
+                random_user["id"], filter={"channel_cid": channel2.cid}
+            )
+
+            assert "drafts" in response
+            assert len(response["drafts"]) == 1
+            draft = response["drafts"][0]
+            assert draft["channel_cid"] == channel2.cid
+            assert draft["message"]["text"] == "Draft in channel 2"
+
+            # Query drafts with sort
+            response = await client.query_drafts(
+                random_user["id"],
+                sort=[{"field": "created_at", "direction": SortOrder.ASC}],
+            )
+
+            assert "drafts" in response
+            assert len(response["drafts"]) == 2
+            assert response["drafts"][0]["channel_cid"] == channel.cid
+            assert response["drafts"][1]["channel_cid"] == channel2.cid
+
+            # Query drafts with pagination
+            response = await client.query_drafts(
+                random_user["id"],
+                options={"limit": 1},
+            )
+
+            assert "drafts" in response
+            assert len(response["drafts"]) == 1
+            assert response["drafts"][0]["channel_cid"] == channel2.cid
+
+            assert response["next"] is not None
+
+            # Query drafts with pagination
+            response = await client.query_drafts(
+                random_user["id"],
+                options={"limit": 1, "next": response["next"]},
+            )
+
+            assert "drafts" in response
+            assert len(response["drafts"]) == 1
+            assert response["drafts"][0]["channel_cid"] == channel.cid
+        finally:
+            # Hard-delete via the synchronous channel.delete (HTTP DELETE)
+            # rather than the async-task delete_channels helper. Failures
+            # surface in CI logs instead of being swallowed silently.
+            try:
+                await channel2.delete(hard=True)
+            except Exception as exc:
+                import sys
+
+                print(
+                    f"[cleanup] channel {channel2.cid} delete failed: "
+                    f"{exc.__class__.__name__}: {exc}",
+                    file=sys.stderr,
+                )