test_async.py 文件源码

def test_async_iter(self):
        cur = self.conn.cursor()

        cur.execute("begin")
        self.wait(cur)
        cur.execute("""
            insert into table1 values (1);
            insert into table1 values (2);
            insert into table1 values (3);
        """)
        self.wait(cur)
        cur.execute("select id from table1 order by id")

        # iteration fails if a query is underway
        self.assertRaises(psycopg2.ProgrammingError, list, cur)

        # but after it's done it should work
        self.wait(cur)
        self.assertEqual(list(cur), [(1, ), (2, ), (3, )])
        self.assertFalse(self.conn.isexecuting())