Add unittest for PubSub reconnect

Author: kristjanvalur

tests/test_asyncio/test_pubsub.py

@@ -1,4 +1,5 @@
 import asyncio
+import functools
 import sys
 from typing import Optional
 
@@ -20,6 +21,18 @@
 pytestmark = pytest.mark.asyncio(forbid_global_loop=True)
 
 
+def with_timeout(t):
+ def wrapper(corofunc):
+ @functools.wraps(corofunc)
+ async def run(*args, **kwargs):
+ async with async_timeout.timeout(t):
+ return await corofunc(*args, **kwargs)
+
+ return run
+
+ return wrapper
+
+
 async def wait_for_message(pubsub, timeout=0.1, ignore_subscribe_messages=False):
  now = asyncio.get_event_loop().time()
  timeout = now + timeout
@@ -603,6 +616,76 @@ async def test_get_message_with_timeout_returns_none(self, r: redis.Redis):
  assert await p.get_message(timeout=0.01) is None
 
 
+@pytest.mark.onlynoncluster
+class TestPubSubReconnect:
+ # @pytest.mark.xfail
+ @with_timeout(2)
+ async def test_reconnect_listen(self, r: redis.Redis):
+ """
+ Test that a loop processing PubSub messages can survive
+ a disconnect, by issuing a connect() call.
+ """
+ messages = asyncio.Queue()
+ pubsub = r.pubsub()
+ interrupt = False
+
+ async def loop():
+ # must make sure the task exits
+ async with async_timeout.timeout(2):
+ nonlocal interrupt
+ await pubsub.subscribe("foo")
+ while True:
+ # print("loop")
+ try:
+ try:
+ await pubsub.connect()
+ await loop_step()
+ # print("succ")
+ except redis.ConnectionError:
+ err = True
+ # print("err")
+ await asyncio.sleep(0.1)
+ except asyncio.CancelledError:
+ # we use a cancel to interrupt the "listen" when we perform a disconnect
+ # print("cancel", interrupt)
+ if interrupt:
+ interrupt = False
+ else:
+ raise
+
+ async def loop_step():
+ # get a single message via listen()
+ async for message in pubsub.listen():
+ await messages.put(message)
+ break
+
+ task = asyncio.get_event_loop().create_task(loop())
+ # get the initial connect message
+ async with async_timeout.timeout(1):
+ message = await messages.get()
+ assert message == {
+ "channel": b"foo",
+ "data": 1,
+ "pattern": None,
+ "type": "subscribe",
+ }
+ # now, disconnect the connection.
+ await pubsub.connection.disconnect()
+ interrupt = True
+ task.cancel() # interrupt the listen call
+ # await another auto-connect message
+ message = await messages.get()
+ assert message == {
+ "channel": b"foo",
+ "data": 1,
+ "pattern": None,
+ "type": "subscribe",
+ }
+ task.cancel()
+ with pytest.raises(asyncio.CancelledError):
+ await task
+
+
 @pytest.mark.onlynoncluster
 class TestPubSubRun:
  async def _subscribe(self, p, *args, **kwargs):