sjs/vdirsyncer
1
0
Fork 0
mirror of https://github.com/samsonjs/vdirsyncer.git synced 2026-03-25 08:55:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Fixed a lot of tests

Browse source
This commit is contained in:
Markus Unterwaditzer 2014-03-04 17:23:30 +01:00
parent 6aa227e12b
commit 273cc4d6fd
8 changed files with 83 additions and 62 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
tests/__init__.py Normal file
View file

@ -0,0 +1,8 @@
def normalize_item(x):
return set(x for x in x.raw.splitlines() if
not x.startswith('X-RADICALE-NAME') and
not x.startswith('PRODID'))
def assert_item_equals(a, b):
assert normalize_item(a) == normalize_item(b)

22
tests/storage/__init__.py
View file

@ -9,12 +9,16 @@
from vdirsyncer.storage.base import Item
import vdirsyncer.exceptions as exceptions
from .. import assert_item_equals
import random
class StorageTests(object):
item_template = u'UID:{uid}\nX-SOMETHING:{r}'
def _create_bogus_item(self, uid):
return Item(u'UID:{}'.format(uid))
r = random.random()
return Item(self.item_template.format(uid=uid, r=r))
def _get_storage(self, **kwargs):
raise NotImplementedError()
@ -45,25 +49,25 @@ class StorageTests(object):
s = self._get_storage()
item = self._create_bogus_item(1)
href, etag = s.upload(item)
assert s.get(href)[0].raw == item.raw
assert_item_equals(s.get(href)[0], item)
def test_update(self):
s = self._get_storage()
item = self._create_bogus_item(1)
href, etag = s.upload(item)
assert s.get(href)[0].raw == item.raw
assert_item_equals(s.get(href)[0], item)
new_item = Item(item.raw + u'\nX-SOMETHING: YES\n')
new_item = self._create_bogus_item(1)
s.update(href, new_item, etag)
assert s.get(href)[0].raw == new_item.raw
assert_item_equals(s.get(href)[0], new_item)
def test_update_nonexisting(self):
s = self._get_storage()
item = self._create_bogus_item(1)
self.assertRaises(
exceptions.PreconditionFailed, s.update, s._get_href('1'), item, 123)
self.assertRaises(
exceptions.PreconditionFailed, s.update, 'huehue', item, 123)
self.assertRaises(exceptions.PreconditionFailed,
s.update, s._get_href('1'), item, 123)
self.assertRaises(exceptions.PreconditionFailed,
s.update, 'huehue', item, 123)
def test_wrong_etag(self):
s = self._get_storage()

9
tests/storage/dav/__init__.py
View file

@ -60,6 +60,7 @@ class Response(object):
self.status_code = x.status_code
self.content = x.get_data(as_text=False)
self.headers = x.headers
self.encoding = x.charset
def raise_for_status(self):
'''copied from requests itself'''
@ -69,6 +70,7 @@ class Response(object):
class DavStorageTests(StorageTests):
'''hrefs are paths without scheme or netloc'''
tmpdir = None
storage_class = None
radicale_path = None
@ -84,10 +86,9 @@ class DavStorageTests(StorageTests):
server = 'http://127.0.0.1'
full_url = server + self.radicale_path
def x(method, item, data=None, headers=None):
assert '/' not in item
url = self.radicale_path + item
r = c.open(path=url, method=method, data=data, headers=headers)
def x(method, path, data=None, headers=None):
path = path or self.radicale_path
r = c.open(path=path, method=method, data=data, headers=headers)
r = Response(r)
return r
return self.storage_class(url=full_url, _request_func=x, **kwargs)

26
tests/storage/dav/test_caldav.py
View file

@ -19,16 +19,16 @@ class CaldavStorageTests(TestCase, DavStorageTests):
storage_class = CaldavStorage
radicale_path = '/bob/test.ics/'
def _create_bogus_item(self, uid):
return Item(u'BEGIN:VCALENDAR\n'
u'VERSION:2.0\n'
u'PRODID:-//dmfs.org//mimedir.icalendar//EN\n'
u'BEGIN:VTODO\n'
u'CREATED:20130721T142233Z\n'
u'DTSTAMP:20130730T074543Z\n'
u'LAST-MODIFIED;VALUE=DATE-TIME:20140122T151338Z\n'
u'SEQUENCE:2\n'
u'SUMMARY:Book: Kowlani - Tödlicher Staub\n'
u'UID:{}\n'
u'END:VTODO\n'
u'END:VCALENDAR'.format(uid))
item_template = (u'BEGIN:VCALENDAR\n'
u'VERSION:2.0\n'
u'PRODID:-//dmfs.org//mimedir.icalendar//EN\n'
u'BEGIN:VTODO\n'
u'CREATED:20130721T142233Z\n'
u'DTSTAMP:20130730T074543Z\n'
u'LAST-MODIFIED;VALUE=DATE-TIME:20140122T151338Z\n'
u'SEQUENCE:2\n'
u'SUMMARY:Book: Kowlani - Tödlicher Staub\n'
u'UID:{uid}\n'
u'X-SOMETHING:{r}\n'
u'END:VTODO\n'
u'END:VCALENDAR')

32
tests/storage/dav/test_carddav.py
View file

@ -19,19 +19,19 @@ class CarddavStorageTests(TestCase, DavStorageTests):
storage_class = CarddavStorage
radicale_path = '/bob/test.vcf/'
def _create_bogus_item(self, uid):
return Item(u'BEGIN:VCARD\n'
u'VERSION:3.0\n'
u'FN:Cyrus Daboo\n'
u'N:Daboo;Cyrus\n'
u'ADR;TYPE=POSTAL:;2822 Email HQ;' # address continuing
u'Suite 2821;RFCVille;PA;15213;USA\n' # on next line
u'EMAIL;TYPE=INTERNET,PREF:cyrus@example.com\n'
u'NICKNAME:me\n'
u'NOTE:Example VCard.\n'
u'ORG:Self Employed\n'
u'TEL;TYPE=WORK,VOICE:412 605 0499\n'
u'TEL;TYPE=FAX:412 605 0705\n'
u'URL:http://www.example.com\n'
u'UID:{}\n'
u'END:VCARD'.format(uid))
item_template = (u'BEGIN:VCARD\n'
u'VERSION:3.0\n'
u'FN:Cyrus Daboo\n'
u'N:Daboo;Cyrus\n'
u'ADR;TYPE=POSTAL:;2822 Email HQ;' # address continuing
u'Suite 2821;RFCVille;PA;15213;USA\n' # on next line
u'EMAIL;TYPE=INTERNET,PREF:cyrus@example.com\n'
u'NICKNAME:me\n'
u'NOTE:Example VCard.\n'
u'ORG:Self Employed\n'
u'TEL;TYPE=WORK,VOICE:412 605 0499\n'
u'TEL;TYPE=FAX:412 605 0705\n'
u'URL:http://www.example.com\n'
u'UID:{uid}\n'
u'X-SOMETHING:{r}\n'
u'END:VCARD')

16
tests/test_sync.py
View file

@ -11,6 +11,7 @@ from unittest import TestCase
from vdirsyncer.storage.base import Item
from vdirsyncer.storage.memory import MemoryStorage
from vdirsyncer.sync import sync
from . import assert_item_equals, normalize_item
import vdirsyncer.exceptions as exceptions
@ -52,7 +53,7 @@ class SyncTests(TestCase):
b.upload(item2)
sync(a, b, status)
assert status
assert a.get('1.txt')[0].raw == b.get('1.txt')[0].raw
assert_item_equals(a.get('1.txt')[0], b.get('1.txt')[0])
def test_upload_and_update(self):
a = MemoryStorage()
@ -62,22 +63,22 @@ class SyncTests(TestCase):
item = Item(u'UID:1') # new item 1 in a
a.upload(item)
sync(a, b, status)
assert b.get('1.txt')[0].raw == item.raw
assert_item_equals(b.get('1.txt')[0], item)
item = Item(u'UID:1\nASDF:YES') # update of item 1 in b
b.update('1.txt', item, b.get('1.txt')[1])
sync(a, b, status)
assert a.get('1.txt')[0].raw == item.raw
assert_item_equals(a.get('1.txt')[0], item)
item2 = Item(u'UID:2') # new item 2 in b
b.upload(item2)
sync(a, b, status)
assert a.get('2.txt')[0].raw == item2.raw
assert_item_equals(a.get('2.txt')[0], item2)
item2 = Item(u'UID:2\nASDF:YES') # update of item 2 in a
a.update('2.txt', item2, a.get('2.txt')[1])
sync(a, b, status)
assert b.get('2.txt')[0].raw == item2.raw
assert_item_equals(b.get('2.txt')[0], item2)
def test_deletion(self):
a = MemoryStorage()
@ -129,7 +130,10 @@ class SyncTests(TestCase):
sync(a, b, status, conflict_resolution='a wins')
obj_a, _ = a.get(href_a)
obj_b, _ = b.get(href_b)
assert obj_a.raw == obj_b.raw == u'UID:1\nASDASD'
assert_item_equals(obj_a, obj_b)
n = normalize_item(obj_a)
assert u'UID:1' in n
assert u'ASDASD' in n
def tset_conflict_resolution_new_etags_without_changes(self):
a = MemoryStorage()

22
vdirsyncer/storage/dav/base.py
View file

@ -72,10 +72,13 @@ class DavStorage(Storage):
'''Used to strip hrefs off the collection's URL, to leave only the
filename.'''
href = urlparse.urlparse(href).path
if href.startswith(self.parsed_url.path):
href = href[len(self.parsed_url.path):]
assert '/' not in href, href
return href
if href.startswith('/'):
return href
assert '/' not in href
return self.parsed_url.path + href
def _get_href(self, uid):
return self._simplify_href(super(DavStorage, self)._get_href(uid))
def _default_headers(self):
return {
@ -83,11 +86,10 @@ class DavStorage(Storage):
'Content-Type': 'application/xml; charset=UTF-8'
}
def _request(self, method, item, data=None, headers=None):
def _request(self, method, path, data=None, headers=None):
if self._session is None:
self._session = requests.session()
assert '/' not in item
url = self.url + item
url = self.parsed_url.scheme + '://' + self.parsed_url.netloc + path
return self._session.request(method, url, data=data, headers=headers,
**self._settings)
@ -105,11 +107,11 @@ class DavStorage(Storage):
def get_multi(self, hrefs):
if not hrefs:
return ()
hrefs = [self._simplify_href(href) for href in hrefs]
href_xml = []
for href in hrefs:
assert '/' not in href, href
href_xml.append('<D:href>{}</D:href>'.format(self.url + href))
href_xml.append('<D:href>{}</D:href>'.format(href))
data = self.get_multi_template.format(hrefs='\n'.join(href_xml))
response = self._request(
'REPORT',
@ -151,6 +153,7 @@ class DavStorage(Storage):
return True
def update(self, href, obj, etag):
href = self._simplify_href(href)
headers = self._default_headers()
headers.update({
'Content-Type': self.item_mimetype,
@ -192,6 +195,7 @@ class DavStorage(Storage):
return href, etag
def delete(self, href, etag):
href = self._simplify_href(href)
headers = self._default_headers()
headers.update({
'If-Match': etag

10
vdirsyncer/storage/dav/caldav.py
View file

@ -64,7 +64,9 @@ class CaldavStorage(DavStorage):
</D:prop>
<C:filter>
<C:comp-filter name="VCALENDAR">
{caldavfilter}
<C:comp-filter name="VTODO">
{caldavfilter}
</C:comp-filter>
</C:comp-filter>
</C:filter>
</C:calendar-query>'''
@ -73,9 +75,7 @@ class CaldavStorage(DavStorage):
if start and end:
start = start.strftime(CALDAV_DT_FORMAT)
end = end.strftime(CALDAV_DT_FORMAT)
caldavfilter = ('<C:comp-filter name="VTODO">'
'<C:time-range start="{start}" end="{end}"/>'
'</C:comp-filter>').format(start=start,
end=end)
caldavfilter = ('<C:time-range start="{start}" end="{end}"/>'
.format(start=start, end=end))
return data.format(caldavfilter=caldavfilter)
return data.format(caldavfilter='')