test.py topology: test with asyncio

Run test async using a wrapper for Cassandra python driver's future.

The wrapper was suggested by a user and brought forward by @fruch.
It's based on https://stackoverflow.com/a/49351069 .

Redefine pytest event_loop fixture to avoid issues with fixtures with
scope bigger than function (like keyspace).
See pytest-dev/pytest-asyncio#68

Convert sample test_null to async. More useful test cases will come
afterwards.

Signed-off-by: Alejo Sanchez <[email protected]>

Author: Alejo Sanchez

test/conftest.py

@@ -10,9 +10,11 @@
 # fixtures directly from test/pylib).
 #
 #
+import asyncio
 from cassandra.auth import PlainTextAuthProvider                         # type: ignore
 from cassandra.cluster import Cluster, ConsistencyLevel                  # type: ignore
 from cassandra.cluster import ExecutionProfile, EXEC_PROFILE_DEFAULT     # type: ignore
+from cassandra.cluster import Session, ResponseFuture                    # type: ignore
 from cassandra.policies import RoundRobinPolicy                          # type: ignore
 from test.pylib.util import unique_name                                  # type: ignore
 import pytest
@@ -30,6 +32,47 @@ def pytest_addoption(parser):
     parser.addoption('--ssl', action='store_true',
                      help='Connect to CQL via an encrypted TLSv1.2 connection')
 
+
+# Change default pytest-asyncio event_loop fixture scope to session to
+# allow async fixtures with scope larger than function. (e.g. keyspace fixture)
+# See https://github.com/pytest-dev/pytest-asyncio/issues/68
+@pytest.fixture(scope="session")
+def event_loop(request):
+    loop = asyncio.get_event_loop_policy().new_event_loop()
+    yield loop
+    loop.close()
+
+
+def _wrap_future(f: ResponseFuture) -> asyncio.Future:
+    """Wrap a cassandra Future into an asyncio.Future object.
+
+    Args:
+        f: future to wrap
+
+    Returns:
+        And asyncio.Future object which can be awaited.
+    """
+    loop = asyncio.get_event_loop()
+    aio_future = loop.create_future()
+
+    def on_result(result):
+        loop.call_soon_threadsafe(aio_future.set_result, result)
+
+    def on_error(exception, *_):
+        loop.call_soon_threadsafe(aio_future.set_exception, exception)
+
+    f.add_callback(on_result)
+    f.add_errback(on_error)
+    return aio_future
+
+
+def run_async(self, *args, **kwargs) -> asyncio.Future:
+    return _wrap_future(self.execute_async(*args, **kwargs))
+
+
+Session.run_async = run_async
+
+
 # "cql" fixture: set up client object for communicating with the CQL API.
 # The host/port combination of the server are determined by the --host and
 # --port options, and defaults to localhost and 9042, respectively.

test/topology/test_null.py

@@ -9,18 +9,19 @@
 
 
 @pytest.fixture(scope="module")
-def table1(cql, test_keyspace):
+async def table1(cql, test_keyspace):
     table = test_keyspace + "." + unique_name()
-    cql.execute(f"CREATE TABLE {table} (p text, c text, v text, primary key (p, c))")
+    await cql.run_async(f"CREATE TABLE {table} (p text, c text, v text, primary key (p, c))")
     yield table
-    cql.execute("DROP TABLE " + table)
+    await cql.run_async("DROP TABLE " + table)
 
 
-def test_delete_empty_string_key(cql, table1):
+@pytest.mark.asyncio
+async def test_delete_empty_string_key(cql, table1):
     s = "foobar"
     # An empty-string clustering *is* allowed:
-    cql.execute(f"DELETE FROM {table1} WHERE p='{s}' AND c=''")
+    await cql.run_async(f"DELETE FROM {table1} WHERE p='{s}' AND c=''")
     # But an empty-string partition key is *not* allowed, with a specific
     # error that a "Key may not be empty":
     with pytest.raises(InvalidRequest, match='Key may not be empty'):
-        cql.execute(f"DELETE FROM {table1} WHERE p='' AND c='{s}'")
+        await cql.run_async(f"DELETE FROM {table1} WHERE p='' AND c='{s}'")