Fixed Engine.release method to release connection in any way. Added…

… `test_release_broken_connection` to reproduce this case.

aiopg/sa/engine.py

@@ -171,9 +171,6 @@ async def _acquire(self):
 
     def release(self, conn):
         """Revert back connection to pool."""
-        if conn.in_transaction:
-            raise InvalidRequestError("Cannot release a connection with "
-                                      "not finished transaction")
         raw = conn.connection
         fut = self._pool.release(raw)
         return fut

tests/test_sa_engine.py

@@ -1,4 +1,5 @@
 import asyncio
+import psycopg2
 from aiopg.connection import TIMEOUT
 from psycopg2.extensions import parse_dsn
 
@@ -78,10 +79,10 @@ def test_not_context_manager(engine):
 async def test_release_transacted(engine):
     conn = await engine.acquire()
     tr = await conn.begin()
-    with pytest.raises(sa.InvalidRequestError):
+    with pytest.warns(ResourceWarning, match='Invalid transaction status'):
         engine.release(conn)
     del tr
-    await conn.close()
+    assert conn.closed
 
 
 def test_timeout(engine):
@@ -145,3 +146,23 @@ async def test_terminate_with_acquired_connections(make_engine):
     await engine.wait_closed()
 
     assert conn.closed
+
+
+async def test_release_broken_connection(make_engine):
+    engine = await make_engine()
+    # FIXME: it is better to see psycopg2.OperationalError("server closed
+    #  the connection unexpectedly ...") here instead of this InterfaceError
+    with pytest.raises(psycopg2.InterfaceError,
+                       match='connection already closed'):
+        async with engine.acquire() as conn:
+            async with conn.begin():
+                await conn.execute('select 1;')
+                psycopg2.connect(engine.dsn).cursor().execute("""
+                SELECT pg_terminate_backend(pg_stat_activity.pid)
+                FROM pg_stat_activity
+                WHERE pg_stat_activity.datname = 'postgres'
+                  AND pid <> pg_backend_pid();
+                """)
+                await conn.execute('select 1;')  # should issue a rollback
+            # connection should be closed and released on exit
+    assert engine.size == 0  # means that the pool is in a valid state