fix linters + test

Author: pinguingman

taskiq_redis/redis_cluster_broker.py

@@ -60,7 +60,7 @@ async def kick(self, message: BrokerMessage) -> None:
 
         :param message: message to append.
         """
-        await self.redis.lpush(self.queue_name, message.message)
+        await self.redis.lpush(self.queue_name, message.message)  # type: ignore[attr-defined]
 
     async def listen(self) -> AsyncGenerator[bytes, None]:
         """
@@ -73,6 +73,5 @@ async def listen(self) -> AsyncGenerator[bytes, None]:
         """
         redis_brpop_data_position = 1
         while True:
-            yield (await self.redis.brpop([self.queue_name]))[
-                redis_brpop_data_position
-            ]
+            value = await self.redis.brpop([self.queue_name])  # type: ignore[attr-defined]
+            yield value[redis_brpop_data_position]

tests/test_broker.py

@@ -5,7 +5,7 @@
 import pytest
 from taskiq import AckableMessage, AsyncBroker, BrokerMessage
 
-from taskiq_redis import ListQueueBroker, PubSubBroker, ListQueueClusterBroker
+from taskiq_redis import ListQueueBroker, ListQueueClusterBroker, PubSubBroker
 
 
 def test_no_url_should_raise_typeerror() -> None:
@@ -109,21 +109,17 @@ async def test_list_queue_cluster_broker(
     We create two workers that listen and send a message to them.
     Expect only one worker to receive the same message we sent.
     """
-
-    print(f"redis_cluster_url: {redis_cluster_url}")
     broker = ListQueueClusterBroker(
-        url=redis_cluster_url, queue_name=uuid.uuid4().hex
+        url=redis_cluster_url,
+        queue_name=uuid.uuid4().hex,
     )
-    worker1_task = asyncio.create_task(get_message(broker))
-    worker2_task = asyncio.create_task(get_message(broker))
+    worker_task = asyncio.create_task(get_message(broker))
     await asyncio.sleep(0.3)
 
     await broker.kick(valid_broker_message)
     await asyncio.sleep(0.3)
 
-    assert worker1_task.done() != worker2_task.done()
-    message = worker1_task.result() if worker1_task.done() else worker2_task.result()
-    assert message == valid_broker_message.message
-    worker1_task.cancel()
-    worker2_task.cancel()
+    assert worker_task.done()
+    assert worker_task.result() == valid_broker_message.message
+    worker_task.cancel()
     await broker.shutdown()