1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
|
import logging
import os
import shutil
import sys
import unittest
from os import path
from webdav3.client import Client
root = logging.getLogger()
root.setLevel(logging.DEBUG)
log_handler = logging.StreamHandler(sys.stdout)
log_handler.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
log_handler.setFormatter(formatter)
root.addHandler(log_handler)
class BaseClientTestCase(unittest.TestCase):
remote_path_file = 'test_dir/test.txt'
remote_compressed_path_file = 'test_dir/compressed.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_compressed_file_path = local_base_dir + 'compressed.txt'
local_path_dir = local_base_dir + 'res/test_dir'
options = {
'webdav_hostname': 'http://localhost:8585',
'webdav_login': 'alice',
'webdav_password': 'secret1234',
'webdav_timeout': 10,
'webdav_override_methods': {
'check': 'GET'
}
}
# options = {
# 'webdav_hostname': 'https://demo1.nextcloud.com/remote.php/dav/files/RCw8Y9XXFnzkLJbN/',
# 'webdav_login': 'RCw8Y9XXFnzkLJbN',
# 'webdav_password': 'demo',
# '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()
|
