test_connection.py 文件源码

def test_cleanup_on_badconn_close(self):
        # ticket #148
        conn = self.conn
        cur = conn.cursor()
        try:
            cur.execute("select pg_terminate_backend(pg_backend_pid())")
        except psycopg2.OperationalError, e:
            if e.pgcode != psycopg2.errorcodes.ADMIN_SHUTDOWN:
                raise
        except psycopg2.DatabaseError, e:
            # curiously when disconnected in green mode we get a DatabaseError
            # without pgcode.
            if e.pgcode is not None:
                raise

        self.assertEqual(conn.closed, 2)
        conn.close()
        self.assertEqual(conn.closed, 1)