port pep492 tests to pytest

tests/pep492/test_async_iter.py

@@ -1,5 +1,5 @@
-import asyncio
-from tests.base import AIOPyMySQLTestCase
+import pytest
+
 from aiomysql import SSCursor
 from aiomysql import sa
 
@@ -8,77 +8,58 @@
 
 meta = MetaData()
 tbl = Table('tbl', meta,
-            Column('id', Integer, nullable=False,
-                   primary_key=True),
+            Column('id', Integer, nullable=False, primary_key=True),
             Column('name', String(255)))
 
 
-class TestAsyncIter(AIOPyMySQLTestCase):
-
-    @asyncio.coroutine
-    def _prepare(self, conn):
-        cur = yield from conn.cursor()
-        yield from cur.execute("DROP TABLE IF EXISTS tbl;")
-
-        yield from cur.execute("""CREATE TABLE tbl (
+@pytest.fixture
+def table(loop, connection_creator, table_cleanup):
+    async def f():
+        connection = await connection_creator()
+        cursor = await connection.cursor()
+        await cursor.execute("DROP TABLE IF EXISTS tbl;")
+        await cursor.execute("""CREATE TABLE tbl (
                  id MEDIUMINT NOT NULL AUTO_INCREMENT,
                  name VARCHAR(255) NOT NULL,
                  PRIMARY KEY (id));""")
 
         for i in [(1, 'a'), (2, 'b'), (3, 'c')]:
-            yield from cur.execute("INSERT INTO tbl VALUES(%s, %s)", i)
-        yield from conn.commit()
-
-    def test_async_cursor(self):
-
-        async def go():
-
-            ret = []
-            conn = self.connections[0]
-            await self._prepare(conn)
-
-            cur = await conn.cursor()
-            await cur.execute('SELECT * from tbl;')
-            async for i in cur:
-                ret.append(i)
-
-            self.assertEqual([(1, 'a'), (2, 'b'), (3, 'c')], ret)
-
-        self.loop.run_until_complete(go())
-
-    def test_async_cursor_server_side(self):
+            await cursor.execute("INSERT INTO tbl VALUES(%s, %s)", i)
 
-        async def go():
-            ret = []
-            conn = self.connections[0]
-            await self._prepare(conn)
+        await cursor.execute("commit;")
+        await cursor.close()
 
-            cur = await conn.cursor(SSCursor)
-            await cur.execute('SELECT * from tbl;')
-            async for i in cur:
-                ret.append(i)
+    table_cleanup('tbl')
+    loop.run_until_complete(f())
 
-            self.assertEqual([(1, 'a'), (2, 'b'), (3, 'c')], ret)
 
-        self.loop.run_until_complete(go())
+@pytest.mark.run_loop
+async def test_async_cursor(cursor, table):
+    ret = []
+    await cursor.execute('SELECT * from tbl;')
+    async for i in cursor:
+        ret.append(i)
+    assert [(1, 'a'), (2, 'b'), (3, 'c')] == ret
 
-    def test_async_iter_over_sa_result(self):
 
-        async def go():
+@pytest.mark.run_loop
+async def test_async_cursor_server_side(connection, table):
+    ret = []
+    cursor = await connection.cursor(SSCursor)
+    await cursor.execute('SELECT * from tbl;')
+    async for i in cursor:
+        ret.append(i)
+    assert [(1, 'a'), (2, 'b'), (3, 'c')] == ret
 
-            ret = []
-            engine = await sa.create_engine(loop=self.loop,
-                                            db=self.db,
-                                            user=self.user,
-                                            password=self.password,
-                                            host=self.host)
-            conn = await engine.acquire()
-            await self._prepare(conn.connection)
 
-            async for i in (await conn.execute(tbl.select())):
-                ret.append(i)
+@pytest.mark.run_loop
+async def test_async_iter_over_sa_result(mysql_params, table, loop):
+    ret = []
+    engine = await sa.create_engine(**mysql_params, loop=loop)
+    conn = await engine.acquire()
 
-            self.assertEqual([(1, 'a'), (2, 'b'), (3, 'c')], ret)
-            engine.terminate()
+    async for i in (await conn.execute(tbl.select())):
+        ret.append(i)
 
-        self.loop.run_until_complete(go())
+    assert [(1, 'a'), (2, 'b'), (3, 'c')] == ret
+    engine.terminate()