test_error_recovery.py 文件源码

def testClosedConnection(self):
        warnings.simplefilter("ignore")
        my_db = pg8000.connect(**db_connect)
        cursor = my_db.cursor()
        my_db.close()
        try:
            cursor.execute("VALUES ('hw1'::text)")
            self.fail("Should have raised an exception")
        except:
            e = exc_info()[1]
            self.assertTrue(isinstance(e, self.db.InterfaceError))
            self.assertEqual(str(e), 'connection is closed')

        warnings.resetwarnings()