import os import json import time from qswiss import qts_get_suid, qts_get_hwsn from ucc.q2 import Q2Client, Q2APIClient if os.name == "posix": import fcntl import pytest import random import logging import logging.config import tempfile import hashlib import shutil import requests from urllib3.exceptions import (InsecureRequestWarning, InsecurePlatformWarning) from ucc.s3 import (S3Client, AmazonS3Client, AmazonS3GovClient, AmazonS3FIPSClient) from ucc.amazonclouddrive import AmazonCloudDriveClient from ucc.dropbox import DropboxClient from ucc.googledrive import GoogleDriveClient from ucc.onedrivegraph import OnedriveGraphClient from ucc.webdav import WebDAVDriveClient from ucc.yandexdisk import YandexDiskClient from ucc.hidrive import HiDriveClient from ucc.box import BoxClient from ucc.swift import (SwiftClient, SwiftAuthV1Credentials, KeystoneV2Client, KeystoneV3Client) from ucc.local import LocalClient, QTSClient from ucc.qnapauth import (QnapAuthCredentials, QnapAuthResult, decrypt_auth_result, encrypt_auth_result) from ucc.auth import HubicCredentials from ucc.backblaze import BackblazeCredentials from ucc.worm2obj import WormToObjectClient from ucc.obj2path import ObjToPathDriveClient from ucc.id2path import IdToPathDriveClient from ucc.backblaze import BackblazeClient from ucc.aliyun import AliyunClient, AliyunRegionClient from ucc.huawei import HuaweiClient from ucc.azure import AzureClient from ucc.googlecloudstorage import (GoogleCloudStorageClient, GoogleCloudStoragePrivateKeyCredentials, JsonKey, P12Key) from ucc.directcloud import (DirectCloudCredentials, DirectCloudClient, DirectCloudAPIClient) from ucc.sharefile import ShareFileClient from ucc.glacier import AmazonGlacierClient from ucc.glacier2object import GlacierToObjectClient from ucc.rtrr import RtrrClient from ucc.qiniu import QiNiuPublicClient, QiNiuPrivateClient from ucc.wasabi import WasabiClient from ucc.oraclecloud import OracleCloudClient from ucc.tencent import TencentClient from ucc.baidu import (BaiduClient, BaiduAPIClient, BaiduExpringDict, BaiduSDKClient) from ucc.filebase import FilebaseClient from ucc.lyve import LyveClient from ucc.ibmcloud import IbmCloudClient from ucc.qnap_object import QNAPObjectClient from ucc import NotFoundError requests.packages.urllib3.disable_warnings(InsecureRequestWarning) requests.packages.urllib3.disable_warnings(InsecurePlatformWarning) OBJECT = ["amazons3", "s3c", "s3_gov", "s3_gov_fips", "hubic", "backblaze", "aliyun", "aliyun_region", "azure", "googlecloudstorage", "swift", 'huaweicloudobs', 'qiniu', 'qiniu_private', 'wasabi', 'oraclecloud', "tencent", "filebase", "lyve", 'ibmcloud', 'qnap_object'] WEBDAV = ["qnapwebdav", "opendrive", "drivehq", "cubby", "mydrive", "safesync", "storagemadeeasy"] PATH_DRIVE = ["dropbox", "local", "qts", "yandex", "hidrive", "sharefile", 'webdav', 'rtrr', "directcloud", "baidu", "baidu_sdk"] ID_DRIVE = ["amazonclouddrive", "box", "googledrive", "qnapclouddrive", "onedrivegraph", "onedrivegraph_germany", "onedrivegraph_china", "q2"] ARCHIVE = ['glacier'] MULTIPART_UPLOAD_THRESHOLD = 5*1024*1024 MULTIPART_PART_SIZE = 5*1024*1024 NOW = time.time() LOG = logging.getLogger(__name__) def pytest_addoption(parser): parser.addoption("--provider", action="store", default="all") parser.addoption("--auth-file", action="store", default="auth.json") def pytest_generate_tests(metafunc): if 'ucc_client' in metafunc.fixturenames: option = metafunc.config.option.provider provider_list = list() provider_list.extend(option.split(',')) metafunc.parametrize('ucc_client', provider_list, indirect=True, scope="session") auth_file = metafunc.config.option.auth_file with open(auth_file, "r") as fp: auth_file_config = json.load(fp) provider_name = metafunc.config.option.provider auth_config = auth_file_config[provider_name] provider_type = auth_config['provider_type'] module_name = metafunc.module.__name__ if 's3' not in provider_type and 's3' in module_name: pytest.skip('skip module: %s ' % module_name) if 'azure' not in provider_type and 'azure' in module_name: pytest.skip('skip module: %s ' % module_name) if 'googledrive' not in provider_type and 'gshortcut' in module_name: pytest.skip('skip module: %s ' % module_name) if 'tencent' not in provider_type and 'tencent' in module_name: pytest.skip('skip module: %s ' % module_name) if 'object_lock' not in provider_name and 'object_lock' in module_name: pytest.skip('skip module: %s ' % module_name) if 'worm' not in provider_name and 'worm' in module_name: pytest.skip('skip module: %s ' % module_name) @pytest.fixture(scope='session', params=[64*1024, 128*1024, 256*1024, 512*1024, 1024*1024]) def bandwidth(request): return request.param @pytest.fixture def unicode_string(): # wiki link: https://en.wikipedia.org/wiki/Specials_(Unicode_block) # wiki link: https://en.wikipedia.org/wiki/Plane_(Unicode) length = int(random.uniform(5, 20)) unicode_range = list(range(0x0100, 0x02AF)) unicodes = [chr(random.choice(unicode_range)) for x in range(0, length)] s = u''.join(unicodes) print(s) return s @pytest.fixture(scope="session") def auth_file(request): return request.config.getoption('--auth-file') @pytest.fixture(scope='session') def auth_file_config(auth_file): with open(auth_file, "r") as fp: return json.load(fp) class FileConfigStore(object): def __init__(self, _file, key, provider): self._file = _file self._key = key self._provider = provider self._fd = None def __enter__(self): fd = open(self._file) fcntl.flock(fd, fcntl.LOCK_EX) self._fd = fd return self def __exit__(self, *args): self._fd.close() def load(self): with open(self._file) as fp: conf = json.load(fp) auth = conf[self._provider]['auth'] result = decrypt_auth_result(auth, self._key) LOG.info(f'auth: {auth}, result: {result}') return result def save(self, result): auth = encrypt_auth_result(json.dumps(result), self._key) LOG.info(f'auth: {auth}, result: {result}') with open(self._file) as fp: conf = json.load(fp) conf[self._provider]['auth'] = auth tmp_fd, temp_auth_file = tempfile.mkstemp() with os.fdopen(tmp_fd, "w") as tmp_fp: json.dump(conf, tmp_fp, indent=4) os.rename(temp_auth_file, self._file) class WindowsFileConfigStore(FileConfigStore): def __enter__(self): fd = open(self._file, "r") self._fd = fd return self def save(self, result): auth = encrypt_auth_result(json.dumps(result), self._key) LOG.info(f'auth: {auth}, result: {result}') fp = open(self._file, "r") conf = json.load(fp) conf[self._provider]['auth'] = auth fp.close() tmp_fd, temp_auth_file = tempfile.mkstemp() tmp_fp = os.fdopen(tmp_fd, "w") json.dump(conf, tmp_fp, indent=4) tmp_fp.close() from shutil import move move(temp_auth_file, self._file) class FileAuthResult(QnapAuthResult): def __init__(self, store): super(FileAuthResult, self).__init__(dict()) self._store = store def __enter__(self): super(FileAuthResult, self).__enter__() self._store.__enter__() result = self._store.load() self.update(result) return self def __exit__(self, *args): result = self.dict() self._store.save(result) self._store.__exit__() super(FileAuthResult, self).__exit__() def dict(self): result = { 'access_token': self.access_token, 'refresh_token': self.refresh_token, 'grant_at': self.grant_at, } result.update(self.auth_config) return result def get_chroot(auth_file_config, provider_name, path: str=None): if path is None: chroot = auth_file_config[provider_name].get('chroot', u'/ucc_bft_') chroot = chroot + str(NOW) else: chroot = auth_file_config[provider_name].get('chroot', '/') chroot = os.path.join(chroot, path.lstrip('/')) return chroot @pytest.fixture(scope='session') def ucc_client(request, auth_file, auth_file_config): provider_name = request.param auth_config = auth_file_config[provider_name] verify_ssl = auth_config.get('verify_ssl', False) app_id = auth_config.get('app_id', 'P5b5rWaBBqeAghWWCvnUCfca') decrypt_key = auth_config.get('decrypt_key', 'KgNQzwKztvICkJj7LK55ASYgB58kyS1y') refresh_url = auth_config.get('refresh_url', 'https://connector.myqnapcloud.com/oauth2/refresh') provider_type = auth_config['provider_type'] def get_cred(): if os.name == "posix": store = FileConfigStore(auth_file, decrypt_key, provider_name) else: store = WindowsFileConfigStore(auth_file, decrypt_key, provider_name) if provider_type == 'onedrivegraph': auth_provider_type = 'microsoftgraph' else: auth_provider_type = provider_type with FileAuthResult(store) as auth_result: return QnapAuthCredentials({ 'app_id': app_id, 'decrypt_key': decrypt_key, 'refresh_url': refresh_url, 'provider': auth_provider_type }, auth_result) if provider_type == 'amazons3': client = AmazonS3Client( access_key_id=auth_config['access_key_id'], secret_access_key=auth_config['access_key_secret'], multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, region='eu-west-1', verify_ssl=verify_ssl ) elif provider_type == 's3_gov': client = AmazonS3GovClient( access_key_id=auth_config['access_key_id'], secret_access_key=auth_config['access_key_secret'], multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, region='us-gov-east-1', verify_ssl=verify_ssl ) elif provider_type == 's3_gov_fips': client = AmazonS3FIPSClient( access_key_id=auth_config['access_key_id'], secret_access_key=auth_config['access_key_secret'], multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, region='us-gov-east-1', verify_ssl=verify_ssl ) elif provider_type == 's3c': client = S3Client( access_key_id=auth_config['access_key_id'], secret_access_key=auth_config['access_key_secret'], multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, host=auth_config['host'], is_secure=auth_config['is_secure'], signature_version=auth_config.get('signature_version', 'v2'), region=auth_config.get('region', ''), verify_ssl=verify_ssl ) elif provider_type == 'amazonclouddrive': cred = get_cred() client = AmazonCloudDriveClient(credentials=cred, verify_ssl=verify_ssl) elif provider_type == 'dropbox': cred = get_cred() client = DropboxClient( credentials=cred, verify_ssl=verify_ssl, multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, ) elif provider_type == 'googledrive': cred = get_cred() client = GoogleDriveClient( credentials=cred, verify_ssl=verify_ssl, multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, ) elif provider_type == 'directcloud': cred = DirectCloudCredentials(auth_config) api_client = DirectCloudAPIClient(credentials=cred, verify_ssl=verify_ssl) client = DirectCloudClient(api_client) elif provider_type == 'onedrivegraph': cred = get_cred() endpoint = OnedriveGraphClient.get_endpoint('global') client = OnedriveGraphClient( endpoint=endpoint, credentials=cred, verify_ssl=verify_ssl, multipart_part_size=MULTIPART_PART_SIZE, ) elif provider_type == 'onedrivegraph_germany': cred = get_cred() endpoint = OnedriveGraphClient.get_endpoint('germany') client = OnedriveGraphClient(endpoint=endpoint, credentials=cred, verify_ssl=verify_ssl) elif provider_type == 'onedrivegraph_china': cred = get_cred() endpoint = OnedriveGraphClient.get_endpoint('china') client = OnedriveGraphClient(endpoint=endpoint, credentials=cred, verify_ssl=verify_ssl) elif provider_type == 'webdav': cred = auth_config # XXX(harry) client = WebDAVDriveClient(credentials=cred) elif provider_type == 'box': cred = get_cred() client = BoxClient( credentials=cred, verify_ssl=verify_ssl, multipart_upload_threshold=20*1024*1024, ) elif provider_type == 'yandex': cred = get_cred() client = YandexDiskClient(credentials=cred, verify_ssl=verify_ssl) elif provider_type == 'hidrive': cred = get_cred() client = HiDriveClient( credentials=cred, chroot='/', verify_ssl=verify_ssl, multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, ) elif provider_type == 'local': platform = os.uname()[0] chroot = tempfile.mkdtemp() client = LocalClient(use_xattr_uuid=False, chroot=chroot) client.UUID_NAME = 'user.ucc.uuid' def fin(): shutil.rmtree(chroot, ignore_errors=True) request.addfinalizer(fin) elif provider_type == 'qts': config = { 'version': '4.2', 'qpkg_home': '/share/syncengine' } platform = os.uname()[0] client = QTSClient(use_xattr_uuid=(platform == 'Darwin'), **config) elif provider_type == "backblaze": account_id = auth_config["account_id"] application_key = auth_config["application_key"] cred = BackblazeCredentials(account_id, application_key) cred._min_part_size = MULTIPART_PART_SIZE client = BackblazeClient(credentials=cred, verify_ssl=verify_ssl, multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, ) elif provider_type == 'azure': chroot = get_chroot(auth_file_config, provider_name) client = AzureClient( account_id=auth_config["account_id"], access_key=auth_config["access_key"], multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, verify_ssl=verify_ssl, copy_archived_blob_folder=f'{chroot}/.restore' ) elif provider_type == 'aliyun': client = AliyunClient( access_key_id=auth_config['access_key_id'], secret_access_key=auth_config['access_key_secret'], multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, region='oss-cn-hangzhou' ) elif provider_type == 'aliyun_region': client = AliyunRegionClient( access_key_id=auth_config['access_key_id'], secret_access_key=auth_config['access_key_secret'], multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, bucket_name='jacksu' ) elif provider_type == 'ibmcloud': client = IbmCloudClient( access_key_id=auth_config['access_key_id'], secret_access_key=auth_config['access_key_secret'], multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, bucket_name='tzuche-test-worm' ) elif provider_type == 'huaweicloudobs': client = HuaweiClient( access_key_id=auth_config['access_key_id'], secret_access_key=auth_config['access_key_secret'], multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, region='cn-north-1' ) elif provider_type == 'googlecloudstorage': email = auth_config.get('email') p12_file = auth_config.get('p12_key_path') json_key_file = auth_config.get('json_key_path') if not email: cred = get_cred() else: if p12_file: with open(p12_file, 'rb') as key_file: key = P12Key(email, key_file.read()) elif json_key_file: with open(json_key_file, 'rb') as key_file: key = JsonKey(email, key_file.read()) else: raise ValueError('email and key_file must be provided') cred = GoogleCloudStoragePrivateKeyCredentials(key) client = GoogleCloudStorageClient( credentials=cred, project_id=auth_config['project_id'], multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, verify_ssl=verify_ssl ) elif provider_type == 'hubic': cred = get_cred() credentials = HubicCredentials(cred) def cb(container, obj, part_num, timestamp=None, total_size=None): h = hashlib.md5() h.update(obj) _hash_name = h.hexdigest() part_name = '%s/%s/%s/%08d' % (_hash_name, str(timestamp), str(total_size), part_num) _container = 'pytest_segments' return _container, part_name client = SwiftClient(credentials=credentials, multipart_use_slo=True, multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, multipart_objname_callback=cb, verify_ssl=verify_ssl) elif provider_type == 'swift': auth_info = auth_config['auth'] version = auth_info.get('version', 1) if version == 1: user_name = auth_info['user_name'] api_key = auth_info['api_key'] auth_url = auth_info['auth_url'] cred = SwiftAuthV1Credentials(user_name, api_key, auth_url) elif version == 2: keystone_client = KeystoneV2Client( auth_endpoint=auth_info['auth_url'], username=auth_info['auth']['username'], password=auth_info['auth']['password'], tenant_id=auth_info['auth'].get('tenant_id'), tenant_name=auth_info['auth'].get('tenant_name')) cred = keystone_client.create_crendetial() elif version == 3: keystone_client = KeystoneV3Client( auth_endpoint=auth_info['auth_url'], username=auth_info['auth']['username'], password=auth_info['auth']['password'], domain_name=auth_info['auth'].get('domain_name'), project_id=auth_info['auth'].get('project_id'), project_name=auth_info['auth'].get('project_name')) cred = keystone_client.create_crendetial() def cb(container, obj, part_num, timestamp=None, total_size=None): h = hashlib.md5() h.update(obj.encode()) _hash_name = h.hexdigest() part_name = '%s/%s/%s/%08d' % (_hash_name, str(timestamp), str(total_size), part_num) _container = 'pytest_segments' return _container, part_name client = SwiftClient(credentials=cred, multipart_objname_callback=cb, multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, verify_ssl=verify_ssl) elif provider_type == 'sharefile': cred = get_cred() client = ShareFileClient(credentials=cred, verify_ssl=verify_ssl, chroot=u'/Shared Folders') elif provider_type == 'glacier': client = AmazonGlacierClient( access_key_id=auth_config['access_key_id'], secret_access_key=auth_config['access_key_secret'], multipart_upload_threshold=5 * 1024 * 1024, multipart_part_size=5 * 1024 * 1024, verify_ssl=verify_ssl ) elif provider_type == 'qiniu': client = QiNiuPublicClient( access_key_id=auth_config['access_key_id'], secret_access_key=auth_config['access_key_secret'], multipart_upload_threshold=5 * 1024 * 1024 ) elif provider_type == 'qiniu_private': rs_host = auth_config['rs_host'] rsf_host = auth_config['rsf_host'] uc_host = auth_config['uc_host'] api_host = auth_config['api_host'] up_host = auth_config['up_host'] client = QiNiuPrivateClient( access_key_id=auth_config['access_key_id'], secret_access_key=auth_config['access_key_secret'], rs_host=rs_host, rsf_host=rsf_host, uc_host=uc_host, api_host=api_host, up_host=up_host, multipart_upload_threshold=5 * 1024 * 1024 ) elif provider_type == 'rtrr': cred = auth_config client = RtrrClient(credentials=cred) elif provider_type == 'wasabi': client = WasabiClient( access_key_id=auth_config['access_key_id'], secret_access_key=auth_config['access_key_secret'], multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, region='eu-west-1', verify_ssl=verify_ssl ) elif provider_type == 'oraclecloud': private_key_content = auth_config.get('private_key_content') if private_key_content: client = OracleCloudClient( access_key_id=auth_config['access_key_id'], secret_access_key=auth_config['access_key_secret'], namespace=auth_config['namespace'], region=auth_config['region'], multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, user=auth_config['user'], fingerprint=auth_config['fingerprint'], tenancy=auth_config['tenancy'], private_key_content=private_key_content, ) else: client = OracleCloudClient( access_key_id=auth_config['access_key_id'], secret_access_key=auth_config['access_key_secret'], namespace=auth_config['namespace'], region=auth_config['region'], multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE ) elif provider_type == 'tencent': client = TencentClient( app_id=auth_config['app_id'], access_key_id=auth_config['access_key_id'], secret_access_key=auth_config['access_key_secret'], multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, verify_ssl=verify_ssl ) elif provider_type == 'baidu': cred = get_cred() api_client = BaiduAPIClient(credentials=cred, verify_ssl=verify_ssl) cache = BaiduExpringDict(10000) client = BaiduClient(api_client=api_client, cache=cache) elif provider_type == 'baidu_sdk': cred = get_cred() api_client = BaiduAPIClient(credentials=cred, verify_ssl=verify_ssl) cache = BaiduExpringDict(10000) digest_str = qts_get_suid() + qts_get_hwsn() temp_dir = tempfile.gettempdir() options = { 'sdk_path': '/mnt/ext/opt/CloudConnector3/baidu_sdk/', 'device_addr': hashlib.md5(digest_str.encode()).hexdigest(), 'temp_dir': temp_dir, 'chroot': get_chroot(auth_file_config, provider_type) } client = BaiduSDKClient(api_client, cache, **options) elif provider_type == 'filebase': client = FilebaseClient( access_key_id=auth_config['access_key_id'], secret_access_key=auth_config['access_key_secret'], multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, verify_ssl=verify_ssl ) elif provider_type == 'lyve': client = LyveClient( region=auth_config.get('region', 'us-east-1'), access_key_id=auth_config['access_key_id'], secret_access_key=auth_config['access_key_secret'], multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, verify_ssl=verify_ssl) elif provider_type == 'q2': api_client = Q2APIClient(credentials=get_cred(), site=auth_config['site']) client = Q2Client( api_client, space_id=auth_config['space_id'], multipart_upload_threshold = MULTIPART_UPLOAD_THRESHOLD, multipart_part_size = MULTIPART_PART_SIZE) elif provider_type == 'qnap_object': client = QNAPObjectClient( access_key_id=auth_config['access_key_id'], secret_access_key=auth_config['access_key_secret'], multipart_upload_threshold=MULTIPART_UPLOAD_THRESHOLD, multipart_part_size=MULTIPART_PART_SIZE, region='us-east-1', verify_ssl=verify_ssl ) else: assert False, "Unknown provider %s" % request.param setattr(client, 'provider_name', provider_name) setattr(client, 'provider_type', provider_type) return client @pytest.fixture(scope="session") def path_client(request, ucc_client, auth_file_config): provider_name = ucc_client.provider_name provider_type = ucc_client.provider_type container = auth_file_config[provider_name].get('container', 'default') chroot = get_chroot(auth_file_config, provider_name) fin = None if provider_type in ID_DRIVE: node = ucc_client.create_folder(ucc_client.root_id, os.path.basename(chroot)) client = IdToPathDriveClient(drive_client=ucc_client, chroot=chroot) # work around Amazon Drive timing issue while True: try: client.get_metadata('/') break except NotFoundError: pass def id_fin(): ucc_client.delete(node.id) fin = id_fin elif provider_type in OBJECT: if provider_type == "hubic": container = 'default' client = ucc_client try: if client.is_object_worm_enabled(container): client = WormToObjectClient(client) except NotImplementedError: pass client = ObjToPathDriveClient(object_client=client, container=container, chroot=chroot) elif provider_type in ARCHIVE: obj_client = GlacierToObjectClient(ucc_client, container=container) client = ObjToPathDriveClient(object_client=obj_client, container=container, chroot=chroot) elif provider_type in PATH_DRIVE: if isinstance(ucc_client, LocalClient): chroot = tempfile.mkdtemp() fin = None elif isinstance(ucc_client, WebDAVDriveClient): recover_baseurl = ucc_client.baseurl recover_chroot = ucc_client._chroot def webdav_reset(): ucc_client.baseurl = recover_baseurl ucc_client._chroot = recover_chroot ucc_client._is_check_chroot = False ucc_client.mkdir(chroot) webdav_reset() def webdav_path_fin(): webdav_reset() ucc_client.rmdirs(chroot) fin = webdav_path_fin else: old_chroot = ucc_client._chroot ucc_client._chroot = '/' ucc_client.mkdir(chroot) ucc_client._chroot = old_chroot def path_fin(): ucc_client._chroot = '/' ucc_client.rmdirs(chroot) fin = path_fin ucc_client._chroot = chroot client = ucc_client else: assert 0, 'Unknown provider type %s' % provider_type if fin: request.addfinalizer(fin) client.provider_name = ucc_client.provider_name client.provider_type = ucc_client.provider_type return client def setup_logging(): log_dict = { 'version': 1, 'disable_existing_loggers': False, 'root': { 'level': 'DEBUG', 'handlers': ['console'] }, 'handlers': { 'console': { 'class': 'logging.StreamHandler', 'formatter': 'brief', 'level': 'DEBUG', }, }, 'formatters': { 'precise': { 'format': '[%(asctime)s][%(threadName)10.10s][%(levelname).1s]' '[%(name)s][%(filename)s:%(funcName)s:%(lineno)s] :' ' %(message)s' }, 'brief': { 'format': '[%(asctime)s][%(threadName)10.10s][%(levelname).1s]' '[%(name)s.%(funcName)s] : %(message)s' }, } } logging.config.dictConfig(log_dict) setup_logging()