Fixed unclosed transactions + pool (see #32)

Author: popravich

aioredis/commands/__init__.py

@@ -50,6 +50,11 @@ def connection(self):
         """:class:`aioredis.RedisConnection` instance."""
         return self._conn
 
+    @property
+    def in_transaction(self):
+        """Set to True when MULTI command was issued."""
+        return self._conn.in_transaction
+
     @property
     def closed(self):
         """True if connection is closed."""

aioredis/pool.py

@@ -136,7 +136,10 @@ def release(self, conn):
         assert conn in self._used, "Invalid connection, maybe from other pool"
         self._used.remove(conn)
         if not conn.closed:
-            if conn.db == self.db:
+            if conn.in_transaction:
+                # TODO: log warning
+                conn.close()
+            elif conn.db == self.db:
                 try:
                     self._pool.put_nowait(conn)
                 except asyncio.QueueFull:

tests/_testutil.py

@@ -1,7 +1,5 @@
 import asyncio
 import unittest
-import socket
-import random
 import os
 
 from functools import wraps
@@ -34,18 +32,6 @@ def tearDown(self):
         self.loop.close()
         del self.loop
 
-    def _find_port(self):
-        s = socket.socket()
-        while True:
-            port = random.randint(1024, 65535)
-            try:
-                s.bind(('127.0.0.1', port))
-            except OSError:
-                pass
-            else:
-                s.close()
-                return port
-
 
 class RedisTest(BaseTest):
 

tests/pool_test.py

@@ -243,6 +243,7 @@ def test_crappy_multiexec(self):
             yield from redis.set('abc', 'def')
             yield from redis.multi()
             yield from redis.set('abc', 'fgh')
+        self.assertTrue(redis.closed)
         with (yield from pool) as redis:
             value = yield from redis.get('abc')
         self.assertEquals(value, 'def')