test_async.py 文件源码

def test_async_cursor_gone(self):
        import gc
        cur = self.conn.cursor()
        cur.execute("select 42;")
        del cur
        gc.collect()
        self.assertRaises(psycopg2.InterfaceError, self.wait, self.conn)

        # The connection is still usable
        cur = self.conn.cursor()
        cur.execute("select 42;")
        self.wait(self.conn)
        self.assertEqual(cur.fetchone(), (42,))