Improve testing connection closes for parallel test run (#600)

* close connection used to clear database
get connection count from proper database instead of whole db

* add connection closing

Author: Lagovas

tests/test.py

@@ -202,6 +202,7 @@
     TEST_MYSQL = True
     connect_args = {
         'user': DB_USER, 'password': DB_USER_PASSWORD,
+        'database': DB_NAME,
         'read_timeout': SOCKET_CONNECT_TIMEOUT,
         'write_timeout': SOCKET_CONNECT_TIMEOUT,
     }
@@ -764,6 +765,7 @@ def drop_tables():
                 '{}://{}:{}/{}'.format(DB_DRIVER, DB_HOST, DB_PORT, DB_NAME),
                 connect_args=connect_args)
     metadata.drop_all(engine_raw)
+    engine_raw.dispose()
 
 
 # Set this to False to not rebuild binaries on setup.
@@ -2605,11 +2607,11 @@ def tearDown(self):
 
     def getActiveConnectionCount(self, cursor):
         if TEST_MYSQL:
-            query = "SHOW STATUS WHERE `variable_name` = 'Threads_connected';"
-            cursor.execute(query)
-            return int(cursor.fetchone()[1])
+            query = "select count(*) from information_schema.processlist where db=%s;"
+            cursor.execute(query, [DB_NAME])
+            return int(cursor.fetchone()[0])
         else:
-            cursor.execute('select count(*) from pg_stat_activity;')
+            cursor.execute('SELECT numbackends FROM pg_stat_database where datname=%s;', [DB_NAME])
             return int(cursor.fetchone()[0])
 
     def getConnectionLimit(self, connection=None):
@@ -2705,7 +2707,6 @@ def testClosingConnectionsWithDB(self):
             connection.autocommit = True
             with TestConnectionClosing.mysql_closing(connection.cursor()) as cursor:
                 current_connection_count = self.getActiveConnectionCount(cursor)
-
                 with self.get_connection():
                     self.assertEqual(self.getActiveConnectionCount(cursor),
                                      current_connection_count+1)
@@ -10814,7 +10815,7 @@ def setUp(self):
 
     def testPlainConnectionAfterDeny(self):
         async def _testPlainConnectionAfterDeny():
-            # We use raw connecitons to specify ssl='prefer'
+            # We use raw connections to specify ssl='prefer'
             # which would ask for ssl connection first.
             # And then after receiving a deny, it would ask for a plain connection
             conn = await asyncpg.connect(
@@ -10824,6 +10825,7 @@ async def _testPlainConnectionAfterDeny():
                 **asyncpg_connect_args
             )
             await conn.fetch('SELECT 1', timeout=STATEMENT_TIMEOUT)
+            await conn.close()
 
         loop = asyncio.new_event_loop()  # create new to avoid concurrent usage of the loop in the current thread and allow parallel execution in the future
         loop.run_until_complete(_testPlainConnectionAfterDeny())
@@ -11025,6 +11027,7 @@ async def test():
             await conn.execute(insert_query, data['id'], data['value_bytes'])
             row = await conn.fetchrow(select_query, data['id'])
             self.assertEqual(data['value_bytes'], row['value_bytes'])
+            await conn.close()
 
         loop = asyncio.new_event_loop()
         loop.run_until_complete(test())
@@ -11077,6 +11080,7 @@ async def test():
             # that our data is not saved due to the rollback.
             row = await conn.fetchrow(select_query, data['id'])
             self.assertEqual(row, None)
+            await conn.close()
 
         loop = asyncio.new_event_loop()
         loop.run_until_complete(test())
@@ -11143,6 +11147,7 @@ async def test():
 
             row = await conn.fetchrow(select_query, data['id'])
             self.assertEqual(row, None)
+            await conn.close()
 
         loop = asyncio.new_event_loop()
         loop.run_until_complete(test())