Use blockbuster to detect blocking calls (#324)

Author: cbornet

astrapy/data/database.py

@@ -3226,7 +3226,7 @@ async def _list_tables_ctx(
         driver_commander = self._get_driver_commander(keyspace=keyspace)
         lt_payload = {"listTables": {"options": {"explain": True}}}
         logger.info("listTables")
-        lt_response = driver_commander.request(
+        lt_response = await driver_commander.async_request(
             payload=lt_payload,
             timeout_context=timeout_context,
         )

poetry.lock

@@ -50,6 +50,7 @@ pytest-httpserver = "~1.0.8"
 testcontainers = "~3.7.1"
 ruff = "^0.6.8"
 types-toml = "^0.10.8.7"
+blockbuster = "~1.5.5"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]

pyproject.toml

@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import time
 from typing import Any, Awaitable, Callable
 
@@ -72,7 +73,7 @@ async def await_until_true(
     while not (await acondition()):
         if time.time() - ini_time > max_seconds:
             raise ValueError("Timed out on condition.")
-        time.sleep(poll_interval)
+        await asyncio.sleep(poll_interval)
 
 
 @pytest.mark.skipif(not IS_ASTRA_DB, reason="Not supported outside of Astra DB")

tests/admin/integration/test_admin.py

@@ -14,7 +14,7 @@
 
 from __future__ import annotations
 
-import time
+import asyncio
 
 import pytest
 
@@ -146,7 +146,7 @@ async def test_collection_default_id_type_async(
         assert isinstance(doc["_id"], UUID)
         await acol.drop()
 
-        time.sleep(2)
+        await asyncio.sleep(2)
         acol = await async_database.create_collection(
             ID_TEST_COLLECTION_NAME_ROOT + DefaultIdType.UUIDV6,
             definition=CollectionDefinition(
@@ -168,7 +168,7 @@ async def test_collection_default_id_type_async(
         assert doc["_id"].version == 6
         await acol.drop()
 
-        time.sleep(2)
+        await asyncio.sleep(2)
         acol = await async_database.create_collection(
             ID_TEST_COLLECTION_NAME_ROOT + DefaultIdType.UUIDV7,
             definition=CollectionDefinition(
@@ -190,7 +190,7 @@ async def test_collection_default_id_type_async(
         assert doc["_id"].version == 7
         await acol.drop()
 
-        time.sleep(2)
+        await asyncio.sleep(2)
         acol = await async_database.create_collection(
             ID_TEST_COLLECTION_NAME_ROOT + DefaultIdType.DEFAULT,
             definition=CollectionDefinition(
@@ -205,7 +205,7 @@ async def test_collection_default_id_type_async(
         assert acol_options.default_id.default_id_type == DefaultIdType.DEFAULT
         await acol.drop()
 
-        time.sleep(2)
+        await asyncio.sleep(2)
         acol = await async_database.create_collection(
             ID_TEST_COLLECTION_NAME_ROOT + DefaultIdType.OBJECTID,
             definition=CollectionDefinition(
@@ -429,7 +429,7 @@ async def test_tokenless_client_async(
     @pytest.mark.describe(
         "test database-from-admin default keyspace per environment, async"
     )
-    async def test_database_from_admin_default_keyspace_per_environment_async(
+    def test_async_database_from_admin_default_keyspace_per_environment(
         self,
         data_api_credentials_kwargs: DataAPICredentials,
         data_api_credentials_info: DataAPICredentialsInfo,
@@ -448,7 +448,7 @@ async def test_database_from_admin_default_keyspace_per_environment_async(
     @pytest.mark.describe(
         "test database-from-astradbadmin default keyspace per environment, async"
     )
-    async def test_database_from_astradbadmin_default_keyspace_per_environment_async(
+    def test_async_database_from_astradbadmin_default_keyspace_per_environment(
         self,
         data_api_credentials_kwargs: DataAPICredentials,
         data_api_credentials_info: DataAPICredentialsInfo,

tests/base/integration/collections/test_collection_ddl_async.py

@@ -80,11 +80,11 @@ async def test_cursor_overalltimeout_exceptions_async(
         acol = async_empty_collection
         await acol.insert_many([{"a": 1}] * 1000)
 
-        await acol.distinct("a", timeout_ms=20000)
+        await acol.distinct("a", timeout_ms=60000)
         with pytest.raises(DataAPITimeoutException):
             await acol.distinct("a", timeout_ms=1)
 
-        await acol.distinct("a", timeout_ms=20000)
+        await acol.distinct("a", timeout_ms=60000)
         with pytest.raises(DataAPITimeoutException):
             await acol.distinct("a", timeout_ms=1)
 

tests/base/integration/collections/test_collection_timeout_async.py

@@ -29,7 +29,7 @@ async def test_collection_methods_vectorize_async(
         async_database: AsyncDatabase,
     ) -> None:
         database_admin = async_database.get_database_admin()
-        ep_result = database_admin.find_embedding_providers()
+        ep_result = await database_admin.async_find_embedding_providers()
 
         assert isinstance(ep_result, FindEmbeddingProvidersResult)
 

tests/base/integration/misc/test_vectorize_ops_async.py

@@ -20,9 +20,11 @@
 
 import functools
 import warnings
+from collections.abc import Iterator
 from typing import Any, Awaitable, Callable, Iterable, TypedDict
 
 import pytest
+from blockbuster import BlockBuster, blockbuster_ctx
 from deprecation import UnsupportedWarning
 
 from astrapy import AsyncDatabase, DataAPIClient, Database
@@ -51,6 +53,14 @@
 )
 
 
+@pytest.fixture(autouse=True)
+def blockbuster() -> Iterator[BlockBuster]:
+    with blockbuster_ctx() as bb:
+        # TODO: follow discussion in https://github.com/encode/httpx/discussions/3456
+        bb.functions["os.stat"].can_block_in("httpx/_client.py", "_init_transport")
+        yield bb
+
+
 class DataAPICredentials(TypedDict):
     token: str | TokenProvider
     api_endpoint: str