Method test_nested_isolation_level

tests/test_transaction.py:214–250

(self)

                    await self.con.reset()

    async def test_nested_isolation_level(self):
        set_sql = 'SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL '
        isolation_levels = {
            'read_committed': 'read committed',
            'read_uncommitted': 'read uncommitted',
            'repeatable_read': 'repeatable read',
            'serializable': 'serializable',
        }
        for inner in [None] + list(isolation_levels):
            for outer, outer_sql_level in isolation_levels.items():
                for implicit in [False, True]:
                    with self.subTest(
                        implicit=implicit, outer=outer, inner=inner,
                    ):
                        if implicit:
                            await self.con.execute(set_sql + outer_sql_level)
                            outer_level = None
                        else:
                            outer_level = outer

                        async with self.con.transaction(isolation=outer_level):
                            if inner and outer != inner:
                                with self.assertRaisesRegex(
                                    asyncpg.InterfaceError,
                                    'current {!r} != outer {!r}'.format(
                                        inner, outer
                                    )
                                ):
                                    async with self.con.transaction(
                                            isolation=inner,
                                    ):
                                        pass
                            else:
                                async with self.con.transaction(
                                        isolation=inner,
                                ):
                                    pass