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
|
# -*- coding: utf-8 -*-
import hypothesis.strategies as st
from hypothesis import example, given, settings
import pytest
from vdirsyncer.metasync import MetaSyncConflict, metasync
from vdirsyncer.storage.base import normalize_meta_value
from vdirsyncer.storage.memory import MemoryStorage
from . import blow_up
def test_irrelevant_status():
a = MemoryStorage()
b = MemoryStorage()
status = {'foo': 'bar'}
metasync(a, b, status, keys=())
assert not status
def test_basic(monkeypatch):
a = MemoryStorage()
b = MemoryStorage()
status = {}
a.set_meta('foo', 'bar')
metasync(a, b, status, keys=['foo'])
assert a.get_meta('foo') == b.get_meta('foo') == 'bar'
a.set_meta('foo', 'baz')
metasync(a, b, status, keys=['foo'])
assert a.get_meta('foo') == b.get_meta('foo') == 'baz'
monkeypatch.setattr(a, 'set_meta', blow_up)
monkeypatch.setattr(b, 'set_meta', blow_up)
metasync(a, b, status, keys=['foo'])
assert a.get_meta('foo') == b.get_meta('foo') == 'baz'
monkeypatch.undo()
monkeypatch.undo()
b.set_meta('foo', None)
metasync(a, b, status, keys=['foo'])
assert not a.get_meta('foo') and not b.get_meta('foo')
def test_conflict():
a = MemoryStorage()
b = MemoryStorage()
status = {}
a.set_meta('foo', 'bar')
b.set_meta('foo', 'baz')
with pytest.raises(MetaSyncConflict):
metasync(a, b, status, keys=['foo'])
assert a.get_meta('foo') == 'bar'
assert b.get_meta('foo') == 'baz'
assert not status
def test_conflict_same_content():
a = MemoryStorage()
b = MemoryStorage()
status = {}
a.set_meta('foo', 'bar')
b.set_meta('foo', 'bar')
metasync(a, b, status, keys=['foo'])
assert a.get_meta('foo') == b.get_meta('foo') == status['foo'] == 'bar'
@pytest.mark.parametrize('wins', 'ab')
def test_conflict_x_wins(wins):
a = MemoryStorage()
b = MemoryStorage()
status = {}
a.set_meta('foo', 'bar')
b.set_meta('foo', 'baz')
metasync(a, b, status, keys=['foo'],
conflict_resolution='a wins' if wins == 'a' else 'b wins')
assert a.get_meta('foo') == b.get_meta('foo') == status['foo'] == (
'bar' if wins == 'a' else 'baz'
)
keys = st.text(min_size=1).filter(lambda x: x.strip() == x)
values = st.text().filter(lambda x: normalize_meta_value(x) == x)
metadata = st.dictionaries(keys, values)
@given(
a=metadata, b=metadata,
status=metadata, keys=st.sets(keys),
conflict_resolution=st.just('a wins') | st.just('b wins')
)
@settings(suppress_health_check=["HealthCheck.too_slow"])
@example(a={u'0': u'0'}, b={}, status={u'0': u'0'}, keys={u'0'},
conflict_resolution='a wins')
def test_fuzzing(a, b, status, keys, conflict_resolution):
def _get_storage(m, instance_name):
s = MemoryStorage(instance_name=instance_name)
s.metadata = m
return s
a = _get_storage(a, 'A')
b = _get_storage(b, 'B')
winning_storage = (a if conflict_resolution == 'a wins' else b)
expected_values = dict((key, winning_storage.get_meta(key))
for key in keys)
metasync(a, b, status,
keys=keys, conflict_resolution=conflict_resolution)
for key in keys:
s = status.get(key, '')
assert a.get_meta(key) == b.get_meta(key) == s
assert s == expected_values[key] or not expected_values[key] or not s
|
