Better tests for sync

This commit is contained in:
Markus Unterwaditzer 2014-02-16 23:52:01 +01:00
parent 58ee2af43c
commit 9b9f17f69f

View file

@ -13,67 +13,85 @@ from vdirsyncer.storage.memory import MemoryStorage
from vdirsyncer.sync import sync
import vdirsyncer.exceptions as exceptions
def only(x):
x = list(x)
assert len(x) == 1, x
return x[0]
def empty_storage(x):
return list(x.list()) == []
class SyncTests(TestCase):
def test_basic(self):
a = MemoryStorage()
b = MemoryStorage()
status = {}
sync(a, b, status)
assert not status
assert empty_storage(a)
assert empty_storage(b)
# creation
item = Item('UID:1')
a.upload(item)
sync(a, b, status)
obj_a, etag_a = a.get('1')
obj_b, etag_b = b.get('1')
assert only(status) == '1'
assert obj_a.raw == obj_b.raw == item.raw
# creation and deletion
item2 = Item('UID:2')
b.upload(item2)
b.delete('1', etag_b)
sync(a, b, status)
assert list(status) == ['2']
assert next(a.list())[0] == '2'
assert next(b.list())[0] == '2'
obj2_a, etag2_a = a.get('2')
assert obj2_a.raw == item2.raw
new_item2 = Item('UID:2\nHUEHUEHUE:PRECISELY')
old_status = status.copy()
a.update(new_item2, next(a.list())[1])
sync(a, b, status)
assert status != old_status
assert list(status) == list(old_status)
assert next(a.list())[0] == '2'
assert next(b.list())[0] == '2'
obj, etag = b.get('2')
assert obj.raw == new_item2.raw
def test_irrelevant_status(self):
a = MemoryStorage()
b = MemoryStorage()
status = {'1': ('UID:1', 1234)}
sync(a, b, status)
assert not status
assert empty_storage(a)
assert empty_storage(b)
def test_new_item(self):
def test_missing_status(self):
a = MemoryStorage()
b = MemoryStorage()
status = {}
a.upload(Item('UID:1'))
item = Item('UID:1')
a.upload(item)
b.upload(item)
sync(a, b, status)
obj, etag = b.get('1')
assert obj.raw == 'UID:1'
assert list(status) == ['1']
assert a.has('1')
assert b.has('1')
def test_upload_and_update(self):
a = MemoryStorage()
b = MemoryStorage()
status = {}
item = Item('UID:1') # new item 1 in a
a.upload(item)
sync(a, b, status)
assert b.get('1')[0].raw == item.raw
item = Item('UID:1\nASDF:YES') # update of item 1 in b
b.update(item, b.get('1')[1])
sync(a, b, status)
assert a.get('1')[0].raw == item.raw
item2 = Item('UID:2') # new item 2 in b
b.upload(item2)
sync(a, b, status)
assert a.get('2')[0].raw == item2.raw
item2 = Item('UID:2\nASDF:YES') # update of item 2 in a
a.update(item2, a.get('2')[1])
sync(a, b, status)
assert b.get('2')[0].raw == item2.raw
def test_deletion(self):
a = MemoryStorage()
b = MemoryStorage()
status = {}
item = Item('UID:1')
a.upload(item)
sync(a, b, status)
b.delete('1', b.get('1')[1])
sync(a, b, status)
assert not a.has('1') and not b.has('1')
a.upload(item)
sync(a, b, status)
assert a.has('1') and b.has('1')
a.delete('1', a.get('1')[1])
sync(a, b, status)
assert not a.has('1') and not b.has('1')
def test_already_synced(self):
a = MemoryStorage()
b = MemoryStorage()
item = Item('UID:1')
a.upload(item)
b.upload(item)
status = {'1': (a.get('1')[1], b.get('1')[1])}
old_status = dict(status)
a.update = b.update = a.upload = b.upload = \
lambda *a, **kw: self.fail('Method shouldn\'t have been called.')
sync(a, b, status)
assert status == old_status
assert a.has('1') and b.has('1')