-
Notifications
You must be signed in to change notification settings - Fork 49
/
test_triggerversionfield.py
95 lines (71 loc) · 2.63 KB
/
test_triggerversionfield.py
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
# -*- coding: utf-8 -*-
from django.core import signals
from django.db import connections, IntegrityError
import mock
import pytest
from concurrency.exceptions import RecordModifiedError
from concurrency.utils import refetch
from tests.models import TriggerConcurrentModel
from django.core.signals import request_started
# Register an event to reset saved queries when a Django request is started.
from tests.util import nextname
def reset_queries(**kwargs):
for conn in connections.all():
conn.queries = []
signals.request_started.connect(reset_queries)
class CaptureQueriesContext(object):
"""
Context manager that captures queries executed by the specified connection.
"""
def __init__(self, connection):
self.connection = connection
def __iter__(self):
return iter(self.captured_queries)
def __getitem__(self, index):
return self.captured_queries[index]
def __len__(self):
return len(self.captured_queries)
@property
def captured_queries(self):
return self.connection.queries[self.initial_queries:self.final_queries]
def __enter__(self):
self.use_debug_cursor = self.connection.use_debug_cursor
self.connection.use_debug_cursor = True
self.initial_queries = len(self.connection.queries)
self.final_queries = None
request_started.disconnect(reset_queries)
return self
def __exit__(self, exc_type, exc_value, traceback):
self.connection.use_debug_cursor = self.use_debug_cursor
request_started.connect(reset_queries)
if exc_type is not None:
return
self.final_queries = len(self.connection.queries)
@pytest.mark.django_db
def test_trigger():
instance = TriggerConcurrentModel()
assert instance.pk is None
assert instance.version == 0
instance.username = next(nextname)
instance.save() # insert
instance = refetch(instance)
assert instance.version == 1
instance.username = next(nextname)
instance.save() # update
assert instance.version == 2
instance.username = next(nextname)
instance.save() # update
assert instance.version == 3
copy = refetch(instance)
copy.save()
with pytest.raises(RecordModifiedError):
instance.save()
@pytest.mark.django_db
def test_trigger_do_not_increase_version_if_error():
instance = TriggerConcurrentModel()
assert instance.pk is None
assert instance.version == 0
with mock.patch('tests.models.TriggerConcurrentModel.save', side_effect=IntegrityError):
with pytest.raises(IntegrityError):
instance.save()
assert instance.version == 0