Compare commits

...

67 commits

Author SHA1 Message Date
Alex Jordan
a3c75b7155 Fixes ezhov-evgeny#43 - local certificate options ignored
If the webdav_cert_path and webdav_key_path options are set, they
should be passed to the call to requests in Client.execute_request
as the cert parameter.
2020-04-06 15:40:35 -07:00
Evgeny Ezhov
39afefffc5 Fixed #49: Remove tailing slashes in web_hostname 2020-04-05 17:56:56 -07:00
Evgeny Ezhov
eeaf66a278 Update Travis CI config for SonarQube 2020-04-05 17:15:34 -07:00
Martin
af64110364 Fix request calling to use session's auth. 2020-04-05 11:43:51 -07:00
Evgeny Ezhov
3aab20b02a Bump version to 3.14.1 before publish 2020-02-23 17:43:12 -08:00
Evgeny Ezhov
0ac3c5e55f Small refactoring 2020-02-19 16:49:52 -08:00
Daniel Loader
3c5ba516af Add Oauth2 Bearer Token support
Basic implementation of authenticating with token instead of basic auth, as the auth=(login,password) flag on requests sends junk authentication as it overrides the bearer authentication header with a base64 encoded blank string.
2020-02-19 16:49:52 -08:00
Evgeny Ezhov
f64efedd96 Fixed issue #40 - error during coping and moving files with cyrillic names 2020-02-18 16:56:36 -08:00
Evgeny Ezhov
eced3b5fe5 Bump version to 3.14 and prepare to release 2020-01-31 21:02:43 -08:00
Evgeny Ezhov
fe4e56caad Cancel support Python earlie 3.5 2020-01-31 21:00:10 -08:00
Evgeny Ezhov
8c7ee2e372 Fix issues and tests 2020-01-31 21:00:10 -08:00
Evgeny Ezhov
85b10d53c9 Update release notes 2020-01-31 21:00:10 -08:00
Ishak BELAHMAR
b3c753c126 Handle 2-way sync when files are modified 2020-01-31 21:00:10 -08:00
Evgeny Ezhov
3d4751cdd0 Support Python 3.8 2020-01-26 19:50:01 -08:00
Evgeny Ezhov
c380dc56c7 Update readme release notes 2020-01-26 19:05:33 -08:00
Evgeny Ezhov
f3e7d44276 Support multiple clients simultaneously
Fixed #34
2020-01-26 19:00:25 -08:00
Evgeny Ezhov
fc14ed2be1 Update dependencies 2020-01-20 18:12:24 -08:00
Evgeny Ezhov
336db8ae19 Override methods for customizing communication with WebDAV servers.
Fix #31 and #30
2020-01-20 18:12:24 -08:00
Evgeny Ezhov
95bea2182c Remove release date 2019-12-17 13:57:29 +03:00
Evgeny Ezhov
7db1cfd274 Bump version to 0.14 2019-12-17 13:52:58 +03:00
Evgeny Ezhov
5e2a9468c1 Increase test coverage 2019-12-17 13:49:10 +03:00
Evgeny Ezhov
5aad617fbf Update tests and fix issues 2019-12-02 11:52:04 +03:00
Evgeny Ezhov
5f84bb1343 Fixed #24 an issue with checking resources on Yandex WebDAV 2019-12-02 11:52:04 +03:00
Evgeny Ezhov
0b9c61e7e7 Fixed SonarQube analysis issues 2019-11-28 14:50:39 +03:00
Evgeny Ezhov
1859e43daf Add manual for using API 2019-11-28 14:20:29 +03:00
Evgeny Ezhov
120a6adfab Added Coverage in SonarQube 2019-11-28 00:43:38 +03:00
Evgeny Ezhov
a3a8e31059 Update readme and setup 2019-11-27 22:57:42 +03:00
Evgeny Ezhov
c3a15dd145 Added SonarQube analysis 2019-11-27 22:28:32 +03:00
Evgeny Ezhov
fe3296f203 Support Python 2.7 in tests 2019-11-27 21:49:35 +03:00
Evgeny Ezhov
dc7d908462 Added an option to disable check in case WebDAV server is not support it 2019-11-27 16:18:50 +03:00
dzhuang
64bbd967bc Disable client.check by default. 2019-11-27 16:18:50 +03:00
Evgeny Ezhov
c90ee856fd Update Readme 2019-11-27 15:53:56 +03:00
Evgeny Ezhov
13ba29604a Style fixes 2019-11-27 15:52:31 +03:00
Tobias Küchel
8a919ac866 on pull() return a boolean, if something was changed on disk or not 2019-11-27 15:52:31 +03:00
Tobias Küchel
6b90e1d16c remove a costy is_dir-check on obvious directories to speed up a pull 2019-11-27 15:52:31 +03:00
Evgeny Ezhov
6415617852 Tests for push and pull methods 2019-11-27 15:52:31 +03:00
Evgeny Ezhov
5b1285392c Changed HEAD to GET method for 'check' request due of not all servers support HEAD 2019-11-27 15:18:02 +03:00
Evgeny Ezhov
78a979715d Adapt to use Apache WebDAV Container for CI 2019-11-27 15:00:52 +03:00
Evgeny Ezhov
ba2453800d Readme update 2019-11-27 15:00:52 +03:00
Evgeny Ezhov
64a3b66efb Add webdav container 2019-11-27 15:00:52 +03:00
Evgeny Ezhov
c238c281dd Merge branch 'master' into develop 2019-11-27 13:24:35 +03:00
Emil 'Skeen' Madsen
1fb79c36be Stripping suburl from paths in extract_response_for_path (#15)
When accessing a server on a suburl, such as an Alfresco WebDav:

http://172.17.0.8:8080/alfresco/webdav
The file paths processed by extract_response_for_path, specifically href contains the suburl as a prefix, for instance:
/alfresco/webdav/Sites/ rather than just Sites/.
I am not sure if this is a specific problem with Alfresco WebDav, or a common issue.
Either way, the provided code fixes this issue by removing the suburl from paths, if it exists.
2019-11-26 14:05:33 +03:00
Evgeny Ezhov
6118801098
Merge pull request #12 from delrey1/master
Updated to use sessions
2019-11-26 13:26:16 +03:00
Evgeny Ezhov
fdea5d1b70
Merge branch 'develop' into master 2019-10-13 17:56:46 +03:00
Evgeny Ezhov
7e409f1f1c Update git ignore 2019-10-13 17:54:45 +03:00
Evgeny Ezhov
c606ae3875
Merge branch 'develop' into master 2019-10-13 17:51:19 +03:00
Jorge
75a4241201 [Jorge] Updated to add verify to workaround request 2019-10-13 17:44:44 +03:00
Jorge
0f0721554b [Jorge] Updated to run a GET with proxy auth as overridden in subs request 2019-10-13 17:44:44 +03:00
Jorge
5a01cbf823 [Jorge] Switch to use python sessions rather than requests 2019-10-13 17:44:44 +03:00
Jorge
dc617d43aa [Jorge] Updated to add venv/.idea/build 2019-10-13 17:44:44 +03:00
Evgeny Ezhov
035c7d315e Update git ignore 2019-10-13 17:43:30 +03:00
Evgeny Ezhov
b663345e9d Tests are fixed 2019-10-13 17:27:14 +03:00
Evgeny Ezhov
d4edab9044 Main version of Python is updated up to 3.7
And small style fixes
2019-10-09 10:35:23 +03:00
Jorge
af85e056bd [Jorge] Updated to add verify to workaround request 2019-09-30 20:50:14 +01:00
Jorge
3632c2a1e0 [Jorge] Updated to run a GET with proxy auth as overridden in subs request 2019-09-30 20:41:01 +01:00
Jorge
8491fdec4c [Jorge] Switch to use python sessions rather than requests 2019-09-30 19:54:25 +01:00
Jorge
2e6284af1b [Jorge] Updated to add venv/.idea/build 2019-09-30 19:54:06 +01:00
Evgeny Ezhov
da46592c6f Update before release 0.12 2019-06-21 13:39:55 +03:00
Jesper Håkansson
991e645fdb Added verify attribute to execute_request method
To be able to control whether to verify the server's TLS certificate
or not, the verify attribute has been added to the execute_request
method.
2019-06-21 13:31:01 +03:00
Jesper Håkansson
ea139d2ec7 Added depth_to_copy argument in copy method in client.py
Implemented support to control the depth http header for the copy
command.
2019-06-21 13:29:41 +03:00
Evgeny Ezhov
1c43adb5bd Update before release 0.11 2019-03-30 09:37:03 +03:00
Evgeny Ezhov
e576958f4b Update before release 0.11 2019-03-30 09:35:21 +03:00
Eric Van Horn
d38aded83d fixed missing requirement; incorrect import 2019-03-30 09:29:19 +03:00
Benjamin Böhmke
460160ed04 fixed download of large files 2019-03-27 17:43:31 +03:00
Evgeny Ezhov
51458a439b Update before release 0.10 2019-01-31 15:44:03 +03:00
Zimmermann, Stefan
f63a7dc48e Remove the upload filesize limit which is a remnant of using PyCurl 2019-01-31 11:34:05 +03:00
Zimmermann, Stefan
17fae10abc Fixing problems of integration and unit tests
- AssertEquals deprecation warnings
- Problems with byte/UTF strings and xml library
2019-01-31 11:33:40 +03:00
27 changed files with 1295 additions and 433 deletions

7
.gitignore vendored
View file

@ -2,3 +2,10 @@
/**/*.pyc
/dist
/*.egg-info
/.eggs/
.project
.pydevproject
/.settings/
venv/
.idea/
build

View file

@ -1,7 +1,32 @@
language: python
dist: xenial
addons:
sonarcloud:
organization: "ezhov-evgeny"
token: f0f714f3bea6bd103e3eb82724ef3bb0d3b54d1d
services:
- docker
python:
- "2.7"
- "3.5"
- "3.6"
- "3.7"
- "3.8"
before_install:
- docker pull bytemark/webdav
- docker run -d --name webdav -e AUTH_TYPE=Basic -e USERNAME=alice -e PASSWORD=secret1234 -p 8585:80 bytemark/webdav
- docker ps -a
install:
- python setup.py develop
- pip install coverage
script:
- python setup.py test
- coverage run setup.py test
- coverage xml
- |
if [[ $TRAVIS_PYTHON_VERSION == "3.8" ]]; then
sonar-scanner
fi

378
README.md Normal file
View file

@ -0,0 +1,378 @@
webdavclient3
=========
[![Build Status](https://travis-ci.com/ezhov-evgeny/webdav-client-python-3.svg?branch=develop)](https://travis-ci.com/ezhov-evgeny/webdav-client-python-3)
[![Quality Gate Status](https://sonarcloud.io/api/project_badges/measure?project=ezhov-evgeny_webdav-client-python-3&metric=alert_status)](https://sonarcloud.io/dashboard?id=ezhov-evgeny_webdav-client-python-3)
[![Coverage](https://sonarcloud.io/api/project_badges/measure?project=ezhov-evgeny_webdav-client-python-3&metric=coverage)](https://sonarcloud.io/dashboard?id=ezhov-evgeny_webdav-client-python-3)
[![PyPI](https://img.shields.io/pypi/v/webdavclient3)](https://pypi.org/project/webdavclient3/) ![PyPI - Python Version](https://img.shields.io/pypi/pyversions/webdavclient3)
Package webdavclient3 based on https://github.com/designerror/webdav-client-python but uses `requests` instead of `PyCURL`.
It provides easy way to work with WebDAV-servers.
Installation
------------
```bash
$ pip install webdavclient3
```
Sample Usage
------------
```python
from webdav3.client import Client
options = {
'webdav_hostname': "https://webdav.server.ru",
'webdav_login': "login",
'webdav_password': "password"
}
client = Client(options)
client.verify = False # To not check SSL certificates (Default = True)
client.session.proxies(...) # To set proxy directly into the session (Optional)
client.session.auth(...) # To set proxy auth directly into the session (Optional)
client.execute_request("mkdir", 'directory_name')
```
Webdav API
==========
Webdav API is a set of webdav actions of work with cloud storage. This set includes the following actions:
`check`, `free`, `info`, `list`, `mkdir`, `clean`, `copy`, `move`, `download`, `upload`, `publish` and `unpublish`.
**Configuring the client**
Required keys for configuring client connection with WevDAV-server are `webdav_hostname` and `webdav_login`, `webdav_password`.
```python
from webdav3.client import Client
options = {
'webdav_hostname': "https://webdav.server.ru",
'webdav_login': "login",
'webdav_password': "password"
}
client = Client(options)
```
If your server does not support `HEAD` method or there are other reasons to override default WebDAV methods for actions use a dictionary option `webdav_override_methods`.
The key should be in the following list: `check`, `free`, `info`, `list`, `mkdir`, `clean`, `copy`, `move`, `download`, `upload`,
`publish` and `unpublish`. The value should a string name of WebDAV method, for example `GET`.
```python
from webdav3.client import Client
options = {
'webdav_hostname': "https://webdav.server.ru",
'webdav_login': "login",
'webdav_password': "password",
'webdav_override_methods': {
'check': 'GET'
}
}
client = Client(options)
```
When a proxy server you need to specify settings to connect through it.
```python
from webdav3.client import Client
options = {
'webdav_hostname': "https://webdav.server.ru",
'webdav_login': "w_login",
'webdav_password': "w_password",
'proxy_hostname': "http://127.0.0.1:8080",
'proxy_login': "p_login",
'proxy_password': "p_password"
}
client = Client(options)
```
If you want to use the certificate path to certificate and private key is defined as follows:
```python
from webdav3.client import Client
options = {
'webdav_hostname': "https://webdav.server.ru",
'webdav_login': "w_login",
'webdav_password': "w_password",
'cert_path': "/etc/ssl/certs/certificate.crt",
'key_path': "/etc/ssl/private/certificate.key"
}
client = Client(options)
```
Or you want to limit the speed or turn on verbose mode:
```python
options = {
...
'recv_speed' : 3000000,
'send_speed' : 3000000,
'verbose' : True
}
client = Client(options)
```
recv_speed: rate limit data download speed in Bytes per second. Defaults to unlimited speed.
send_speed: rate limit data upload speed in Bytes per second. Defaults to unlimited speed.
verbose: set verbose mode on/off. By default verbose mode is off.
Also if your server does not support `check` it is possible to disable it:
```python
options = {
...
'disable_check': True
}
client = Client(options)
```
By default checking of remote resources is enabled.
**Synchronous methods**
```python
# Checking existence of the resource
client.check("dir1/file1")
client.check("dir1")
```
```python
# Get information about the resource
client.info("dir1/file1")
client.info("dir1/")
```
```python
# Check free space
free_size = client.free()
```
```python
# Get a list of resources
files1 = client.list()
files2 = client.list("dir1")
```
```python
# Create directory
client.mkdir("dir1/dir2")
```
```python
# Delete resource
client.clean("dir1/dir2")
```
```python
# Copy resource
client.copy(remote_path_from="dir1/file1", remote_path_to="dir2/file1")
client.copy(remote_path_from="dir2", remote_path_to="dir3")
```
```python
# Move resource
client.move(remote_path_from="dir1/file1", remote_path_to="dir2/file1")
client.move(remote_path_from="dir2", remote_path_to="dir3")
```
```python
# Move resource
client.download_sync(remote_path="dir1/file1", local_path="~/Downloads/file1")
client.download_sync(remote_path="dir1/dir2/", local_path="~/Downloads/dir2/")
```
```python
# Unload resource
client.upload_sync(remote_path="dir1/file1", local_path="~/Documents/file1")
client.upload_sync(remote_path="dir1/dir2/", local_path="~/Documents/dir2/")
```
```python
# Publish the resource
link = client.publish("dir1/file1")
link = client.publish("dir2")
```
```python
# Unpublish resource
client.unpublish("dir1/file1")
client.unpublish("dir2")
```
```python
# Exception handling
from webdav3.client import WebDavException
try:
...
except WebDavException as exception:
...
```
```python
# Get the missing files
client.pull(remote_directory='dir1', local_directory='~/Documents/dir1')
```
```python
# Send missing files
client.push(remote_directory='dir1', local_directory='~/Documents/dir1')
```
**Asynchronous methods**
```python
# Load resource
kwargs = {
'remote_path': "dir1/file1",
'local_path': "~/Downloads/file1",
'callback': callback
}
client.download_async(**kwargs)
kwargs = {
'remote_path': "dir1/dir2/",
'local_path': "~/Downloads/dir2/",
'callback': callback
}
client.download_async(**kwargs)
```
```python
# Unload resource
kwargs = {
'remote_path': "dir1/file1",
'local_path': "~/Downloads/file1",
'callback': callback
}
client.upload_async(**kwargs)
kwargs = {
'remote_path': "dir1/dir2/",
'local_path': "~/Downloads/dir2/",
'callback': callback
}
client.upload_async(**kwargs)
```
Resource API
============
Resource API using the concept of OOP that enables cloud-level resources.
```python
# Get a resource
res1 = client.resource("dir1/file1")
```
```python
# Work with the resource
res1.rename("file2")
res1.move("dir1/file2")
res1.copy("dir2/file1")
info = res1.info()
res1.read_from(buffer)
res1.read(local_path="~/Documents/file1")
res1.read_async(local_path="~/Documents/file1", callback)
res1.write_to(buffer)
res1.write(local_path="~/Downloads/file1")
res1.write_async(local_path="~/Downloads/file1", callback)
```
Release Notes
-------------
**Version 3.14.1**
* Fixed issue during coping and moving files with cyrillic names
* Support OAuth2 bearer tokens by https://github.com/danielloader
**Version 3.14**
* Override methods for customizing communication with WebDAV servers
* Support multiple clients simultaneously
* Sync modified files during pull and push by https://github.com/mont5piques
**Version 0.14**
* Fixed an issue with checking resources on Yandex WebDAV server
**Version 0.13 27.11.2019**
* Main version of Python is updated up to 3.7
* Switch to use python sessions rather than requests by https://github.com/delrey1
* Stripping suburl from paths in extract_response_for_path by https://github.com/Skeen
* Added Docker Web DAV for CI
* Changed HEAD to GET method for 'check' request due of not all servers support HEAD by request of https://github.com/danieleTrimarchi
* Removed a costy is_dir-check on obvious directories to speed up a pull by https://github.com/jolly-jump
* Added an option to disable check in case WebDAV server is not support it by request of https://github.com/dzhuang
**Version 0.12 - 21.06.2019**
* Added depth argument in copy method in client.py by https://github.com/JesperHakansson
* Added verify attribute to execute_request method by https://github.com/JesperHakansson
**Version 0.11 30.03.2019**
* Fixed MemoryError if a large file is downloaded with a 32 bit python by https://github.com/bboehmke
* Fixed argcomplete is required to run wdc but was not included in the requirements by https://github.com/evanhorn
* Fixed wdc tries to import webdav instead of webdav3 by https://github.com/evanhorn
**Version 0.10 31.01.2019**
* AssertEquals deprecation warnings by https://github.com/StefanZi
* Problems with byte/UTF strings and xml library by https://github.com/StefanZi
* Add some Eclipse specific files to gitignore by https://github.com/StefanZi
* Remove filesize limit by https://github.com/StefanZi
**Version 0.9 10.05.2018**
* Client.mkdir now accepts 201 HTTP-code by https://github.com/a1ezzz
* Tests are updated
* Added Travis-CI
**Version 0.8 07.05.2018**
* Fixed issue in extract_response_for_path when a link in "href" attribute is an absolute link by https://github.com/a1ezzz
**Version 0.7 16.03.2018**
* Fixed issue with wrong argument for resource creation by https://github.com/janLo
**Version 0.6 21.02.2018**
* Fixed issue with in extracting response for path by https://github.com/mightydok
**Version 0.5 03.12.2017**
* Added method for setting of WebDAV resource property values in batch
**Version 0.4 - 27.11.2017**
* Refactoring of WebDAV client and making it works in following methods:
- Checking is remote resource directory
- Fixed problem when connection lost during request executing and nothing was happened, now it raises an exception
**Version 0.3 - 18.10.2017**
* 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
- Deleting of resource on WebDAV server
- Getting of information about WebDAV resource
**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

View file

@ -1,55 +0,0 @@
webdavclient3
=============
.. image:: https://travis-ci.com/ezhov-evgeny/webdav-client-python-3.svg?branch=master
:target: https://travis-ci.com/ezhov-evgeny/webdav-client-python-3
Based on https://github.com/designerror/webdav-client-python
But uses `requests` instead of `PyCURL`
Release Notes
=============
**Version 0.9 10.05.2018**
* Client.mkdir now accepts 201 HTTP-code by https://github.com/a1ezzz
* Tests are updated
* Added Travis-CI
**Version 0.8 07.05.2018**
* Fixed issue in extract_response_for_path when a link in "href" attribute is an absolute link by https://github.com/a1ezzz
**Version 0.7 16.03.2018**
* Fixed issue with wrong argument for resource creation by https://github.com/janLo
**Version 0.6 21.02.2018**
* Fixed issue with in extracting response for path by https://github.com/mightydok
**Version 0.5 03.12.2017**
* Added method for setting of WebDAV resource property values in batch
**Version 0.4 - 27.11.2017**
* Refactoring of WebDAV client and making it works in following methods:
- Checking is remote resource directory
- Fixed problem when connection lost during request executing and nothing was happened, now it raises an exception
**Version 0.3 - 18.10.2017**
* 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
- Deleting of resource on WebDAV server
- Getting of information about WebDAV resource
**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

View file

@ -6,7 +6,7 @@ from setuptools import setup, find_packages
from setuptools.command.install import install as InstallCommand
from setuptools.command.test import test as TestCommand
version = "0.9"
version = "3.14.1"
requirements = "libxml2-dev libxslt-dev python-dev"
@ -42,18 +42,25 @@ class Test(TestCommand):
errno = pytest.main(self.pytest_args)
sys.exit(errno)
try:
long_description = open('README.md', encoding="utf-8").read()
except TypeError:
long_description = open('README.md').read()
setup(
name='webdavclient3',
version=version,
packages=find_packages(),
requires=['python (>= 2.7.6)'],
install_requires=['requests', 'lxml'],
requires=['python (>= 3.3.0)'],
install_requires=['requests', 'lxml', 'python-dateutil'],
scripts=['wdc'],
tests_require=['pytest', 'pyhamcrest', 'junit-xml', 'pytest-allure-adaptor'],
test_suite='tests',
tests_require=['pytest'],
cmdclass={'install': Install, 'test': Test},
description='WebDAV client, based on original package https://github.com/designerror/webdav-client-python but '
'uses requests instead of PyCURL',
long_description=open('README.rst').read(),
long_description=long_description,
long_description_content_type='text/markdown',
author='Evgeny Ezhov',
author_email='ezhov.evgeny@gmail.com',
url='https://github.com/ezhov-evgeny/webdav-client-python-3',
@ -66,13 +73,10 @@ setup(
'Operating System :: MacOS',
'Operating System :: Microsoft',
'Operating System :: Unix',
'Programming Language :: Python :: 2.6',
'Programming Language :: Python :: 2.7',
'Programming Language :: Python :: 3.0',
'Programming Language :: Python :: 3.1',
'Programming Language :: Python :: 3.2',
'Programming Language :: Python :: 3.3',
'Programming Language :: Python :: 3.4',
'Programming Language :: Python :: 3.5',
'Programming Language :: Python :: 3.6',
'Programming Language :: Python :: 3.7',
'Programming Language :: Python :: 3.8',
'Topic :: Internet',
'Topic :: Software Development :: Libraries :: Python Modules',
],

11
sonar-project.properties Normal file
View file

@ -0,0 +1,11 @@
sonar.projectKey=ezhov-evgeny_webdav-client-python-3
sonar.organization=ezhov-evgeny
# This is the name and version displayed in the SonarCloud UI.
sonar.projectName=webdav-client-python-3
sonar.projectVersion=0.13
# Path is relative to the sonar-project.properties file. Replace "\" by "/" on Windows.
sonar.sources=webdav3
sonar.tests=tests
sonar.python.coverage.reportPaths=coverage.xml
# Encoding of the source code. Default is default system encoding
sonar.sourceEncoding=UTF-8

0
tests/__init__.py Normal file
View file

83
tests/base_client_it.py Normal file
View file

@ -0,0 +1,83 @@
import os
import shutil
import unittest
from os import path
from webdav3.client import Client
class BaseClientTestCase(unittest.TestCase):
remote_path_file = 'test_dir/test.txt'
remote_path_file2 = 'test_dir2/test.txt'
remote_inner_path_file = 'test_dir/inner/test.txt'
remote_path_dir = 'test_dir'
remote_path_dir2 = 'test_dir2'
remote_inner_path_dir = 'test_dir/inner'
inner_dir_name = 'inner'
local_base_dir = 'tests/'
local_file = 'test.txt'
local_file_path = local_base_dir + 'test.txt'
local_path_dir = local_base_dir + 'res/test_dir'
options = {
'webdav_hostname': 'http://localhost:8585',
'webdav_login': 'alice',
'webdav_password': 'secret1234',
'webdav_override_methods': {
'check': 'GET'
}
}
# options = {
# 'webdav_hostname': 'https://webdav.yandex.ru',
# 'webdav_login': 'webdavclient.test2',
# 'webdav_password': 'Qwerty123!'
# }
def setUp(self):
self.client = Client(self.options)
self.clean_local_dir(self.local_path_dir)
def tearDown(self):
self.clean_local_dir(self.local_path_dir)
self.clean_remote_dir(self.remote_path_dir)
self.clean_remote_dir(self.remote_path_dir2)
def clean_remote_dir(self, remote_path_dir):
if self.client.check(remote_path=remote_path_dir):
self.client.clean(remote_path=remote_path_dir)
@staticmethod
def clean_local_dir(local_path_dir):
if path.exists(path=local_path_dir):
shutil.rmtree(path=local_path_dir)
def _prepare_for_downloading(self, inner_dir=False, base_path=''):
if base_path:
self._create_remote_dir_if_needed(base_path)
self._prepare_dir_for_downloading(base_path + self.remote_path_dir, base_path + self.remote_path_file, self.local_file_path)
if not path.exists(self.local_path_dir):
os.makedirs(self.local_path_dir)
if inner_dir:
self._prepare_dir_for_downloading(base_path + self.remote_inner_path_dir, base_path + self.remote_inner_path_file, self.local_file_path)
def _prepare_dir_for_downloading(self, remote_path_dir, remote_path_file, local_file_path):
self._create_remote_dir_if_needed(remote_path_dir)
if not self.client.check(remote_path=remote_path_file):
self.client.upload_file(remote_path=remote_path_file, local_path=local_file_path)
def _create_remote_dir_if_needed(self, remote_dir):
if not self.client.check(remote_path=remote_dir):
self.client.mkdir(remote_path=remote_dir)
def _prepare_for_uploading(self):
if not self.client.check(remote_path=self.remote_path_dir):
self.client.mkdir(remote_path=self.remote_path_dir)
if not path.exists(path=self.local_path_dir):
os.makedirs(self.local_path_dir)
if not path.exists(path=self.local_path_dir + os.sep + self.local_file):
shutil.copy(src=self.local_file_path, dst=self.local_path_dir + os.sep + self.local_file)
if __name__ == '__main__':
unittest.main()

View file

@ -1,32 +0,0 @@
__author__ = 'designerror'
from hamcrest.core.base_matcher import BaseMatcher
from hamcrest.core.helpers.hasmethod import hasmethod
class Valid(BaseMatcher):
def _matches(self, item):
return False if not hasmethod(item, 'valid') else item.valid()
class NotValid(BaseMatcher):
def _matches(self, item):
return False if not hasmethod(item, 'valid') else not item.valid()
class Success(BaseMatcher):
def _matches(self, item):
return item
class NotSuccess(BaseMatcher):
def _matches(self, item):
return not item
def valid():
return Valid()
def not_valid():
return NotValid()
def success():
return Success()
def not_success():
return NotSuccess()

136
tests/response_dir.xml Normal file
View file

@ -0,0 +1,136 @@
<?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:creationdate>2012-04-04T20:00:00Z</d:creationdate>
<d:displayname>disk</d:displayname>
<d:getlastmodified>Wed, 04 Apr 2012 20:00:00 GMT</d:getlastmodified>
<d:resourcetype>
<d:collection/>
</d:resourcetype>
</d:prop>
</d:propstat>
</d:response>
<d:response>
<d:href>/test_dir/</d:href>
<d:propstat>
<d:status>HTTP/1.1 200 OK</d:status>
<d:prop>
<d:creationdate>2018-05-10T07:31:13Z</d:creationdate>
<d:displayname>test_dir</d:displayname>
<d:getlastmodified>Thu, 10 May 2018 07:31:13 GMT</d:getlastmodified>
<d:resourcetype>
<d:collection/>
</d:resourcetype>
</d:prop>
</d:propstat>
</d:response>
<d:response>
<d:href>/%D0%93%D0%BE%D1%80%D1%8B.jpg</d:href>
<d:propstat>
<d:status>HTTP/1.1 200 OK</d:status>
<d:prop>
<d:getetag>1392851f0668017168ee4b5a59d66e7b</d:getetag>
<d:creationdate>2018-05-09T14:44:28Z</d:creationdate>
<d:displayname>Горы.jpg</d:displayname>
<d:getlastmodified>Wed, 09 May 2018 14:44:28 GMT</d:getlastmodified>
<d:getcontenttype>image/jpeg</d:getcontenttype>
<d:getcontentlength>1762478</d:getcontentlength>
<d:resourcetype/>
</d:prop>
</d:propstat>
</d:response>
<d:response>
<d:href>/%D0%97%D0%B8%D0%BC%D0%B0.jpg</d:href>
<d:propstat>
<d:status>HTTP/1.1 200 OK</d:status>
<d:prop>
<d:getetag>a64146fee5e15b3b94c204e544426d43</d:getetag>
<d:creationdate>2018-05-09T14:44:28Z</d:creationdate>
<d:displayname>Зима.jpg</d:displayname>
<d:getlastmodified>Wed, 09 May 2018 14:44:28 GMT</d:getlastmodified>
<d:getcontenttype>image/jpeg</d:getcontenttype>
<d:getcontentlength>1394575</d:getcontentlength>
<d:resourcetype/>
</d:prop>
</d:propstat>
</d:response>
<d:response>
<d:href>/%D0%9C%D0%B8%D1%88%D0%BA%D0%B8.jpg</d:href>
<d:propstat>
<d:status>HTTP/1.1 200 OK</d:status>
<d:prop>
<d:getetag>569a1c98696050439b5b2a1ecfa52d19</d:getetag>
<d:creationdate>2018-05-09T14:44:27Z</d:creationdate>
<d:displayname>Мишки.jpg</d:displayname>
<d:getlastmodified>Wed, 09 May 2018 14:44:27 GMT</d:getlastmodified>
<d:getcontenttype>image/jpeg</d:getcontenttype>
<d:getcontentlength>1555830</d:getcontentlength>
<d:resourcetype/>
</d:prop>
</d:propstat>
</d:response>
<d:response>
<d:href>/%D0%9C%D0%BE%D1%80%D0%B5.jpg</d:href>
<d:propstat>
<d:status>HTTP/1.1 200 OK</d:status>
<d:prop>
<d:getetag>ab903d9cab031eca2a8f12f37bbc9d37</d:getetag>
<d:creationdate>2018-05-09T14:44:27Z</d:creationdate>
<d:displayname>Море.jpg</d:displayname>
<d:getlastmodified>Wed, 09 May 2018 14:44:27 GMT</d:getlastmodified>
<d:getcontenttype>image/jpeg</d:getcontenttype>
<d:getcontentlength>1080301</d:getcontentlength>
<d:resourcetype/>
</d:prop>
</d:propstat>
</d:response>
<d:response>
<d:href>/%D0%9C%D0%BE%D1%81%D0%BA%D0%B2%D0%B0.jpg</d:href>
<d:propstat>
<d:status>HTTP/1.1 200 OK</d:status>
<d:prop>
<d:getetag>d27d72a3059ad5ebed7a5470459d2670</d:getetag>
<d:creationdate>2018-05-09T14:44:27Z</d:creationdate>
<d:displayname>Москва.jpg</d:displayname>
<d:getlastmodified>Wed, 09 May 2018 14:44:27 GMT</d:getlastmodified>
<d:getcontenttype>image/jpeg</d:getcontenttype>
<d:getcontentlength>1454228</d:getcontentlength>
<d:resourcetype/>
</d:prop>
</d:propstat>
</d:response>
<d:response>
<d:href>/%D0%A1%D0%B0%D0%BD%D0%BA%D1%82-%D0%9F%D0%B5%D1%82%D0%B5%D1%80%D0%B1%D1%83%D1%80%D0%B3.jpg</d:href>
<d:propstat>
<d:status>HTTP/1.1 200 OK</d:status>
<d:prop>
<d:getetag>f1abe3b27b410128623fd1ca00a45c29</d:getetag>
<d:creationdate>2018-05-09T14:44:27Z</d:creationdate>
<d:displayname>Санкт-Петербург.jpg</d:displayname>
<d:getlastmodified>Wed, 09 May 2018 14:44:27 GMT</d:getlastmodified>
<d:getcontenttype>image/jpeg</d:getcontenttype>
<d:getcontentlength>2573704</d:getcontentlength>
<d:resourcetype/>
</d:prop>
</d:propstat>
</d:response>
<d:response>
<d:href>/%D0%A5%D0%BB%D0%B5%D0%B1%D0%BD%D1%8B%D0%B5%20%D0%BA%D1%80%D0%BE%D1%88%D0%BA%D0%B8.mp4</d:href>
<d:propstat>
<d:status>HTTP/1.1 200 OK</d:status>
<d:prop>
<d:getetag>ea977f513074d5524bee3638798183b9</d:getetag>
<d:creationdate>2018-05-09T14:44:28Z</d:creationdate>
<d:displayname>Хлебные крошки.mp4</d:displayname>
<d:getlastmodified>Wed, 09 May 2018 14:44:28 GMT</d:getlastmodified>
<d:getcontenttype>video/mp4</d:getcontenttype>
<d:getcontentlength>31000079</d:getcontentlength>
<d:resourcetype/>
</d:prop>
</d:propstat>
</d:response>
</d:multistatus>

View file

@ -3,38 +3,14 @@ import shutil
import unittest
from io import BytesIO, StringIO
from os import path
from unittest import TestCase
from time import sleep
from webdav3.client import Client
from tests.base_client_it import BaseClientTestCase
from webdav3.exceptions import MethodNotSupported, OptionNotValid, RemoteResourceNotFound
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_base_dir = 'tests/'
local_file = 'test.txt'
local_file_path = local_base_dir + 'test.txt'
local_path_dir = local_base_dir + 'res/test_dir'
def setUp(self):
options = {
'webdav_hostname': 'https://webdav.yandex.ru',
'webdav_login': 'webdavclient.test2',
'webdav_password': 'Qwerty123!'
}
self.client = Client(options)
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)
class ClientTestCase(BaseClientTestCase):
pulled_file = BaseClientTestCase.local_path_dir + os.sep + BaseClientTestCase.local_file
def test_list(self):
self._prepare_for_downloading()
@ -43,6 +19,10 @@ class ClientTestCase(TestCase):
self.assertGreater(file_list.__len__(), 0, 'Expected that amount of files more then 0')
def test_free(self):
if 'localhost' in self.options['webdav_hostname']:
with self.assertRaises(MethodNotSupported):
self.client.free()
else:
self.assertGreater(self.client.free(), 0, 'Expected that free space on WebDAV server is more then 0 bytes')
def test_check(self):
@ -54,16 +34,29 @@ class ClientTestCase(TestCase):
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.')
@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_from(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')
self.assertEqual(buff.getvalue(), b'test content for testing of webdav client')
def test_download_from_dir(self):
self._prepare_for_downloading()
buff = BytesIO()
with self.assertRaises(OptionNotValid):
self.client.download_from(buff=buff, remote_path=self.remote_path_dir)
def test_download_from_wrong_file(self):
self._prepare_for_downloading()
buff = BytesIO()
with self.assertRaises(RemoteResourceNotFound):
self.client.download_from(buff=buff, remote_path='wrong')
def test_download_directory_wrong(self):
self._prepare_for_downloading()
with self.assertRaises(RemoteResourceNotFound):
self.client.download_directory(remote_path=self.remote_path_file, local_path=self.local_path_dir)
@unittest.skip("Yandex brakes response for file it contains property resourcetype as collection but it should "
"be empty for file")
def test_download(self):
self._prepare_for_downloading()
self.client.download(local_path=self.local_path_dir, remote_path=self.remote_path_dir)
@ -71,14 +64,11 @@ class ClientTestCase(TestCase):
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),
'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')
@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):
self._prepare_for_downloading()
os.mkdir(self.local_path_dir)
def callback():
self.assertTrue(path.exists(self.local_path_dir + os.path.sep + self.local_file),
@ -91,11 +81,8 @@ class ClientTestCase(TestCase):
self.assertTrue(path.exists(self.local_path_dir + os.path.sep + self.local_file),
'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):
self._prepare_for_downloading()
os.mkdir(self.local_path_dir)
def callback():
self.assertTrue(path.exists(self.local_path_dir + os.path.sep + self.local_file),
@ -107,10 +94,13 @@ class ClientTestCase(TestCase):
remote_path=self.remote_path_file, callback=callback)
self.assertFalse(path.exists(self.local_path_dir + os.path.sep + self.local_file),
'Expected the file has not been downloaded yet')
# It needs for ending download before environment will be cleaned in tearDown
sleep(0.4)
def test_upload_from(self):
self._prepare_for_uploading()
buff = StringIO(u'test content for testing of webdav client')
buff = StringIO()
buff.write(u'test content for testing of webdav client')
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.')
@ -165,8 +155,7 @@ class ClientTestCase(TestCase):
def test_info(self):
self._prepare_for_downloading()
result = self.client.info(remote_path=self.remote_path_file)
self.assertEquals(result['name'], 'test.txt')
self.assertEquals(result['size'], '41')
self.assertEqual(result['size'], '41')
self.assertTrue('created' in result)
self.assertTrue('modified' in result)
@ -181,7 +170,7 @@ class ClientTestCase(TestCase):
def test_get_property_of_non_exist(self):
self._prepare_for_downloading()
result = self.client.get_property(remote_path=self.remote_path_file, option={'name': 'aProperty'})
self.assertEquals(result, None, 'For not found property should return value as None')
self.assertEqual(result, None, 'For not found property should return value as None')
def test_set_property(self):
self._prepare_for_downloading()
@ -192,7 +181,7 @@ class ClientTestCase(TestCase):
})
result = self.client.get_property(remote_path=self.remote_path_file,
option={'namespace': 'test', 'name': 'aProperty'})
self.assertEquals(result, 'aValue', 'Property value should be set')
self.assertEqual(result, 'aValue', 'Property value should be set')
def test_set_property_batch(self):
self._prepare_for_downloading()
@ -210,26 +199,84 @@ class ClientTestCase(TestCase):
])
result = self.client.get_property(remote_path=self.remote_path_file,
option={'namespace': 'test', 'name': 'aProperty'})
self.assertEquals(result, 'aValue', 'First property value should be set')
self.assertEqual(result, 'aValue', 'First property value should be set')
result = self.client.get_property(remote_path=self.remote_path_file,
option={'namespace': 'test', 'name': 'aProperty2'})
self.assertEquals(result, 'aValue2', 'Second property value should be set')
self.assertEqual(result, 'aValue2', 'Second property value should be set')
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)
if not self.client.check(remote_path=self.remote_path_file):
self.client.upload_file(remote_path=self.remote_path_file, local_path=self.local_file_path)
if not path.exists(self.local_path_dir):
os.makedirs(self.local_path_dir)
def test_pull(self):
self._prepare_for_downloading(True)
self.client.pull(self.remote_path_dir, self.local_path_dir)
self.assertTrue(path.exists(self.local_path_dir), 'Expected the directory is downloaded.')
self.assertTrue(path.exists(self.local_path_dir + os.path.sep + self.inner_dir_name), 'Expected the directory is downloaded.')
self.assertTrue(path.exists(self.local_path_dir + os.path.sep + self.inner_dir_name), 'Expected the directory is downloaded.')
self.assertTrue(path.isdir(self.local_path_dir), 'Expected this is a directory.')
self.assertTrue(path.isdir(self.local_path_dir + os.path.sep + self.inner_dir_name), 'Expected this is a directory.')
self.assertTrue(path.exists(self.local_path_dir + os.path.sep + self.local_file),
'Expected the file is downloaded')
self.assertTrue(path.isfile(self.local_path_dir + os.path.sep + self.local_file),
'Expected this is a file')
def _prepare_for_uploading(self):
if not self.client.check(remote_path=self.remote_path_dir):
self.client.mkdir(remote_path=self.remote_path_dir)
def test_push(self):
self._prepare_for_uploading()
self.client.push(self.remote_path_dir, self.local_path_dir)
self.assertTrue(self.client.check(self.remote_path_dir), 'Expected the directory is created.')
self.assertTrue(self.client.check(self.remote_path_file), 'Expected the file is uploaded.')
def test_valid(self):
self.assertTrue(self.client.valid())
def test_check_is_overridden(self):
self.assertEqual('GET', self.client.requests['check'])
def test_pull_newer(self):
init_modification_time = int(self._prepare_local_test_file_and_get_modification_time())
sleep(1)
self._prepare_for_downloading(base_path='time/')
result = self.client.pull('time/' + self.remote_path_dir, self.local_path_dir)
update_modification_time = int(os.path.getmtime(self.pulled_file))
self.assertTrue(result)
self.assertGreater(update_modification_time, init_modification_time)
self.client.clean(remote_path='time/' + self.remote_path_dir)
def test_pull_older(self):
self._prepare_for_downloading(base_path='time/')
sleep(1)
init_modification_time = int(self._prepare_local_test_file_and_get_modification_time())
result = self.client.pull('time/' + self.remote_path_dir, self.local_path_dir)
update_modification_time = int(os.path.getmtime(self.pulled_file))
self.assertFalse(result)
self.assertEqual(update_modification_time, init_modification_time)
self.client.clean(remote_path='time/' + self.remote_path_dir)
def test_push_newer(self):
self._prepare_for_downloading(base_path='time/')
sleep(1)
self._prepare_for_uploading()
init_modification_time = self.client.info('time/' + self.remote_path_file)['modified']
result = self.client.push('time/' + self.remote_path_dir, self.local_path_dir)
update_modification_time = self.client.info('time/' + self.remote_path_file)['modified']
self.assertTrue(result)
self.assertNotEqual(init_modification_time, update_modification_time)
self.client.clean(remote_path='time/' + self.remote_path_dir)
def test_push_older(self):
self._prepare_for_uploading()
sleep(1)
self._prepare_for_downloading(base_path='time/')
init_modification_time = self.client.info('time/' + self.remote_path_file)['modified']
result = self.client.push('time/' + self.remote_path_dir, self.local_path_dir)
update_modification_time = self.client.info('time/' + self.remote_path_file)['modified']
self.assertFalse(result)
self.assertEqual(init_modification_time, update_modification_time)
self.client.clean(remote_path='time/' + self.remote_path_dir)
def _prepare_local_test_file_and_get_modification_time(self):
if not path.exists(path=self.local_path_dir):
os.makedirs(self.local_path_dir)
os.mkdir(self.local_path_dir)
if not path.exists(path=self.local_path_dir + os.sep + self.local_file):
shutil.copy(src=self.local_file_path, dst=self.local_path_dir + os.sep + self.local_file)
shutil.copy(src=self.local_file_path, dst=self.pulled_file)
return os.path.getmtime(self.pulled_file)
if __name__ == '__main__':

View file

@ -0,0 +1,75 @@
import unittest
from tests.base_client_it import BaseClientTestCase
from webdav3.client import Resource
from webdav3.urn import Urn
class ResourceTestCase(BaseClientTestCase):
def test_str(self):
self._prepare_for_downloading()
resource = Resource(self.client, Urn(self.remote_path_file))
self.assertEqual('resource /test_dir/test.txt', resource.__str__())
def test_is_not_dir(self):
self._prepare_for_downloading()
resource = Resource(self.client, Urn(self.remote_path_file))
self.assertFalse(resource.is_dir())
def test_is_dir(self):
self._prepare_for_downloading()
resource = Resource(self.client, Urn(self.remote_path_dir))
self.assertTrue(resource.is_dir())
def test_rename(self):
self._prepare_for_downloading()
resource = Resource(self.client, Urn(self.remote_path_file))
resource.rename('new_name.text')
self.assertTrue(self.client.check(self.remote_path_dir + '/new_name.text'))
def test_move(self):
self._prepare_for_downloading()
resource = Resource(self.client, Urn(self.remote_path_file))
self.client.mkdir(self.remote_path_dir2)
resource.move(self.remote_path_file2)
self.assertFalse(self.client.check(self.remote_path_file))
self.assertTrue(self.client.check(self.remote_path_file2))
def test_copy(self):
self._prepare_for_downloading()
resource = Resource(self.client, Urn(self.remote_path_file))
self.client.mkdir(self.remote_path_dir2)
resource.copy(self.remote_path_file2)
self.assertTrue(self.client.check(self.remote_path_file))
self.assertTrue(self.client.check(self.remote_path_file2))
def test_info(self):
self._prepare_for_downloading()
resource = Resource(self.client, Urn(self.remote_path_file))
info = resource.info()
self.assertIsNotNone(info)
self.assertGreater(len(info), 0)
def test_info_params(self):
self._prepare_for_downloading()
resource = Resource(self.client, Urn(self.remote_path_file))
info = resource.info(['size'])
self.assertIsNotNone(info)
self.assertEqual(1, len(info))
self.assertTrue('size' in info)
def test_clean(self):
self._prepare_for_downloading()
resource = Resource(self.client, Urn(self.remote_path_file))
resource.clean()
self.assertFalse(self.client.check(self.remote_path_file))
def test_check(self):
self._prepare_for_downloading()
resource = Resource(self.client, Urn(self.remote_path_file))
self.assertTrue(resource.check())
if __name__ == '__main__':
unittest.main()

View file

@ -4,7 +4,7 @@ from unittest import TestCase
from lxml.etree import ElementTree, Element
from webdav3.client import WebDavXmlUtils as utils
from webdav3.client import WebDavXmlUtils as utils, listdir
class ClientTestCase(TestCase):
@ -18,21 +18,21 @@ class ClientTestCase(TestCase):
'<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)
result = utils.parse_get_list_response(content.encode('utf-8'))
self.assertEqual(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>')
self.assertEqual(result, b'<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<propfind xmlns="DAV:"><prop>'
b'<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, 'localhost')
self.assertEquals(result, 10737417543)
result = utils.parse_free_space_response(content.encode('utf-8'), 'localhost')
self.assertEqual(result, 10737417543)
def test_parse_info_response(self):
content = '<?xml version="1.0" encoding="utf-8"?><d:multistatus xmlns:d="DAV:"><d:response>' \
@ -42,11 +42,11 @@ class ClientTestCase(TestCase):
'</d:getcontenttype><d:getcontentlength>41</d:getcontentlength><d:displayname>test.txt' \
'</d:displayname><d:creationdate>2017-10-18T15:16:04Z</d:creationdate></d:prop></d:propstat>' \
'</d:response></d:multistatus>'
result = utils.parse_info_response(content, '/test_dir/test.txt', 'localhost')
self.assertEquals(result['created'], '2017-10-18T15:16:04Z')
self.assertEquals(result['name'], 'test.txt')
self.assertEquals(result['modified'], 'Wed, 18 Oct 2017 15:16:04 GMT')
self.assertEquals(result['size'], '41')
result = utils.parse_info_response(content.encode('utf-8'), '/test_dir/test.txt', 'localhost')
self.assertEqual(result['created'], '2017-10-18T15:16:04Z')
self.assertEqual(result['name'], 'test.txt')
self.assertEqual(result['modified'], 'Wed, 18 Oct 2017 15:16:04 GMT')
self.assertEqual(result['size'], '41')
def test_create_get_property_request_content(self):
option = {
@ -54,24 +54,24 @@ class ClientTestCase(TestCase):
'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>')
self.assertEqual(result, b'<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<propfind xmlns="DAV:"><prop>'
b'<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>')
self.assertEqual(result, b'<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<propfind xmlns="DAV:"><prop>'
b'<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')
result = utils.parse_get_property_response(content=content.encode('utf-8'), name='aProperty')
self.assertEqual(result, 'aValue')
def test_create_set_one_property_request_content(self):
option = {
@ -80,16 +80,16 @@ class ClientTestCase(TestCase):
'value': 'aValue'
}
result = utils.create_set_property_batch_request_content(options=[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>')
self.assertEqual(result, b'<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<propertyupdate xmlns="DAV:"><set><prop>'
b'<aProperty xmlns="test">aValue</aProperty></prop></set></propertyupdate>')
def test_create_set_one_property_request_content_name_only(self):
option = {
'name': 'aProperty'
}
result = utils.create_set_property_batch_request_content(options=[option])
self.assertEquals(result, '<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<propertyupdate xmlns="DAV:"><set><prop>'
'<aProperty xmlns=""></aProperty></prop></set></propertyupdate>')
self.assertEqual(result, b'<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<propertyupdate xmlns="DAV:"><set><prop>'
b'<aProperty xmlns=""></aProperty></prop></set></propertyupdate>')
def test_create_set_property_batch_request_content(self):
options = [
@ -105,9 +105,9 @@ class ClientTestCase(TestCase):
}
]
result = utils.create_set_property_batch_request_content(options=options)
self.assertEquals(result, '<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<propertyupdate xmlns="DAV:"><set><prop>'
'<aProperty xmlns="test">aValue</aProperty><aProperty2 xmlns="test2">aValue2'
'</aProperty2></prop></set></propertyupdate>')
self.assertEqual(result, b'<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<propertyupdate xmlns="DAV:"><set><prop>'
b'<aProperty xmlns="test">aValue</aProperty><aProperty2 xmlns="test2">aValue2'
b'</aProperty2></prop></set></propertyupdate>')
def test_create_set_property_batch_request_content_name_only(self):
options = [
@ -119,66 +119,29 @@ class ClientTestCase(TestCase):
}
]
result = utils.create_set_property_batch_request_content(options=options)
self.assertEquals(result, '<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<propertyupdate xmlns="DAV:"><set><prop>'
'<aProperty xmlns=""></aProperty><aProperty2 xmlns=""></aProperty2></prop></set>'
'</propertyupdate>')
self.assertEqual(result, b'<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<propertyupdate xmlns="DAV:"><set><prop>'
b'<aProperty xmlns=""></aProperty><aProperty2 xmlns=""></aProperty2></prop></set>'
b'</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/>')
self.assertEqual(result, b'<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<test/>')
def test_parse_is_dir_response_directory(self):
content = '<?xml version=\'1.0\' encoding=\'UTF-8\'?><d:multistatus xmlns:d="DAV:"><d:response><d:href>/</d:h' \
'ref><d:propstat><d:status>HTTP/1.1 200 OK</d:status><d:prop><d:creationdate>2012-04-04T20:00:00Z</' \
'd:creationdate><d:displayname>disk</d:displayname><d:getlastmodified>Wed, 04 Apr 2012 20:00:00 GMT' \
'</d:getlastmodified><d:resourcetype><d:collection/></d:resourcetype></d:prop></d:propstat></d:resp' \
'onse><d:response><d:href>/test_dir/</d:href><d:propstat><d:status>HTTP/1.1 200 OK</d:status><d:pro' \
'p><d:creationdate>2018-05-10T07:31:13Z</d:creationdate><d:displayname>test_dir</d:displayname><d:g' \
'etlastmodified>Thu, 10 May 2018 07:31:13 GMT</d:getlastmodified><d:resourcetype><d:collection/></d' \
':resourcetype></d:prop></d:propstat></d:response><d:response><d:href>/%D0%93%D0%BE%D1%80%D1%8B.jpg' \
'</d:href><d:propstat><d:status>HTTP/1.1 200 OK</d:status><d:prop><d:getetag>1392851f0668017168ee4b' \
'5a59d66e7b</d:getetag><d:creationdate>2018-05-09T14:44:28Z</d:creationdate><d:displayname>Горы.jpg' \
'</d:displayname><d:getlastmodified>Wed, 09 May 2018 14:44:28 GMT</d:getlastmodified><d:getcontentt' \
'ype>image/jpeg</d:getcontenttype><d:getcontentlength>1762478</d:getcontentlength><d:resourcetype/>' \
'</d:prop></d:propstat></d:response><d:response><d:href>/%D0%97%D0%B8%D0%BC%D0%B0.jpg</d:href><d:pr' \
'opstat><d:status>HTTP/1.1 200 OK</d:status><d:prop><d:getetag>a64146fee5e15b3b94c204e544426d43</d:' \
'getetag><d:creationdate>2018-05-09T14:44:28Z</d:creationdate><d:displayname>Зима.jpg</d:displaynam' \
'e><d:getlastmodified>Wed, 09 May 2018 14:44:28 GMT</d:getlastmodified><d:getcontenttype>image/jpeg' \
'</d:getcontenttype><d:getcontentlength>1394575</d:getcontentlength><d:resourcetype/></d:prop></d:p' \
'ropstat></d:response><d:response><d:href>/%D0%9C%D0%B8%D1%88%D0%BA%D0%B8.jpg</d:href><d:propstat><' \
'd:status>HTTP/1.1 200 OK</d:status><d:prop><d:getetag>569a1c98696050439b5b2a1ecfa52d19</d:getetag>' \
'<d:creationdate>2018-05-09T14:44:27Z</d:creationdate><d:displayname>Мишки.jpg</d:displayname><d:ge' \
'tlastmodified>Wed, 09 May 2018 14:44:27 GMT</d:getlastmodified><d:getcontenttype>image/jpeg</d:get' \
'contenttype><d:getcontentlength>1555830</d:getcontentlength><d:resourcetype/></d:prop></d:propstat' \
'></d:response><d:response><d:href>/%D0%9C%D0%BE%D1%80%D0%B5.jpg</d:href><d:propstat><d:status>HTTP' \
'/1.1 200 OK</d:status><d:prop><d:getetag>ab903d9cab031eca2a8f12f37bbc9d37</d:getetag><d:creationda' \
'te>2018-05-09T14:44:27Z</d:creationdate><d:displayname>Море.jpg</d:displayname><d:getlastmodified>' \
'Wed, 09 May 2018 14:44:27 GMT</d:getlastmodified><d:getcontenttype>image/jpeg</d:getcontenttype><d' \
':getcontentlength>1080301</d:getcontentlength><d:resourcetype/></d:prop></d:propstat></d:response>' \
'<d:response><d:href>/%D0%9C%D0%BE%D1%81%D0%BA%D0%B2%D0%B0.jpg</d:href><d:propstat><d:status>HTTP/1' \
'.1 200 OK</d:status><d:prop><d:getetag>d27d72a3059ad5ebed7a5470459d2670</d:getetag><d:creationdate' \
'>2018-05-09T14:44:27Z</d:creationdate><d:displayname>Москва.jpg</d:displayname><d:getlastmodified>' \
'Wed, 09 May 2018 14:44:27 GMT</d:getlastmodified><d:getcontenttype>image/jpeg</d:getcontenttype><d' \
':getcontentlength>1454228</d:getcontentlength><d:resourcetype/></d:prop></d:propstat></d:response>' \
'<d:response><d:href>/%D0%A1%D0%B0%D0%BD%D0%BA%D1%82-%D0%9F%D0%B5%D1%82%D0%B5%D1%80%D0%B1%D1%83%D1%' \
'80%D0%B3.jpg</d:href><d:propstat><d:status>HTTP/1.1 200 OK</d:status><d:prop><d:getetag>f1abe3b27b' \
'410128623fd1ca00a45c29</d:getetag><d:creationdate>2018-05-09T14:44:27Z</d:creationdate><d:displayn' \
'ame>Санкт-Петербург.jpg</d:displayname><d:getlastmodified>Wed, 09 May 2018 14:44:27 GMT</d:getlast' \
'modified><d:getcontenttype>image/jpeg</d:getcontenttype><d:getcontentlength>2573704</d:getcontentl' \
'ength><d:resourcetype/></d:prop></d:propstat></d:response><d:response><d:href>/%D0%A5%D0%BB%D0%B5%' \
'D0%B1%D0%BD%D1%8B%D0%B5%20%D0%BA%D1%80%D0%BE%D1%88%D0%BA%D0%B8.mp4</d:href><d:propstat><d:status>H' \
'TTP/1.1 200 OK</d:status><d:prop><d:getetag>ea977f513074d5524bee3638798183b9</d:getetag><d:creatio' \
'ndate>2018-05-09T14:44:28Z</d:creationdate><d:displayname>Хлебные крошки.mp4</d:displayname><d:get' \
'lastmodified>Wed, 09 May 2018 14:44:28 GMT</d:getlastmodified><d:getcontenttype>video/mp4</d:getco' \
'ntenttype><d:getcontentlength>31000079</d:getcontentlength><d:resourcetype/></d:prop></d:propstat>' \
'</d:response></d:multistatus>'
try:
f = open('./tests/response_dir.xml', encoding='utf-8')
content = f.read().encode('utf-8')
except:
f = open('./tests/response_dir.xml')
content = f.read().decode('utf-8').encode('utf-8')
f.close()
path = '/test_dir'
hostname = 'https://webdav.yandex.ru'
result = utils.parse_is_dir_response(content, path, hostname)
self.assertTrue(result, 'It should be directory')
def test_parse_is_dir_response_directory(self):
def test_parse_is_dir_response_file(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:creationdate>2018-05-10T07' \
':40:11Z</d:creationdate><d:displayname>test_dir</d:displayname><d:getlastmodified>Thu, 10 May 2018' \
@ -190,9 +153,14 @@ class ClientTestCase(TestCase):
'th><d:resourcetype/></d:prop></d:propstat></d:response></d:multistatus>'
path = '/test_dir/test.txt'
hostname = 'https://webdav.yandex.ru'
result = utils.parse_is_dir_response(content, path, hostname)
result = utils.parse_is_dir_response(content.encode('utf-8'), path, hostname)
self.assertFalse(result, 'It should be file')
def test_listdir_inner_dir(self):
file_names = listdir('.')
self.assertGreater(len(file_names), 0)
self.assertTrue('webdav3/' in file_names)
if __name__ == '__main__':
unittest.main()

91
tests/test_connection.py Normal file
View file

@ -0,0 +1,91 @@
import unittest
from webdav3.client import get_options
from webdav3.connection import WebDAVSettings, OptionNotValid, ConnectionSettings
class ConnectionTestCase(unittest.TestCase):
def test_connection_settings_valid(self):
options = {
'webdav_hostname': 'http://localhost:8585',
'webdav_login': 'alice',
'webdav_password': 'secret1234'
}
webdav_options = get_options(option_type=WebDAVSettings, from_options=options)
settings = WebDAVSettings(webdav_options)
self.assertTrue(settings.is_valid())
self.assertTrue(settings.valid())
def test_connection_settings_no_hostname(self):
options = {
'webdav_login': 'alice',
'webdav_password': 'secret1234'
}
webdav_options = get_options(option_type=WebDAVSettings, from_options=options)
settings = WebDAVSettings(webdav_options)
self.assertRaises(OptionNotValid, settings.is_valid)
self.assertFalse(settings.valid())
def test_connection_settings_no_login(self):
options = {
'webdav_hostname': 'http://localhost:8585',
'webdav_password': 'secret1234'
}
webdav_options = get_options(option_type=WebDAVSettings, from_options=options)
settings = WebDAVSettings(webdav_options)
self.assertRaises(OptionNotValid, settings.is_valid)
self.assertFalse(settings.valid())
def test_connection_settings_no_token_and_no_login(self):
options = {
'webdav_hostname': 'http://localhost:8585'
}
webdav_options = get_options(option_type=WebDAVSettings, from_options=options)
settings = WebDAVSettings(webdav_options)
self.assertRaises(OptionNotValid, settings.is_valid)
self.assertFalse(settings.valid())
def test_connection_settings_wrong_cert_path(self):
options = {
'webdav_hostname': 'http://localhost:8585',
'webdav_login': 'alice',
'cert_path': './wrong.file',
'webdav_password': 'secret1234'
}
webdav_options = get_options(option_type=WebDAVSettings, from_options=options)
settings = WebDAVSettings(webdav_options)
self.assertRaises(OptionNotValid, settings.is_valid)
self.assertFalse(settings.valid())
def test_connection_settings_wrong_key_path(self):
options = {
'webdav_hostname': 'http://localhost:8585',
'webdav_login': 'alice',
'key_path': './wrong.file',
'webdav_password': 'secret1234'
}
webdav_options = get_options(option_type=WebDAVSettings, from_options=options)
settings = WebDAVSettings(webdav_options)
self.assertRaises(OptionNotValid, settings.is_valid)
self.assertFalse(settings.valid())
def test_connection_settings_with_key_path_an_no_cert_path(self):
options = {
'webdav_hostname': 'http://localhost:8585',
'webdav_login': 'alice',
'key_path': './publish.sh',
'webdav_password': 'secret1234'
}
webdav_options = get_options(option_type=WebDAVSettings, from_options=options)
settings = WebDAVSettings(webdav_options)
self.assertRaises(OptionNotValid, settings.is_valid)
self.assertFalse(settings.valid())
def test_connection_settings_does_nothing(self):
settings = ConnectionSettings()
settings.is_valid()
self.assertTrue(settings.valid())
if __name__ == '__main__':
unittest.main()

View file

@ -0,0 +1,23 @@
import os
import unittest
from tests.test_client_it import ClientTestCase
class MultiClientTestCase(ClientTestCase):
remote_path_file = 'директория/тестовый.txt'
remote_path_file2 = 'директория/тестовый2.txt'
remote_inner_path_file = 'директория/вложенная/тестовый.txt'
remote_path_dir = 'директория'
remote_path_dir2 = 'директория2'
remote_inner_path_dir = 'директория/вложенная'
inner_dir_name = 'вложенная'
local_base_dir = 'tests/'
local_file = 'тестовый.txt'
local_file_path = local_base_dir + 'тестовый.txt'
local_path_dir = local_base_dir + 'res/директория'
pulled_file = local_path_dir + os.sep + local_file
if __name__ == '__main__':
unittest.main()

50
tests/test_exceptions.py Normal file
View file

@ -0,0 +1,50 @@
import unittest
from webdav3.exceptions import OptionNotValid, LocalResourceNotFound, RemoteResourceNotFound, MethodNotSupported, ConnectionException, NoConnection, \
RemoteParentNotFound, NotConnection, ResponseErrorCode, NotEnoughSpace
class ExceptionsTestCase(unittest.TestCase):
def test_option_not_valid(self):
exception = OptionNotValid('Name', 'Value', 'Namespace/')
self.assertEqual("Option (Namespace/Name=Value) have invalid name or value", exception.__str__())
def test_local_resource_not_found(self):
exception = LocalResourceNotFound('Path')
self.assertEqual("Local file: Path not found", exception.__str__())
def test_remote_resource_not_found(self):
exception = RemoteResourceNotFound('Path')
self.assertEqual("Remote resource: Path not found", exception.__str__())
def test_remote_parent_not_found(self):
exception = RemoteParentNotFound('Path')
self.assertEqual("Remote parent for: Path not found", exception.__str__())
def test_method_not_supported(self):
exception = MethodNotSupported('HEAD', 'Server')
self.assertEqual("Method 'HEAD' not supported for Server", exception.__str__())
def test_connection_exception(self):
exception = ConnectionException(MethodNotSupported('HEAD', 'Server'))
self.assertEqual("Method 'HEAD' not supported for Server", exception.__str__())
def test_no_connection(self):
exception = NoConnection('Server')
self.assertEqual("No connection with Server", exception.__str__())
def test_not_connection_legacy(self):
exception = NotConnection('Server')
self.assertEqual("No connection with Server", exception.__str__())
def test_response_error_code(self):
exception = ResponseErrorCode('http://text/', 502, 'Service Unavailable')
self.assertEqual("Request to http://text/ failed with code 502 and message: Service Unavailable", exception.__str__())
def test_not_enough_space(self):
exception = NotEnoughSpace()
self.assertEqual("Not enough space on the server", exception.__str__())
if __name__ == '__main__':
unittest.main()

View file

@ -0,0 +1,26 @@
import unittest
from tests.test_client_it import ClientTestCase
from webdav3.client import Client
class MultiClientTestCase(ClientTestCase):
options2 = {
'webdav_hostname': 'https://wrong.url.ru',
'webdav_login': 'webdavclient.test2',
'webdav_password': 'Qwerty123!',
'webdav_override_methods': {
'check': 'FAKE',
'download': 'FAKE',
'upload': 'FAKE',
'clean': 'FAKE',
}
}
def setUp(self):
super(ClientTestCase, self).setUp()
self.second_client = Client(self.options2)
if __name__ == '__main__':
unittest.main()

View file

@ -0,0 +1,26 @@
import unittest
from tests.test_client_it import ClientTestCase
class TailingSlashClientTestCase(ClientTestCase):
options = {
'webdav_hostname': 'http://localhost:8585/',
'webdav_login': 'alice',
'webdav_password': 'secret1234',
'webdav_override_methods': {
'check': 'GET'
}
}
def test_list_inner(self):
self._prepare_for_downloading(True)
file_list = self.client.list(self.remote_inner_path_dir)
self.assertIsNotNone(file_list, 'List of files should not be None')
def test_hostname_no_tailing_slash(self):
self.assertEqual('5', self.client.webdav.hostname[-1])
if __name__ == '__main__':
unittest.main()

View file

@ -1 +0,0 @@
__author__ = 'designerror'

View file

@ -1,11 +0,0 @@
__author__ = 'designerror'
import allure
#from hamcrest import *
class TestRequiredOptions:
def test_without_webdav_hostname(self):
options = { 'webdav_server': "https://webdav.yandex.ru"}
allure.attach('options', options.__str__())
assert 1

View file

@ -1 +0,0 @@
__author__ = 'designerror'

View file

@ -0,0 +1 @@
test content for testing of webdav client

2
wdc
View file

@ -306,7 +306,7 @@ if __name__ == "__main__":
parser = argparse.ArgumentParser(prog='wdc', formatter_class=Formatter, epilog=epilog, usage=usage)
parser.add_argument("action", help=actions_help, choices=actions)
from webdav.client import __version__ as version
from webdav3.client import __version__ as version
version_text = "{name} {version}".format(name="%(prog)s", version=version)
parser.add_argument("-v", '--version', action='version', version=version_text)
parser.add_argument("-r", "--root", help="example: dir1/dir2")

View file

@ -10,18 +10,18 @@ from re import sub
import lxml.etree as etree
import requests
from dateutil import parser as dateutil_parser
from webdav3.connection import *
from webdav3.exceptions import *
from webdav3.urn import Urn
try:
from urllib.parse import unquote, urlsplit
from urllib.parse import unquote, urlsplit, urlparse
except ImportError:
from urllib import unquote
from urlparse import urlsplit
from urlparse import urlsplit, urlparse
__version__ = "0.2"
log = logging.getLogger(__name__)
@ -74,6 +74,7 @@ def wrap_connection_error(fn):
raise ConnectionException(re)
else:
return res
return _wrapper
@ -83,14 +84,14 @@ class Client(object):
# path to root directory of WebDAV
root = '/'
# Max size of file for uploading
large_size = 2 * 1024 * 1024 * 1024
# request timeout in seconds
timeout = 30
# controls whether to verify the server's TLS certificate or not
verify = True
# HTTP headers for different actions
http_header = {
default_http_header = {
'list': ["Accept: */*", "Depth: 1"],
'free': ["Accept: */*", "Depth: 0", "Content-Type: text/xml"],
'copy': ["Accept: */*"],
@ -103,6 +104,59 @@ class Client(object):
'set_property': ["Accept: */*", "Depth: 1", "Content-Type: application/x-www-form-urlencoded"]
}
# mapping of actions to WebDAV methods
default_requests = {
'options': 'OPTIONS',
'download': "GET",
'upload': "PUT",
'copy': "COPY",
'move': "MOVE",
'mkdir': "MKCOL",
'clean': "DELETE",
'check': "HEAD",
'list': "PROPFIND",
'free': "PROPFIND",
'info': "PROPFIND",
'publish': "PROPPATCH",
'unpublish': "PROPPATCH",
'published': "PROPPATCH",
'get_property': "PROPFIND",
'set_property': "PROPPATCH"
}
meta_xmlns = {
'https://webdav.yandex.ru': "urn:yandex:disk:meta",
}
def __init__(self, options):
"""Constructor of WebDAV client
:param options: the dictionary of connection options to WebDAV.
WebDev settings:
`webdav_hostname`: url for WebDAV server should contain protocol and ip address or domain name.
Example: `https://webdav.server.com`.
`webdav_login`: (optional) login name for WebDAV server can be empty in case using of token auth.
`webdav_password`: (optional) password for WebDAV server can be empty in case using of token auth.
`webdav_token': (optional) token for WebDAV server can be empty in case using of login/password auth.
`webdav_root`: (optional) root directory of WebDAV server. Defaults is `/`.
`webdav_cert_path`: (optional) path to certificate.
`webdav_key_path`: (optional) path to private key.
`webdav_recv_speed`: (optional) rate limit data download speed in Bytes per second.
Defaults to unlimited speed.
`webdav_send_speed`: (optional) rate limit data upload speed in Bytes per second.
Defaults to unlimited speed.
`webdav_verbose`: (optional) set verbose mode on.off. By default verbose mode is off.
"""
self.session = requests.Session()
self.http_header = Client.default_http_header.copy()
self.requests = Client.default_requests.copy()
webdav_options = get_options(option_type=WebDAVSettings, from_options=options)
self.webdav = WebDAVSettings(webdav_options)
self.requests.update(self.webdav.override_methods)
self.default_options = {}
def get_headers(self, action, headers_ext=None):
"""Returns HTTP headers of specified WebDAV actions.
@ -111,11 +165,11 @@ class Client(object):
the specified action.
:return: the dictionary of headers for specified action.
"""
if action in Client.http_header:
if action in self.http_header:
try:
headers = Client.http_header[action].copy()
headers = self.http_header[action].copy()
except AttributeError:
headers = Client.http_header[action][:]
headers = self.http_header[action][:]
else:
headers = list()
@ -123,9 +177,9 @@ class Client(object):
headers.extend(headers_ext)
if self.webdav.token:
webdav_token = "Authorization: OAuth {token}".format(token=self.webdav.token)
webdav_token = "Authorization: Bearer {token}".format(token=self.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):
"""Generates url by uri path.
@ -155,80 +209,35 @@ class Client(object):
the specified action.
:return: HTTP response of request.
"""
response = requests.request(
method=Client.requests[action],
if self.session.auth:
self.session.request(method="GET", url=self.webdav.hostname, verify=self.verify) # (Re)Authenticates against the proxy
response = self.session.request(
method=self.requests[action],
url=self.get_url(path),
auth=(self.webdav.login, self.webdav.password),
auth=(self.webdav.login, self.webdav.password) if (not self.webdav.token and not self.session.auth) else None,
headers=self.get_headers(action, headers_ext),
timeout=self.timeout,
data=data
cert=(self.webdav.cert_path, self.webdav.key_path) if (self.webdav.cert_path and self.webdav.key_path) else None,
data=data,
stream=True,
verify=self.verify
)
if response.status_code == 507:
raise NotEnoughSpace()
if response.status_code == 404:
raise RemoteResourceNotFound(path=path)
if response.status_code == 405:
raise MethodNotSupported(name=action, server=self.webdav.hostname)
if response.status_code >= 400:
raise ResponseErrorCode(url=self.get_url(path), code=response.status_code, message=response.content)
return response
# mapping of actions to WebDAV methods
requests = {
'download': "GET",
'upload': "PUT",
'copy': "COPY",
'move': "MOVE",
'mkdir': "MKCOL",
'clean': "DELETE",
'check': "HEAD",
'list': "PROPFIND",
'free': "PROPFIND",
'info': "PROPFIND",
'publish': "PROPPATCH",
'unpublish': "PROPPATCH",
'published': "PROPPATCH",
'get_property': "PROPFIND",
'set_property': "PROPPATCH"
}
meta_xmlns = {
'https://webdav.yandex.ru': "urn:yandex:disk:meta",
}
def __init__(self, options):
"""Constructor of WebDAV client
:param options: the dictionary of connection options to WebDAV can include proxy server options.
WebDev settings:
`webdav_hostname`: url for WebDAV server should contain protocol and ip address or domain name.
Example: `https://webdav.server.com`.
`webdav_login`: (optional) login name for WebDAV server can be empty in case using of token auth.
`webdav_password`: (optional) password for WebDAV server can be empty in case using of token auth.
`webdav_token': (optional) token for WebDAV server can be empty in case using of login/password auth.
`webdav_root`: (optional) root directory of WebDAV server. Defaults is `/`.
`webdav_cert_path`: (optional) path to certificate.
`webdav_key_path`: (optional) path to private key.
`webdav_recv_speed`: (optional) rate limit data download speed in Bytes per second.
Defaults to unlimited speed.
`webdav_send_speed`: (optional) rate limit data upload speed in Bytes per second.
Defaults to unlimited speed.
`webdav_verbose`: (optional) set verbose mode on.off. By default verbose mode is off.
Proxy settings (optional):
`proxy_hostname`: url to proxy server should contain protocol and ip address or domain name and if needed
port. Example: `https://proxy.server.com:8383`.
`proxy_login`: login name for proxy server.
`proxy_password`: password for proxy server.
"""
webdav_options = get_options(option_type=WebDAVSettings, from_options=options)
proxy_options = get_options(option_type=ProxySettings, from_options=options)
self.webdav = WebDAVSettings(webdav_options)
self.proxy = ProxySettings(proxy_options)
self.default_options = {}
def valid(self):
"""Validates of WebDAV and proxy settings.
"""Validates of WebDAV settings.
:return: True in case settings are valid and False otherwise.
"""
return True if self.webdav.valid() and self.proxy.valid() else False
return True if self.webdav.valid() else False
@wrap_connection_error
def list(self, remote_path=root):
@ -239,8 +248,7 @@ class Client(object):
:return: list of nested file or directory names.
"""
directory_urn = Urn(remote_path, directory=True)
if directory_urn.path() != Client.root:
if not self.check(directory_urn.path()):
if directory_urn.path() != Client.root and not self.check(directory_urn.path()):
raise RemoteResourceNotFound(directory_urn.path())
response = self.execute_request(action='list', path=directory_urn.quote())
@ -268,9 +276,14 @@ class Client(object):
:param remote_path: (optional) path to resource on WebDAV server. Defaults is root directory of WebDAV.
:return: True if resource is exist or False otherwise
"""
if self.webdav.disable_check:
return True
urn = Urn(remote_path)
try:
response = self.execute_request(action='check', path=urn.quote())
except RemoteResourceNotFound:
return False
except ResponseErrorCode:
return False
@ -291,7 +304,11 @@ class Client(object):
if not self.check(directory_urn.parent()):
raise RemoteParentNotFound(directory_urn.path())
try:
response = self.execute_request(action='mkdir', path=directory_urn.quote())
except MethodNotSupported:
# Yandex WebDAV returns 405 status code when directory already exists
return True
return response.status_code in (200, 201)
@wrap_connection_error
@ -309,7 +326,7 @@ class Client(object):
raise RemoteResourceNotFound(urn.path())
response = self.execute_request(action='download', path=urn.quote())
buff.write(response.content)
shutil.copyfileobj(response.raw, buff)
def download(self, remote_path, local_path, progress=None):
"""Downloads remote resource from WebDAV and save it in local path.
@ -448,7 +465,7 @@ class Client(object):
self.mkdir(remote_path)
for resource_name in listdir(local_path):
_remote_path = "{parent}{name}".format(parent=urn.path(), name=resource_name)
_remote_path = "{parent}{name}".format(parent=urn.path(), name=resource_name).replace('\\', '')
_local_path = os.path.join(local_path, resource_name)
self.upload(local_path=_local_path, remote_path=_remote_path, progress=progress)
@ -475,10 +492,6 @@ class Client(object):
raise RemoteParentNotFound(urn.path())
with open(local_path, "rb") as local_file:
file_size = os.path.getsize(local_path)
if file_size > self.large_size:
raise ResourceTooBig(path=local_path, size=file_size, max_size=self.large_size)
self.execute_request(action='upload', path=urn.quote(), data=local_file)
def upload_sync(self, remote_path, local_path, callback=None):
@ -506,12 +519,13 @@ class Client(object):
threading.Thread(target=target).start()
@wrap_connection_error
def copy(self, remote_path_from, remote_path_to):
def copy(self, remote_path_from, remote_path_to, depth=1):
"""Copies resource from one place to another on WebDAV server.
More information you can find by link http://webdav.org/specs/rfc4918.html#METHOD_COPY
:param remote_path_from: the path to resource which will be copied,
:param remote_path_to: the path where resource will be copied.
:param depth: folder depth to copy
"""
urn_from = Urn(remote_path_from)
if not self.check(urn_from.path()):
@ -521,8 +535,12 @@ class Client(object):
if not self.check(urn_to.parent()):
raise RemoteParentNotFound(urn_to.path())
header_destination = "Destination: {path}".format(path=self.get_full_path(urn_to))
self.execute_request(action='copy', path=urn_from.quote(), headers_ext=[header_destination])
headers = [
"Destination: {url}".format(url=self.get_url(urn_to.quote()))
]
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
def move(self, remote_path_from, remote_path_to, overwrite=False):
@ -541,7 +559,7 @@ class Client(object):
if not self.check(urn_to.parent()):
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.quote()))
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])
@ -569,13 +587,16 @@ class Client(object):
`modified`: date of resource modification.
"""
urn = Urn(remote_path)
if not self.check(urn.path()) and not self.check(Urn(remote_path, directory=True).path()):
raise RemoteResourceNotFound(remote_path)
self._check_remote_resource(remote_path, urn)
response = self.execute_request(action='info', path=urn.quote())
path = self.get_full_path(urn)
return WebDavXmlUtils.parse_info_response(content=response.content, path=path, hostname=self.webdav.hostname)
def _check_remote_resource(self, remote_path, urn):
if not self.check(urn.path()) and not self.check(Urn(remote_path, directory=True).path()):
raise RemoteResourceNotFound(remote_path)
@wrap_connection_error
def is_dir(self, remote_path):
"""Checks is the remote resource directory.
@ -586,8 +607,7 @@ class Client(object):
"""
urn = Urn(remote_path)
parent_urn = Urn(urn.parent())
if not self.check(urn.path()) and not self.check(Urn(remote_path, directory=True).path()):
raise RemoteResourceNotFound(remote_path)
self._check_remote_resource(remote_path, urn)
response = self.execute_request(action='info', path=parent_urn.quote())
path = self.get_full_path(urn)
@ -652,48 +672,39 @@ class Client(object):
def prune(src, exp):
return [sub(exp, "", item) for item in src]
updated = False
urn = Urn(remote_directory, directory=True)
if not self.is_dir(urn.path()):
raise OptionNotValid(name="remote_path", value=remote_directory)
if not os.path.isdir(local_directory):
raise OptionNotValid(name="local_path", value=local_directory)
if not os.path.exists(local_directory):
raise LocalResourceNotFound(local_directory)
self._validate_remote_directory(urn)
self._validate_local_directory(local_directory)
paths = self.list(urn.path())
expression = "{begin}{end}".format(begin="^", end=urn.path())
remote_resource_names = prune(paths, expression)
for local_resource_name in listdir(local_directory):
local_path = os.path.join(local_directory, local_resource_name)
remote_path = "{remote_directory}{resource_name}".format(remote_directory=urn.path(),
resource_name=local_resource_name)
remote_path = "{remote_directory}{resource_name}".format(remote_directory=urn.path(), resource_name=local_resource_name)
if os.path.isdir(local_path):
if not self.check(remote_path=remote_path):
self.mkdir(remote_path=remote_path)
self.push(remote_directory=remote_path, local_directory=local_path)
result = self.push(remote_directory=remote_path, local_directory=local_path)
updated = updated or result
else:
if local_resource_name in remote_resource_names:
if local_resource_name in remote_resource_names and not self.is_local_more_recent(local_path, remote_path):
continue
self.upload_file(remote_path=remote_path, local_path=local_path)
updated = True
return updated
def pull(self, remote_directory, local_directory):
def prune(src, exp):
return [sub(exp, "", item) for item in src]
updated = False
urn = Urn(remote_directory, directory=True)
if not self.is_dir(urn.path()):
raise OptionNotValid(name="remote_path", value=remote_directory)
if not os.path.exists(local_directory):
raise LocalResourceNotFound(local_directory)
self._validate_remote_directory(urn)
self._validate_local_directory(local_directory)
local_resource_names = listdir(local_directory)
@ -702,27 +713,62 @@ class Client(object):
remote_resource_names = prune(paths, expression)
for remote_resource_name in remote_resource_names:
local_path = os.path.join(local_directory, remote_resource_name)
remote_path = "{remote_directory}{resource_name}".format(remote_directory=urn.path(),
resource_name=remote_resource_name)
remote_path = "{remote_directory}{resource_name}".format(remote_directory=urn.path(), resource_name=remote_resource_name)
remote_urn = Urn(remote_path)
if self.is_dir(remote_urn.path()):
if remote_urn.path().endswith("/"):
if not os.path.exists(local_path):
updated = True
os.mkdir(local_path)
self.pull(remote_directory=remote_path, local_directory=local_path)
result = self.pull(remote_directory=remote_path, local_directory=local_path)
updated = updated or result
else:
if remote_resource_name in local_resource_names:
if remote_resource_name in local_resource_names and self.is_local_more_recent(local_path, remote_path):
continue
self.download_file(remote_path=remote_path, local_path=local_path)
updated = True
return updated
def is_local_more_recent(self, local_path, remote_path):
"""Tells if local resource is more recent that the remote on if possible
:param local_path: the path to local resource.
:param remote_path: the path to remote resource.
:return: True if local resource is more recent, False if the remote one is
None if comparison is not possible
"""
try:
remote_info = self.info(remote_path)
remote_last_mod_date = remote_info['modified']
remote_last_mod_date = dateutil_parser.parse(remote_last_mod_date)
remote_last_mod_date_unix_ts = int(remote_last_mod_date.timestamp())
local_last_mod_date_unix_ts = int(os.stat(local_path).st_mtime)
return remote_last_mod_date_unix_ts < local_last_mod_date_unix_ts
except ValueError or RuntimeWarning or KeyError:
# If there is problem when parsing dates, or cannot get
# last modified information, return None
return None
def sync(self, remote_directory, local_directory):
self.pull(remote_directory=remote_directory, local_directory=local_directory)
self.push(remote_directory=remote_directory, local_directory=local_directory)
def _validate_remote_directory(self, urn):
if not self.is_dir(urn.path()):
raise OptionNotValid(name="remote_path", value=urn.path())
@staticmethod
def _validate_local_directory(local_directory):
if not os.path.isdir(local_directory):
raise OptionNotValid(name="local_path", value=local_directory)
if not os.path.exists(local_directory):
raise LocalResourceNotFound(local_directory)
class Resource(object):
def __init__(self, client, urn):
@ -959,10 +1005,10 @@ class WebDavXmlUtils:
:param hostname: the server hostname.
:return: XML object of response for the remote resource defined by path.
"""
prefix = urlparse(hostname).path
try:
tree = etree.fromstring(content)
responses = tree.findall("{DAV:}response")
n_path = Urn.normalize_path(path)
for resp in responses:
@ -970,6 +1016,9 @@ class WebDavXmlUtils:
if Urn.compare_path(n_path, href) is True:
return resp
href_without_prefix = href[len(prefix):] if href.startswith(prefix) else href
if Urn.compare_path(n_path, href_without_prefix) is True:
return resp
raise RemoteResourceNotFound(path)
except etree.XMLSyntaxError:
raise MethodNotSupported(name="is_dir", server=hostname)

View file

@ -6,10 +6,13 @@ from webdav3.urn import Urn
class ConnectionSettings:
def is_valid(self):
"""
Method checks is settings are valid
:return: True if settings are valid otherwise False
"""
pass
def valid(self):
try:
self.is_valid()
except OptionNotValid:
@ -22,20 +25,21 @@ class WebDAVSettings(ConnectionSettings):
ns = "webdav:"
prefix = "webdav_"
keys = {'hostname', 'login', 'password', 'token', 'root', 'cert_path', 'key_path', 'recv_speed', 'send_speed',
'verbose'}
hostname = None
login = None
password = None
token = None
root = None
cert_path = None
key_path = None
recv_speed = None
send_speed = None
verbose = None
'verbose', 'disable_check', 'override_methods'}
def __init__(self, options):
self.hostname = None
self.login = None
self.password = None
self.token = None
self.root = None
self.cert_path = None
self.key_path = None
self.recv_speed = None
self.send_speed = None
self.verbose = None
self.disable_check = False
self.override_methods = {}
self.options = dict()
@ -46,9 +50,9 @@ class WebDAVSettings(ConnectionSettings):
self.root = Urn(self.root).quote() if self.root else ''
self.root = self.root.rstrip(Urn.separate)
self.hostname = self.hostname.rstrip(Urn.separate)
def is_valid(self):
if not self.hostname:
raise OptionNotValid(name="hostname", value=self.hostname, ns=self.ns)
@ -66,31 +70,4 @@ class WebDAVSettings(ConnectionSettings):
if not self.token and not self.login:
raise OptionNotValid(name="login", value=self.login, ns=self.ns)
class ProxySettings(ConnectionSettings):
ns = "proxy:"
prefix = "proxy_"
keys = {'hostname', 'login', 'password'}
hostname = None
login = None
password = None
def __init__(self, options):
self.options = dict()
for key in self.keys:
value = options.get(key, '')
self.options[key] = value
self.__dict__[key] = value
def is_valid(self):
if self.password and not self.login:
raise OptionNotValid(name="login", value=self.login, ns=self.ns)
if self.login or self.password:
if not self.hostname:
raise OptionNotValid(name="hostname", value=self.hostname, ns=self.ns)
return True

View file

@ -49,26 +49,13 @@ class RemoteParentNotFound(NotFound):
return "Remote parent for: {path} not found".format(path=self.path)
class ResourceTooBig(WebDavException):
def __init__(self, path, size, max_size):
self.path = path
self.size = size
self.max_size = max_size
def __str__(self):
return "Resource {path} is too big, it should be less then {max_size} but actually: {size}".format(
path=self.path,
max_size=self.max_size,
size=self.size)
class MethodNotSupported(WebDavException):
def __init__(self, name, server):
self.name = name
self.server = server
def __str__(self):
return "Method {name} not supported for {server}".format(name=self.name, server=self.server)
return "Method '{name}' not supported for {server}".format(name=self.name, server=self.server)
class ConnectionException(WebDavException):
@ -84,7 +71,7 @@ class NoConnection(WebDavException):
self.hostname = hostname
def __str__(self):
return "Not connection with {hostname}".format(hostname=self.hostname)
return "No connection with {hostname}".format(hostname=self.hostname)
# This exception left only for supporting original library interface.
@ -109,7 +96,7 @@ class ResponseErrorCode(WebDavException):
class NotEnoughSpace(WebDavException):
def __init__(self):
pass
self.message = "Not enough space on the server"
def __str__(self):
return "Not enough space on the server"
return self.message

View file

@ -34,13 +34,11 @@ class Urn(object):
return self._path
def filename(self):
path_split = self._path.split(Urn.separate)
name = path_split[-2] + Urn.separate if path_split[-1] == '' else path_split[-1]
return unquote(name)
def parent(self):
path_split = self._path.split(Urn.separate)
nesting_level = self.nesting_level()
parent_path_split = path_split[:nesting_level]