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

Fixing and refactoring of copy/move resource methods

Browse source
This commit is contained in:
evgeny.ezhov 2017-10-18 17:15:49 +03:00
parent 735a649d57
commit 825ae24a72
4 changed files with 63 additions and 82 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
README.rst
View file

@ -11,6 +11,8 @@ Version 0.3 - TBD
* Refactoring of WebDAV client and making it works in following methods:
- Getting of WebDAV resource property value
- Setting of WebDAV resource property value
- Coping of resource on WebDAV server
- Moving of resource on WebDAV server
Version 0.2 - 11.09.2017
* Refactoring of WebDAV client and making it works in following methods:

27
tests/test_client_it.py
View file

@ -9,7 +9,9 @@ from webdav3.client import Client
class ClientTestCase(TestCase):
remote_path_file = 'test_dir/test.txt'
remote_path_file2 = 'test_dir2/test.txt'
remote_path_dir = 'test_dir'
remote_path_dir2 = 'test_dir2'
local_path_file = 'test.txt'
local_path_dir = 'res/test_dir'
@ -23,7 +25,16 @@ class ClientTestCase(TestCase):
if path.exists(path=self.local_path_dir):
shutil.rmtree(path=self.local_path_dir)
def tearDown(self):
if path.exists(path=self.local_path_dir):
shutil.rmtree(path=self.local_path_dir)
if self.client.check(remote_path=self.remote_path_dir):
self.client.clean(remote_path=self.remote_path_dir)
if self.client.check(remote_path=self.remote_path_dir2):
self.client.clean(remote_path=self.remote_path_dir2)
def test_list(self):
self._prepare_for_downloading()
file_list = self.client.list()
self.assertIsNotNone(file_list, 'List of files should not be None')
self.assertGreater(file_list.__len__(), 0, 'Expected that amount of files more then 0')
@ -41,15 +52,14 @@ class ClientTestCase(TestCase):
self.assertTrue(self.client.check(remote_path=self.remote_path_dir), 'Expected the directory is created.')
def test_download_to(self):
self._prepare_for_downloading()
buff = BytesIO()
self.client.download_from(buff=buff, remote_path=self.remote_path_file)
self.assertEquals(buff.getvalue(), 'test content for testing of webdav client')
def test_download(self):
self._prepare_for_downloading()
self.client.download(local_path=self.local_path_dir, remote_path=self.remote_path_dir)
self.assertTrue(path.exists(self.local_path_dir), 'Expected the directory is downloaded.')
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_path_file),
@ -123,6 +133,19 @@ class ClientTestCase(TestCase):
self.client.upload(remote_path=self.remote_path_file, local_path=self.local_path_dir)
def test_copy(self):
self._prepare_for_downloading()
self.client.mkdir(remote_path=self.remote_path_dir2)
self.client.copy(remote_path_from=self.remote_path_file, remote_path_to=self.remote_path_file2)
self.assertTrue(self.client.check(remote_path=self.remote_path_file2))
def test_move(self):
self._prepare_for_downloading()
self.client.mkdir(remote_path=self.remote_path_dir2)
self.client.move(remote_path_from=self.remote_path_file, remote_path_to=self.remote_path_file2)
self.assertFalse(self.client.check(remote_path=self.remote_path_file))
self.assertTrue(self.client.check(remote_path=self.remote_path_file2))
def test_get_property(self):
self._prepare_for_downloading()
result = self.client.get_property(remote_path=self.remote_path_file, option={'name': 'aProperty'})

4
tests/test_client_unit.py
View file

@ -29,7 +29,7 @@ class ClientTestCase(TestCase):
'<d:propstat><d:status>HTTP/1.1 200 OK</d:status><d:prop><d:quota-used-bytes>697' \
'</d:quota-used-bytes><d:quota-available-bytes>10737417543</d:quota-available-bytes></d:prop>' \
'</d:propstat></d:response></d:multistatus>'
result = utils.parse_free_space_response(content)
result = utils.parse_free_space_response(content, 'localhost')
self.assertEquals(result, 10737417543)
def test_create_get_property_request_content(self):
@ -37,7 +37,7 @@ class ClientTestCase(TestCase):
'namespace': 'test',
'name': 'aProperty'
}
result = utils.create_get_property_request_content(option=option)
result = utils.create_get_property_request_content(option=option, )
self.assertEquals(result, '<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<propfind xmlns="DAV:"><prop>'
'<aProperty xmlns="test"/></prop></propfind>')

112
webdav3/client.py
View file

@ -98,24 +98,29 @@ class Client(object):
'set_property': ["Accept: */*", "Depth: 1", "Content-Type: application/x-www-form-urlencoded"]
}
def get_header(self, action):
"""Returns HTTP headers of specified WebDAV actions
def get_headers(self, action, headers_ext=None):
"""Returns HTTP headers of specified WebDAV actions.
:param action: the identifier of action
:return: the dictionary of headers for specified action
:param action: the identifier of action.
:param headers_ext: (optional) the addition headers list witch sgould be added to basic HTTP headers for
the specified action.
:return: the dictionary of headers for specified action.
"""
if action in Client.http_header:
try:
header = Client.http_header[action].copy()
headers = Client.http_header[action].copy()
except AttributeError:
header = Client.http_header[action][:]
headers = Client.http_header[action][:]
else:
header = list()
headers = list()
if headers_ext:
headers.extend(headers_ext)
if self.webdav.token:
webdav_token = "Authorization: OAuth {token}".format(token=self.webdav.token)
header.append(webdav_token)
return dict([map(lambda s: s.strip(), i.split(':')) for i in header])
headers.append(webdav_token)
return dict([map(lambda s: s.strip(), i.split(':')) for i in headers])
def get_url(self, path):
"""Generates url by uri path.
@ -126,20 +131,22 @@ class Client(object):
url = {'hostname': self.webdav.hostname, 'root': self.webdav.root, 'path': path}
return "{hostname}{root}{path}".format(**url)
def execute_request(self, action, path, data=None):
def execute_request(self, action, path, data=None, headers_ext=None):
"""Generate request to WebDAV server for specified action and path and execute it.
:param action: the action for WebDAV server which should be executed.
:param path: the path to resource for action
:param data: (optional) Dictionary or list of tuples ``[(key, value)]`` (will be form-encoded), bytes,
or file-like object to send in the body of the :class:`Request`.
:param headers_ext: (optional) the addition headers list witch should be added to basic HTTP headers for
the specified action.
:return: HTTP response of request.
"""
response = requests.request(
method=Client.requests[action],
url=self.get_url(path),
auth=(self.webdav.login, self.webdav.password),
headers=self.get_header(action),
headers=self.get_headers(action, headers_ext),
data=data
)
if response.status_code == 507:
@ -266,12 +273,10 @@ class Client(object):
"""
directory_urn = Urn(remote_path, directory=True)
if not self.check(directory_urn.parent()):
raise RemoteParentNotFound(directory_urn.path())
response = self.execute_request(action='mkdir', path=directory_urn.quote())
return response.status_code == 200
@wrap_connection_error
@ -282,7 +287,6 @@ class Client(object):
:param remote_path: path to file on WebDAV server.
"""
urn = Urn(remote_path)
if self.is_dir(urn.path()):
raise OptionNotValid(name="remote_path", value=remote_path)
@ -290,7 +294,6 @@ class Client(object):
raise RemoteResourceNotFound(urn.path())
response = self.execute_request(action='download_to', path=urn.quote())
buff.write(response.content)
def download(self, remote_path, local_path, progress=None):
@ -315,7 +318,6 @@ class Client(object):
:param progress: Progress function. Not supported now.
"""
urn = Urn(remote_path, directory=True)
if not self.is_dir(urn.path()):
raise OptionNotValid(name="remote_path", value=remote_path)
@ -338,7 +340,6 @@ class Client(object):
:param progress: progress function. Not supported now.
"""
urn = Urn(remote_path)
if self.is_dir(urn.path()):
raise OptionNotValid(name="remote_path", value=remote_path)
@ -350,7 +351,6 @@ class Client(object):
with open(local_path, 'wb') as local_file:
response = self.execute_request('download_file', urn.quote())
for block in response.iter_content(1024):
local_file.write(block)
@ -362,7 +362,6 @@ class Client(object):
:param callback: the callback which will be invoked when downloading is complete.
"""
self.download(local_path=local_path, remote_path=remote_path)
if callback:
callback()
@ -384,7 +383,6 @@ class Client(object):
:param remote_path: the path to save file remotely on WebDAV server.
"""
urn = Urn(remote_path)
if urn.is_dir():
raise OptionNotValid(name="remote_path", value=remote_path)
@ -416,7 +414,6 @@ class Client(object):
:param progress: Progress function. Not supported now.
"""
urn = Urn(remote_path, directory=True)
if not urn.is_dir():
raise OptionNotValid(name="remote_path", value=remote_path)
@ -449,7 +446,6 @@ class Client(object):
raise LocalResourceNotFound(local_path)
urn = Urn(remote_path)
if urn.is_dir():
raise OptionNotValid(name="remote_path", value=remote_path)
@ -490,7 +486,6 @@ class Client(object):
target = (lambda: self.upload_sync(local_path=local_path, remote_path=remote_path, callback=callback))
threading.Thread(target=target).start()
# TODO refactor code below and write tests for it.
@wrap_connection_error
def copy(self, remote_path_from, remote_path_to):
"""Copies resource from one place to another on WebDAV server.
@ -498,78 +493,38 @@ class Client(object):
:param remote_path_from: the path to resource which will be copied,
:param remote_path_to: the path where resource will be copied.
"""
def header(remote_path_to):
path = Urn(remote_path_to).path()
destination = "{root}{path}".format(root=self.webdav.root, path=path)
header = self.get_header('copy')
header["Destination"] = destination
return header
urn_from = Urn(remote_path_from)
if not self.check(urn_from.path()):
raise RemoteResourceNotFound(urn_from.path())
urn_to = Urn(remote_path_to)
if not self.check(urn_to.parent()):
raise RemoteParentNotFound(urn_to.path())
url = {'hostname': self.webdav.hostname, 'root': self.webdav.root, 'path': urn_from.quote()}
options = {
'URL': "{hostname}{root}{path}".format(**url),
'CUSTOMREQUEST': Client.requests['copy'],
'HTTPHEADER': header(remote_path_to)
}
h = header(remote_path_to)
response = requests.request(
options["CUSTOMREQUEST"],
options["URL"],
auth=(self.webdav.login, self.webdav.password),
headers=h,
)
# TODO: check response status
header_destination = "Destination: {root}{path}".format(root=self.webdav.root, path=urn_to.path())
self.execute_request(action='copy', path=urn_from.quote(), headers_ext=[header_destination])
@wrap_connection_error
def move(self, remote_path_from, remote_path_to, overwrite=False):
"""Moves resource from one place to another on WebDAV server.
def header(remote_path_to):
path = Urn(remote_path_to).path()
destination = "{root}{path}".format(root=self.webdav.root, path=path)
header = self.get_header('move')
header["Destination"] = destination
header["Overwrite"] = "T" if overwrite else "F"
return header
:param remote_path_from: the path to resource which will be moved,
:param remote_path_to: the path where resource will be moved.
:param overwrite: (optional) the flag, overwrite file if it exists. Defaults is False
"""
urn_from = Urn(remote_path_from)
if not self.check(urn_from.path()):
raise RemoteResourceNotFound(urn_from.path())
urn_to = Urn(remote_path_to)
if not self.check(urn_to.parent()):
raise RemoteParentNotFound(urn_to.path())
url = {'hostname': self.webdav.hostname, 'root': self.webdav.root, 'path': urn_from.quote()}
options = {
'URL': "{hostname}{root}{path}".format(**url),
'CUSTOMREQUEST': Client.requests['move'],
}
h = header(remote_path_to)
response = requests.request(
options["CUSTOMREQUEST"],
options["URL"],
auth=(self.webdav.login, self.webdav.password),
headers=h,
)
# TODO: check response status
header_destination = "Destination: {root}{path}".format(root=self.webdav.root, path=urn_to.path())
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])
# TODO refactor code below and write tests for it.
@wrap_connection_error
def clean(self, remote_path):
@ -579,14 +534,14 @@ class Client(object):
options = {
'URL': "{hostname}{root}{path}".format(**url),
'CUSTOMREQUEST': Client.requests['clean'],
'HTTPHEADER': self.get_header('clean')
'HTTPHEADER': self.get_headers('clean', None)
}
response = requests.request(
options["CUSTOMREQUEST"],
options["URL"],
auth=(self.webdav.login, self.webdav.password),
headers=self.get_header('clean'),
headers=self.get_headers('clean', None),
)
# TODO: check response status
@ -645,7 +600,7 @@ class Client(object):
options["CUSTOMREQUEST"],
options["URL"],
auth=(self.webdav.login, self.webdav.password),
headers=self.get_header('info')
headers=self.get_headers('info', None)
)
path = "{root}{path}".format(root=self.webdav.root, path=urn.path())
return parse(response, path)
@ -701,7 +656,7 @@ class Client(object):
options["CUSTOMREQUEST"],
options["URL"],
auth=(self.webdav.login, self.webdav.password),
headers=self.get_header('info')
headers=self.get_headers('info', None)
)
path = "{root}{path}".format(root=self.webdav.root, path=urn.path())
@ -944,6 +899,7 @@ class WebDavXmlUtils:
Parses of response content XML from WebDAV server and extract na amount of free space.
:param content: the XML content of HTTP response from WebDAV server for getting free space.
:param hostname: the server hostname.
:return: an amount of free space in bytes.
"""
try: