Method test_isolation_level

tests/test_transaction.py:183–212

(self)

        self.assertFalse(self.con.is_in_transaction())

    async def test_isolation_level(self):
        await self.con.reset()
        default_isolation = await self.con.fetchval(
            'SHOW default_transaction_isolation'
        )
        isolation_levels = {
            None: default_isolation,
            'read_committed': 'read committed',
            'read_uncommitted': 'read uncommitted',
            'repeatable_read': 'repeatable read',
            'serializable': 'serializable',
        }
        set_sql = 'SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL '
        get_sql = 'SHOW TRANSACTION ISOLATION LEVEL'
        for tx_level in isolation_levels:
            for conn_level in isolation_levels:
                with self.subTest(conn=conn_level, tx=tx_level):
                    if conn_level:
                        await self.con.execute(
                            set_sql + isolation_levels[conn_level]
                        )
                    level = await self.con.fetchval(get_sql)
                    self.assertEqual(level, isolation_levels[conn_level])
                    async with self.con.transaction(isolation=tx_level):
                        level = await self.con.fetchval(get_sql)
                        self.assertEqual(
                            level,
                            isolation_levels[tx_level or conn_level],
                        )
                    await self.con.reset()

    async def test_nested_isolation_level(self):
        set_sql = 'SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL '