mirror of
https://github.com/samsonjs/vdirsyncer.git
synced 2026-04-27 14:57:41 +00:00
WIP
This commit is contained in:
parent
fa5112126f
commit
d61c4fc31b
4 changed files with 79 additions and 25 deletions
|
|
@ -20,4 +20,12 @@ radicale_storage_database() { pip install --use-mirrors sqlalchemy pysqlite; }
|
||||||
radicale_storage_filesystem() { true; }
|
radicale_storage_filesystem() { true; }
|
||||||
|
|
||||||
|
|
||||||
|
davserver_owncloud() {
|
||||||
|
pip install paste
|
||||||
|
git clone git@github.com:untitaker/owncloud-testserver.git
|
||||||
|
cd ./owncloud-testserver/
|
||||||
|
sh install.sh
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
davserver_$DAV_SERVER
|
davserver_$DAV_SERVER
|
||||||
|
|
|
||||||
|
|
@ -18,6 +18,8 @@ from vdirsyncer.storage.base import Item
|
||||||
dav_server = os.environ.get('DAV_SERVER', '').strip() or 'radicale'
|
dav_server = os.environ.get('DAV_SERVER', '').strip() or 'radicale'
|
||||||
if dav_server in ('radicale', 'radicale_git'):
|
if dav_server in ('radicale', 'radicale_git'):
|
||||||
from ._radicale import ServerMixin
|
from ._radicale import ServerMixin
|
||||||
|
elif dav_server == 'owncloud':
|
||||||
|
from ._owncloud import ServerMixin
|
||||||
else:
|
else:
|
||||||
raise RuntimeError('{} is not a known DAV server.'.format(dav_server))
|
raise RuntimeError('{} is not a known DAV server.'.format(dav_server))
|
||||||
|
|
||||||
|
|
|
||||||
50
tests/storage/dav/_owncloud.py
Normal file
50
tests/storage/dav/_owncloud.py
Normal file
|
|
@ -0,0 +1,50 @@
|
||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
'''
|
||||||
|
vdirsyncer.tests.storage.dav._owncloud
|
||||||
|
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||||
|
|
||||||
|
Using utilities from paste to wrap the PHP application into WSGI.
|
||||||
|
|
||||||
|
:copyright: (c) 2014 Markus Unterwaditzer
|
||||||
|
:license: MIT, see LICENSE for more details.
|
||||||
|
'''
|
||||||
|
|
||||||
|
from paste.cgiapp import CGIApplication
|
||||||
|
from vdirsyncer.utils import expand_path
|
||||||
|
from ._radicale import wsgi_setup
|
||||||
|
import subprocess
|
||||||
|
import os
|
||||||
|
|
||||||
|
owncloud_repo = expand_path(os.path.join(os.path.dirname(__file__), '../../../owncloud-testserver/'))
|
||||||
|
app = CGIApplication(None, 'php.cgi', [owncloud_repo], include_os_environ=False, query_string=None)
|
||||||
|
|
||||||
|
def middleware(environ, start_response):
|
||||||
|
print(environ)
|
||||||
|
environ['REQUEST_URI'] = 'http://127.0.0.1' + environ['PATH_INFO']
|
||||||
|
return app(environ, start_response)
|
||||||
|
|
||||||
|
class ServerMixin(object):
|
||||||
|
storage_class = None
|
||||||
|
wsgi_teardown = None
|
||||||
|
|
||||||
|
def setup_method(self, method):
|
||||||
|
subprocess.call([os.path.join(owncloud_repo, 'install.sh')])
|
||||||
|
self.wsgi_teardown = wsgi_setup(middleware)
|
||||||
|
|
||||||
|
def get_storage_args(self, collection='test'):
|
||||||
|
assert self.storage_class.fileext in ('.ics', '.vcf')
|
||||||
|
if self.storage_class.fileext == '.vcf':
|
||||||
|
url = 'http://127.0.0.1/remote.php/carddav/addressbooks/asdf/'
|
||||||
|
else:
|
||||||
|
url = 'http://127.0.0.1/remote.php/carddav/addressbooks/asdf/'
|
||||||
|
if collection is not None:
|
||||||
|
assert collection in ('test', 'test1', 'test2', 'test3', 'test4',
|
||||||
|
'test5', 'test6', 'test7', 'test8', 'test9',
|
||||||
|
'test10')
|
||||||
|
|
||||||
|
return {'url': url, 'collection': collection}
|
||||||
|
|
||||||
|
def teardown_method(self, method):
|
||||||
|
if self.wsgi_teardown is not None:
|
||||||
|
self.wsgi_teardown()
|
||||||
|
self.wsgi_teardown = None
|
||||||
|
|
@ -113,10 +113,24 @@ class Response(object):
|
||||||
raise HTTPError(str(self.status_code))
|
raise HTTPError(str(self.status_code))
|
||||||
|
|
||||||
|
|
||||||
|
def wsgi_setup(app):
|
||||||
|
c = Client(app, WerkzeugResponse)
|
||||||
|
def x(session, method, url, data=None, headers=None, **kw):
|
||||||
|
path = urlparse.urlparse(url).path
|
||||||
|
assert isinstance(data, bytes) or data is None
|
||||||
|
r = c.open(path=path, method=method, data=data, headers=headers)
|
||||||
|
r = Response(r)
|
||||||
|
return r
|
||||||
|
|
||||||
|
p = mock.patch('requests.Session.request', new=x)
|
||||||
|
p.start()
|
||||||
|
return p.stop
|
||||||
|
|
||||||
|
|
||||||
class ServerMixin(object):
|
class ServerMixin(object):
|
||||||
'''hrefs are paths without scheme or netloc'''
|
'''hrefs are paths without scheme or netloc'''
|
||||||
storage_class = None
|
storage_class = None
|
||||||
patcher = None
|
wsgi_teardown = None
|
||||||
tmpdir = None
|
tmpdir = None
|
||||||
|
|
||||||
def setup_method(self, method):
|
def setup_method(self, method):
|
||||||
|
|
@ -124,18 +138,7 @@ class ServerMixin(object):
|
||||||
do_the_radicale_dance(self.tmpdir)
|
do_the_radicale_dance(self.tmpdir)
|
||||||
from radicale import Application
|
from radicale import Application
|
||||||
app = Application()
|
app = Application()
|
||||||
|
self.wsgi_teardown = wsgi_setup(app)
|
||||||
c = Client(app, WerkzeugResponse)
|
|
||||||
|
|
||||||
def x(session, method, url, data=None, headers=None, **kw):
|
|
||||||
path = urlparse.urlparse(url).path
|
|
||||||
assert isinstance(data, bytes) or data is None
|
|
||||||
r = c.open(path=path, method=method, data=data, headers=headers)
|
|
||||||
r = Response(r)
|
|
||||||
return r
|
|
||||||
|
|
||||||
self.patcher = p = mock.patch('requests.Session.request', new=x)
|
|
||||||
p.start()
|
|
||||||
|
|
||||||
def get_storage_args(self, collection='test'):
|
def get_storage_args(self, collection='test'):
|
||||||
url = 'http://127.0.0.1/bob/'
|
url = 'http://127.0.0.1/bob/'
|
||||||
|
|
@ -148,15 +151,6 @@ class ServerMixin(object):
|
||||||
if self.tmpdir is not None:
|
if self.tmpdir is not None:
|
||||||
shutil.rmtree(self.tmpdir)
|
shutil.rmtree(self.tmpdir)
|
||||||
self.tmpdir = None
|
self.tmpdir = None
|
||||||
if self.patcher is not None:
|
if self.wsgi_teardown is not None:
|
||||||
self.patcher.stop()
|
self.wsgi_teardown()
|
||||||
self.patcher = None
|
self.wsgi_teardown = None
|
||||||
|
|
||||||
def test_dav_broken_item(self):
|
|
||||||
item = Item(u'UID:1')
|
|
||||||
s = self._get_storage()
|
|
||||||
try:
|
|
||||||
s.upload(item)
|
|
||||||
except exceptions.Error:
|
|
||||||
pass
|
|
||||||
assert not list(s.list())
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue