Add _lock to BlockingConnectionPool

CHANGES/1068.bugfix

@@ -0,0 +1 @@
+Adds a lock to ``aioredis.connection.ConnectionPool``

aioredis/connection.py

@@ -1267,7 +1267,7 @@ def __init__(
         # will notice the first thread already did the work and simply
         # release the lock.
         self._fork_lock = threading.Lock()
-        self._lock: asyncio.Lock
+        self._lock = asyncio.Lock()
         self._created_connections: int
         self._available_connections: List[Connection]
         self._in_use_connections: Set[Connection]

tests/test_connection_pool.py

@@ -24,6 +24,9 @@ def __init__(self, **kwargs):
     async def connect(self):
         pass
 
+    async def disconnect(self):
+        pass
+
     async def can_read(self, timeout: float = 0):
         return False
 
@@ -122,6 +125,13 @@ async def test_connection_creation(self, master_host):
         assert isinstance(connection, DummyConnection)
         assert connection.kwargs == connection_kwargs
 
+    async def test_disconnect(self, master_host):
+        """A regression test for #1047"""
+        connection_kwargs = {"foo": "bar", "biz": "baz", "host": master_host}
+        pool = self.get_pool(connection_kwargs=connection_kwargs)
+        await pool.get_connection("_")
+        await pool.disconnect()
+
     async def test_multiple_connections(self, master_host):
         connection_kwargs = {"host": master_host}
         pool = self.get_pool(connection_kwargs=connection_kwargs)