-
Notifications
You must be signed in to change notification settings - Fork 49
/
test_base.py
72 lines (56 loc) · 2.19 KB
/
test_base.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
import pytest
import django
from concurrency.api import apply_concurrency_check
import concurrency.config
from concurrency.core import _set_version
from concurrency.exceptions import RecordModifiedError
from concurrency.utils import refetch
from demo.util import (with_all_models, concurrent_model, unique_id,
with_std_models)
pytest.mark.django_db(transaction=False)
def pytest_generate_tests(metafunc):
if 'protocol' in metafunc.fixturenames:
if django.VERSION[1] in [4, 5]:
metafunc.parametrize("protocol", [1])
else:
metafunc.parametrize("protocol", [1, 2])
@with_all_models
# @pytest.mark.parametrize("protocol", [1, 2])
@pytest.mark.django_db
def test_standard_save(model_class, protocol, monkeypatch):
# this test pass if executed alone,
# produce a Duplicate Key (only django 1.4) if executed with other tests
monkeypatch.setattr(concurrency.config.conf, 'PROTOCOL', protocol)
if django.VERSION[:2] == (1, 4):
model_class.objects.all().delete()
instance = model_class(username=concurrent_model.__name__)
instance.save()
assert instance.get_concurrency_version() > 0
@pytest.mark.django_db(transaction=False)
@with_std_models
def test_conflict(model_class, protocol, monkeypatch):
monkeypatch.setattr(concurrency.config.conf, 'PROTOCOL', protocol)
id = next(unique_id)
instance = model_class.objects.get_or_create(pk=id)[0]
instance.save()
copy = refetch(instance)
copy.save()
with pytest.raises(RecordModifiedError):
instance.save()
assert copy.get_concurrency_version() > instance.get_concurrency_version()
@pytest.mark.django_db(transaction=False)
@with_std_models
def test_do_not_check_if_no_version(model_class):
id = next(unique_id)
instance, __ = model_class.objects.get_or_create(pk=id)
instance.save()
copy = refetch(instance)
copy.save()
with pytest.raises(RecordModifiedError):
_set_version(instance, 1)
instance.version = 1
instance.save()
_set_version(instance, 0)
instance.save()
assert instance.get_concurrency_version() > 0
assert instance.get_concurrency_version() != copy.get_concurrency_version()