Better tests for async connect / close

README.md

@@ -10,7 +10,7 @@ ORM powered by **[asyncio](https://docs.python.org/3/library/asyncio.html)**.
 Overview
 --------
 
-* Works on Python 3.4+
+* Requires Python 3.4+
 * Has support for PostgreSQL via [aiopg](https://github.com/aio-libs/aiopg)
 * Has support for MySQL via [aiomysql](https://github.com/aio-libs/aiomysql)
 * Single point for high-level async API

tests/__init__.py

@@ -399,13 +399,33 @@ class ManagerTestCase(BaseManagerTestCase):
 
     def test_connect_close(self):
         @asyncio.coroutine
-        def test(objects):
-            yield from objects.connect()
-            self.assertTrue(objects.is_connected)
+        def get_conn(objects):
             yield from objects.connect()
+            yield from asyncio.sleep(0.125, loop=self.loop)
+            # NOTE: "private" member access
+            return objects.database._async_conn
+
+        @asyncio.coroutine
+        def test(objects):
+            c1 = yield from get_conn(objects)
+            c2 = yield from get_conn(objects)
+            self.assertEqual(c1, c2)
             self.assertTrue(objects.is_connected)
+
             yield from objects.close()
             self.assertTrue(not objects.is_connected)
+
+            done, not_done = yield from asyncio.wait([
+                get_conn(objects),
+                get_conn(objects),
+                get_conn(objects),
+            ], loop=self.loop)
+
+            conn = next(iter(done)).result()
+            self.assertEqual(len(done), 3)
+            self.assertTrue(objects.is_connected)
+            self.assertTrue(all(map(lambda t: t.result() == conn, done)))
+
             yield from objects.close()
             self.assertTrue(not objects.is_connected)