test_transaction.py

# Copyright (C) 2016-present the asyncpg authors and contributors
# <see AUTHORS file>
#
# This module is part of asyncpg and is released under
# the Apache 2.0 License: http://www.apache.org/licenses/LICENSE-2.0


importasyncpg

fromasyncpgimport_testbaseastb


classTestTransaction(tb.ConnectedTestCase):

asyncdeftest_transaction_regular(self):
self.assertIsNone(self.con._top_xact)
self.assertFalse(self.con.is_in_transaction())
tr=self.con.transaction()
self.assertIsNone(self.con._top_xact)
self.assertFalse(self.con.is_in_transaction())

withself.assertRaises(ZeroDivisionError):
asyncwithtraswith_tr:
self.assertIs(self.con._top_xact, tr)
self.assertTrue(self.con.is_in_transaction())

# We don't return the transaction object from __aenter__,
# to make it harder for people to use '.rollback()' and
# '.commit()' from within an 'async with' block.
self.assertIsNone(with_tr)

awaitself.con.execute('''
 CREATE TABLE mytab (a int);
 ''')

1/0

self.assertIsNone(self.con._top_xact)
self.assertFalse(self.con.is_in_transaction())

withself.assertRaisesRegex(asyncpg.PostgresError,
'"mytab" does not exist'):
awaitself.con.prepare('''
 SELECT * FROM mytab
 ''')

asyncdeftest_transaction_nested(self):
self.assertIsNone(self.con._top_xact)
self.assertFalse(self.con.is_in_transaction())

tr=self.con.transaction()

self.assertIsNone(self.con._top_xact)
self.assertFalse(self.con.is_in_transaction())

withself.assertRaises(ZeroDivisionError):
asyncwithtr:
self.assertIs(self.con._top_xact, tr)
self.assertTrue(self.con.is_in_transaction())

awaitself.con.execute('''
 CREATE TABLE mytab (a int);
 ''')

asyncwithself.con.transaction():
self.assertIs(self.con._top_xact, tr)
self.assertTrue(self.con.is_in_transaction())

awaitself.con.execute('''
 INSERT INTO mytab (a) VALUES (1), (2);
 ''')

self.assertIs(self.con._top_xact, tr)
self.assertTrue(self.con.is_in_transaction())

withself.assertRaises(ZeroDivisionError):
in_tr=self.con.transaction()
asyncwithin_tr:

self.assertIs(self.con._top_xact, tr)
self.assertTrue(self.con.is_in_transaction())

awaitself.con.execute('''
 INSERT INTO mytab (a) VALUES (3), (4);
 ''')

1/0

st=awaitself.con.prepare('SELECT * FROM mytab;')

recs= []
asyncforrecinst.cursor():
recs.append(rec)

self.assertEqual(len(recs), 2)
self.assertEqual(recs[0][0], 1)
self.assertEqual(recs[1][0], 2)

self.assertIs(self.con._top_xact, tr)
self.assertTrue(self.con.is_in_transaction())

1/0

self.assertIs(self.con._top_xact, None)
self.assertFalse(self.con.is_in_transaction())

withself.assertRaisesRegex(asyncpg.PostgresError,
'"mytab" does not exist'):
awaitself.con.prepare('''
 SELECT * FROM mytab
 ''')

asyncdeftest_transaction_interface_errors(self):
self.assertIsNone(self.con._top_xact)
self.assertFalse(self.con.is_in_transaction())

tr=self.con.transaction(readonly=True, isolation='serializable')
withself.assertRaisesRegex(asyncpg.InterfaceError,
'cannot start; .* already started'):
asyncwithtr:
awaittr.start()

self.assertTrue(repr(tr).startswith(
'<asyncpg.Transaction state:rolledback serializable readonly'))

self.assertIsNone(self.con._top_xact)
self.assertFalse(self.con.is_in_transaction())

withself.assertRaisesRegex(asyncpg.InterfaceError,
'cannot start; .* already rolled back'):
asyncwithtr:
pass

self.assertIsNone(self.con._top_xact)
self.assertFalse(self.con.is_in_transaction())

tr=self.con.transaction()
withself.assertRaisesRegex(asyncpg.InterfaceError,
'cannot manually commit.*async with'):
asyncwithtr:
awaittr.commit()

self.assertIsNone(self.con._top_xact)
self.assertFalse(self.con.is_in_transaction())

tr=self.con.transaction()
withself.assertRaisesRegex(asyncpg.InterfaceError,
'cannot manually rollback.*async with'):
asyncwithtr:
awaittr.rollback()

self.assertIsNone(self.con._top_xact)
self.assertFalse(self.con.is_in_transaction())

tr=self.con.transaction()
withself.assertRaisesRegex(asyncpg.InterfaceError,
'cannot enter context:.*async with'):
asyncwithtr:
asyncwithtr:
pass

asyncdeftest_transaction_within_manual_transaction(self):
self.assertIsNone(self.con._top_xact)
self.assertFalse(self.con.is_in_transaction())

awaitself.con.execute('BEGIN')

tr=self.con.transaction()
self.assertIsNone(self.con._top_xact)
self.assertTrue(self.con.is_in_transaction())

withself.assertRaisesRegex(asyncpg.InterfaceError,
'cannot use Connection.transaction'):
awaittr.start()

withself.assertLoopErrorHandlerCalled(
'Resetting connection with an active transaction'):
awaitself.con.reset()

self.assertIsNone(self.con._top_xact)
self.assertFalse(self.con.is_in_transaction())

asyncdeftest_isolation_level(self):
awaitself.con.reset()
default_isolation=awaitself.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'
fortx_levelinisolation_levels:
forconn_levelinisolation_levels:
withself.subTest(conn=conn_level, tx=tx_level):
ifconn_level:
awaitself.con.execute(
set_sql+isolation_levels[conn_level]
 )
level=awaitself.con.fetchval(get_sql)
self.assertEqual(level, isolation_levels[conn_level])
asyncwithself.con.transaction(isolation=tx_level):
level=awaitself.con.fetchval(get_sql)
self.assertEqual(
level,
isolation_levels[tx_levelorconn_level],
 )
awaitself.con.reset()

asyncdeftest_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',
 }
forinnerin [None] +list(isolation_levels):
forouter, outer_sql_levelinisolation_levels.items():
forimplicitin [False, True]:
withself.subTest(
implicit=implicit, outer=outer, inner=inner,
 ):
ifimplicit:
awaitself.con.execute(set_sql+outer_sql_level)
outer_level=None
else:
outer_level=outer

asyncwithself.con.transaction(isolation=outer_level):
ifinnerandouter!=inner:
withself.assertRaisesRegex(
asyncpg.InterfaceError,
'current {!r} != outer {!r}'.format(
inner, outer
 )
 ):
asyncwithself.con.transaction(
isolation=inner,
 ):
pass
else:
asyncwithself.con.transaction(
isolation=inner,
 ):
pass