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

Fixing and refactoring of set/get property methods

Browse source
This commit is contained in:
evgeny.ezhov 2017-10-17 09:04:47 +03:00
parent 6c2aecd277
commit 735a649d57
4 changed files with 274 additions and 132 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
README.rst
View file

@ -1,5 +1,25 @@
webdavclient3
============
=============
Based on https://github.com/designerror/webdav-client-python
But uses `requests` instead of `PyCURL`
Release Notes
=============
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
Version 0.2 - 11.09.2017
* Refactoring of WebDAV client and making it works in following methods:
- Constructor with connecting to WebDAV
- Getting a list of resources on WebDAV
- Getting an information about free space on WebDAV
- Checking of existence of resource on WebDAV
- Making a directory on WebDAV
- Downloading of files and directories from WebDAV
- Asynchronously downloading of files and directories from WebDAV
- Uploading of files and directories to WebDAV
- Asynchronously uploading of files and directories to WebDAV

22
tests/test_client.py → tests/test_client_it.py
View file

@ -11,7 +11,7 @@ class ClientTestCase(TestCase):
remote_path_file = 'test_dir/test.txt'
remote_path_dir = 'test_dir'
local_path_file = 'test.txt'
local_path_dir = u'res/test_dir'
local_path_dir = 'res/test_dir'
def setUp(self):
options = {
@ -42,7 +42,7 @@ class ClientTestCase(TestCase):
def test_download_to(self):
buff = BytesIO()
self.client.download_to(buff=buff, remote_path=self.remote_path_file)
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):
@ -90,7 +90,7 @@ class ClientTestCase(TestCase):
def test_upload_from(self):
self._prepare_for_uploading()
buff = StringIO(u'test content for testing of webdav client')
self.client.upload_from(buff=buff, remote_path=self.remote_path_file)
self.client.upload_to(buff=buff, remote_path=self.remote_path_file)
self.assertTrue(self.client.check(self.remote_path_file), 'Expected the file is uploaded.')
self.test_download_to()
@ -123,6 +123,22 @@ class ClientTestCase(TestCase):
self.client.upload(remote_path=self.remote_path_file, local_path=self.local_path_dir)
def test_get_property(self):
self._prepare_for_downloading()
result = self.client.get_property(remote_path=self.remote_path_file, option={'name': 'aProperty'})
self.assertEquals(result, None)
def test_set_property(self):
self._prepare_for_downloading()
self.client.set_property(remote_path=self.remote_path_file, option={
'namespace': 'test',
'name': 'aProperty',
'value': 'aValue'
})
result = self.client.get_property(remote_path=self.remote_path_file,
option={'namespace': 'test', 'name': 'aProperty'})
self.assertEquals(result, "aValue")
def _prepare_for_downloading(self):
if not self.client.check(remote_path=self.remote_path_dir):
self.client.mkdir(remote_path=self.remote_path_dir)

81
tests/test_client_unit.py Normal file
View file

@ -0,0 +1,81 @@
from unittest import TestCase
from lxml.etree import ElementTree, Element
from webdav3.client import WebDavXmlUtils as utils
class ClientTestCase(TestCase):
def test_parse_get_list_response(self):
content = '<?xml version="1.0" encoding="utf-8"?><d:multistatus xmlns:d="DAV:"><d:response><d:href>/test_dir/' \
'</d:href><d:propstat><d:status>HTTP/1.1 200 OK</d:status><d:prop><d:resourcetype><d:collection/>' \
'</d:resourcetype><d:getlastmodified>Mon, 16 Oct 2017 04:18:00 GMT</d:getlastmodified>' \
'<d:displayname>test_dir</d:displayname><d:creationdate>2017-10-16T04:18:00Z</d:creationdate>' \
'</d:prop></d:propstat></d:response><d:response><d:href>/test_dir/test.txt/</d:href><d:propstat>' \
'<d:status>HTTP/1.1 200 OK</d:status><d:prop><d:resourcetype><d:collection/></d:resourcetype>' \
'<d:getlastmodified>Mon, 16 Oct 2017 04:18:18 GMT</d:getlastmodified><d:displayname>test.txt' \
'</d:displayname><d:creationdate>2017-10-16T04:18:18Z</d:creationdate></d:prop></d:propstat>' \
'</d:response></d:multistatus>'
result = utils.parse_get_list_response(content)
self.assertEquals(result.__len__(), 2)
def test_create_free_space_request_content(self):
result = utils.create_free_space_request_content()
self.assertEquals(result, '<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<propfind xmlns="DAV:"><prop>'
'<quota-available-bytes/><quota-used-bytes/></prop></propfind>')
def test_parse_free_space_response(self):
content = '<?xml version="1.0" encoding="utf-8"?><d:multistatus xmlns:d="DAV:"><d:response><d:href>/</d:href>' \
'<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)
self.assertEquals(result, 10737417543)
def test_create_get_property_request_content(self):
option = {
'namespace': 'test',
'name': 'aProperty'
}
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>')
def test_create_get_property_request_content_name_only(self):
option = {
'name': 'aProperty'
}
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=""/></prop></propfind>')
def test_parse_get_property_response(self):
content = '<?xml version="1.0" encoding="utf-8"?><d:multistatus xmlns:d="DAV:"><d:response>' \
'<d:href>/test_dir/test.txt</d:href><d:propstat><d:status>HTTP/1.1 200 OK</d:status><d:prop>' \
'<aProperty xmlns="test">aValue</aProperty></d:prop></d:propstat></d:response></d:multistatus>'
result = utils.parse_get_property_response(content=content, name='aProperty')
self.assertEquals(result, 'aValue')
def test_create_set_property_request_content(self):
option = {
'namespace': 'test',
'name': 'aProperty',
'value': 'aValue'
}
result = utils.create_set_property_request_content(option=option)
self.assertEquals(result, '<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<propertyupdate xmlns="DAV:"><set><prop>'
'<aProperty xmlns="test">aValue</aProperty></prop></set></propertyupdate>')
def test_create_set_property_request_content_name_only(self):
option = {
'name': 'aProperty'
}
result = utils.create_set_property_request_content(option=option)
self.assertEquals(result, '<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<propertyupdate xmlns="DAV:"><set><prop>'
'<aProperty xmlns=""></aProperty></prop></set></propertyupdate>')
def test_etree_to_string(self):
tree = ElementTree(Element('test'))
result = utils.etree_to_string(tree)
self.assertEquals(result, '<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<test/>')

279
webdav3/client.py
View file

@ -94,8 +94,8 @@ class Client(object):
'clean': ["Accept: */*", "Connection: Keep-Alive"],
'check': ["Accept: */*"],
'info': ["Accept: */*", "Depth: 1"],
'get_metadata': ["Accept: */*", "Depth: 1", "Content-Type: application/x-www-form-urlencoded"],
'set_metadata': ["Accept: */*", "Depth: 1", "Content-Type: application/x-www-form-urlencoded"]
'get_property': ["Accept: */*", "Depth: 1", "Content-Type: application/x-www-form-urlencoded"],
'set_property': ["Accept: */*", "Depth: 1", "Content-Type: application/x-www-form-urlencoded"]
}
def get_header(self, action):
@ -166,8 +166,8 @@ class Client(object):
'publish': "PROPPATCH",
'unpublish': "PROPPATCH",
'published': "PROPPATCH",
'get_metadata': "PROPFIND",
'set_metadata': "PROPPATCH"
'get_property': "PROPFIND",
'set_property': "PROPPATCH"
}
meta_xmlns = {
@ -219,29 +219,13 @@ class Client(object):
:param remote_path: path to remote directory.
:return: list of nested file or directory names.
"""
def parse(list_response):
"""Parses of response from WebDAV server and extract file and directory names.
:param list_response: HTTP response from WebDAV server for getting list of files by remote path.
:return: list of extracted file or directory names.
"""
try:
response_str = list_response.content
tree = etree.fromstring(response_str)
hrees = [unquote(hree.text) for hree in tree.findall(".//{DAV:}href")]
return [Urn(hree) for hree in hrees]
except etree.XMLSyntaxError:
return list()
directory_urn = Urn(remote_path, directory=True)
if directory_urn.path() != Client.root:
if not self.check(directory_urn.path()):
raise RemoteResourceNotFound(directory_urn.path())
response = self.execute_request(action='list', path=directory_urn.quote())
urns = parse(response)
urns = WebDavXmlUtils.parse_get_list_response(response.content)
path = "{root}{path}".format(root=self.webdav.root, path=directory_urn.path())
return [urn.filename() for urn in urns if urn.path() != path and urn.path() != path[:-1]]
@ -252,39 +236,9 @@ class Client(object):
:return: an amount of free space in bytes.
"""
def parse(free_response):
"""Parses of response from WebDAV server and extract na amount of free space.
:param free_response: HTTP response from WebDAV server for getting free space.
:return: an amount of free space in bytes.
"""
try:
response_str = free_response.content
tree = etree.fromstring(response_str)
node = tree.find('.//{DAV:}quota-available-bytes')
if node is not None:
return int(node.text)
else:
raise MethodNotSupported(name='free', server=self.webdav.hostname)
except TypeError:
raise MethodNotSupported(name='free', server=self.webdav.hostname)
except etree.XMLSyntaxError:
return str()
def data():
root = etree.Element("propfind", xmlns="DAV:")
prop = etree.SubElement(root, "prop")
etree.SubElement(prop, "quota-available-bytes")
etree.SubElement(prop, "quota-used-bytes")
tree = etree.ElementTree(root)
buff = BytesIO()
tree.write(buff)
return buff.getvalue()
response = self.execute_request(action='free', path='', data=data())
return parse(response)
data = WebDavXmlUtils.create_free_space_request_content()
response = self.execute_request(action='free', path='', data=data)
return WebDavXmlUtils.parse_free_space_response(response.content, self.webdav.hostname)
@wrap_connection_error
def check(self, remote_path=root):
@ -321,7 +275,7 @@ class Client(object):
return response.status_code == 200
@wrap_connection_error
def download_to(self, buff, remote_path):
def download_from(self, buff, remote_path):
"""Downloads file from WebDAV and writes it in buffer.
:param buff: buffer object for writing of downloaded file content.
@ -423,7 +377,7 @@ class Client(object):
threading.Thread(target=target).start()
@wrap_connection_error
def upload_from(self, buff, remote_path):
def upload_to(self, buff, remote_path):
"""Uploads file from buffer to remote path on WebDAV server.
:param buff: the buffer with content for file.
@ -544,6 +498,7 @@ 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()
@ -753,91 +708,48 @@ class Client(object):
return parse(response, path)
def resource(self, remote_path):
urn = Urn(remote_path)
return Resource(self, urn.path())
@wrap_connection_error
def get_property(self, remote_path, option):
"""
Gets metadata property of remote resource on WebDAV server.
More information you can find by link http://webdav.org/specs/rfc4918.html#METHOD_PROPFIND
def parse(response, option):
response_str = response.content
tree = etree.fromstring(response_str)
xpath = "{xpath_prefix}{xpath_exp}".format(xpath_prefix=".//", xpath_exp=option['name'])
return tree.findtext(xpath)
def data(option):
root = etree.Element("propfind", xmlns="DAV:")
prop = etree.SubElement(root, "prop")
etree.SubElement(prop, option.get('name', ""), xmlns=option.get('namespace', ""))
tree = etree.ElementTree(root)
buff = BytesIO()
tree.write(buff)
return buff.getvalue()
:param remote_path: the path to remote resource.
:param option: the property attribute as dictionary with following keys:
`namespace`: (optional) the namespace for XML property which will be set,
`name`: the name of property which will be set.
:return: the value of property or None if property is not found.
"""
urn = Urn(remote_path)
if not self.check(urn.path()):
raise RemoteResourceNotFound(urn.path())
url = {'hostname': self.webdav.hostname, 'root': self.webdav.root, 'path': urn.quote()}
options = {
'URL': "{hostname}{root}{path}".format(**url),
'CUSTOMREQUEST': Client.requests['get_metadata'],
'HTTPHEADER': self.get_header('get_metadata'),
'POSTFIELDS': data(option),
'NOBODY': 0
}
response = requests.request(
options["CUSTOMREQUEST"],
options["URL"],
auth=(self.webdav.login, self.webdav.password),
headers=self.get_header('get_metadata'),
data=data(option)
)
return parse(response, option)
data = WebDavXmlUtils.create_get_property_request_content(option)
response = self.execute_request(action='get_property', path=urn.quote(), data=data)
return WebDavXmlUtils.parse_get_property_response(response.content, option['name'])
@wrap_connection_error
def set_property(self, remote_path, option):
"""
Sets metadata property of remote resource on WebDAV server.
More information you can find by link http://webdav.org/specs/rfc4918.html#METHOD_PROPPATCH
def data(option):
root_node = etree.Element("propertyupdate", xmlns="DAV:")
root_node.set('xmlns:u', option.get('namespace', ""))
set_node = etree.SubElement(root_node, "set")
prop_node = etree.SubElement(set_node, "prop")
opt_node = etree.SubElement(prop_node, "{namespace}:{name}".format(namespace='u', name=option['name']))
opt_node.text = option.get('value', "")
tree = etree.ElementTree(root_node)
buff = BytesIO()
tree.write(buff)
return buff.getvalue()
:param remote_path: the path to remote resource.
:param option: the property attribute as dictionary with following keys:
`namespace`: (optional) the namespace for XML property which will be set,
`name`: the name of property which will be set,
`value`: (optional) the value of property which will be set. Defaults is empty string.
"""
urn = Urn(remote_path)
if not self.check(urn.path()):
raise RemoteResourceNotFound(urn.path())
url = {'hostname': self.webdav.hostname, 'root': self.webdav.root, 'path': urn.quote()}
options = {
'URL': "{hostname}{root}{path}".format(**url),
'CUSTOMREQUEST': Client.requests['set_metadata'],
'HTTPHEADER': self.get_header('get_metadata'),
'POSTFIELDS': data(option)
}
data = WebDavXmlUtils.create_set_property_request_content(option)
self.execute_request(action='set_property', path=urn.quote(), data=data)
response = requests.request(
options["CUSTOMREQUEST"],
options["URL"],
auth=(self.webdav.login, self.webdav.password),
headers=self.get_header('get_metadata'),
data=data(option)
)
def resource(self, remote_path):
urn = Urn(remote_path)
return Resource(self, urn.path())
def push(self, remote_directory, local_directory):
@ -960,7 +872,7 @@ class Resource(object):
return self.client.check(self.urn.path())
def read_from(self, buff):
self.client.upload_from(buff=buff, remote_path=self.urn.path())
self.client.upload_to(buff=buff, remote_path=self.urn.path())
def read(self, local_path):
return self.client.upload_sync(local_path=local_path, remote_path=self.urn.path())
@ -969,7 +881,7 @@ class Resource(object):
return self.client.upload_async(local_path=local_path, remote_path=self.urn.path(), callback=callback)
def write_to(self, buff):
return self.client.download_to(buff=buff, remote_path=self.urn.path())
return self.client.download_from(buff=buff, remote_path=self.urn.path())
def write(self, local_path):
return self.client.download_sync(local_path=local_path, remote_path=self.urn.path())
@ -991,3 +903,116 @@ class Resource(object):
def property(self, option, value):
option['value'] = value.__str__()
self.client.set_property(remote_path=self.urn.path(), option=option)
class WebDavXmlUtils:
def __init__(self):
pass
@staticmethod
def parse_get_list_response(content):
"""
Parses of response content XML from WebDAV server and extract file and directory names.
:param content: the XML content of HTTP response from WebDAV server for getting list of files by remote path.
:return: list of extracted file or directory names.
"""
try:
tree = etree.fromstring(content)
hrees = [unquote(hree.text) for hree in tree.findall(".//{DAV:}href")]
return [Urn(hree) for hree in hrees]
except etree.XMLSyntaxError:
return list()
@staticmethod
def create_free_space_request_content():
"""
Creates an XML for requesting of free space on remote WebDAV server.
:return: the XML string of request content.
"""
root = etree.Element("propfind", xmlns="DAV:")
prop = etree.SubElement(root, "prop")
etree.SubElement(prop, "quota-available-bytes")
etree.SubElement(prop, "quota-used-bytes")
tree = etree.ElementTree(root)
return WebDavXmlUtils.etree_to_string(tree)
@staticmethod
def parse_free_space_response(content, hostname):
"""
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.
:return: an amount of free space in bytes.
"""
try:
tree = etree.fromstring(content)
node = tree.find('.//{DAV:}quota-available-bytes')
if node is not None:
return int(node.text)
else:
raise MethodNotSupported(name='free', server=hostname)
except TypeError:
raise MethodNotSupported(name='free', server=hostname)
except etree.XMLSyntaxError:
return str()
@staticmethod
def create_get_property_request_content(option):
"""
Creates an XML for requesting of getting a property value of remote WebDAV resource.
:param option: the property attributes as dictionary with following keys:
`namespace`: (optional) the namespace for XML property which will be get,
`name`: the name of property which will be get.
:return: the XML string of request content.
"""
root = etree.Element("propfind", xmlns="DAV:")
prop = etree.SubElement(root, "prop")
etree.SubElement(prop, option.get('name', ""), xmlns=option.get('namespace', ""))
tree = etree.ElementTree(root)
return WebDavXmlUtils.etree_to_string(tree)
@staticmethod
def parse_get_property_response(content, name):
"""
Parses of response content XML from WebDAV server for getting metadata property value for some resource.
:param content: the XML content of response as string.
:param name: the name of property for finding a value in response
:return: the value of property if it has been found or None otherwise.
"""
tree = etree.fromstring(content)
return tree.xpath('//*[local-name() = $name]', name=name)[0].text
@staticmethod
def create_set_property_request_content(option):
"""
Creates an XML for requesting of setting a property value for remote WebDAV resource.
:param option: the property attributes as dictionary with following keys:
`namespace`: (optional) the namespace for XML property which will be set,
`name`: the name of property which will be set,
`value`: (optional) the value of property which will be set. Defaults is empty string.
:return: the XML string of request content.
"""
root_node = etree.Element('propertyupdate', xmlns='DAV:')
set_node = etree.SubElement(root_node, 'set')
prop_node = etree.SubElement(set_node, 'prop')
opt_node = etree.SubElement(prop_node, option['name'], xmlns=option.get('namespace', ''))
opt_node.text = option.get('value', '')
tree = etree.ElementTree(root_node)
return WebDavXmlUtils.etree_to_string(tree)
@staticmethod
def etree_to_string(tree):
"""
Creates string from lxml.etree.ElementTree with XML declaration and UTF-8 encoding.
:param tree: the instance of ElementTree
:return: the string of XML.
"""
buff = BytesIO()
tree.write(buff, xml_declaration=True, encoding='UTF-8')
return buff.getvalue()