Use blockbuster to detect blocking calls

pyproject.toml

@@ -45,6 +45,7 @@ types-requests = "^2.31.0.20240311"
 testcontainers = "~3.7.1"
 python-dotenv = "~1.0.1"
 isort = "^5.13.2"
+blockbuster = "^1.1.2"
 
 [tool.poetry.scripts]
 cassio-create-init-string = "cassio.config.bundle_management:create_init_string_utility"

src/cassio/table/cql.py

@@ -22,7 +22,9 @@ class CQLOpType(Enum):
 
 INSERT_ROW_CQL_TEMPLATE = """INSERT INTO {{table_fqname}} ({columns_desc}) VALUES ({value_placeholders}) {ttl_spec} ;"""  # noqa: E501
 
-CREATE_INDEX_CQL_PREFIX = "CREATE CUSTOM INDEX IF NOT EXISTS {index_name}_{{table_name}} ON {{table_fqname}} "  # noqa: E501
+CREATE_INDEX_CQL_PREFIX = (
+    "CREATE CUSTOM INDEX IF NOT EXISTS {index_name}_{{table_name}} ON {{table_fqname}} "  # noqa: E501
+)
 
 CREATE_INDEX_CQL_TEMPLATE = (
     CREATE_INDEX_CQL_PREFIX

src/cassio/table/mixins/metadata.py

@@ -34,7 +34,7 @@ def __init__(
 
     @staticmethod
     def _normalize_metadata_indexing_policy(
-        metadata_indexing: Union[Tuple[str, Iterable[str]], str]
+        metadata_indexing: Union[Tuple[str, Iterable[str]], str],
     ) -> MetadataIndexingPolicy:
         mode: MetadataIndexingMode
         fields: Set[str]

src/cassio/table/mixins/vector.py

@@ -50,7 +50,7 @@ async def _aschema_da(self) -> List[ColumnSpecType]:
 
     @staticmethod
     def _get_create_vector_index_cql(
-        vector_index_options: List[Tuple[str, Any]]
+        vector_index_options: List[Tuple[str, Any]],
     ) -> str:
         index_name = "idx_vector"
         index_column = "vector"

tests/conftest.py

@@ -1,12 +1,15 @@
 """
 fixtures for testing
 """
-
+import _thread
 import os
+import threading
 from tempfile import TemporaryDirectory
-from typing import Dict, Iterator, List, Optional, Tuple
+from typing import Dict, Iterator, List, Optional, Tuple, Any
 
 import pytest
+from blockbuster import BlockBuster, blockbuster_ctx
+from blockbuster.blockbuster import BlockBusterFunction
 from cassandra.auth import PlainTextAuthProvider
 from cassandra.cluster import Cluster, Session
 from cassandra.protocol import ProtocolVersion
@@ -24,6 +27,29 @@
 # Fixtures
 
 
+@pytest.fixture(autouse=True)
+def blockbuster() -> Iterator[BlockBuster]:
+    def lock_acquire_exclude(lock: threading.Lock, blocking=True, timeout=-1) -> bool:
+        return not blocking or timeout == 0 or not lock.locked()
+
+    with blockbuster_ctx() as bb:
+        bb.functions["threading.Lock.acquire"] = BlockBusterFunction(
+            _thread.LockType,
+            "acquire",
+            is_immutable=True,
+            can_block_predicate=lock_acquire_exclude,
+        )
+        bb.functions["threading.Lock.acquire"].activate()
+        bb.functions["threading.Lock.acquire_lock"] = BlockBusterFunction(
+            _thread.LockType,
+            "acquire_lock",
+            is_immutable=True,
+            can_block_predicate=lock_acquire_exclude,
+        )
+        bb.functions["threading.Lock.acquire_lock"].activate()
+        yield bb
+
+
 @pytest.fixture(scope="session", autouse=True)
 def cassandra_port(db_keyspace: str) -> Iterator[int]:
     if os.getenv("TEST_DB_MODE", "LOCAL_CASSANDRA") == "TESTCONTAINERS_CASSANDRA":

tests/integration/test_tableclasses_metadatacassandratable.py

@@ -3,6 +3,7 @@
 """
 import asyncio
 import os
+import threading
 
 import pytest
 from cassandra.cluster import Session
@@ -160,6 +161,7 @@ def test_md_routing(self, db_session: Session, db_keyspace: str) -> None:
         assert gotten_deny["metadata"] == test_md_allowdeny_string
         t_deny.clear()
 
+    @pytest.mark.asyncio
     def test_metadata_update_sync(self, db_session: Session, db_keyspace: str) -> None:
         """Consistent behaviour when writing new metadata to an existing row."""
         table_name_fad = "m_ct"
@@ -238,13 +240,17 @@ async def test_metadata_update_asyncio(
     ) -> None:
         """Consistent behaviour when writing new metadata to an existing row."""
         table_name_fad = "m_ct"
-        db_session.execute(f"DROP TABLE IF EXISTS {db_keyspace}.{table_name_fad};")
+        await call_wrapped_async(
+            db_session.execute_async,
+            f"DROP TABLE IF EXISTS {db_keyspace}.{table_name_fad};",
+        )
         t_fad = MetadataCassandraTable(
             session=db_session,
             keyspace=db_keyspace,
             table=table_name_fad,
             primary_key_type="TEXT",
             metadata_indexing=("allow", {"idx", "idx2"}),
+            async_setup=True,
         )
         row_id_to_put_args = {
             f"{'I' if has_idx else '_'}{'U' if has_uid else '_'}": {