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
124
125
126
127
128
129
130
131
132
133
134
|
from django.db import connection, IntegrityError
from django.db.transaction import atomic
from django.test import TransactionTestCase
from cacheops.transaction import queue_when_in_transaction
from .models import Category, Post
from .utils import run_in_thread
def get_category():
return Category.objects.cache().get(pk=1)
class IntentionalRollback(Exception):
pass
class TransactionSupportTests(TransactionTestCase):
databases = ('default', 'slave')
fixtures = ['basic']
def test_atomic(self):
with atomic():
obj = get_category()
obj.title = 'Changed'
obj.save()
self.assertEqual('Changed', get_category().title)
self.assertEqual('Django', run_in_thread(get_category).title)
self.assertEqual('Changed', run_in_thread(get_category).title)
self.assertEqual('Changed', get_category().title)
def test_nested(self):
with atomic():
with atomic():
obj = get_category()
obj.title = 'Changed'
obj.save()
self.assertEqual('Changed', get_category().title)
self.assertEqual('Django', run_in_thread(get_category).title)
self.assertEqual('Changed', get_category().title)
self.assertEqual('Django', run_in_thread(get_category).title)
self.assertEqual('Changed', run_in_thread(get_category).title)
self.assertEqual('Changed', get_category().title)
def test_rollback(self):
try:
with atomic():
obj = get_category()
obj.title = 'Changed'
obj.save()
self.assertEqual('Changed', get_category().title)
self.assertEqual('Django', run_in_thread(get_category).title)
raise IntentionalRollback()
except IntentionalRollback:
pass
self.assertEqual('Django', get_category().title)
self.assertEqual('Django', run_in_thread(get_category).title)
def test_nested_rollback(self):
with atomic():
try:
with atomic():
obj = get_category()
obj.title = 'Changed'
obj.save()
self.assertEqual('Changed', get_category().title)
self.assertEqual('Django', run_in_thread(get_category).title)
raise IntentionalRollback()
except IntentionalRollback:
pass
self.assertEqual('Django', get_category().title)
self.assertEqual('Django', run_in_thread(get_category).title)
self.assertEqual('Django', get_category().title)
self.assertEqual('Django', run_in_thread(get_category).title)
def test_smart_transactions(self):
with atomic():
get_category()
with self.assertNumQueries(0):
get_category()
with atomic():
with self.assertNumQueries(0):
get_category()
obj = get_category()
obj.title += ' changed'
obj.save()
get_category()
with self.assertNumQueries(1):
get_category()
def test_rollback_during_integrity_error(self):
# store category in cache
get_category()
# Make current DB be "dirty" by write
with self.assertRaises(IntegrityError):
with atomic():
Post.objects.create(category_id=-1, title='')
# however, this write should be rolled back and current DB should
# not be "dirty"
with self.assertNumQueries(0):
get_category()
def test_call_cacheops_cbs_before_on_commit_cbs(self):
calls = []
with atomic():
def django_commit_handler():
calls.append('django')
connection.on_commit(django_commit_handler)
@queue_when_in_transaction
def cacheops_commit_handler(using):
calls.append('cacheops')
cacheops_commit_handler('default')
self.assertEqual(calls, ['cacheops', 'django'])
def test_multidb(self):
try:
with atomic('slave'):
with atomic():
obj = get_category()
obj.title = 'Changed'
obj.save()
raise IntentionalRollback()
except IntentionalRollback:
pass
self.assertEqual('Changed', get_category().title)
|
