1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
|
# This file is part of Tryton. The COPYRIGHT file at the top level of
# this repository contains the full copyright notices and license terms.
import unittest
from unittest.mock import Mock
from trytond.tests.test_tryton import CONTEXT, DB_NAME, USER, activate_module
from trytond.transaction import Transaction
def empty_transaction(*args, **kwargs):
'''
Just starts a transaction in the context manager and returns `True`
and stops transaction for the given arguments.
All positional arguments are passed to `start` method of transaction
'''
with Transaction().start(*args, **kwargs):
return True
class TransactionTestCase(unittest.TestCase):
'Test the Transaction Context manager'
@classmethod
def setUpClass(cls):
activate_module('tests')
def test_nonexistdb(self):
'''Attempt opening a transaction with a non existant DB
and ensure that it stops cleanly and allows starting of next
transaction'''
self.assertRaises(
Exception, empty_transaction, "Non existant DB", USER,
context=CONTEXT)
self.assertTrue(empty_transaction(DB_NAME, USER, context=CONTEXT))
def test_set_user(self):
'Test set_user'
with Transaction().start(DB_NAME, USER, context=CONTEXT) \
as transaction:
self.assertEqual(transaction.user, USER)
self.assertEqual(transaction.context.get('user'), None)
with Transaction().set_user(0):
self.assertEqual(transaction.user, 0)
self.assertEqual(transaction.context.get('user'), None)
with Transaction().set_user(0, set_context=True):
self.assertEqual(transaction.user, 0)
self.assertEqual(transaction.context.get('user'), USER)
# Nested same set_user should keep original context user
with Transaction().set_user(0, set_context=True):
self.assertEqual(transaction.user, 0)
self.assertEqual(transaction.context.get('user'), USER)
# Unset user context
with Transaction().set_user(0, set_context=False):
self.assertEqual(transaction.user, 0)
self.assertEqual(transaction.context.get('user'), None)
# set context for non root
self.assertRaises(ValueError,
Transaction().set_user, 2, set_context=True)
# not set context for non root
with Transaction().set_user(2):
self.assertEqual(transaction.user, 2)
def test_stacked_transactions(self):
'Test that transactions are stacked / unstacked correctly'
with Transaction().start(DB_NAME, USER, context=CONTEXT) \
as transaction:
with transaction.new_transaction() as new_transaction:
self.assertIsNot(new_transaction, transaction)
self.assertIsNot(Transaction(), transaction)
self.assertIs(Transaction(), new_transaction)
self.assertIs(Transaction(), transaction)
def test_two_phase_commit(self):
# A successful transaction
dm = Mock()
with Transaction().start(DB_NAME, USER, context=CONTEXT) \
as transaction:
transaction.join(dm)
dm.tpc_begin.assert_called_once_with(transaction)
dm.commit.assert_called_once_with(transaction)
dm.tpc_vote.assert_called_once_with(transaction)
dm.tpc_abort.assert_not_called()
dm.tpc_finish.assert_called_once_with(transaction)
# Failing in the datamanager
dm.reset_mock()
dm.tpc_vote.side_effect = ValueError('Failing the datamanager')
try:
with Transaction().start(DB_NAME, USER, context=CONTEXT) \
as transaction:
transaction.join(dm)
except ValueError:
pass
dm.tpc_begin.assert_called_once_with(transaction)
dm.commit.assert_called_once_with(transaction)
dm.tpc_vote.assert_called_once_with(transaction)
dm.tpc_abort.assert_called_once_with(transaction)
dm.tpc_finish.assert_not_called()
# Failing in tryton
dm.reset_mock()
try:
with Transaction().start(DB_NAME, USER, context=CONTEXT) \
as transaction:
transaction.join(dm)
raise ValueError('Failing in tryton')
except ValueError:
pass
dm.tpc_begin.assert_not_called()
dm.commit.assert_not_called()
dm.tpc_vote.assert_not_called()
dm.tpc_abort.assert_called_once_with(transaction)
dm.tpc_finish.assert_not_called()
|
