jotbe/webdav-client-python-3
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Adapt to use Apache WebDAV Container for CI

Browse source
This commit is contained in:
Evgeny Ezhov 2019-11-27 14:57:22 +03:00
parent ba2453800d
commit 78a979715d
3 changed files with 17 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
tests/test_client_it.py
View file

@ -6,6 +6,7 @@ from os import path
from unittest import TestCase from unittest import TestCase
from webdav3.client import Client from webdav3.client import Client
from webdav3.exceptions import MethodNotSupported
class ClientTestCase(TestCase): class ClientTestCase(TestCase):
@ -43,6 +44,7 @@ class ClientTestCase(TestCase):
self.assertGreater(file_list.__len__(), 0, 'Expected that amount of files more then 0') self.assertGreater(file_list.__len__(), 0, 'Expected that amount of files more then 0')
def test_free(self): def test_free(self):
with self.assertRaises(MethodNotSupported):
self.assertGreater(self.client.free(), 0, 'Expected that free space on WebDAV server is more then 0 bytes') self.assertGreater(self.client.free(), 0, 'Expected that free space on WebDAV server is more then 0 bytes')
def test_check(self): def test_check(self):
@ -54,16 +56,12 @@ class ClientTestCase(TestCase):
self.client.mkdir(remote_path=self.remote_path_dir) self.client.mkdir(remote_path=self.remote_path_dir)
self.assertTrue(self.client.check(remote_path=self.remote_path_dir), 'Expected the directory is created.') self.assertTrue(self.client.check(remote_path=self.remote_path_dir), 'Expected the directory is created.')
@unittest.skip("Yandex brakes response for file it contains property resourcetype as collection but it should "
"be empty for file")
def test_download_to(self): def test_download_to(self):
self._prepare_for_downloading() self._prepare_for_downloading()
buff = BytesIO() buff = BytesIO()
self.client.download_from(buff=buff, remote_path=self.remote_path_file) self.client.download_from(buff=buff, remote_path=self.remote_path_file)
self.assertEqual(buff.getvalue(), 'test content for testing of webdav client') self.assertEqual(buff.getvalue(), b'test content for testing of webdav client')
@unittest.skip("Yandex brakes response for file it contains property resourcetype as collection but it should "
"be empty for file")
def test_download(self): def test_download(self):
self._prepare_for_downloading() self._prepare_for_downloading()
self.client.download(local_path=self.local_path_dir, remote_path=self.remote_path_dir) self.client.download(local_path=self.local_path_dir, remote_path=self.remote_path_dir)
@ -71,14 +69,11 @@ class ClientTestCase(TestCase):
self.assertTrue(path.isdir(self.local_path_dir), 'Expected this is a directory.') self.assertTrue(path.isdir(self.local_path_dir), 'Expected this is a directory.')
self.assertTrue(path.exists(self.local_path_dir + os.path.sep + self.local_file), self.assertTrue(path.exists(self.local_path_dir + os.path.sep + self.local_file),
'Expected the file is downloaded') 'Expected the file is downloaded')
self.assertTrue(path.isfile(self.local_path_dir + os.path.sep + self.local_path_file), self.assertTrue(path.isfile(self.local_path_dir + os.path.sep + self.local_file),
'Expected this is a file') 'Expected this is a file')
@unittest.skip("Yandex brakes response for file it contains property resourcetype as collection but it should "
"be empty for file")
def test_download_sync(self): def test_download_sync(self):
self._prepare_for_downloading() self._prepare_for_downloading()
os.mkdir(self.local_path_dir)
def callback(): def callback():
self.assertTrue(path.exists(self.local_path_dir + os.path.sep + self.local_file), self.assertTrue(path.exists(self.local_path_dir + os.path.sep + self.local_file),
@ -91,11 +86,8 @@ class ClientTestCase(TestCase):
self.assertTrue(path.exists(self.local_path_dir + os.path.sep + self.local_file), self.assertTrue(path.exists(self.local_path_dir + os.path.sep + self.local_file),
'Expected the file has already been downloaded') 'Expected the file has already been downloaded')
@unittest.skip("Yandex brakes response for file it contains property resourcetype as collection but it should "
"be empty for file")
def test_download_async(self): def test_download_async(self):
self._prepare_for_downloading() self._prepare_for_downloading()
os.mkdir(self.local_path_dir)
def callback(): def callback():
self.assertTrue(path.exists(self.local_path_dir + os.path.sep + self.local_file), self.assertTrue(path.exists(self.local_path_dir + os.path.sep + self.local_file),
@ -166,7 +158,6 @@ class ClientTestCase(TestCase):
def test_info(self): def test_info(self):
self._prepare_for_downloading() self._prepare_for_downloading()
result = self.client.info(remote_path=self.remote_path_file) result = self.client.info(remote_path=self.remote_path_file)
self.assertEqual(result['name'], 'test.txt')
self.assertEqual(result['size'], '41') self.assertEqual(result['size'], '41')
self.assertTrue('created' in result) self.assertTrue('created' in result)
self.assertTrue('modified' in result) self.assertTrue('modified' in result)

2
tests/test_client_unit.py
View file

@ -129,7 +129,7 @@ class ClientTestCase(TestCase):
self.assertEqual(result, b'<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<test/>') self.assertEqual(result, b'<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<test/>')
def test_parse_is_dir_response_directory(self): def test_parse_is_dir_response_directory(self):
f = open('./tests/response_dir.xml') f = open('./tests/response_dir.xml', encoding='utf-8')
content = f.read() content = f.read()
path = '/test_dir' path = '/test_dir'
hostname = 'https://webdav.yandex.ru' hostname = 'https://webdav.yandex.ru'

15
webdav3/client.py
View file

@ -21,7 +21,6 @@ except ImportError:
from urllib import unquote from urllib import unquote
from urlparse import urlsplit, urlparse from urlparse import urlsplit, urlparse
__version__ = "0.2"
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@ -74,6 +73,7 @@ def wrap_connection_error(fn):
raise ConnectionException(re) raise ConnectionException(re)
else: else:
return res return res
return _wrapper return _wrapper
@ -128,7 +128,7 @@ class Client(object):
if self.webdav.token: if self.webdav.token:
webdav_token = "Authorization: OAuth {token}".format(token=self.webdav.token) webdav_token = "Authorization: OAuth {token}".format(token=self.webdav.token)
headers.append(webdav_token) headers.append(webdav_token)
return dict([map(lambda s: s.strip(), i.split(':')) for i in headers]) return dict([map(lambda s: s.strip(), i.split(':', 1)) for i in headers])
def get_url(self, path): def get_url(self, path):
"""Generates url by uri path. """Generates url by uri path.
@ -519,9 +519,12 @@ class Client(object):
if not self.check(urn_to.parent()): if not self.check(urn_to.parent()):
raise RemoteParentNotFound(urn_to.path()) raise RemoteParentNotFound(urn_to.path())
header_destination = "Destination: {path}".format(path=self.get_full_path(urn_to)) headers = [
header_depth = "Depth: {depth}".format(depth=depth) "Destination: {url}".format(url=self.get_url(urn_to))
self.execute_request(action='copy', path=urn_from.quote(), headers_ext=[header_destination, header_depth]) ]
if self.is_dir(urn_from.path()):
headers.append("Depth: {depth}".format(depth=depth))
self.execute_request(action='copy', path=urn_from.quote(), headers_ext=headers)
@wrap_connection_error @wrap_connection_error
def move(self, remote_path_from, remote_path_to, overwrite=False): def move(self, remote_path_from, remote_path_to, overwrite=False):
@ -540,7 +543,7 @@ class Client(object):
if not self.check(urn_to.parent()): if not self.check(urn_to.parent()):
raise RemoteParentNotFound(urn_to.path()) raise RemoteParentNotFound(urn_to.path())
header_destination = "Destination: {path}".format(path=self.get_full_path(urn_to)) header_destination = "Destination: {path}".format(path=self.get_url(urn_to))
header_overwrite = "Overwrite: {flag}".format(flag="T" if overwrite else "F") header_overwrite = "Overwrite: {flag}".format(flag="T" if overwrite else "F")
self.execute_request(action='move', path=urn_from.quote(), headers_ext=[header_destination, header_overwrite]) self.execute_request(action='move', path=urn_from.quote(), headers_ext=[header_destination, header_overwrite])