Merge branch 'main' into fix-linter-pyteststyle

Author: mergify[bot]

tests/datastore_redis/test_asyncio.py

@@ -129,24 +129,32 @@ async def _test_pipeline(client):
 @background_task()
 def test_async_pubsub(client, loop):
     messages_received = []
+    message_received = asyncio.Event()
 
     async def reader(pubsub):
         while True:
             message = await pubsub.get_message(ignore_subscribe_messages=True)
             if message:
+                message_received.set()
                 messages_received.append(message["data"].decode())
                 if message["data"].decode() == "NOPE":
                     break
 
+    async def _publish(client, channel, message):
+        """Publish a message and wait for the reader to receive it."""
+        await client.publish(channel, message)
+        await asyncio.wait_for(message_received.wait(), timeout=10)
+        message_received.clear()
+
     async def _test_pubsub():
         async with client.pubsub() as pubsub:
             await pubsub.psubscribe("channel:*")
 
             future = asyncio.create_task(reader(pubsub))
 
-            await client.publish("channel:1", "Hello")
-            await client.publish("channel:2", "World")
-            await client.publish("channel:1", "NOPE")
+            await _publish(client, "channel:1", "Hello")
+            await _publish(client, "channel:2", "World")
+            await _publish(client, "channel:1", "NOPE")
 
             await future
 

tests/datastore_valkey/test_asyncio.py

@@ -121,24 +121,32 @@ async def _test_pipeline(client):
 @background_task()
 def test_async_pubsub(client, loop):
     messages_received = []
+    message_received = asyncio.Event()
 
     async def reader(pubsub):
         while True:
             message = await pubsub.get_message(ignore_subscribe_messages=True)
             if message:
+                message_received.set()
                 messages_received.append(message["data"].decode())
                 if message["data"].decode() == "NOPE":
                     break
 
+    async def _publish(client, channel, message):
+        """Publish a message and wait for the reader to receive it."""
+        await client.publish(channel, message)
+        await asyncio.wait_for(message_received.wait(), timeout=10)
+        message_received.clear()
+
     async def _test_pubsub():
         async with client.pubsub() as pubsub:
             await pubsub.psubscribe("channel:*")
 
             future = asyncio.create_task(reader(pubsub))
 
-            await client.publish("channel:1", "Hello")
-            await client.publish("channel:2", "World")
-            await client.publish("channel:1", "NOPE")
+            await _publish(client, "channel:1", "Hello")
+            await _publish(client, "channel:2", "World")
+            await _publish(client, "channel:1", "NOPE")
 
             await future