Compare commits
No commits in common. "develop" and "release-0.2" have entirely different histories.
develop
...
release-0.
30 changed files with 708 additions and 1942 deletions
9
.gitignore
vendored
9
.gitignore
vendored
|
|
@ -1,11 +1,2 @@
|
|||
*.pyc
|
||||
/**/*.pyc
|
||||
/dist
|
||||
/*.egg-info
|
||||
/.eggs/
|
||||
.project
|
||||
.pydevproject
|
||||
/.settings/
|
||||
venv/
|
||||
.idea/
|
||||
build
|
||||
|
|
|
|||
32
.travis.yml
32
.travis.yml
|
|
@ -1,32 +0,0 @@
|
|||
language: python
|
||||
dist: xenial
|
||||
|
||||
addons:
|
||||
sonarcloud:
|
||||
organization: "ezhov-evgeny"
|
||||
token: f0f714f3bea6bd103e3eb82724ef3bb0d3b54d1d
|
||||
|
||||
services:
|
||||
- docker
|
||||
|
||||
python:
|
||||
- "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:
|
||||
- coverage run setup.py test
|
||||
- coverage xml
|
||||
- |
|
||||
if [[ $TRAVIS_PYTHON_VERSION == "3.8" ]]; then
|
||||
sonar-scanner
|
||||
fi
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
COPYRIGHT AND PERMISSION NOTICE
|
||||
python setup.py register -r pypitestCOPYRIGHT AND PERMISSION NOTICE
|
||||
|
||||
Copyright (c) 2016, The WDC Project, and many
|
||||
contributors, see the THANKS file.
|
||||
|
|
|
|||
378
README.md
378
README.md
|
|
@ -1,378 +0,0 @@
|
|||
webdavclient3
|
||||
=========
|
||||
[](https://travis-ci.com/ezhov-evgeny/webdav-client-python-3)
|
||||
[](https://sonarcloud.io/dashboard?id=ezhov-evgeny_webdav-client-python-3)
|
||||
[](https://sonarcloud.io/dashboard?id=ezhov-evgeny_webdav-client-python-3)
|
||||
[](https://pypi.org/project/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
|
||||
5
README.rst
Normal file
5
README.rst
Normal file
|
|
@ -0,0 +1,5 @@
|
|||
webdavclient3
|
||||
============
|
||||
|
||||
Based on https://github.com/designerror/webdav-client-python
|
||||
But uses `requests` instead of `PyCURL`
|
||||
|
|
@ -1,8 +0,0 @@
|
|||
# Clean old dists
|
||||
rm -r dist
|
||||
|
||||
# Create new packege
|
||||
python setup.py sdist
|
||||
|
||||
# Upload to pypi
|
||||
twine upload dist/*
|
||||
33
setup.py
33
setup.py
|
|
@ -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 = "3.14.1"
|
||||
version = "0.2"
|
||||
requirements = "libxml2-dev libxslt-dev python-dev"
|
||||
|
||||
|
||||
|
|
@ -42,28 +42,20 @@ 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 (>= 3.3.0)'],
|
||||
install_requires=['requests', 'lxml', 'python-dateutil'],
|
||||
requires=['python (>= 2.7.6)'],
|
||||
install_requires=['requests', 'lxml'],
|
||||
scripts=['wdc'],
|
||||
test_suite='tests',
|
||||
tests_require=['pytest'],
|
||||
tests_require=['pytest', 'pyhamcrest', 'junit-xml', 'pytest-allure-adaptor'],
|
||||
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=long_description,
|
||||
long_description_content_type='text/markdown',
|
||||
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(),
|
||||
author='Evgeny Ezhov',
|
||||
author_email='ezhov.evgeny@gmail.com',
|
||||
url='https://github.com/ezhov-evgeny/webdav-client-python-3',
|
||||
url='https://github.com/ezhov-evgeny/webdav-client-python-2',
|
||||
license='MIT License',
|
||||
keywords='webdav, client, python, module, library, packet, Yandex.Disk, Dropbox, Google Disk, Box, 4shared',
|
||||
classifiers=[
|
||||
|
|
@ -73,10 +65,13 @@ setup(
|
|||
'Operating System :: MacOS',
|
||||
'Operating System :: Microsoft',
|
||||
'Operating System :: Unix',
|
||||
'Programming Language :: Python :: 3.5',
|
||||
'Programming Language :: Python :: 3.6',
|
||||
'Programming Language :: Python :: 3.7',
|
||||
'Programming Language :: Python :: 3.8',
|
||||
'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',
|
||||
'Topic :: Internet',
|
||||
'Topic :: Software Development :: Libraries :: Python Modules',
|
||||
],
|
||||
|
|
|
|||
|
|
@ -1,11 +0,0 @@
|
|||
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
|
||||
|
|
@ -1,83 +0,0 @@
|
|||
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()
|
||||
32
tests/conftest.py
Normal file
32
tests/conftest.py
Normal file
|
|
@ -0,0 +1,32 @@
|
|||
__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()
|
||||
|
|
@ -1,136 +0,0 @@
|
|||
<?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>
|
||||
140
tests/test_client.py
Normal file
140
tests/test_client.py
Normal file
|
|
@ -0,0 +1,140 @@
|
|||
import os.path
|
||||
import shutil
|
||||
from io import BytesIO, StringIO
|
||||
from os import path
|
||||
from unittest import TestCase
|
||||
|
||||
from webdav3.client import Client
|
||||
|
||||
|
||||
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'
|
||||
|
||||
def setUp(self):
|
||||
options = {
|
||||
'webdav_hostname': 'https://webdav.yandex.ru',
|
||||
'webdav_login': 'webdavclient.test',
|
||||
'webdav_password': 'Qwerty123!'
|
||||
}
|
||||
self.client = Client(options)
|
||||
if path.exists(path=self.local_path_dir):
|
||||
shutil.rmtree(path=self.local_path_dir)
|
||||
|
||||
def test_list(self):
|
||||
file_list = self.client.list()
|
||||
self.assertIsNotNone(file_list, 'List of files should not be None')
|
||||
self.assertGreater(file_list.__len__(), 0, 'Expected that amount of files more then 0')
|
||||
|
||||
def test_free(self):
|
||||
self.assertGreater(self.client.free(), 0, 'Expected that free space on WebDAV server is more then 0 bytes')
|
||||
|
||||
def test_check(self):
|
||||
self.assertTrue(self.client.check(), 'Expected that root directory is exist')
|
||||
|
||||
def test_mkdir(self):
|
||||
if self.client.check(remote_path=self.remote_path_dir):
|
||||
self.client.clean(remote_path=self.remote_path_dir)
|
||||
self.client.mkdir(remote_path=self.remote_path_dir)
|
||||
self.assertTrue(self.client.check(remote_path=self.remote_path_dir), 'Expected the directory is created.')
|
||||
|
||||
def test_download_to(self):
|
||||
buff = BytesIO()
|
||||
self.client.download_to(buff=buff, remote_path=self.remote_path_file)
|
||||
self.assertEquals(buff.getvalue(), 'test content for testing of webdav client')
|
||||
|
||||
def test_download(self):
|
||||
self._prepare_for_downloading()
|
||||
|
||||
self.client.download(local_path=self.local_path_dir, remote_path=self.remote_path_dir)
|
||||
|
||||
self.assertTrue(path.exists(self.local_path_dir), 'Expected the directory is downloaded.')
|
||||
self.assertTrue(path.isdir(self.local_path_dir), 'Expected this is a directory.')
|
||||
self.assertTrue(path.exists(self.local_path_dir + os.path.sep + self.local_path_file),
|
||||
'Expected the file is downloaded')
|
||||
self.assertTrue(path.isfile(self.local_path_dir + os.path.sep + self.local_path_file),
|
||||
'Expected this is a 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_path_file),
|
||||
'Expected the file is downloaded')
|
||||
self.assertTrue(path.isfile(self.local_path_dir + os.path.sep + self.local_path_file),
|
||||
'Expected this is a file')
|
||||
|
||||
self.client.download_sync(local_path=self.local_path_dir + os.path.sep + self.local_path_file,
|
||||
remote_path=self.remote_path_file, callback=callback)
|
||||
self.assertTrue(path.exists(self.local_path_dir + os.path.sep + self.local_path_file),
|
||||
'Expected the file has already been downloaded')
|
||||
|
||||
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_path_file),
|
||||
'Expected the file is downloaded')
|
||||
self.assertTrue(path.isfile(self.local_path_dir + os.path.sep + self.local_path_file),
|
||||
'Expected this is a file')
|
||||
|
||||
self.client.download_async(local_path=self.local_path_dir + os.path.sep + self.local_path_file,
|
||||
remote_path=self.remote_path_file, callback=callback)
|
||||
self.assertFalse(path.exists(self.local_path_dir + os.path.sep + self.local_path_file),
|
||||
'Expected the file has not been downloaded yet')
|
||||
|
||||
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.assertTrue(self.client.check(self.remote_path_file), 'Expected the file is uploaded.')
|
||||
self.test_download_to()
|
||||
|
||||
def test_upload(self):
|
||||
self._prepare_for_uploading()
|
||||
self.client.upload(remote_path=self.remote_path_file, local_path=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_upload_file(self):
|
||||
self._prepare_for_uploading()
|
||||
self.client.upload_file(remote_path=self.remote_path_file, local_path=self.local_path_file)
|
||||
self.assertTrue(self.client.check(remote_path=self.remote_path_file), 'Expected the file is uploaded.')
|
||||
|
||||
def test_upload_sync(self):
|
||||
self._prepare_for_uploading()
|
||||
|
||||
def callback():
|
||||
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.')
|
||||
|
||||
self.client.upload(remote_path=self.remote_path_file, local_path=self.local_path_dir)
|
||||
|
||||
def test_upload_async(self):
|
||||
self._prepare_for_uploading()
|
||||
|
||||
def callback():
|
||||
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.')
|
||||
|
||||
self.client.upload(remote_path=self.remote_path_file, local_path=self.local_path_dir)
|
||||
|
||||
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_path_file)
|
||||
|
||||
def _prepare_for_uploading(self):
|
||||
if self.client.check(remote_path=self.remote_path_file):
|
||||
self.client.clean(remote_path=self.remote_path_file)
|
||||
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.mkdir(self.local_path_dir)
|
||||
if not path.exists(path=self.local_path_dir + os.sep + self.local_path_file):
|
||||
shutil.copy(src=self.local_path_file, dst=self.local_path_dir + os.sep + self.local_path_file)
|
||||
|
|
@ -1,283 +0,0 @@
|
|||
import os.path
|
||||
import shutil
|
||||
import unittest
|
||||
from io import BytesIO, StringIO
|
||||
from os import path
|
||||
from time import sleep
|
||||
|
||||
from tests.base_client_it import BaseClientTestCase
|
||||
from webdav3.exceptions import MethodNotSupported, OptionNotValid, RemoteResourceNotFound
|
||||
|
||||
|
||||
class ClientTestCase(BaseClientTestCase):
|
||||
pulled_file = BaseClientTestCase.local_path_dir + os.sep + BaseClientTestCase.local_file
|
||||
|
||||
def test_list(self):
|
||||
self._prepare_for_downloading()
|
||||
file_list = self.client.list()
|
||||
self.assertIsNotNone(file_list, 'List of files should not be None')
|
||||
self.assertGreater(file_list.__len__(), 0, 'Expected that amount of files more then 0')
|
||||
|
||||
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):
|
||||
self.assertTrue(self.client.check(), 'Expected that root directory is exist')
|
||||
|
||||
def test_mkdir(self):
|
||||
if self.client.check(remote_path=self.remote_path_dir):
|
||||
self.client.clean(remote_path=self.remote_path_dir)
|
||||
self.client.mkdir(remote_path=self.remote_path_dir)
|
||||
self.assertTrue(self.client.check(remote_path=self.remote_path_dir), 'Expected the directory is created.')
|
||||
|
||||
def test_download_from(self):
|
||||
self._prepare_for_downloading()
|
||||
buff = BytesIO()
|
||||
self.client.download_from(buff=buff, remote_path=self.remote_path_file)
|
||||
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)
|
||||
|
||||
def test_download(self):
|
||||
self._prepare_for_downloading()
|
||||
self.client.download(local_path=self.local_path_dir, remote_path=self.remote_path_dir)
|
||||
self.assertTrue(path.exists(self.local_path_dir), 'Expected the directory is downloaded.')
|
||||
self.assertTrue(path.isdir(self.local_path_dir), 'Expected this is a directory.')
|
||||
self.assertTrue(path.exists(self.local_path_dir + os.path.sep + self.local_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 test_download_sync(self):
|
||||
self._prepare_for_downloading()
|
||||
|
||||
def callback():
|
||||
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')
|
||||
|
||||
self.client.download_sync(local_path=self.local_path_dir + os.path.sep + self.local_file,
|
||||
remote_path=self.remote_path_file, callback=callback)
|
||||
self.assertTrue(path.exists(self.local_path_dir + os.path.sep + self.local_file),
|
||||
'Expected the file has already been downloaded')
|
||||
|
||||
def test_download_async(self):
|
||||
self._prepare_for_downloading()
|
||||
|
||||
def callback():
|
||||
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')
|
||||
|
||||
self.client.download_async(local_path=self.local_path_dir + os.path.sep + self.local_file,
|
||||
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()
|
||||
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.')
|
||||
|
||||
def test_upload(self):
|
||||
self._prepare_for_uploading()
|
||||
self.client.upload(remote_path=self.remote_path_file, local_path=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_upload_file(self):
|
||||
self._prepare_for_uploading()
|
||||
self.client.upload_file(remote_path=self.remote_path_file, local_path=self.local_file_path)
|
||||
self.assertTrue(self.client.check(remote_path=self.remote_path_file), 'Expected the file is uploaded.')
|
||||
|
||||
def test_upload_sync(self):
|
||||
self._prepare_for_uploading()
|
||||
|
||||
def callback():
|
||||
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.')
|
||||
|
||||
self.client.upload(remote_path=self.remote_path_file, local_path=self.local_path_dir)
|
||||
|
||||
def test_upload_async(self):
|
||||
self._prepare_for_uploading()
|
||||
|
||||
def callback():
|
||||
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.')
|
||||
|
||||
self.client.upload(remote_path=self.remote_path_file, local_path=self.local_path_dir)
|
||||
|
||||
def test_copy(self):
|
||||
self._prepare_for_downloading()
|
||||
self.client.mkdir(remote_path=self.remote_path_dir2)
|
||||
self.client.copy(remote_path_from=self.remote_path_file, remote_path_to=self.remote_path_file2)
|
||||
self.assertTrue(self.client.check(remote_path=self.remote_path_file2))
|
||||
|
||||
def test_move(self):
|
||||
self._prepare_for_downloading()
|
||||
self.client.mkdir(remote_path=self.remote_path_dir2)
|
||||
self.client.move(remote_path_from=self.remote_path_file, remote_path_to=self.remote_path_file2)
|
||||
self.assertFalse(self.client.check(remote_path=self.remote_path_file))
|
||||
self.assertTrue(self.client.check(remote_path=self.remote_path_file2))
|
||||
|
||||
def test_clean(self):
|
||||
self._prepare_for_downloading()
|
||||
self.client.clean(remote_path=self.remote_path_dir)
|
||||
self.assertFalse(self.client.check(remote_path=self.remote_path_file))
|
||||
self.assertFalse(self.client.check(remote_path=self.remote_path_dir))
|
||||
|
||||
def test_info(self):
|
||||
self._prepare_for_downloading()
|
||||
result = self.client.info(remote_path=self.remote_path_file)
|
||||
self.assertEqual(result['size'], '41')
|
||||
self.assertTrue('created' in result)
|
||||
self.assertTrue('modified' in result)
|
||||
|
||||
def test_directory_is_dir(self):
|
||||
self._prepare_for_downloading()
|
||||
self.assertTrue(self.client.is_dir(self.remote_path_dir), 'Should return True for directory')
|
||||
|
||||
def test_file_is_not_dir(self):
|
||||
self._prepare_for_downloading()
|
||||
self.assertFalse(self.client.is_dir(self.remote_path_file), 'Should return False for file')
|
||||
|
||||
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.assertEqual(result, None, 'For not found property should return value as 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.assertEqual(result, 'aValue', 'Property value should be set')
|
||||
|
||||
def test_set_property_batch(self):
|
||||
self._prepare_for_downloading()
|
||||
self.client.set_property_batch(remote_path=self.remote_path_file, option=[
|
||||
{
|
||||
'namespace': 'test',
|
||||
'name': 'aProperty',
|
||||
'value': 'aValue'
|
||||
},
|
||||
{
|
||||
'namespace': 'test',
|
||||
'name': 'aProperty2',
|
||||
'value': 'aValue2'
|
||||
}
|
||||
])
|
||||
result = self.client.get_property(remote_path=self.remote_path_file,
|
||||
option={'namespace': 'test', 'name': 'aProperty'})
|
||||
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.assertEqual(result, 'aValue2', 'Second property value should be set')
|
||||
|
||||
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 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.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.pulled_file)
|
||||
return os.path.getmtime(self.pulled_file)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
|
@ -1,75 +0,0 @@
|
|||
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()
|
||||
|
|
@ -1,166 +0,0 @@
|
|||
# coding=utf-8
|
||||
import unittest
|
||||
from unittest import TestCase
|
||||
|
||||
from lxml.etree import ElementTree, Element
|
||||
|
||||
from webdav3.client import WebDavXmlUtils as utils, listdir
|
||||
|
||||
|
||||
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.encode('utf-8'))
|
||||
self.assertEqual(result.__len__(), 2)
|
||||
|
||||
def test_create_free_space_request_content(self):
|
||||
result = utils.create_free_space_request_content()
|
||||
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.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>' \
|
||||
'<d:href>/test_dir/test.txt</d:href><d:propstat><d:status>HTTP/1.1 200 OK</d:status><d:prop>' \
|
||||
'<d:resourcetype/><d:getlastmodified>Wed, 18 Oct 2017 15:16:04 GMT</d:getlastmodified>' \
|
||||
'<d:getetag>ab0b4b7973803c03639b848682b5f38c</d:getetag><d:getcontenttype>text/plain' \
|
||||
'</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.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 = {
|
||||
'namespace': 'test',
|
||||
'name': 'aProperty'
|
||||
}
|
||||
result = utils.create_get_property_request_content(option=option, )
|
||||
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.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.encode('utf-8'), name='aProperty')
|
||||
self.assertEqual(result, 'aValue')
|
||||
|
||||
def test_create_set_one_property_request_content(self):
|
||||
option = {
|
||||
'namespace': 'test',
|
||||
'name': 'aProperty',
|
||||
'value': 'aValue'
|
||||
}
|
||||
result = utils.create_set_property_batch_request_content(options=[option])
|
||||
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.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 = [
|
||||
{
|
||||
'namespace': 'test',
|
||||
'name': 'aProperty',
|
||||
'value': 'aValue'
|
||||
},
|
||||
{
|
||||
'namespace': 'test2',
|
||||
'name': 'aProperty2',
|
||||
'value': 'aValue2'
|
||||
}
|
||||
]
|
||||
result = utils.create_set_property_batch_request_content(options=options)
|
||||
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 = [
|
||||
{
|
||||
'name': 'aProperty'
|
||||
},
|
||||
{
|
||||
'name': 'aProperty2'
|
||||
}
|
||||
]
|
||||
result = utils.create_set_property_batch_request_content(options=options)
|
||||
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.assertEqual(result, b'<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<test/>')
|
||||
|
||||
def test_parse_is_dir_response_directory(self):
|
||||
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_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' \
|
||||
' 07:40:11 GMT</d:getlastmodified><d:resourcetype/></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:getet' \
|
||||
'ag>ab0b4b7973803c03639b848682b5f38c</d:getetag><d:creationdate>2018-05-10T07:40:12Z</d:creationdat' \
|
||||
'e><d:displayname>test.txt</d:displayname><d:getlastmodified>Thu, 10 May 2018 07:40:12 GMT</d:getla' \
|
||||
'stmodified><d:getcontenttype>text/plain</d:getcontenttype><d:getcontentlength>41</d:getcontentleng' \
|
||||
'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.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()
|
||||
|
|
@ -1,91 +0,0 @@
|
|||
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()
|
||||
|
|
@ -1,23 +0,0 @@
|
|||
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()
|
||||
|
|
@ -1,50 +0,0 @@
|
|||
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()
|
||||
|
|
@ -1,26 +0,0 @@
|
|||
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()
|
||||
|
|
@ -1,26 +0,0 @@
|
|||
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()
|
||||
1
tests/webdav/test_authenticate.py
Normal file
1
tests/webdav/test_authenticate.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
__author__ = 'designerror'
|
||||
11
tests/webdav/test_connection.py
Normal file
11
tests/webdav/test_connection.py
Normal file
|
|
@ -0,0 +1,11 @@
|
|||
__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
|
||||
1
tests/webdav/test_methods.py
Normal file
1
tests/webdav/test_methods.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
__author__ = 'designerror'
|
||||
|
|
@ -1 +0,0 @@
|
|||
test content for testing of webdav client
|
||||
2
wdc
2
wdc
|
|
@ -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 webdav3.client import __version__ as version
|
||||
from webdav.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")
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -6,13 +6,10 @@ 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:
|
||||
|
|
@ -25,21 +22,20 @@ class WebDAVSettings(ConnectionSettings):
|
|||
ns = "webdav:"
|
||||
prefix = "webdav_"
|
||||
keys = {'hostname', 'login', 'password', 'token', 'root', 'cert_path', 'key_path', 'recv_speed', 'send_speed',
|
||||
'verbose', 'disable_check', 'override_methods'}
|
||||
'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
|
||||
|
||||
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()
|
||||
|
||||
|
|
@ -50,9 +46,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)
|
||||
|
||||
|
|
@ -70,4 +66,31 @@ class WebDAVSettings(ConnectionSettings):
|
|||
|
||||
if not self.token and not self.login:
|
||||
raise OptionNotValid(name="login", value=self.login, ns=self.ns)
|
||||
return True
|
||||
|
||||
|
||||
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)
|
||||
|
|
|
|||
|
|
@ -49,38 +49,34 @@ 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):
|
||||
def __init__(self, exception):
|
||||
self.exception = exception
|
||||
|
||||
def __str__(self):
|
||||
return self.exception.__str__()
|
||||
|
||||
|
||||
class NoConnection(WebDavException):
|
||||
def __init__(self, hostname):
|
||||
self.hostname = hostname
|
||||
|
||||
def __str__(self):
|
||||
return "No connection with {hostname}".format(hostname=self.hostname)
|
||||
|
||||
|
||||
# This exception left only for supporting original library interface.
|
||||
class NotConnection(WebDavException):
|
||||
def __init__(self, hostname):
|
||||
self.hostname = hostname
|
||||
|
||||
def __str__(self):
|
||||
return "No connection with {hostname}".format(hostname=self.hostname)
|
||||
return "Not connection with {hostname}".format(hostname=self.hostname)
|
||||
|
||||
|
||||
class ResponseErrorCode(WebDavException):
|
||||
|
|
@ -96,7 +92,7 @@ class ResponseErrorCode(WebDavException):
|
|||
|
||||
class NotEnoughSpace(WebDavException):
|
||||
def __init__(self):
|
||||
self.message = "Not enough space on the server"
|
||||
pass
|
||||
|
||||
def __str__(self):
|
||||
return self.message
|
||||
return "Not enough space on the server"
|
||||
|
|
|
|||
|
|
@ -1,8 +1,7 @@
|
|||
try:
|
||||
from urllib.parse import unquote, quote, urlsplit
|
||||
from urllib.parse import unquote, quote
|
||||
except ImportError:
|
||||
from urllib import unquote, quote
|
||||
from urlparse import urlsplit
|
||||
|
||||
from re import sub
|
||||
|
||||
|
|
@ -34,11 +33,13 @@ 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]
|
||||
|
|
@ -53,13 +54,3 @@ class Urn(object):
|
|||
|
||||
def is_dir(self):
|
||||
return self._path[-1] == Urn.separate
|
||||
|
||||
@staticmethod
|
||||
def normalize_path(path):
|
||||
result = sub('/{2,}', '/', path)
|
||||
return result if len(result) < 1 or result[-1] != Urn.separate else result[:-1]
|
||||
|
||||
@staticmethod
|
||||
def compare_path(path_a, href):
|
||||
unqouted_path = Urn.separate + unquote(urlsplit(href).path)
|
||||
return Urn.normalize_path(path_a) == Urn.normalize_path(unqouted_path)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue