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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
|
from __future__ import annotations
import asyncio
import hypothesis.strategies as st
import pytest
import pytest_asyncio
from hypothesis import example
from hypothesis import given
from tests import blow_up
from vdirsyncer.exceptions import UserError
from vdirsyncer.metasync import MetaSyncConflict
from vdirsyncer.metasync import logger
from vdirsyncer.metasync import metasync
from vdirsyncer.storage.base import normalize_meta_value
from vdirsyncer.storage.memory import MemoryStorage
@pytest.mark.asyncio
async def test_irrelevant_status():
a = MemoryStorage()
b = MemoryStorage()
status = {"foo": "bar"}
await metasync(a, b, status, keys=())
assert not status
@pytest.mark.asyncio
async def test_basic(monkeypatch):
a = MemoryStorage()
b = MemoryStorage()
status = {}
await a.set_meta("foo", None)
await metasync(a, b, status, keys=["foo"])
assert await a.get_meta("foo") is None and await b.get_meta("foo") is None
await a.set_meta("foo", "bar")
await metasync(a, b, status, keys=["foo"])
assert await a.get_meta("foo") == await b.get_meta("foo") == "bar"
await a.set_meta("foo", "baz")
await metasync(a, b, status, keys=["foo"])
assert await a.get_meta("foo") == await b.get_meta("foo") == "baz"
monkeypatch.setattr(a, "set_meta", blow_up)
monkeypatch.setattr(b, "set_meta", blow_up)
await metasync(a, b, status, keys=["foo"])
assert await a.get_meta("foo") == await b.get_meta("foo") == "baz"
monkeypatch.undo()
monkeypatch.undo()
await b.set_meta("foo", None)
await metasync(a, b, status, keys=["foo"])
assert not await a.get_meta("foo") and not await b.get_meta("foo")
@pytest_asyncio.fixture
async def conflict_state(request):
a = MemoryStorage()
b = MemoryStorage()
status = {}
await a.set_meta("foo", "bar")
await b.set_meta("foo", "baz")
async def do_cleanup():
assert await a.get_meta("foo") == "bar"
assert await b.get_meta("foo") == "baz"
assert not status
request.addfinalizer(lambda: asyncio.run(do_cleanup()))
return a, b, status
@pytest_asyncio.fixture
async def test_conflict(conflict_state):
a, b, status = conflict_state
with pytest.raises(MetaSyncConflict):
await metasync(a, b, status, keys=["foo"])
@pytest.mark.asyncio
async def test_invalid_conflict_resolution(conflict_state):
a, b, status = conflict_state
with pytest.raises(UserError) as excinfo:
await metasync(a, b, status, keys=["foo"], conflict_resolution="foo")
assert "Invalid conflict resolution setting" in str(excinfo.value)
@pytest.mark.asyncio
async def test_warning_on_custom_conflict_commands(conflict_state, monkeypatch):
a, b, status = conflict_state
warnings = []
monkeypatch.setattr(logger, "warning", warnings.append)
with pytest.raises(MetaSyncConflict):
await metasync(
a,
b,
status,
keys=["foo"],
conflict_resolution=lambda *a, **kw: None,
)
assert warnings == ["Custom commands don't work on metasync."]
@pytest.mark.asyncio
async def test_conflict_same_content():
a = MemoryStorage()
b = MemoryStorage()
status = {}
await a.set_meta("foo", "bar")
await b.set_meta("foo", "bar")
await metasync(a, b, status, keys=["foo"])
assert await a.get_meta("foo") == await b.get_meta("foo") == status["foo"] == "bar"
@pytest.mark.parametrize("wins", "ab")
@pytest.mark.asyncio
async def test_conflict_x_wins(wins):
a = MemoryStorage()
b = MemoryStorage()
status = {}
await a.set_meta("foo", "bar")
await b.set_meta("foo", "baz")
await metasync(
a,
b,
status,
keys=["foo"],
conflict_resolution="a wins" if wins == "a" else "b wins",
)
assert (
await a.get_meta("foo")
== await 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"),
)
@example(
a={"0": "0"}, b={}, status={"0": "0"}, keys={"0"}, conflict_resolution="a wins"
)
@example(
a={"0": "0"},
b={"0": "1"},
status={"0": "0"},
keys={"0"},
conflict_resolution="a wins",
)
@pytest.mark.asyncio
async 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 = {
key: await winning_storage.get_meta(key) for key in keys if key not in status
}
await metasync(a, b, status, keys=keys, conflict_resolution=conflict_resolution)
for key in keys:
s = status.get(key)
assert await a.get_meta(key) == await b.get_meta(key) == s
if expected_values.get(key) and s:
assert s == expected_values[key]
|
