chore: Improve client connection checking in pytests

src/facade/dragonfly_connection.cc

@@ -499,9 +499,13 @@ string Connection::GetClientInfo() const {
   absl::StrAppend(&res, " laddr=", le.address().to_string(), ":", le.port());
   absl::StrAppend(&res, " fd=", lsb->native_handle(), " name=", name_);
   absl::StrAppend(&res, " age=", now - creation_time_, " idle=", now - last_interaction_);
-  absl::StrAppend(&res, " phase=", phase_, " ");
+  absl::StrAppend(&res, " phase=", phase_);
+
   if (cc_) {
-    absl::StrAppend(&res, service_->GetContextInfo(cc_.get()));
+    string cc_info = service_->GetContextInfo(cc_.get());
+    if (!cc_info.empty()) {
+      absl::StrAppend(&res, " ", cc_info);
+    }
   }
 
   return res;

tests/dragonfly/cluster_test.py

@@ -9,13 +9,13 @@
 
 @dfly_args({})
 class TestNotEmulated:
-    def test_cluster_commands_fails_when_not_emulate(self, client: redis.Redis):
-        with pytest.raises(redis.ResponseError) as respErr:
-            client.execute_command("CLUSTER HELP")
+    async def test_cluster_commands_fails_when_not_emulate(self, async_client: aioredis.Redis):
+        with pytest.raises(aioredis.ResponseError) as respErr:
+            await async_client.execute_command("CLUSTER HELP")
         assert "cluster_mode" in str(respErr.value)
 
-        with pytest.raises(redis.ResponseError) as respErr:
-            client.execute_command("CLUSTER SLOTS")
+        with pytest.raises(aioredis.ResponseError) as respErr:
+            await async_client.execute_command("CLUSTER SLOTS")
         assert "emulated" in str(respErr.value)
 
 
@@ -76,7 +76,6 @@ def is_local_host(ip: str) -> bool:
 
 
 @dfly_args({"proactor_threads": 4, "cluster_mode": "emulated"})
-@pytest.mark.asyncio
 async def test_cluster_slots_in_replicas(df_local_factory):
     master = df_local_factory.create(port=BASE_PORT)
     replica = df_local_factory.create(port=BASE_PORT+1, logtostdout=True)
@@ -107,11 +106,8 @@ async def test_cluster_slots_in_replicas(df_local_factory):
 
 
 @dfly_args({"cluster_mode": "emulated", "cluster_announce_ip": "127.0.0.2"})
-@pytest.mark.asyncio
-async def test_cluster_info(async_pool):
-    conn = aioredis.Redis(connection_pool=async_pool)
-
-    res = await conn.execute_command("CLUSTER INFO")
+async def test_cluster_info(async_client):
+    res = await async_client.execute_command("CLUSTER INFO")
     assert len(res) == 16
     assert res == {'cluster_current_epoch': '1',
                    'cluster_known_nodes': '1',
@@ -134,10 +130,8 @@ async def test_cluster_info(async_pool):
 
 @dfly_args({"cluster_mode": "emulated", "cluster_announce_ip": "127.0.0.2"})
 @pytest.mark.asyncio
-async def test_cluster_nodes(async_pool):
-    conn = aioredis.Redis(connection_pool=async_pool)
-
-    res = await conn.execute_command("CLUSTER NODES")
+async def test_cluster_nodes(async_client):
+    res = await async_client.execute_command("CLUSTER NODES")
     assert len(res) == 1
     info = res['127.0.0.2:6379@6379']
     assert res is not None

tests/dragonfly/conftest.py

@@ -4,6 +4,7 @@
 
 import os
 import sys
+from time import sleep
 import pytest
 import pytest_asyncio
 import redis
@@ -88,38 +89,39 @@ def df_server(df_factory: DflyInstanceFactory) -> DflyInstance:
     instance.start()
 
     yield instance
-
     clients_left = None
     try:
         client = redis.Redis(port=instance.port)
-        clients_left = client.execute_command("INFO")['connected_clients']
+        client.client_setname("mgr")
+        sleep(0.1)
+        clients_left = [x for x in client.client_list() if x["name"] != "mgr"]
     except Exception as e:
         print(e, file=sys.stderr)
 
     instance.stop()
-    assert clients_left == 1
+    assert clients_left == []
 
 
 @pytest.fixture(scope="class")
 def connection(df_server: DflyInstance):
     return redis.Connection(port=df_server.port)
 
 
-@pytest.fixture(scope="class")
-def sync_pool(df_server: DflyInstance):
-    pool = redis.ConnectionPool(decode_responses=True, port=df_server.port)
-    yield pool
-    pool.disconnect()
+# @pytest.fixture(scope="class")
+# def sync_pool(df_server: DflyInstance):
+#     pool = redis.ConnectionPool(decode_responses=True, port=df_server.port)
+#     yield pool
+#     pool.disconnect()
 
 
-@pytest.fixture(scope="class")
-def client(sync_pool):
-    """
-    Return a client to the default instance with all entries flushed.
-    """
-    client = redis.Redis(connection_pool=sync_pool)
-    client.flushall()
-    return client
+# @pytest.fixture(scope="class")
+# def client(sync_pool):
+#     """
+#     Return a client to the default instance with all entries flushed.
+#     """
+#     client = redis.Redis(connection_pool=sync_pool)
+#     client.flushall()
+#     return client
 
 
 @pytest.fixture(scope="function")
@@ -149,8 +151,9 @@ async def async_client(async_pool):
     Return an async client to the default instance with all entries flushed.
     """
     client = aioredis.Redis(connection_pool=async_pool)
+    await client.client_setname("test")
     await client.flushall()
-    return client
+    yield client
 
 
 def pytest_addoption(parser):

tests/dragonfly/connection_test.py

@@ -281,7 +281,7 @@ def generate(max):
 
 
 @pytest.mark.asyncio
-async def test_subsribers_with_active_publisher(df_server: DflyInstance, max_connections=100):
+async def test_subscribers_with_active_publisher(df_server: DflyInstance, max_connections=100):
     # TODO: I am not how to customize the max connections for the pool.
     async_pool = aioredis.ConnectionPool(host="localhost", port=df_server.port,
                                          db=0, decode_responses=True, max_connections=max_connections)
@@ -316,7 +316,6 @@ async def subscribe_worker():
     await async_pool.disconnect()
 
 
-@pytest.mark.asyncio
 async def test_big_command(df_server, size=8 * 1024):
     reader, writer = await asyncio.open_connection('127.0.0.1', df_server.port)
 
@@ -328,7 +327,7 @@ async def test_big_command(df_server, size=8 * 1024):
     writer.close()
     await writer.wait_closed()
 
-@pytest.mark.asyncio
+
 async def test_subscribe_pipelined(async_client: aioredis.Redis):
     pipe = async_client.pipeline(transaction=False)
     pipe.execute_command('subscribe channel').execute_command('subscribe channel')

tests/dragonfly/eval_test.py

@@ -93,7 +93,6 @@ def DJANGO_CACHEOPS_SCHEMA(vs): return {
 
 @dfly_multi_test_args({'default_lua_config': 'allow-undeclared-keys', 'proactor_threads': 4},
                       {'default_lua_config': 'allow-undeclared-keys disable-atomicity', 'proactor_threads': 4})
-@pytest.mark.asyncio
 async def test_django_cacheops_script(async_client, num_keys=500):
     script = async_client.register_script(DJANGO_CACHEOPS_SCRIPT)
 
@@ -161,7 +160,6 @@ async def test_django_cacheops_script(async_client, num_keys=500):
 
 @dfly_multi_test_args({'default_lua_config': 'allow-undeclared-keys', 'proactor_threads': 4},
                       {'default_lua_config': 'allow-undeclared-keys disable-atomicity', 'proactor_threads': 4})
-@pytest.mark.asyncio
 async def test_golang_asynq_script(async_pool, num_queues=10, num_tasks=100):
     async def enqueue_worker(queue):
         client = aioredis.Redis(connection_pool=async_pool)

tests/dragonfly/generic_test.py

@@ -7,12 +7,12 @@
 
 @dfly_multi_test_args({'keys_output_limit': 512}, {'keys_output_limit': 1024})
 class TestKeys:
-    def test_max_keys(self, client, df_server):
+    async def test_max_keys(self, async_client: aioredis.Redis, df_server):
         max_keys = df_server['keys_output_limit']
-
-        batch_fill_data(client, gen_test_data(max_keys * 3))
-
-        keys = client.keys()
+        pipe = async_client.pipeline()
+        batch_fill_data(pipe, gen_test_data(max_keys * 3))
+        await pipe.execute()
+        keys = await async_client.keys()
         assert len(keys) in range(max_keys, max_keys+512)
 
 @pytest.fixture(scope="function")
@@ -22,7 +22,6 @@ def export_dfly_password() -> str:
     yield pwd
     del os.environ['DFLY_PASSWORD']
 
-@pytest.mark.asyncio
 async def test_password(df_local_factory, export_dfly_password):
     dfly = df_local_factory.create()
     dfly.start()

tests/dragonfly/redis_replicaiton_test.py

@@ -85,7 +85,6 @@ def redis_server() -> RedisServer:
 ]
 
 
-@pytest.mark.asyncio
 @pytest.mark.parametrize("t_replicas, seeder_config", full_sync_replication_specs)
 async def test_replication_full_sync(df_local_factory, df_seeder_factory, redis_server, t_replicas, seeder_config):
     master = redis_server
@@ -117,7 +116,6 @@ async def test_replication_full_sync(df_local_factory, df_seeder_factory, redis_
 ]
 
 
-@pytest.mark.asyncio
 @pytest.mark.parametrize("t_replicas, seeder_config", stable_sync_replication_specs)
 async def test_replication_stable_sync(df_local_factory, df_seeder_factory, redis_server, t_replicas, seeder_config):
     master = redis_server
@@ -153,7 +151,6 @@ async def test_replication_stable_sync(df_local_factory, df_seeder_factory, redi
 ]
 
 
-@pytest.mark.asyncio
 @pytest.mark.parametrize("t_replicas, seeder_config", replication_specs)
 async def test_redis_replication_all(df_local_factory, df_seeder_factory, redis_server, t_replicas, seeder_config):
     master = redis_server
@@ -210,7 +207,6 @@ async def run_replication(c_replica):
 ]
 
 
-@pytest.mark.asyncio
 @pytest.mark.parametrize("t_replicas, t_disconnect, seeder_config", master_disconnect_cases)
 async def test_disconnect_master(df_local_factory, df_seeder_factory, redis_server, t_replicas, t_disconnect, seeder_config):
 

tests/dragonfly/replication_test.py

@@ -7,7 +7,7 @@
 import re
 
 from .utility import *
-from . import dfly_args
+from . import DflyInstanceFactory, dfly_args
 
 
 BASE_PORT = 1111
@@ -147,7 +147,7 @@ async def check_data(seeder, replicas, c_replicas):
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("t_master, t_crash_fs, t_crash_ss, t_disonnect, n_keys", disconnect_cases)
-async def test_disconnect_replica(df_local_factory, df_seeder_factory, t_master, t_crash_fs, t_crash_ss, t_disonnect, n_keys):
+async def test_disconnect_replica(df_local_factory: DflyInstanceFactory, df_seeder_factory, t_master, t_crash_fs, t_crash_ss, t_disonnect, n_keys):
     master = df_local_factory.create(port=BASE_PORT, proactor_threads=t_master)
     replicas = [
         (df_local_factory.create(
@@ -252,7 +252,7 @@ async def disconnect(replica, c_replica, crash_type):
 
     # Check master survived all disconnects
     assert await c_master.ping()
-
+    await c_master.close()
 
 """
 Test stopping master during different phases.
@@ -381,6 +381,7 @@ async def test_rotating_masters(df_local_factory, df_seeder_factory, t_replica,
         fill_seeder.stop()
         fill_task.cancel()
 
+
 """
 Test flushall command. Set data to master send flashall and set more data.
 Check replica keys at the end.

tests/dragonfly/server_family_test.py

@@ -12,7 +12,6 @@ def test_quit(connection):
 
 
 def test_quit_after_sub(connection):
-    connection = redis.Connection()
     connection.send_command("SUBSCRIBE", "foo")
     connection.read_response()
 
@@ -23,11 +22,11 @@ def test_quit_after_sub(connection):
         connection.read_response()
 
 
-def test_multi_exec(client):
-    pipeline = client.pipeline()
+async def test_multi_exec(async_client: aioredis.Redis):
+    pipeline = async_client.pipeline()
     pipeline.set("foo", "bar")
     pipeline.get("foo")
-    val = pipeline.execute()
+    val = await pipeline.execute()
     assert val == [True, "bar"]
 
 
@@ -42,43 +41,39 @@ def test_multi_exec(client):
 '''
 
 
-def test_multi_eval(client):
+async def test_multi_eval(async_client: aioredis.Redis):
     try:
-        pipeline = client.pipeline()
+        pipeline = async_client.pipeline()
         pipeline.set("foo", "bar")
         pipeline.get("foo")
         pipeline.eval("return 43", 0)
-        assert True, "This part should not executed due to issue #457"
-
-        val = pipeline.execute()
+        val = await pipeline.execute()
         assert val == "foo"
     except Exception as e:
         msg = str(e)
         assert "Dragonfly does not allow execution of" in msg
 
 
-def test_connection_name(client):
-    name = client.execute_command("CLIENT GETNAME")
-    assert not name
-    client.execute_command("CLIENT SETNAME test_conn_name")
-    name = client.execute_command("CLIENT GETNAME")
+async def test_connection_name(async_client: aioredis.Redis):
+    name = await async_client.execute_command("CLIENT GETNAME")
+    assert name == "test"
+    await async_client.execute_command("CLIENT SETNAME test_conn_name")
+    name = await async_client.execute_command("CLIENT GETNAME")
     assert name == "test_conn_name"
 
 
 '''
 make sure that the scan command is working with python
 '''
-
-
-def test_scan(client):
+async def test_scan(async_client: aioredis.Redis):
     def gen_test_data():
         for i in range(10):
             yield f"key-{i}", f"value-{i}"
 
     for key, val in gen_test_data():
-        res = client.set(key, val)
+        res = await async_client.set(key, val)
         assert res is not None
-        cur, keys = client.scan(cursor=0, match=key, count=2)
+        cur, keys = await async_client.scan(cursor=0, match=key, count=2)
         assert cur == 0
         assert len(keys) == 1
         assert keys[0] == key

tests/pytest.ini

@@ -4,3 +4,4 @@ log_cli_format = [%(asctime)s.%(msecs)03d %(levelname)s] %(message)s
 log_date_format = %Y-%m-%d %H:%M:%S
 log_file_level=DEBUG
 log_cli = true
+asyncio_mode=auto