test_connection.py 文件源码

def test_setattr_isolation_level_int(self):
        cur = self.conn.cursor()
        self.conn.isolation_level = ext.ISOLATION_LEVEL_SERIALIZABLE
        self.assertEqual(self.conn.isolation_level, ext.ISOLATION_LEVEL_SERIALIZABLE)

        cur.execute("SHOW transaction_isolation;")
        self.assertEqual(cur.fetchone()[0], 'serializable')
        self.conn.rollback()

        self.conn.isolation_level = ext.ISOLATION_LEVEL_REPEATABLE_READ
        cur.execute("SHOW transaction_isolation;")
        if self.conn.server_version > 80000:
            self.assertEqual(self.conn.isolation_level,
                ext.ISOLATION_LEVEL_REPEATABLE_READ)
            self.assertEqual(cur.fetchone()[0], 'repeatable read')
        else:
            self.assertEqual(self.conn.isolation_level,
                ext.ISOLATION_LEVEL_SERIALIZABLE)
            self.assertEqual(cur.fetchone()[0], 'serializable')
        self.conn.rollback()

        self.conn.isolation_level = ext.ISOLATION_LEVEL_READ_COMMITTED
        self.assertEqual(self.conn.isolation_level,
            ext.ISOLATION_LEVEL_READ_COMMITTED)
        cur.execute("SHOW transaction_isolation;")
        self.assertEqual(cur.fetchone()[0], 'read committed')
        self.conn.rollback()

        self.conn.isolation_level = ext.ISOLATION_LEVEL_READ_UNCOMMITTED
        cur.execute("SHOW transaction_isolation;")
        if self.conn.server_version > 80000:
            self.assertEqual(self.conn.isolation_level,
                ext.ISOLATION_LEVEL_READ_UNCOMMITTED)
            self.assertEqual(cur.fetchone()[0], 'read uncommitted')
        else:
            self.assertEqual(self.conn.isolation_level,
                ext.ISOLATION_LEVEL_READ_COMMITTED)
            self.assertEqual(cur.fetchone()[0], 'read committed')
        self.conn.rollback()

        self.assertEqual(ext.ISOLATION_LEVEL_DEFAULT, None)
        self.conn.isolation_level = ext.ISOLATION_LEVEL_DEFAULT
        self.assertEqual(self.conn.isolation_level, None)
        cur.execute("SHOW transaction_isolation;")
        isol = cur.fetchone()[0]
        cur.execute("SHOW default_transaction_isolation;")
        self.assertEqual(cur.fetchone()[0], isol)