Fix wrong default transaction isolation level (MagicStack#622)

* Fix wrong default transaction isolation level

This fixes the issue when the default_transaction_isolation is
not "read committed", `transaction(isolation='read_committed')`
won't start a transaction in "read committed" isolation level.

asyncpg/connection.py

@@ -228,7 +228,7 @@ def get_settings(self):
         """
         return self._protocol.get_settings()
 
-    def transaction(self, *, isolation='read_committed', readonly=False,
+    def transaction(self, *, isolation=None, readonly=False,
                     deferrable=False):
         """Create a :class:`~transaction.Transaction` object.
 
@@ -237,7 +237,9 @@ def transaction(self, *, isolation='read_committed', readonly=False,
 
         :param isolation: Transaction isolation mode, can be one of:
                           `'serializable'`, `'repeatable_read'`,
-                          `'read_committed'`.
+                          `'read_committed'`. If not specified, the behavior
+                          is up to the server and session, which is usually
+                          ``read_committed``.
 
         :param readonly: Specifies whether or not this transaction is
                          read-only.

asyncpg/transaction.py

@@ -20,6 +20,11 @@ class TransactionState(enum.Enum):
 
 
 ISOLATION_LEVELS = {'read_committed', 'serializable', 'repeatable_read'}
+ISOLATION_LEVELS_BY_VALUE = {
+    'read committed': 'read_committed',
+    'serializable': 'serializable',
+    'repeatable read': 'repeatable_read',
+}
 
 
 class Transaction(connresource.ConnectionResource):
@@ -36,12 +41,12 @@ class Transaction(connresource.ConnectionResource):
     def __init__(self, connection, isolation, readonly, deferrable):
         super().__init__(connection)
 
-        if isolation not in ISOLATION_LEVELS:
+        if isolation and isolation not in ISOLATION_LEVELS:
             raise ValueError(
                 'isolation is expected to be either of {}, '
                 'got {!r}'.format(ISOLATION_LEVELS, isolation))
 
-        if isolation != 'serializable':
+        if isolation and isolation != 'serializable':
             if readonly:
                 raise ValueError(
                     '"readonly" is only supported for '
@@ -110,20 +115,27 @@ async def start(self):
             con._top_xact = self
         else:
             # Nested transaction block
-            top_xact = con._top_xact
-            if self._isolation != top_xact._isolation:
-                raise apg_errors.InterfaceError(
-                    'nested transaction has a different isolation level: '
-                    'current {!r} != outer {!r}'.format(
-                        self._isolation, top_xact._isolation))
+            if self._isolation:
+                top_xact_isolation = con._top_xact._isolation
+                if top_xact_isolation is None:
+                    top_xact_isolation = ISOLATION_LEVELS_BY_VALUE[
+                        await self._connection.fetchval(
+                            'SHOW transaction_isolation;')]
+                if self._isolation != top_xact_isolation:
+                    raise apg_errors.InterfaceError(
+                        'nested transaction has a different isolation level: '
+                        'current {!r} != outer {!r}'.format(
+                            self._isolation, top_xact_isolation))
             self._nested = True
 
         if self._nested:
             self._id = con._get_unique_id('savepoint')
             query = 'SAVEPOINT {};'.format(self._id)
         else:
-            if self._isolation == 'read_committed':
+            if self._isolation is None:
                 query = 'BEGIN;'
+            elif self._isolation == 'read_committed':
+                query = 'BEGIN ISOLATION LEVEL READ COMMITTED;'
             elif self._isolation == 'repeatable_read':
                 query = 'BEGIN ISOLATION LEVEL REPEATABLE READ;'
             else:

tests/test_transaction.py

@@ -179,3 +179,70 @@ async def test_transaction_within_manual_transaction(self):
 
         self.assertIsNone(self.con._top_xact)
         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',
+            '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 '
+        isolation_levels = {
+            'read_committed': 'read committed',
+            '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