diff --git libcloud/common/openstack.py libcloud/common/openstack.py index b9952ef..76c3e40 100644 --- libcloud/common/openstack.py +++ libcloud/common/openstack.py @@ -40,9 +40,13 @@ __all__ = [ # @TODO: Refactor for re-use by other openstack drivers -class OpenStackAuthResponse(Response): +class OpenStackResponse(Response): + valid_response_codes = [httplib.NOT_FOUND, httplib.CONFLICT] + driver = None + def success(self): - return True + i = int(self.status) + return i >= 200 and i <= 299 or i in self.valid_response_codes def parse_body(self): if not self.body: @@ -54,7 +58,7 @@ class OpenStackAuthResponse(Response): key = 'Content-Type' else: raise LibcloudError('Missing content-type header', - driver=OpenStackAuthConnection) + driver=self.driver) content_type = self.headers[key] if content_type.find(';') != -1: @@ -66,7 +70,7 @@ class OpenStackAuthResponse(Response): except: raise MalformedResponseError('Failed to parse JSON', body=self.body, - driver=OpenStackAuthConnection) + driver=self.driver) elif content_type == 'text/plain': data = self.body else: @@ -75,14 +79,20 @@ class OpenStackAuthResponse(Response): return data -class OpenStackAuthConnection(ConnectionUserAndKey): +class OpenStackAuthResponse(OpenStackResponse): + def success(self): + return True + +class OpenStackAuthConnection(ConnectionUserAndKey): responseCls = OpenStackAuthResponse name = 'OpenStack Auth' timeout = None + action = None def __init__(self, parent_conn, auth_url, auth_version, user_id, key, tenant_name=None, timeout=None): + self.responseCls.driver = self self.parent_conn = parent_conn # enable tests to use the same mock connection classes. self.conn_classes = parent_conn.conn_classes @@ -109,18 +119,30 @@ class OpenStackAuthConnection(ConnectionUserAndKey): def authenticate(self): if self.auth_version == "1.0": + if self.action is None: + self.action = "/v1.0" + return self.authenticate_1_0() elif self.auth_version == "1.1": + if self.action is None: + self.action = "/v1.1/auth" + return self.authenticate_1_1() elif self.auth_version == "2.0" or self.auth_version == "2.0_apikey": + if self.action is None: + self.action = "/v2.0/tokens" + return self.authenticate_2_0_with_apikey() elif self.auth_version == "2.0_password": + if self.action is None: + self.action = "/v2.0/tokens" + return self.authenticate_2_0_with_password() else: raise LibcloudError('Unsupported Auth Version requested') def authenticate_1_0(self): - resp = self.request("/v1.0", + resp = self.request(self.action, headers={ 'X-Auth-User': self.user_id, 'X-Auth-Key': self.key, @@ -130,7 +152,7 @@ class OpenStackAuthConnection(ConnectionUserAndKey): if resp.status == httplib.UNAUTHORIZED: # HTTP UNAUTHORIZED (401): auth failed raise InvalidCredsError() - elif resp.status != httplib.NO_CONTENT: + elif resp.status not in [ httplib.OK, httplib.NO_CONTENT ]: raise MalformedResponseError('Malformed response', body='code: %s body:%s headers:%s' % (resp.status, resp.body, @@ -147,6 +169,7 @@ class OpenStackAuthConnection(ConnectionUserAndKey): self.urls['cloudFiles'] = \ [{'publicURL': headers.get('x-storage-url', None)}] self.auth_token = headers.get('x-auth-token', None) + self.auth_token_expires = None self.auth_user_info = None if not self.auth_token: @@ -156,7 +179,7 @@ class OpenStackAuthConnection(ConnectionUserAndKey): def authenticate_1_1(self): reqbody = json.dumps({'credentials': {'username': self.user_id, 'key': self.key}}) - resp = self.request("/v1.1/auth", + resp = self.request(self.action, data=reqbody, headers={}, method='POST') @@ -208,7 +231,7 @@ class OpenStackAuthConnection(ConnectionUserAndKey): return self.authenticate_2_0_with_body(reqbody) def authenticate_2_0_with_body(self, reqbody): - resp = self.request('/v2.0/tokens', + resp = self.request(self.action, data=reqbody, headers={'Content-Type': 'application/json'}, method='POST') @@ -294,9 +317,7 @@ class OpenStackServiceCatalog(object): return {} def _parse_auth_v1(self, service_catalog): - for service, endpoints in service_catalog.items(): - self._service_catalog[service] = {} for endpoint in endpoints: @@ -389,6 +410,7 @@ class OpenStackBaseConnection(ConnectionUserAndKey): service_name = None service_region = None _auth_version = None + connCls = OpenStackAuthConnection def __init__(self, user_id, key, secure=True, host=None, port=None, timeout=None, @@ -476,10 +498,10 @@ class OpenStackBaseConnection(ConnectionUserAndKey): raise LibcloudError('OpenStack instance must ' + 'have auth_url set') - osa = OpenStackAuthConnection(self, aurl, self._auth_version, - self.user_id, self.key, - tenant_name=self._ex_tenant_name, - timeout=self.timeout) + osa = self.connCls(self, aurl, self._auth_version, + self.user_id, self.key, + tenant_name=self._ex_tenant_name, + timeout=self.timeout) # may throw InvalidCreds, etc osa.authenticate() diff --git libcloud/storage/drivers/cloudfiles.py libcloud/storage/drivers/cloudfiles.py index 5b63b52..cfbe14a 100644 --- libcloud/storage/drivers/cloudfiles.py +++ libcloud/storage/drivers/cloudfiles.py @@ -47,6 +47,8 @@ from libcloud.storage.types import ObjectHashMismatchError from libcloud.storage.types import InvalidContainerNameError from libcloud.common.openstack import OpenStackBaseConnection from libcloud.common.openstack import OpenStackDriverMixin +from libcloud.storage.drivers.openstack import ( + OpenStackFileResponse, OpenStackFileRawResponse, OpenStackStorageDriver) from libcloud.common.rackspace import ( AUTH_URL_US, AUTH_URL_UK) @@ -54,42 +56,8 @@ from libcloud.common.rackspace import ( CDN_HOST = 'cdn.clouddrive.com' API_VERSION = 'v1.0' - -class CloudFilesResponse(Response): - valid_response_codes = [httplib.NOT_FOUND, httplib.CONFLICT] - - def success(self): - i = int(self.status) - return i >= 200 and i <= 299 or i in self.valid_response_codes - - def parse_body(self): - if not self.body: - return None - - if 'content-type' in self.headers: - key = 'content-type' - elif 'Content-Type' in self.headers: - key = 'Content-Type' - else: - raise LibcloudError('Missing content-type header') - - content_type = self.headers[key] - if content_type.find(';') != -1: - content_type = content_type.split(';')[0] - - if content_type == 'application/json': - try: - data = json.loads(self.body) - except: - raise MalformedResponseError('Failed to parse JSON', - body=self.body, - driver=CloudFilesStorageDriver) - elif content_type == 'text/plain': - data = self.body - else: - data = self.body - - return data +class CloudFilesResponse(OpenStackFileResponse): + pass class CloudFilesRawResponse(CloudFilesResponse, RawResponse): @@ -190,7 +158,7 @@ class CloudFilesSwiftConnection(CloudFilesConnection): raise LibcloudError('Could not find specified endpoint') -class CloudFilesStorageDriver(StorageDriver, OpenStackDriverMixin): +class CloudFilesStorageDriver(OpenStackStorageDriver): """ CloudFiles driver. """ @@ -198,592 +166,6 @@ class CloudFilesStorageDriver(StorageDriver, OpenStackDriverMixin): website = 'http://www.rackspace.com/' connectionCls = CloudFilesConnection - hash_type = 'md5' - supports_chunked_encoding = True - - def __init__(self, key, secret=None, secure=True, host=None, port=None, - datacenter='ord', **kwargs): - """ - @inherits: L{StorageDriver.__init__} - - @param datacenter: Datacenter ID which should be used. - @type datacenter: C{str} - """ - if hasattr(self, '_datacenter'): - datacenter = self._datacenter - - # This is here for backard compatibility - if 'ex_force_service_region' in kwargs: - datacenter = kwargs['ex_force_service_region'] - - self.datacenter = datacenter - - OpenStackDriverMixin.__init__(self, (), **kwargs) - super(CloudFilesStorageDriver, self).__init__(key=key, secret=secret, - secure=secure, host=host, - port=port, **kwargs) - - def iterate_containers(self): - response = self.connection.request('') - - if response.status == httplib.NO_CONTENT: - return [] - elif response.status == httplib.OK: - return self._to_container_list(json.loads(response.body)) - - raise LibcloudError('Unexpected status code: %s' % (response.status)) - - def get_container(self, container_name): - response = self.connection.request('/%s' % (container_name), - method='HEAD') - - if response.status == httplib.NO_CONTENT: - container = self._headers_to_container( - container_name, response.headers) - return container - elif response.status == httplib.NOT_FOUND: - raise ContainerDoesNotExistError(None, self, container_name) - - raise LibcloudError('Unexpected status code: %s' % (response.status)) - - def get_object(self, container_name, object_name): - container = self.get_container(container_name) - response = self.connection.request('/%s/%s' % (container_name, - object_name), - method='HEAD') - if response.status in [httplib.OK, httplib.NO_CONTENT]: - obj = self._headers_to_object( - object_name, container, response.headers) - return obj - elif response.status == httplib.NOT_FOUND: - raise ObjectDoesNotExistError(None, self, object_name) - - raise LibcloudError('Unexpected status code: %s' % (response.status)) - - def get_container_cdn_url(self, container): - container_name = container.name - response = self.connection.request('/%s' % (container_name), - method='HEAD', - cdn_request=True) - - if response.status == httplib.NO_CONTENT: - cdn_url = response.headers['x-cdn-uri'] - return cdn_url - elif response.status == httplib.NOT_FOUND: - raise ContainerDoesNotExistError(value='', - container_name=container_name, - driver=self) - - raise LibcloudError('Unexpected status code: %s' % (response.status)) - - def get_object_cdn_url(self, obj): - container_cdn_url = self.get_container_cdn_url(container=obj.container) - return '%s/%s' % (container_cdn_url, obj.name) - - def enable_container_cdn(self, container, ex_ttl=None): - """ - @inherits: L{StorageDriver.enable_container_cdn} - - @param ex_ttl: cache time to live - @type ex_ttl: C{int} - """ - container_name = container.name - headers = {'X-CDN-Enabled': 'True'} - - if ex_ttl: - headers['X-TTL'] = ex_ttl - - response = self.connection.request('/%s' % (container_name), - method='PUT', - headers=headers, - cdn_request=True) - - return response.status in [httplib.CREATED, httplib.ACCEPTED] - - def create_container(self, container_name): - container_name = self._clean_container_name(container_name) - response = self.connection.request( - '/%s' % (container_name), method='PUT') - - if response.status == httplib.CREATED: - # Accepted mean that container is not yet created but it will be - # eventually - extra = {'object_count': 0} - container = Container(name=container_name, - extra=extra, driver=self) - - return container - elif response.status == httplib.ACCEPTED: - error = ContainerAlreadyExistsError(None, self, container_name) - raise error - - raise LibcloudError('Unexpected status code: %s' % (response.status)) - - def delete_container(self, container): - name = self._clean_container_name(container.name) - - # Only empty container can be deleted - response = self.connection.request('/%s' % (name), method='DELETE') - - if response.status == httplib.NO_CONTENT: - return True - elif response.status == httplib.NOT_FOUND: - raise ContainerDoesNotExistError(value='', - container_name=name, driver=self) - elif response.status == httplib.CONFLICT: - # @TODO: Add "delete_all_objects" parameter? - raise ContainerIsNotEmptyError(value='', - container_name=name, driver=self) - - def download_object(self, obj, destination_path, overwrite_existing=False, - delete_on_failure=True): - container_name = obj.container.name - object_name = obj.name - response = self.connection.request('/%s/%s' % (container_name, - object_name), - method='GET', raw=True) - - return self._get_object( - obj=obj, callback=self._save_object, response=response, - callback_kwargs={'obj': obj, - 'response': response.response, - 'destination_path': destination_path, - 'overwrite_existing': overwrite_existing, - 'delete_on_failure': delete_on_failure}, - success_status_code=httplib.OK) - - def download_object_as_stream(self, obj, chunk_size=None): - container_name = obj.container.name - object_name = obj.name - response = self.connection.request('/%s/%s' % (container_name, - object_name), - method='GET', raw=True) - - return self._get_object(obj=obj, callback=read_in_chunks, - response=response, - callback_kwargs={'iterator': response.response, - 'chunk_size': chunk_size}, - success_status_code=httplib.OK) - - def upload_object(self, file_path, container, object_name, extra=None, - verify_hash=True): - """ - Upload an object. - - Note: This will override file with a same name if it already exists. - """ - upload_func = self._upload_file - upload_func_kwargs = {'file_path': file_path} - - return self._put_object(container=container, object_name=object_name, - upload_func=upload_func, - upload_func_kwargs=upload_func_kwargs, - extra=extra, file_path=file_path, - verify_hash=verify_hash) - - def upload_object_via_stream(self, iterator, - container, object_name, extra=None): - if isinstance(iterator, file): - iterator = iter(iterator) - - upload_func = self._stream_data - upload_func_kwargs = {'iterator': iterator} - - return self._put_object(container=container, object_name=object_name, - upload_func=upload_func, - upload_func_kwargs=upload_func_kwargs, - extra=extra, iterator=iterator) - - def delete_object(self, obj): - container_name = self._clean_container_name(obj.container.name) - object_name = self._clean_object_name(obj.name) - - response = self.connection.request( - '/%s/%s' % (container_name, object_name), method='DELETE') - - if response.status == httplib.NO_CONTENT: - return True - elif response.status == httplib.NOT_FOUND: - raise ObjectDoesNotExistError(value='', object_name=object_name, - driver=self) - - raise LibcloudError('Unexpected status code: %s' % (response.status)) - - def ex_purge_object_from_cdn(self, obj, email=None): - """ - Purge edge cache for the specified object. - - @param email: Email where a notification will be sent when the job - completes. (optional) - @type email: C{str} - """ - container_name = self._clean_container_name(obj.container.name) - object_name = self._clean_object_name(obj.name) - headers = {'X-Purge-Email': email} if email else {} - - response = self.connection.request('/%s/%s' % (container_name, - object_name), - method='DELETE', - headers=headers, - cdn_request=True) - - return response.status == httplib.NO_CONTENT - - def ex_get_meta_data(self): - """ - Get meta data - - @rtype: C{dict} - """ - response = self.connection.request('', method='HEAD') - - if response.status == httplib.NO_CONTENT: - container_count = response.headers.get( - 'x-account-container-count', 'unknown') - object_count = response.headers.get( - 'x-account-object-count', 'unknown') - bytes_used = response.headers.get( - 'x-account-bytes-used', 'unknown') - temp_url_key = response.headers.get( - 'x-account-meta-temp-url-key', None) - - return {'container_count': int(container_count), - 'object_count': int(object_count), - 'bytes_used': int(bytes_used), - 'temp_url_key': temp_url_key} - - raise LibcloudError('Unexpected status code: %s' % (response.status)) - - def ex_multipart_upload_object(self, file_path, container, object_name, - chunk_size=33554432, extra=None, - verify_hash=True): - object_size = os.path.getsize(file_path) - if object_size < chunk_size: - return self.upload_object(file_path, container, object_name, - extra=extra, verify_hash=verify_hash) - - iter_chunk_reader = FileChunkReader(file_path, chunk_size) - - for index, iterator in enumerate(iter_chunk_reader): - self._upload_object_part(container=container, - object_name=object_name, - part_number=index, - iterator=iterator, - verify_hash=verify_hash) - - return self._upload_object_manifest(container=container, - object_name=object_name, - extra=extra, - verify_hash=verify_hash) - - def ex_enable_static_website(self, container, index_file='index.html'): - """ - Enable serving a static website. - - @param container: Container instance - @type container: L{Container} - - @param index_file: Name of the object which becomes an index page for - every sub-directory in this container. - @type index_file: C{str} - - @rtype: C{bool} - """ - container_name = container.name - headers = {'X-Container-Meta-Web-Index': index_file} - - response = self.connection.request('/%s' % (container_name), - method='POST', - headers=headers, - cdn_request=False) - - return response.status in [httplib.CREATED, httplib.ACCEPTED] - - def ex_set_error_page(self, container, file_name='error.html'): - """ - Set a custom error page which is displayed if file is not found and - serving of a static website is enabled. - - @param container: Container instance - @type container: L{Container} - - @param file_name: Name of the object which becomes the error page. - @type file_name: C{str} - - @rtype: C{bool} - """ - container_name = container.name - headers = {'X-Container-Meta-Web-Error': file_name} - - response = self.connection.request('/%s' % (container_name), - method='POST', - headers=headers, - cdn_request=False) - - return response.status in [httplib.CREATED, httplib.ACCEPTED] - - def ex_set_account_metadata_temp_url_key(self, key): - """ - Set the metadata header X-Account-Meta-Temp-URL-Key on your Cloud - Files account. - - @param key: X-Account-Meta-Temp-URL-Key - @type key: C{str} - - @rtype: C{bool} - """ - headers = {'X-Account-Meta-Temp-URL-Key': key} - - response = self.connection.request('', - method='POST', - headers=headers, - cdn_request=False) - - return response.status in [httplib.OK, httplib.NO_CONTENT, - httplib.CREATED, httplib.ACCEPTED] - - def ex_get_object_temp_url(self, obj, method='GET', timeout=60): - """ - Create a temporary URL to allow others to retrieve or put objects - in your Cloud Files account for as long or as short a time as you - wish. This method is specifically for allowing users to retrieve - or update an object. - - @param obj: The object that you wish to make temporarily public - @type obj: L{Object} - - @param method: Which method you would like to allow, 'PUT' or 'GET' - @type method: C{str} - - @param timeout: Time (in seconds) after which you want the TempURL - to expire. - @type timeout: C{int} - - @rtype: C{bool} - """ - self.connection._populate_hosts_and_request_paths() - expires = int(time() + timeout) - path = '%s/%s/%s' % (self.connection.request_path, - obj.container.name, obj.name) - try: - key = self.ex_get_meta_data()['temp_url_key'] - assert key is not None - except Exception: - raise KeyError('You must first set the ' + - 'X-Account-Meta-Temp-URL-Key header on your ' + - 'Cloud Files account using ' + - 'ex_set_account_metadata_temp_url_key before ' + - 'you can use this method.') - hmac_body = '%s\n%s\n%s' % (method, expires, path) - sig = hmac.new(b(key), b(hmac_body), sha1).hexdigest() - params = urlencode({'temp_url_sig': sig, - 'temp_url_expires': expires}) - - temp_url = 'https://%s/%s/%s?%s' %\ - (self.connection.host + self.connection.request_path, - obj.container.name, obj.name, params) - - return temp_url - - def _upload_object_part(self, container, object_name, part_number, - iterator, verify_hash=True): - upload_func = self._stream_data - upload_func_kwargs = {'iterator': iterator} - part_name = object_name + '/%08d' % part_number - extra = {'content_type': 'application/octet-stream'} - - self._put_object(container=container, - object_name=part_name, - upload_func=upload_func, - upload_func_kwargs=upload_func_kwargs, - extra=extra, iterator=iterator, - verify_hash=verify_hash) - - def _upload_object_manifest(self, container, object_name, extra=None, - verify_hash=True): - extra = extra or {} - meta_data = extra.get('meta_data') - - container_name_cleaned = self._clean_container_name(container.name) - object_name_cleaned = self._clean_object_name(object_name) - request_path = '/%s/%s' % (container_name_cleaned, object_name_cleaned) - - headers = {'X-Auth-Token': self.connection.auth_token, - 'X-Object-Manifest': '%s/%s/' % - (container_name_cleaned, - object_name_cleaned)} - - data = '' - response = self.connection.request(request_path, - method='PUT', data=data, - headers=headers, raw=True) - - object_hash = None - - if verify_hash: - hash_function = self._get_hash_function() - hash_function.update(b(data)) - data_hash = hash_function.hexdigest() - object_hash = response.headers.get('etag') - - if object_hash != data_hash: - raise ObjectHashMismatchError( - value=('MD5 hash checksum does not match (expected=%s, ' + - 'actual=%s)') % - (data_hash, object_hash), - object_name=object_name, driver=self) - - obj = Object(name=object_name, size=0, hash=object_hash, extra=None, - meta_data=meta_data, container=container, driver=self) - - return obj - - def iterate_container_objects(self, container): - params = {} - - while True: - response = self.connection.request('/%s' % (container.name), - params=params) - - if response.status == httplib.NO_CONTENT: - # Empty or non-existent container - break - elif response.status == httplib.OK: - objects = self._to_object_list(json.loads(response.body), - container) - - if len(objects) == 0: - break - - for obj in objects: - yield obj - params['marker'] = obj.name - - else: - raise LibcloudError('Unexpected status code: %s' % - (response.status)) - - def _put_object(self, container, object_name, upload_func, - upload_func_kwargs, extra=None, file_path=None, - iterator=None, verify_hash=True): - extra = extra or {} - container_name_cleaned = self._clean_container_name(container.name) - object_name_cleaned = self._clean_object_name(object_name) - content_type = extra.get('content_type', None) - meta_data = extra.get('meta_data', None) - - headers = {} - if meta_data: - for key, value in list(meta_data.items()): - key = 'X-Object-Meta-%s' % (key) - headers[key] = value - - request_path = '/%s/%s' % (container_name_cleaned, object_name_cleaned) - result_dict = self._upload_object( - object_name=object_name, content_type=content_type, - upload_func=upload_func, upload_func_kwargs=upload_func_kwargs, - request_path=request_path, request_method='PUT', - headers=headers, file_path=file_path, iterator=iterator) - - response = result_dict['response'].response - bytes_transferred = result_dict['bytes_transferred'] - server_hash = result_dict['response'].headers.get('etag', None) - - if response.status == httplib.EXPECTATION_FAILED: - raise LibcloudError(value='Missing content-type header', - driver=self) - elif verify_hash and not server_hash: - raise LibcloudError(value='Server didn\'t return etag', - driver=self) - elif (verify_hash and result_dict['data_hash'] != server_hash): - raise ObjectHashMismatchError( - value=('MD5 hash checksum does not match (expected=%s, ' + - 'actual=%s)') % (result_dict['data_hash'], server_hash), - object_name=object_name, driver=self) - elif response.status == httplib.CREATED: - obj = Object( - name=object_name, size=bytes_transferred, hash=server_hash, - extra=None, meta_data=meta_data, container=container, - driver=self) - - return obj - else: - # @TODO: Add test case for this condition (probably 411) - raise LibcloudError('status_code=%s' % (response.status), - driver=self) - - def _clean_container_name(self, name): - """ - Clean container name. - """ - if name.startswith('/'): - name = name[1:] - name = urlquote(name) - - if name.find('/') != -1: - raise InvalidContainerNameError(value='Container name cannot' - ' contain slashes', - container_name=name, driver=self) - - if len(name) > 256: - raise InvalidContainerNameError( - value='Container name cannot be longer than 256 bytes', - container_name=name, driver=self) - - return name - - def _clean_object_name(self, name): - name = urlquote(name) - return name - - def _to_container_list(self, response): - # @TODO: Handle more then 10k containers - use "lazy list"? - for container in response: - extra = {'object_count': int(container['count']), - 'size': int(container['bytes'])} - yield Container(name=container['name'], extra=extra, driver=self) - - def _to_object_list(self, response, container): - objects = [] - - for obj in response: - name = obj['name'] - size = int(obj['bytes']) - hash = obj['hash'] - extra = {'content_type': obj['content_type'], - 'last_modified': obj['last_modified']} - objects.append(Object( - name=name, size=size, hash=hash, extra=extra, - meta_data=None, container=container, driver=self)) - - return objects - - def _headers_to_container(self, name, headers): - size = int(headers.get('x-container-bytes-used', 0)) - object_count = int(headers.get('x-container-object-count', 0)) - - extra = {'object_count': object_count, - 'size': size} - container = Container(name=name, extra=extra, driver=self) - return container - - def _headers_to_object(self, name, container, headers): - size = int(headers.pop('content-length', 0)) - last_modified = headers.pop('last-modified', None) - etag = headers.pop('etag', None) - content_type = headers.pop('content-type', None) - - meta_data = {} - for key, value in list(headers.items()): - if key.find('x-object-meta-') != -1: - key = key.replace('x-object-meta-', '') - meta_data[key] = value - - extra = {'content_type': content_type, 'last_modified': last_modified} - - obj = Object(name=name, size=size, hash=etag, extra=extra, - meta_data=meta_data, container=container, driver=self) - return obj def _ex_connection_class_kwargs(self): kwargs = {'ex_force_service_region': self.datacenter} @@ -834,65 +216,3 @@ class CloudFilesUKStorageDriver(CloudFilesStorageDriver): type = Provider.CLOUDFILES_UK name = 'CloudFiles (UK)' _datacenter = 'lon' - - -class FileChunkReader(object): - def __init__(self, file_path, chunk_size): - self.file_path = file_path - self.total = os.path.getsize(file_path) - self.chunk_size = chunk_size - self.bytes_read = 0 - self.stop_iteration = False - - def __iter__(self): - return self - - def next(self): - if self.stop_iteration: - raise StopIteration - - start_block = self.bytes_read - end_block = start_block + self.chunk_size - if end_block >= self.total: - end_block = self.total - self.stop_iteration = True - self.bytes_read += end_block - start_block - return ChunkStreamReader(file_path=self.file_path, - start_block=start_block, - end_block=end_block, - chunk_size=8192) - - def __next__(self): - return self.next() - - -class ChunkStreamReader(object): - def __init__(self, file_path, start_block, end_block, chunk_size): - self.fd = open(file_path, 'rb') - self.fd.seek(start_block) - self.start_block = start_block - self.end_block = end_block - self.chunk_size = chunk_size - self.bytes_read = 0 - self.stop_iteration = False - - def __iter__(self): - return self - - def next(self): - if self.stop_iteration: - self.fd.close() - raise StopIteration - - block_size = self.chunk_size - if self.bytes_read + block_size > \ - self.end_block - self.start_block: - block_size = self.end_block - self.start_block - self.bytes_read - self.stop_iteration = True - - block = self.fd.read(block_size) - self.bytes_read += block_size - return block - - def __next__(self): - return self.next() diff --git libcloud/storage/drivers/ktucloud.py libcloud/storage/drivers/ktucloud.py new file mode 100644 index 0000000..c43a473 --- /dev/null +++ libcloud/storage/drivers/ktucloud.py @@ -0,0 +1,154 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from hashlib import sha1 +import hmac +import os +from time import time + +from libcloud.utils.py3 import httplib +from libcloud.utils.py3 import urlencode + +try: + import simplejson as json +except ImportError: + import json + +from libcloud.utils.py3 import PY3 +from libcloud.utils.py3 import b +from libcloud.utils.py3 import urlquote + +if PY3: + from io import FileIO as file + +from libcloud.utils.files import read_in_chunks +from libcloud.common.types import MalformedResponseError, LibcloudError +from libcloud.common.base import Response, RawResponse + +from libcloud.storage.providers import Provider +from libcloud.storage.base import Object, Container, StorageDriver +from libcloud.storage.types import ContainerAlreadyExistsError +from libcloud.storage.types import ContainerDoesNotExistError +from libcloud.storage.types import ContainerIsNotEmptyError +from libcloud.storage.types import ObjectDoesNotExistError +from libcloud.storage.types import ObjectHashMismatchError +from libcloud.storage.types import InvalidContainerNameError +from libcloud.common.openstack import OpenStackBaseConnection +from libcloud.common.openstack import OpenStackDriverMixin +from libcloud.common.openstack import OpenStackAuthConnection +from libcloud.common.openstack import OpenStackServiceCatalog +from libcloud.common.openstack import OpenStackAuthResponse +from libcloud.storage.drivers.openstack import ( + OpenStackFileResponse, + OpenStackFileRawResponse, OpenStackStorageDriver) + +API_VERSION = 'v1.0' +AUTH_API_VERSION_KTUCLOUD = '1.0' +AUTH_URL_KTUCLOUD = 'https://api.ucloudbiz.olleh.com/storage/v1/auth' + + +class KTUCloudStorageAuthResponse(OpenStackAuthResponse): + def parse_body(self): + if not self.body: + return None + + content_type = 'application/json' + try: + data = json.loads(self.body) + except: + raise MalformedResponseError('Failed to parse JSON', + body=self.body, + driver=KTUCloudStorageDriver) + return data + + +class KTUCloudStorageResponse(OpenStackFileResponse): + pass + + +class KTUCloudStorageRawResponse(KTUCloudStorageResponse, RawResponse): + pass + + +class KTUCloudStorageAuthConnection(OpenStackAuthConnection): + responseCls = KTUCloudStorageAuthResponse + action = "/storage/v1/auth" + + +class KTUCloudStorageConnection(OpenStackBaseConnection): + """ + Base connection class for the Cloudfiles driver. + """ + + responseCls = KTUCloudStorageResponse + rawResponseCls = KTUCloudStorageRawResponse + connCls = KTUCloudStorageAuthConnection + + def __init__(self, user_id, key, secure=True, auth_url=AUTH_URL_KTUCLOUD, + **kwargs): + super(KTUCloudStorageConnection, self).__init__(user_id, key, + secure=secure, **kwargs) + self.auth_url = auth_url + self.api_version = API_VERSION + self._auth_version = AUTH_API_VERSION_KTUCLOUD + self.accept_format = 'application/json' + self.cdn_request = False + + def request(self, action, params=None, data='', headers=None, method='GET', + raw=False, cdn_request=False): + if not headers: + headers = {} + if not params: + params = {} + + self.cdn_request = cdn_request + params['format'] = 'json' + + if method in ['POST', 'PUT'] and 'Content-Type' not in headers: + headers.update({'Content-Type': 'application/json; charset=UTF-8'}) + + return super(KTUCloudStorageConnection, self).request( + action=action, + params=params, data=data, + method=method, headers=headers, + raw=raw) + + def get_endpoint(self, *args, **kwargs): + eps = self.service_catalog.get_endpoints(name='cloudFiles') + endpoint = eps[0] + + if 'publicURL' in endpoint: + return endpoint['publicURL'] + else: + raise LibcloudError('Could not find specified endpoint') + + +class KTUCloudStorageDriver(OpenStackStorageDriver): + """ + KTCloudStorage driver. + """ + name = 'KTUCloudStorage' + website = 'http://cs.ucloud.com/' + + connectionCls = KTUCloudStorageConnection + + def _ex_connection_class_kwargs(self): + kwargs = {'ex_force_service_region': self.datacenter} + + kwargs['auth_url'] = AUTH_URL_KTUCLOUD + + kwargs.update(self.openstack_connection_kwargs()) + return kwargs + diff --git libcloud/storage/drivers/openstack.py libcloud/storage/drivers/openstack.py new file mode 100644 index 0000000..995c4b8 --- /dev/null +++ libcloud/storage/drivers/openstack.py @@ -0,0 +1,717 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from hashlib import sha1 +import hmac +import os +from time import time + +from libcloud.utils.py3 import httplib +from libcloud.utils.py3 import urlencode + +try: + import simplejson as json +except ImportError: + import json + +from libcloud.utils.py3 import PY3 +from libcloud.utils.py3 import b +from libcloud.utils.py3 import urlquote + +if PY3: + from io import FileIO as file + +from libcloud.utils.files import read_in_chunks +from libcloud.common.types import MalformedResponseError, LibcloudError +from libcloud.common.base import Response, RawResponse + +from libcloud.storage.providers import Provider +from libcloud.storage.base import Object, Container, StorageDriver +from libcloud.storage.types import ContainerAlreadyExistsError +from libcloud.storage.types import ContainerDoesNotExistError +from libcloud.storage.types import ContainerIsNotEmptyError +from libcloud.storage.types import ObjectDoesNotExistError +from libcloud.storage.types import ObjectHashMismatchError +from libcloud.storage.types import InvalidContainerNameError +from libcloud.common.openstack import OpenStackBaseConnection +from libcloud.common.openstack import OpenStackDriverMixin +from libcloud.common.openstack import OpenStackResponse + +class OpenStackFileResponse(OpenStackResponse): + valid_response_codes = [httplib.NOT_FOUND, httplib.CONFLICT] + + +class OpenStackFileRawResponse(OpenStackFileResponse, RawResponse): + pass + + +class OpenStackStorageDriver(StorageDriver, OpenStackDriverMixin): + """ + OpenStack Storage driver. + """ + + connectionCls = None + hash_type = 'md5' + supports_chunked_encoding = True + + def __init__(self, key, secret=None, secure=True, host=None, port=None, + datacenter='ord', **kwargs): + """ + @inherits: L{StorageDriver.__init__} + + @param datacenter: Datacenter ID which should be used. + @type datacenter: C{str} + """ + if hasattr(self, '_datacenter'): + datacenter = self._datacenter + + # This is here for backard compatibility + if 'ex_force_service_region' in kwargs: + datacenter = kwargs['ex_force_service_region'] + + self.datacenter = datacenter + + OpenStackDriverMixin.__init__(self, (), **kwargs) + super(OpenStackStorageDriver, self).__init__(key=key, secret=secret, + secure=secure, host=host, + port=port, **kwargs) + + def iterate_containers(self): + response = self.connection.request('') + + if response.status == httplib.NO_CONTENT: + return [] + elif response.status == httplib.OK: + return self._to_container_list(json.loads(response.body)) + + raise LibcloudError('Unexpected status code: %s' % (response.status)) + + def get_container(self, container_name): + response = self.connection.request('/%s' % (container_name), + method='HEAD') + + if response.status == httplib.NO_CONTENT: + container = self._headers_to_container( + container_name, response.headers) + return container + elif response.status == httplib.NOT_FOUND: + raise ContainerDoesNotExistError(None, self, container_name) + + raise LibcloudError('Unexpected status code: %s' % (response.status)) + + def get_object(self, container_name, object_name): + container = self.get_container(container_name) + response = self.connection.request('/%s/%s' % (container_name, + object_name), + method='HEAD') + if response.status in [httplib.OK, httplib.NO_CONTENT]: + obj = self._headers_to_object( + object_name, container, response.headers) + return obj + elif response.status == httplib.NOT_FOUND: + raise ObjectDoesNotExistError(None, self, object_name) + + raise LibcloudError('Unexpected status code: %s' % (response.status)) + + def get_container_cdn_url(self, container): + container_name = container.name + response = self.connection.request('/%s' % (container_name), + method='HEAD', + cdn_request=True) + + if response.status == httplib.NO_CONTENT: + cdn_url = response.headers['x-cdn-uri'] + return cdn_url + elif response.status == httplib.NOT_FOUND: + raise ContainerDoesNotExistError(value='', + container_name=container_name, + driver=self) + + raise LibcloudError('Unexpected status code: %s' % (response.status)) + + def get_object_cdn_url(self, obj): + container_cdn_url = self.get_container_cdn_url(container=obj.container) + return '%s/%s' % (container_cdn_url, obj.name) + + def enable_container_cdn(self, container, ex_ttl=None): + """ + @inherits: L{StorageDriver.enable_container_cdn} + + @param ex_ttl: cache time to live + @type ex_ttl: C{int} + """ + container_name = container.name + headers = {'X-CDN-Enabled': 'True'} + + if ex_ttl: + headers['X-TTL'] = ex_ttl + + response = self.connection.request('/%s' % (container_name), + method='PUT', + headers=headers, + cdn_request=True) + + return response.status in [httplib.CREATED, httplib.ACCEPTED] + + def create_container(self, container_name): + container_name = self._clean_container_name(container_name) + response = self.connection.request( + '/%s' % (container_name), method='PUT') + + if response.status == httplib.CREATED: + # Accepted mean that container is not yet created but it will be + # eventually + extra = {'object_count': 0} + container = Container(name=container_name, + extra=extra, driver=self) + + return container + elif response.status == httplib.ACCEPTED: + error = ContainerAlreadyExistsError(None, self, container_name) + raise error + + raise LibcloudError('Unexpected status code: %s' % (response.status)) + + def delete_container(self, container): + name = self._clean_container_name(container.name) + + # Only empty container can be deleted + response = self.connection.request('/%s' % (name), method='DELETE') + + if response.status == httplib.NO_CONTENT: + return True + elif response.status == httplib.NOT_FOUND: + raise ContainerDoesNotExistError(value='', + container_name=name, driver=self) + elif response.status == httplib.CONFLICT: + # @TODO: Add "delete_all_objects" parameter? + raise ContainerIsNotEmptyError(value='', + container_name=name, driver=self) + + def download_object(self, obj, destination_path, overwrite_existing=False, + delete_on_failure=True): + container_name = obj.container.name + object_name = obj.name + response = self.connection.request('/%s/%s' % (container_name, + object_name), + method='GET', raw=True) + + return self._get_object( + obj=obj, callback=self._save_object, response=response, + callback_kwargs={'obj': obj, + 'response': response.response, + 'destination_path': destination_path, + 'overwrite_existing': overwrite_existing, + 'delete_on_failure': delete_on_failure}, + success_status_code=httplib.OK) + + def download_object_as_stream(self, obj, chunk_size=None): + container_name = obj.container.name + object_name = obj.name + response = self.connection.request('/%s/%s' % (container_name, + object_name), + method='GET', raw=True) + + return self._get_object(obj=obj, callback=read_in_chunks, + response=response, + callback_kwargs={'iterator': response.response, + 'chunk_size': chunk_size}, + success_status_code=httplib.OK) + + def upload_object(self, file_path, container, object_name, extra=None, + verify_hash=True): + """ + Upload an object. + + Note: This will override file with a same name if it already exists. + """ + upload_func = self._upload_file + upload_func_kwargs = {'file_path': file_path} + + return self._put_object(container=container, object_name=object_name, + upload_func=upload_func, + upload_func_kwargs=upload_func_kwargs, + extra=extra, file_path=file_path, + verify_hash=verify_hash) + + def upload_object_via_stream(self, iterator, + container, object_name, extra=None): + if isinstance(iterator, file): + iterator = iter(iterator) + + upload_func = self._stream_data + upload_func_kwargs = {'iterator': iterator} + + return self._put_object(container=container, object_name=object_name, + upload_func=upload_func, + upload_func_kwargs=upload_func_kwargs, + extra=extra, iterator=iterator) + + def delete_object(self, obj): + container_name = self._clean_container_name(obj.container.name) + object_name = self._clean_object_name(obj.name) + + response = self.connection.request( + '/%s/%s' % (container_name, object_name), method='DELETE') + + if response.status == httplib.NO_CONTENT: + return True + elif response.status == httplib.NOT_FOUND: + raise ObjectDoesNotExistError(value='', object_name=object_name, + driver=self) + + raise LibcloudError('Unexpected status code: %s' % (response.status)) + + def ex_purge_object_from_cdn(self, obj, email=None): + """ + Purge edge cache for the specified object. + + @param email: Email where a notification will be sent when the job + completes. (optional) + @type email: C{str} + """ + container_name = self._clean_container_name(obj.container.name) + object_name = self._clean_object_name(obj.name) + headers = {'X-Purge-Email': email} if email else {} + + response = self.connection.request('/%s/%s' % (container_name, + object_name), + method='DELETE', + headers=headers, + cdn_request=True) + + return response.status == httplib.NO_CONTENT + + def ex_get_meta_data(self): + """ + Get meta data + + @rtype: C{dict} + """ + response = self.connection.request('', method='HEAD') + + if response.status == httplib.NO_CONTENT: + container_count = response.headers.get( + 'x-account-container-count', 'unknown') + object_count = response.headers.get( + 'x-account-object-count', 'unknown') + bytes_used = response.headers.get( + 'x-account-bytes-used', 'unknown') + temp_url_key = response.headers.get( + 'x-account-meta-temp-url-key', None) + + return {'container_count': int(container_count), + 'object_count': int(object_count), + 'bytes_used': int(bytes_used), + 'temp_url_key': temp_url_key} + + raise LibcloudError('Unexpected status code: %s' % (response.status)) + + def ex_multipart_upload_object(self, file_path, container, object_name, + chunk_size=33554432, extra=None, + verify_hash=True): + object_size = os.path.getsize(file_path) + if object_size < chunk_size: + return self.upload_object(file_path, container, object_name, + extra=extra, verify_hash=verify_hash) + + iter_chunk_reader = FileChunkReader(file_path, chunk_size) + + for index, iterator in enumerate(iter_chunk_reader): + self._upload_object_part(container=container, + object_name=object_name, + part_number=index, + iterator=iterator, + verify_hash=verify_hash) + + return self._upload_object_manifest(container=container, + object_name=object_name, + extra=extra, + verify_hash=verify_hash) + + def ex_enable_static_website(self, container, index_file='index.html'): + """ + Enable serving a static website. + + @param container: Container instance + @type container: L{Container} + + @param index_file: Name of the object which becomes an index page for + every sub-directory in this container. + @type index_file: C{str} + + @rtype: C{bool} + """ + container_name = container.name + headers = {'X-Container-Meta-Web-Index': index_file} + + response = self.connection.request('/%s' % (container_name), + method='POST', + headers=headers, + cdn_request=False) + + return response.status in [httplib.CREATED, httplib.ACCEPTED] + + def ex_set_error_page(self, container, file_name='error.html'): + """ + Set a custom error page which is displayed if file is not found and + serving of a static website is enabled. + + @param container: Container instance + @type container: L{Container} + + @param file_name: Name of the object which becomes the error page. + @type file_name: C{str} + + @rtype: C{bool} + """ + container_name = container.name + headers = {'X-Container-Meta-Web-Error': file_name} + + response = self.connection.request('/%s' % (container_name), + method='POST', + headers=headers, + cdn_request=False) + + return response.status in [httplib.CREATED, httplib.ACCEPTED] + + def ex_set_account_metadata_temp_url_key(self, key): + """ + Set the metadata header X-Account-Meta-Temp-URL-Key on your Cloud + Files account. + + @param key: X-Account-Meta-Temp-URL-Key + @type key: C{str} + + @rtype: C{bool} + """ + headers = {'X-Account-Meta-Temp-URL-Key': key} + + response = self.connection.request('', + method='POST', + headers=headers, + cdn_request=False) + + return response.status in [httplib.OK, httplib.NO_CONTENT, + httplib.CREATED, httplib.ACCEPTED] + + def ex_get_object_temp_url(self, obj, method='GET', timeout=60): + """ + Create a temporary URL to allow others to retrieve or put objects + in your Cloud Files account for as long or as short a time as you + wish. This method is specifically for allowing users to retrieve + or update an object. + + @param obj: The object that you wish to make temporarily public + @type obj: L{Object} + + @param method: Which method you would like to allow, 'PUT' or 'GET' + @type method: C{str} + + @param timeout: Time (in seconds) after which you want the TempURL + to expire. + @type timeout: C{int} + + @rtype: C{bool} + """ + self.connection._populate_hosts_and_request_paths() + expires = int(time() + timeout) + path = '%s/%s/%s' % (self.connection.request_path, + obj.container.name, obj.name) + try: + key = self.ex_get_meta_data()['temp_url_key'] + assert key is not None + except Exception: + raise KeyError('You must first set the ' + + 'X-Account-Meta-Temp-URL-Key header on your ' + + 'Cloud Files account using ' + + 'ex_set_account_metadata_temp_url_key before ' + + 'you can use this method.') + hmac_body = '%s\n%s\n%s' % (method, expires, path) + sig = hmac.new(b(key), b(hmac_body), sha1).hexdigest() + params = urlencode({'temp_url_sig': sig, + 'temp_url_expires': expires}) + + temp_url = 'https://%s/%s/%s?%s' %\ + (self.connection.host + self.connection.request_path, + obj.container.name, obj.name, params) + + return temp_url + + def _upload_object_part(self, container, object_name, part_number, + iterator, verify_hash=True): + upload_func = self._stream_data + upload_func_kwargs = {'iterator': iterator} + part_name = object_name + '/%08d' % part_number + extra = {'content_type': 'application/octet-stream'} + + self._put_object(container=container, + object_name=part_name, + upload_func=upload_func, + upload_func_kwargs=upload_func_kwargs, + extra=extra, iterator=iterator, + verify_hash=verify_hash) + + def _upload_object_manifest(self, container, object_name, extra=None, + verify_hash=True): + extra = extra or {} + meta_data = extra.get('meta_data') + + container_name_cleaned = self._clean_container_name(container.name) + object_name_cleaned = self._clean_object_name(object_name) + request_path = '/%s/%s' % (container_name_cleaned, object_name_cleaned) + + headers = {'X-Auth-Token': self.connection.auth_token, + 'X-Object-Manifest': '%s/%s/' % + (container_name_cleaned, + object_name_cleaned)} + + data = '' + response = self.connection.request(request_path, + method='PUT', data=data, + headers=headers, raw=True) + + object_hash = None + + if verify_hash: + hash_function = self._get_hash_function() + hash_function.update(b(data)) + data_hash = hash_function.hexdigest() + object_hash = response.headers.get('etag') + + if object_hash != data_hash: + raise ObjectHashMismatchError( + value=('MD5 hash checksum does not match (expected=%s, ' + + 'actual=%s)') % + (data_hash, object_hash), + object_name=object_name, driver=self) + + obj = Object(name=object_name, size=0, hash=object_hash, extra=None, + meta_data=meta_data, container=container, driver=self) + + return obj + + def iterate_container_objects(self, container): + params = {} + + while True: + response = self.connection.request('/%s' % (container.name), + params=params) + + if response.status == httplib.NO_CONTENT: + # Empty or non-existent container + break + elif response.status == httplib.OK: + objects = self._to_object_list(json.loads(response.body), + container) + + if len(objects) == 0: + break + + for obj in objects: + yield obj + params['marker'] = obj.name + + else: + raise LibcloudError('Unexpected status code: %s' % + (response.status)) + + def _put_object(self, container, object_name, upload_func, + upload_func_kwargs, extra=None, file_path=None, + iterator=None, verify_hash=True): + extra = extra or {} + container_name_cleaned = self._clean_container_name(container.name) + object_name_cleaned = self._clean_object_name(object_name) + content_type = extra.get('content_type', None) + meta_data = extra.get('meta_data', None) + + headers = {} + if meta_data: + for key, value in list(meta_data.items()): + key = 'X-Object-Meta-%s' % (key) + headers[key] = value + + request_path = '/%s/%s' % (container_name_cleaned, object_name_cleaned) + result_dict = self._upload_object( + object_name=object_name, content_type=content_type, + upload_func=upload_func, upload_func_kwargs=upload_func_kwargs, + request_path=request_path, request_method='PUT', + headers=headers, file_path=file_path, iterator=iterator) + + response = result_dict['response'].response + bytes_transferred = result_dict['bytes_transferred'] + server_hash = result_dict['response'].headers.get('etag', None) + + if response.status == httplib.EXPECTATION_FAILED: + raise LibcloudError(value='Missing content-type header', + driver=self) + elif verify_hash and not server_hash: + raise LibcloudError(value='Server didn\'t return etag', + driver=self) + elif (verify_hash and result_dict['data_hash'] != server_hash): + raise ObjectHashMismatchError( + value=('MD5 hash checksum does not match (expected=%s, ' + + 'actual=%s)') % (result_dict['data_hash'], server_hash), + object_name=object_name, driver=self) + elif response.status == httplib.CREATED: + obj = Object( + name=object_name, size=bytes_transferred, hash=server_hash, + extra=None, meta_data=meta_data, container=container, + driver=self) + + return obj + else: + # @TODO: Add test case for this condition (probably 411) + raise LibcloudError('status_code=%s' % (response.status), + driver=self) + + def _clean_container_name(self, name): + """ + Clean container name. + """ + if name.startswith('/'): + name = name[1:] + name = urlquote(name) + + if name.find('/') != -1: + raise InvalidContainerNameError(value='Container name cannot' + ' contain slashes', + container_name=name, driver=self) + + if len(name) > 256: + raise InvalidContainerNameError( + value='Container name cannot be longer than 256 bytes', + container_name=name, driver=self) + + return name + + def _clean_object_name(self, name): + name = urlquote(name) + return name + + def _to_container_list(self, response): + # @TODO: Handle more then 10k containers - use "lazy list"? + for container in response: + extra = {'object_count': int(container['count']), + 'size': int(container['bytes'])} + yield Container(name=container['name'], extra=extra, driver=self) + + def _to_object_list(self, response, container): + objects = [] + + for obj in response: + name = obj['name'] + size = int(obj['bytes']) + hash = obj['hash'] + extra = {'content_type': obj['content_type'], + 'last_modified': obj['last_modified']} + objects.append(Object( + name=name, size=size, hash=hash, extra=extra, + meta_data=None, container=container, driver=self)) + + return objects + + def _headers_to_container(self, name, headers): + size = int(headers.get('x-container-bytes-used', 0)) + object_count = int(headers.get('x-container-object-count', 0)) + + extra = {'object_count': object_count, + 'size': size} + container = Container(name=name, extra=extra, driver=self) + return container + + def _headers_to_object(self, name, container, headers): + size = int(headers.pop('content-length', 0)) + last_modified = headers.pop('last-modified', None) + etag = headers.pop('etag', None) + content_type = headers.pop('content-type', None) + + meta_data = {} + for key, value in list(headers.items()): + if key.find('x-object-meta-') != -1: + key = key.replace('x-object-meta-', '') + meta_data[key] = value + + extra = {'content_type': content_type, 'last_modified': last_modified} + + obj = Object(name=name, size=size, hash=etag, extra=extra, + meta_data=meta_data, container=container, driver=self) + return obj + + def _ex_connection_class_kwargs(self): + return super(OpenStackStorageDriver, self).\ + _ex_connection_class_kwargs() + + +class FileChunkReader(object): + def __init__(self, file_path, chunk_size): + self.file_path = file_path + self.total = os.path.getsize(file_path) + self.chunk_size = chunk_size + self.bytes_read = 0 + self.stop_iteration = False + + def __iter__(self): + return self + + def next(self): + if self.stop_iteration: + raise StopIteration + + start_block = self.bytes_read + end_block = start_block + self.chunk_size + if end_block >= self.total: + end_block = self.total + self.stop_iteration = True + self.bytes_read += end_block - start_block + return ChunkStreamReader(file_path=self.file_path, + start_block=start_block, + end_block=end_block, + chunk_size=8192) + + def __next__(self): + return self.next() + + +class ChunkStreamReader(object): + def __init__(self, file_path, start_block, end_block, chunk_size): + self.fd = open(file_path, 'rb') + self.fd.seek(start_block) + self.start_block = start_block + self.end_block = end_block + self.chunk_size = chunk_size + self.bytes_read = 0 + self.stop_iteration = False + + def __iter__(self): + return self + + def next(self): + if self.stop_iteration: + self.fd.close() + raise StopIteration + + block_size = self.chunk_size + if self.bytes_read + block_size > \ + self.end_block - self.start_block: + block_size = self.end_block - self.start_block - self.bytes_read + self.stop_iteration = True + + block = self.fd.read(block_size) + self.bytes_read += block_size + return block + + def __next__(self): + return self.next() diff --git libcloud/storage/providers.py libcloud/storage/providers.py index 94ee89a..bb4229d 100644 --- libcloud/storage/providers.py +++ libcloud/storage/providers.py @@ -45,6 +45,8 @@ DRIVERS = { ('libcloud.storage.drivers.nimbus', 'NimbusStorageDriver'), Provider.LOCAL: ('libcloud.storage.drivers.local', 'LocalStorageDriver'), + Provider.KTUCLOUDSTORAGE: + ('libcloud.storage.drivers.ktucloud', 'KTUCloudStorageDriver'), # Deprecated Provider.CLOUDFILES_US: diff --git libcloud/storage/types.py libcloud/storage/types.py index d3c3b55..1d839dc 100644 --- libcloud/storage/types.py +++ libcloud/storage/types.py @@ -43,6 +43,7 @@ class Provider(object): @cvar S3_US_WEST_OREGON: Amazon S3 US West 2 (Oregon) @cvar NIMBUS: Nimbus.io driver @cvar LOCAL: Local storage driver + @cvar KTUCLOUDSTORAGE: KTUCloud Storage driver """ DUMMY = 'dummy' S3 = 's3' @@ -57,6 +58,7 @@ class Provider(object): NIMBUS = 'nimbus' LOCAL = 'local' CLOUDFILES = 'cloudfiles' + KTUCLOUDSTORAGE = 'ktucloudstorage' # Deperecated CLOUDFILES_US = 'cloudfiles_us' diff --git libcloud/test/storage/test_cloudfiles.py libcloud/test/storage/test_cloudfiles.py index 5654828..f1739a4 100644 --- libcloud/test/storage/test_cloudfiles.py +++ libcloud/test/storage/test_cloudfiles.py @@ -668,7 +668,7 @@ class CloudFilesTests(unittest.TestCase): result = self.driver.ex_set_account_metadata_temp_url_key("a key") self.assertTrue(result) - @mock.patch("libcloud.storage.drivers.cloudfiles.time") + @mock.patch("libcloud.storage.drivers.openstack.time") def test_ex_get_object_temp_url(self, time): time.return_value = 0 self.driver.ex_get_meta_data = mock.Mock() @@ -685,7 +685,6 @@ class CloudFilesTests(unittest.TestCase): sig = hmac.new(b('foo'), b(hmac_body), sha1).hexdigest() ret = self.driver.ex_get_object_temp_url(obj, 'GET') temp_url = 'https://storage101.%s1.clouddrive.com/v1/MossoCloudFS/foo_bar_container/foo_bar_object?temp_url_expires=60&temp_url_sig=%s' % (self.datacenter, sig) - self.assertEquals(''.join(sorted(ret)), ''.join(sorted(temp_url))) def test_ex_get_object_temp_url_no_key_raises_key_error(self): diff --git libcloud/test/storage/test_ktucloud.py libcloud/test/storage/test_ktucloud.py new file mode 100644 index 0000000..31bbe7d --- /dev/null +++ libcloud/test/storage/test_ktucloud.py @@ -0,0 +1,1013 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from hashlib import sha1 +import hmac +import os +import os.path # pylint: disable-msg=W0404 +import math +import sys +import copy +import unittest + +import mock + +import libcloud.utils.files + +from libcloud.utils.py3 import PY3 +from libcloud.utils.py3 import b +from libcloud.utils.py3 import httplib + +if PY3: + from io import FileIO as file + +from libcloud.common.types import LibcloudError, MalformedResponseError +from libcloud.storage.base import Container, Object +from libcloud.storage.types import ContainerAlreadyExistsError +from libcloud.storage.types import ContainerDoesNotExistError +from libcloud.storage.types import ContainerIsNotEmptyError +from libcloud.storage.types import ObjectDoesNotExistError +from libcloud.storage.types import ObjectHashMismatchError +from libcloud.storage.types import InvalidContainerNameError +from libcloud.storage.drivers.ktucloud import KTUCloudStorageDriver +from libcloud.storage.drivers.dummy import DummyIterator + +from libcloud.test import StorageMockHttp, MockRawResponse # pylint: disable-msg=E0611 +from libcloud.test import MockHttpTestCase # pylint: disable-msg=E0611 +from libcloud.test.file_fixtures import StorageFileFixtures, OpenStackFixtures # pylint: disable-msg=E0611 + +KTUCLOUD_X_STORAGE_URL = 'https://storage4.clouddrive.com/v1/MossoCloudFS' + + +class KTUCloudStorageTests(unittest.TestCase): + driver_klass = KTUCloudStorageDriver + driver_args = ('dummy', 'dummy') + driver_kwargs = {} + datacenter = 'ord' + + def setUp(self): + self.driver_klass.connectionCls.conn_classes = ( + None, KTUCloudStorageMockHttp) + self.driver_klass.connectionCls.rawResponseCls = \ + KTUCloudStorageMockRawResponse + KTUCloudStorageMockHttp.type = None + KTUCloudStorageMockRawResponse.type = None + self.driver = self.driver_klass(*self.driver_args, + **self.driver_kwargs) + + # normally authentication happens lazily, but we force it here + self.driver.connection._populate_hosts_and_request_paths() + self._remove_test_file() + + def tearDown(self): + self._remove_test_file() + + def test_invalid_ex_force_service_region(self): + driver = KTUCloudStorageDriver('driver', 'dummy', + ex_force_service_region='invalid') + + try: + driver.list_containers() + except: + self.fail('Exception was thrown') + + def test_ex_force_service_region(self): + driver = KTUCloudStorageDriver('driver', 'dummy', + ex_force_service_region='ORD') + driver.list_containers() + + def test_force_auth_token_kwargs(self): + base_url = 'https://cdn2.clouddrive.com/v1/MossoCloudFS' + kwargs = { + 'ex_force_auth_token': 'some-auth-token', + 'ex_force_base_url': base_url + } + driver = KTUCloudStorageDriver('driver', 'dummy', **kwargs) + driver.list_containers() + + self.assertEquals(kwargs['ex_force_auth_token'], + driver.connection.auth_token) + self.assertEquals('cdn2.clouddrive.com', + driver.connection.host) + self.assertEquals('/v1/MossoCloudFS', + driver.connection.request_path) + + def test_force_auth_url_kwargs(self): + kwargs = { + 'ex_force_auth_version': '2.0', + 'ex_force_auth_url': 'https://identity.api.rackspace.com' + } + driver = KTUCloudStorageDriver('driver', 'dummy', **kwargs) + + ktucloud_auth_version = '1.0' + self.assertEquals(kwargs['ex_force_auth_url'], + driver.connection._ex_force_auth_url) + self.assertEquals(ktucloud_auth_version, + driver.connection._auth_version) + + def test_invalid_json_throws_exception(self): + KTUCloudStorageMockHttp.type = 'MALFORMED_JSON' + try: + self.driver.list_containers() + except MalformedResponseError: + pass + else: + self.fail('Exception was not thrown') + + def test_service_catalog(self): + url = KTUCLOUD_X_STORAGE_URL + self.assertEqual( + url, + self.driver.connection.get_endpoint()) + + def test_list_containers(self): + KTUCloudStorageMockHttp.type = 'EMPTY' + containers = self.driver.list_containers() + self.assertEqual(len(containers), 0) + + KTUCloudStorageMockHttp.type = None + containers = self.driver.list_containers() + self.assertEqual(len(containers), 3) + + container = [c for c in containers if c.name == 'container2'][0] + self.assertEqual(container.extra['object_count'], 120) + self.assertEqual(container.extra['size'], 340084450) + + def test_list_container_objects(self): + KTUCloudStorageMockHttp.type = 'EMPTY' + container = Container( + name='test_container', extra={}, driver=self.driver) + objects = self.driver.list_container_objects(container=container) + self.assertEqual(len(objects), 0) + + KTUCloudStorageMockHttp.type = None + objects = self.driver.list_container_objects(container=container) + self.assertEqual(len(objects), 4) + + obj = [o for o in objects if o.name == 'foo test 1'][0] + self.assertEqual(obj.hash, '16265549b5bda64ecdaa5156de4c97cc') + self.assertEqual(obj.size, 1160520) + self.assertEqual(obj.container.name, 'test_container') + + def test_list_container_objects_iterator(self): + KTUCloudStorageMockHttp.type = 'ITERATOR' + container = Container( + name='test_container', extra={}, driver=self.driver) + objects = self.driver.list_container_objects(container=container) + self.assertEqual(len(objects), 5) + + obj = [o for o in objects if o.name == 'foo-test-1'][0] + self.assertEqual(obj.hash, '16265549b5bda64ecdaa5156de4c97cc') + self.assertEqual(obj.size, 1160520) + self.assertEqual(obj.container.name, 'test_container') + + def test_get_container(self): + container = self.driver.get_container(container_name='test_container') + self.assertEqual(container.name, 'test_container') + self.assertEqual(container.extra['object_count'], 800) + self.assertEqual(container.extra['size'], 1234568) + + def test_get_container_not_found(self): + try: + self.driver.get_container(container_name='not_found') + except ContainerDoesNotExistError: + pass + else: + self.fail('Exception was not thrown') + + def test_get_object_success(self): + obj = self.driver.get_object(container_name='test_container', + object_name='test_object') + self.assertEqual(obj.container.name, 'test_container') + self.assertEqual(obj.size, 555) + self.assertEqual(obj.hash, '6b21c4a111ac178feacf9ec9d0c71f17') + self.assertEqual(obj.extra['content_type'], 'application/zip') + self.assertEqual( + obj.extra['last_modified'], 'Tue, 25 Jan 2011 22:01:49 GMT') + self.assertEqual(obj.meta_data['foo-bar'], 'test 1') + self.assertEqual(obj.meta_data['bar-foo'], 'test 2') + + def test_get_object_not_found(self): + try: + self.driver.get_object(container_name='test_container', + object_name='not_found') + except ObjectDoesNotExistError: + pass + else: + self.fail('Exception was not thrown') + + def test_create_container_success(self): + container = self.driver.create_container( + container_name='test_create_container') + self.assertTrue(isinstance(container, Container)) + self.assertEqual(container.name, 'test_create_container') + self.assertEqual(container.extra['object_count'], 0) + + def test_create_container_already_exists(self): + KTUCloudStorageMockHttp.type = 'ALREADY_EXISTS' + + try: + self.driver.create_container( + container_name='test_create_container') + except ContainerAlreadyExistsError: + pass + else: + self.fail( + 'Container already exists but an exception was not thrown') + + def test_create_container_invalid_name_too_long(self): + name = ''.join(['x' for x in range(0, 257)]) + try: + self.driver.create_container(container_name=name) + except InvalidContainerNameError: + pass + else: + self.fail( + 'Invalid name was provided (name is too long)' + ', but exception was not thrown') + + def test_create_container_invalid_name_slashes_in_name(self): + try: + self.driver.create_container(container_name='test/slashes/') + except InvalidContainerNameError: + pass + else: + self.fail( + 'Invalid name was provided (name contains slashes)' + ', but exception was not thrown') + + def test_delete_container_success(self): + container = Container(name='foo_bar_container', extra={}, driver=self) + result = self.driver.delete_container(container=container) + self.assertTrue(result) + + def test_delete_container_not_found(self): + KTUCloudStorageMockHttp.type = 'NOT_FOUND' + container = Container(name='foo_bar_container', extra={}, driver=self) + try: + self.driver.delete_container(container=container) + except ContainerDoesNotExistError: + pass + else: + self.fail( + 'Container does not exist but an exception was not thrown') + + def test_delete_container_not_empty(self): + KTUCloudStorageMockHttp.type = 'NOT_EMPTY' + container = Container(name='foo_bar_container', extra={}, driver=self) + try: + self.driver.delete_container(container=container) + except ContainerIsNotEmptyError: + pass + else: + self.fail('Container is not empty but an exception was not thrown') + + def test_download_object_success(self): + container = Container(name='foo_bar_container', extra={}, driver=self) + obj = Object(name='foo_bar_object', size=1000, hash=None, extra={}, + container=container, meta_data=None, + driver=KTUCloudStorageDriver) + destination_path = os.path.abspath(__file__) + '.temp' + result = self.driver.download_object(obj=obj, + destination_path=destination_path, + overwrite_existing=False, + delete_on_failure=True) + self.assertTrue(result) + + def test_download_object_invalid_file_size(self): + KTUCloudStorageMockRawResponse.type = 'INVALID_SIZE' + container = Container(name='foo_bar_container', extra={}, driver=self) + obj = Object(name='foo_bar_object', size=1000, hash=None, extra={}, + container=container, meta_data=None, + driver=KTUCloudStorageDriver) + destination_path = os.path.abspath(__file__) + '.temp' + result = self.driver.download_object(obj=obj, + destination_path=destination_path, + overwrite_existing=False, + delete_on_failure=True) + self.assertFalse(result) + + def test_download_object_success_not_found(self): + KTUCloudStorageMockRawResponse.type = 'NOT_FOUND' + container = Container(name='foo_bar_container', extra={}, driver=self) + + obj = Object(name='foo_bar_object', size=1000, hash=None, extra={}, + container=container, + meta_data=None, + driver=KTUCloudStorageDriver) + destination_path = os.path.abspath(__file__) + '.temp' + try: + self.driver.download_object( + obj=obj, + destination_path=destination_path, + overwrite_existing=False, + delete_on_failure=True) + except ObjectDoesNotExistError: + pass + else: + self.fail('Object does not exist but an exception was not thrown') + + def test_download_object_as_stream(self): + container = Container(name='foo_bar_container', extra={}, driver=self) + obj = Object(name='foo_bar_object', size=1000, hash=None, extra={}, + container=container, meta_data=None, + driver=KTUCloudStorageDriver) + + stream = self.driver.download_object_as_stream(obj=obj, chunk_size=None) + self.assertTrue(hasattr(stream, '__iter__')) + + def test_upload_object_success(self): + def upload_file(self, response, file_path, chunked=False, + calculate_hash=True): + return True, 'hash343hhash89h932439jsaa89', 1000 + + old_func = KTUCloudStorageDriver._upload_file + KTUCloudStorageDriver._upload_file = upload_file + file_path = os.path.abspath(__file__) + container = Container(name='foo_bar_container', extra={}, driver=self) + object_name = 'foo_test_upload' + extra = {'meta_data': {'some-value': 'foobar'}} + obj = self.driver.upload_object(file_path=file_path, container=container, + extra=extra, object_name=object_name) + self.assertEqual(obj.name, 'foo_test_upload') + self.assertEqual(obj.size, 1000) + self.assertTrue('some-value' in obj.meta_data) + KTUCloudStorageDriver._upload_file = old_func + + def test_upload_object_zero_size_object(self): + def upload_file(self, response, file_path, chunked=False, + calculate_hash=True): + return True, 'hash343hhash89h932439jsaa89', 0 + + old_func = KTUCloudStorageDriver._upload_file + old_request = self.driver.connection.request + KTUCloudStorageDriver._upload_file = upload_file + file_path = os.path.join(os.path.dirname(__file__), '__init__.py') + container = Container(name='foo_bar_container', extra={}, driver=self) + object_name = 'empty' + extra = {} + + def func(*args, **kwargs): + self.assertEqual(kwargs['headers']['Content-Length'], 0) + func.called = True + return old_request(*args, **kwargs) + + self.driver.connection.request = func + func.called = False + obj = self.driver.upload_object(file_path=file_path, container=container, + extra=extra, object_name=object_name) + self.assertEqual(obj.name, 'empty') + self.assertEqual(obj.size, 0) + self.assertTrue(func.called) + KTUCloudStorageDriver._upload_file = old_func + self.driver.connection.request = old_request + + def test_upload_object_invalid_hash(self): + def upload_file(self, response, file_path, chunked=False, + calculate_hash=True): + return True, 'hash343hhash89h932439jsaa89', 1000 + + KTUCloudStorageMockRawResponse.type = 'INVALID_HASH' + + old_func = KTUCloudStorageDriver._upload_file + KTUCloudStorageDriver._upload_file = upload_file + file_path = os.path.abspath(__file__) + container = Container(name='foo_bar_container', extra={}, driver=self) + object_name = 'foo_test_upload' + try: + self.driver.upload_object(file_path=file_path, container=container, + object_name=object_name, + verify_hash=True) + except ObjectHashMismatchError: + pass + else: + self.fail( + 'Invalid hash was returned but an exception was not thrown') + finally: + KTUCloudStorageDriver._upload_file = old_func + + def test_upload_object_no_content_type(self): + def no_content_type(name): + return None, None + + old_func = libcloud.utils.files.guess_file_mime_type + libcloud.utils.files.guess_file_mime_type = no_content_type + file_path = os.path.abspath(__file__) + container = Container(name='foo_bar_container', extra={}, driver=self) + object_name = 'foo_test_upload' + try: + self.driver.upload_object(file_path=file_path, container=container, + object_name=object_name) + except AttributeError: + pass + else: + self.fail( + 'File content type not provided' + ' but an exception was not thrown') + finally: + libcloud.utils.files.guess_file_mime_type = old_func + + def test_upload_object_error(self): + def dummy_content_type(name): + return 'application/zip', None + + def send(instance): + raise Exception('') + + old_func1 = libcloud.utils.files.guess_file_mime_type + libcloud.utils.files.guess_file_mime_type = dummy_content_type + old_func2 = KTUCloudStorageMockHttp.send + KTUCloudStorageMockHttp.send = send + + file_path = os.path.abspath(__file__) + container = Container(name='foo_bar_container', extra={}, driver=self) + object_name = 'foo_test_upload' + try: + self.driver.upload_object( + file_path=file_path, + container=container, + object_name=object_name) + except LibcloudError: + pass + else: + self.fail('Timeout while uploading but an exception was not thrown') + finally: + libcloud.utils.files.guess_file_mime_type = old_func1 + KTUCloudStorageMockHttp.send = old_func2 + + def test_upload_object_inexistent_file(self): + def dummy_content_type(name): + return 'application/zip', None + + old_func = libcloud.utils.files.guess_file_mime_type + libcloud.utils.files.guess_file_mime_type = dummy_content_type + + file_path = os.path.abspath(__file__ + '.inexistent') + container = Container(name='foo_bar_container', extra={}, driver=self) + object_name = 'foo_test_upload' + try: + self.driver.upload_object( + file_path=file_path, + container=container, + object_name=object_name) + except OSError: + pass + else: + self.fail('Inexistent but an exception was not thrown') + finally: + libcloud.utils.files.guess_file_mime_type = old_func + + def test_upload_object_via_stream(self): + def dummy_content_type(name): + return 'application/zip', None + + old_func = libcloud.utils.files.guess_file_mime_type + libcloud.utils.files.guess_file_mime_type = dummy_content_type + + container = Container(name='foo_bar_container', extra={}, driver=self) + object_name = 'foo_test_stream_data' + iterator = DummyIterator(data=['2', '3', '5']) + try: + self.driver.upload_object_via_stream(container=container, + object_name=object_name, + iterator=iterator) + finally: + libcloud.utils.files.guess_file_mime_type = old_func + + def test_delete_object_success(self): + container = Container(name='foo_bar_container', extra={}, driver=self) + obj = Object(name='foo_bar_object', size=1000, hash=None, extra={}, + container=container, meta_data=None, + driver=KTUCloudStorageDriver) + status = self.driver.delete_object(obj=obj) + self.assertTrue(status) + + def test_delete_object_not_found(self): + KTUCloudStorageMockHttp.type = 'NOT_FOUND' + container = Container(name='foo_bar_container', extra={}, driver=self) + obj = Object(name='foo_bar_object', size=1000, hash=None, extra={}, + container=container, meta_data=None, + driver=KTUCloudStorageDriver) + try: + self.driver.delete_object(obj=obj) + except ObjectDoesNotExistError: + pass + else: + self.fail('Object does not exist but an exception was not thrown') + + def test_ex_get_meta_data(self): + meta_data = self.driver.ex_get_meta_data() + self.assertTrue(isinstance(meta_data, dict)) + self.assertTrue('object_count' in meta_data) + self.assertTrue('container_count' in meta_data) + self.assertTrue('bytes_used' in meta_data) + self.assertTrue('temp_url_key' in meta_data) + + def test_ex_purge_object_from_cdn(self): + KTUCloudStorageMockHttp.type = 'PURGE_SUCCESS' + container = Container(name='foo_bar_container', extra={}, + driver=self.driver) + obj = Object(name='object', size=1000, hash=None, extra={}, + container=container, meta_data=None, + driver=self) + + self.assertTrue(self.driver.ex_purge_object_from_cdn(obj=obj)) + + def test_ex_purge_object_from_cdn_with_email(self): + KTUCloudStorageMockHttp.type = 'PURGE_SUCCESS_EMAIL' + container = Container(name='foo_bar_container', extra={}, + driver=self.driver) + obj = Object(name='object', size=1000, hash=None, extra={}, + container=container, meta_data=None, + driver=self) + self.assertTrue(self.driver.ex_purge_object_from_cdn(obj=obj, + email='test@test.com')) + + @mock.patch('os.path.getsize') + def test_ex_multipart_upload_object_for_small_files(self, getsize_mock): + getsize_mock.return_value = 0 + + old_func = KTUCloudStorageDriver.upload_object + mocked_upload_object = mock.Mock(return_value="test") + KTUCloudStorageDriver.upload_object = mocked_upload_object + + file_path = os.path.abspath(__file__) + container = Container(name='foo_bar_container', extra={}, driver=self) + object_name = 'foo_test_upload' + obj = self.driver.ex_multipart_upload_object(file_path=file_path, container=container, + object_name=object_name) + KTUCloudStorageDriver.upload_object = old_func + + self.assertTrue(mocked_upload_object.called) + self.assertEqual(obj, "test") + + def test_ex_multipart_upload_object_success(self): + _upload_object_part = KTUCloudStorageDriver._upload_object_part + _upload_object_manifest = KTUCloudStorageDriver._upload_object_manifest + + mocked__upload_object_part = mock.Mock(return_value="test_part") + mocked__upload_object_manifest = mock.Mock(return_value="test_manifest") + + KTUCloudStorageDriver._upload_object_part = mocked__upload_object_part + KTUCloudStorageDriver._upload_object_manifest = mocked__upload_object_manifest + + parts = 5 + file_path = os.path.abspath(__file__) + chunk_size = int(math.ceil(float(os.path.getsize(file_path)) / parts)) + container = Container(name='foo_bar_container', extra={}, driver=self) + object_name = 'foo_test_upload' + self.driver.ex_multipart_upload_object(file_path=file_path, container=container, + object_name=object_name, chunk_size=chunk_size) + + KTUCloudStorageDriver._upload_object_part = _upload_object_part + KTUCloudStorageDriver._upload_object_manifest = _upload_object_manifest + + self.assertEqual(mocked__upload_object_part.call_count, parts) + self.assertTrue(mocked__upload_object_manifest.call_count, 1) + + def test__upload_object_part(self): + _put_object = KTUCloudStorageDriver._put_object + mocked__put_object = mock.Mock(return_value="test") + KTUCloudStorageDriver._put_object = mocked__put_object + + part_number = 7 + object_name = "test_object" + expected_name = object_name + '/%08d' % part_number + container = Container(name='foo_bar_container', extra={}, driver=self) + + self.driver._upload_object_part(container, object_name, + part_number, None) + + KTUCloudStorageDriver._put_object = _put_object + + func_kwargs = tuple(mocked__put_object.call_args)[1] + self.assertEquals(func_kwargs['object_name'], expected_name) + self.assertEquals(func_kwargs['container'], container) + + def test__upload_object_manifest(self): + hash_function = self.driver._get_hash_function() + hash_function.update(b('')) + data_hash = hash_function.hexdigest() + + fake_response = type('KTUCloudStorageResponse', (), {'headers': + {'etag': data_hash} + }) + + _request = self.driver.connection.request + mocked_request = mock.Mock(return_value=fake_response) + self.driver.connection.request = mocked_request + + container = Container(name='foo_bar_container', extra={}, driver=self) + object_name = "test_object" + + self.driver._upload_object_manifest(container, object_name) + + func_args, func_kwargs = tuple(mocked_request.call_args) + + self.driver.connection.request = _request + + self.assertEquals(func_args[0], "/" + container.name + "/" + object_name) + self.assertEquals(func_kwargs["headers"]["X-Object-Manifest"], + container.name + "/" + object_name + "/") + self.assertEquals(func_kwargs["method"], "PUT") + + def test__upload_object_manifest_wrong_hash(self): + fake_response = type('KTUCloudStorageResponse', (), {'headers': + {'etag': '0000000'}}) + + _request = self.driver.connection.request + mocked_request = mock.Mock(return_value=fake_response) + self.driver.connection.request = mocked_request + + container = Container(name='foo_bar_container', extra={}, driver=self) + object_name = "test_object" + + + try: + self.driver._upload_object_manifest(container, object_name) + except ObjectHashMismatchError: + pass + else: + self.fail('Exception was not thrown') + finally: + self.driver.connection.request = _request + + def test_ex_enable_static_website(self): + container = Container(name='foo_bar_container', extra={}, driver=self) + result = self.driver.ex_enable_static_website(container=container, + index_file='index.html') + self.assertTrue(result) + + def test_ex_set_error_page(self): + container = Container(name='foo_bar_container', extra={}, driver=self) + result = self.driver.ex_set_error_page(container=container, + file_name='error.html') + self.assertTrue(result) + + def test_ex_set_account_metadata_temp_url_key(self): + result = self.driver.ex_set_account_metadata_temp_url_key("a key") + self.assertTrue(result) + + @mock.patch("libcloud.storage.drivers.openstack.time") + def test_ex_get_object_temp_url(self, time): + time.return_value = 0 + self.driver.ex_get_meta_data = mock.Mock() + self.driver.ex_get_meta_data.return_value = {'container_count': 1, + 'object_count': 1, + 'bytes_used': 1, + 'temp_url_key': 'foo'} + container = Container(name='foo_bar_container', extra={}, driver=self) + obj = Object(name='foo_bar_object', size=1000, hash=None, extra={}, + container=container, meta_data=None, + driver=self) + hmac_body = "%s\n%s\n%s" % ('GET', 60, + "/v1/MossoCloudFS/foo_bar_container/foo_bar_object") + sig = hmac.new(b('foo'), b(hmac_body), sha1).hexdigest() + ret = self.driver.ex_get_object_temp_url(obj, 'GET') + temp_url = '%s/foo_bar_container/foo_bar_object?temp_url_expires=60&temp_url_sig=%s' % (KTUCLOUD_X_STORAGE_URL, sig) + + self.assertEquals(''.join(sorted(ret)), ''.join(sorted(temp_url))) + + def test_ex_get_object_temp_url_no_key_raises_key_error(self): + self.driver.ex_get_meta_data = mock.Mock() + self.driver.ex_get_meta_data.return_value = {'container_count': 1, + 'object_count': 1, + 'bytes_used': 1, + 'temp_url_key': None} + container = Container(name='foo_bar_container', extra={}, driver=self) + obj = Object(name='foo_bar_object', size=1000, hash=None, extra={}, + container=container, meta_data=None, + driver=self) + self.assertRaises(KeyError, self.driver.ex_get_object_temp_url, obj, 'GET') + + def _remove_test_file(self): + file_path = os.path.abspath(__file__) + '.temp' + + try: + os.unlink(file_path) + except OSError: + pass + + +class KTUCloudStorageMockHttp(StorageMockHttp, MockHttpTestCase): + + fixtures = StorageFileFixtures('cloudfiles') + auth_fixtures = OpenStackFixtures() + base_headers = { 'content-type': 'application/json; charset=UTF-8'} + + # fake auth token response + def _storage_v1_auth(self, method, url, body, headers): + headers = copy.deepcopy(self.base_headers) + headers.update({ 'x-server-management-url': + 'https://servers.api.rackspacecloud.com/v1.0/slug', + 'x-auth-token': 'FE011C19', + 'x-cdn-management-url': + 'https://cdn.clouddrive.com/v1/MossoCloudFS', + 'x-storage-token': 'FE011C19', + 'x-storage-url': + KTUCLOUD_X_STORAGE_URL }) + return (httplib.NO_CONTENT, + "", + headers, + httplib.responses[httplib.NO_CONTENT]) + + def _v1_MossoCloudFS_MALFORMED_JSON(self, method, url, body, headers): + # test_invalid_json_throws_exception + body = 'broken: json /*"' + return (httplib.NO_CONTENT, + body, + self.base_headers, + httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_EMPTY(self, method, url, body, headers): + return (httplib.NO_CONTENT, + body, + self.base_headers, + httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS(self, method, url, body, headers): + headers = copy.deepcopy(self.base_headers) + if method == 'GET': + # list_containers + body = self.fixtures.load('list_containers.json') + status_code = httplib.OK + elif method == 'HEAD': + # get_meta_data + body = self.fixtures.load('meta_data.json') + status_code = httplib.NO_CONTENT + headers.update({ 'x-account-container-count': 10, + 'x-account-object-count': 400, + 'x-account-bytes-used': 1234567 + }) + elif method == 'POST': + body = '' + status_code = httplib.NO_CONTENT + return (status_code, body, headers, httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_not_found(self, method, url, body, headers): + # test_get_object_not_found + if method == 'HEAD': + body = '' + else: + raise ValueError('Invalid method') + + return (httplib.NOT_FOUND, + body, + self.base_headers, + httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_test_container_EMPTY(self, method, url, body, headers): + body = self.fixtures.load('list_container_objects_empty.json') + return (httplib.OK, + body, + self.base_headers, + httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_test_container(self, method, url, body, headers): + headers = copy.deepcopy(self.base_headers) + if method == 'GET': + # list_container_objects + if url.find('marker') == -1: + body = self.fixtures.load('list_container_objects.json') + status_code = httplib.OK + else: + body = '' + status_code = httplib.NO_CONTENT + elif method == 'HEAD': + # get_container + body = self.fixtures.load('list_container_objects_empty.json') + status_code = httplib.NO_CONTENT + headers.update({ 'x-container-object-count': 800, + 'x-container-bytes-used': 1234568 + }) + return (status_code, body, headers, httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_test_container_ITERATOR(self, method, url, body, headers): + headers = copy.deepcopy(self.base_headers) + # list_container_objects + if url.find('foo-test-3') != -1: + body = self.fixtures.load('list_container_objects_not_exhausted2.json') + status_code = httplib.OK + elif url.find('foo-test-5') != -1: + body = '' + status_code = httplib.NO_CONTENT + else: + # First request + body = self.fixtures.load('list_container_objects_not_exhausted1.json') + status_code = httplib.OK + + return (status_code, body, headers, httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_test_container_not_found( + self, method, url, body, headers): + # test_get_container_not_found + if method == 'HEAD': + body = '' + else: + raise ValueError('Invalid method') + + return (httplib.NOT_FOUND, body, + self.base_headers, + httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_test_container_test_object( + self, method, url, body, headers): + headers = copy.deepcopy(self.base_headers) + if method == 'HEAD': + # get_object + body = self.fixtures.load('list_container_objects_empty.json') + status_code = httplib.NO_CONTENT + headers.update({ 'content-length': 555, + 'last-modified': 'Tue, 25 Jan 2011 22:01:49 GMT', + 'etag': '6b21c4a111ac178feacf9ec9d0c71f17', + 'x-object-meta-foo-bar': 'test 1', + 'x-object-meta-bar-foo': 'test 2', + 'content-type': 'application/zip'}) + return (status_code, body, headers, httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_test_create_container( + self, method, url, body, headers): + # test_create_container_success + headers = copy.deepcopy(self.base_headers) + body = self.fixtures.load('list_container_objects_empty.json') + headers = copy.deepcopy(self.base_headers) + headers.update({ 'content-length': 18, + 'date': 'Mon, 28 Feb 2011 07:52:57 GMT' + }) + status_code = httplib.CREATED + return (status_code, body, headers, httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_test_create_container_ALREADY_EXISTS( + self, method, url, body, headers): + # test_create_container_already_exists + headers = copy.deepcopy(self.base_headers) + body = self.fixtures.load('list_container_objects_empty.json') + headers.update({ 'content-type': 'text/plain' }) + status_code = httplib.ACCEPTED + return (status_code, body, headers, httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_foo_bar_container(self, method, url, body, headers): + if method == 'DELETE': + # test_delete_container_success + body = self.fixtures.load('list_container_objects_empty.json') + headers = self.base_headers + status_code = httplib.NO_CONTENT + elif method == 'POST': + # test_ex_enable_static_website + body = '' + headers = self.base_headers + status_code = httplib.ACCEPTED + return (status_code, body, headers, httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_foo_bar_container_object_PURGE_SUCCESS( + self, method, url, body, headers): + + if method == 'DELETE': + # test_ex_purge_from_cdn + headers = self.base_headers + status_code = httplib.NO_CONTENT + return (status_code, body, headers, httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_foo_bar_container_object_PURGE_SUCCESS_EMAIL( + self, method, url, body, headers): + + if method == 'DELETE': + # test_ex_purge_from_cdn_with_email + self.assertEqual(headers['X-Purge-Email'], 'test@test.com') + headers = self.base_headers + status_code = httplib.NO_CONTENT + return (status_code, body, headers, httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_foo_bar_container_NOT_FOUND( + self, method, url, body, headers): + + if method == 'DELETE': + # test_delete_container_not_found + body = self.fixtures.load('list_container_objects_empty.json') + headers = self.base_headers + status_code = httplib.NOT_FOUND + return (status_code, body, headers, httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_foo_bar_container_NOT_EMPTY( + self, method, url, body, headers): + + if method == 'DELETE': + # test_delete_container_not_empty + body = self.fixtures.load('list_container_objects_empty.json') + headers = self.base_headers + status_code = httplib.CONFLICT + return (status_code, body, headers, httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_foo_bar_container_foo_bar_object( + self, method, url, body, headers): + + if method == 'DELETE': + # test_delete_object_success + body = self.fixtures.load('list_container_objects_empty.json') + headers = self.base_headers + status_code = httplib.NO_CONTENT + return (status_code, body, headers, httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_foo_bar_container_foo_bar_object_NOT_FOUND( + self, method, url, body, headers): + + if method == 'DELETE': + # test_delete_object_success + body = self.fixtures.load('list_container_objects_empty.json') + headers = self.base_headers + status_code = httplib.NOT_FOUND + + return (status_code, body, headers, httplib.responses[httplib.OK]) + + def _v1_1_auth(self, method, url, body, headers): + body = self.auth_fixtures.load('_v1_1__auth.json') + return (httplib.OK, body, {'content-type': 'application/json; charset=UTF-8'}, httplib.responses[httplib.OK]) + + +class KTUCloudStorageMockRawResponse(MockRawResponse): + + fixtures = StorageFileFixtures('cloudfiles') + base_headers = { 'content-type': 'application/json; charset=UTF-8'} + + def _v1_MossoCloudFS_foo_bar_container_foo_test_upload( + self, method, url, body, headers): + # test_object_upload_success + + body = '' + headers = {} + headers.update(self.base_headers) + headers['etag'] = 'hash343hhash89h932439jsaa89' + return (httplib.CREATED, body, headers, httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_foo_bar_container_empty(self, method, url, body, + headers): + # test_upload_object_zero_size_object + body = '' + headers = {} + headers.update(self.base_headers) + headers['etag'] = 'hash343hhash89h932439jsaa89' + return (httplib.CREATED, body, headers, httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_foo_bar_container_foo_test_upload_INVALID_HASH( + self, method, url, body, headers): + # test_object_upload_invalid_hash + body = '' + headers = {} + headers.update(self.base_headers) + headers['etag'] = 'foobar' + return (httplib.CREATED, body, headers, + httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_foo_bar_container_foo_bar_object( + self, method, url, body, headers): + + # test_download_object_success + body = 'test' + self._data = self._generate_random_data(1000) + return (httplib.OK, + body, + self.base_headers, + httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_foo_bar_container_foo_bar_object_INVALID_SIZE( + self, method, url, body, headers): + # test_download_object_invalid_file_size + body = 'test' + self._data = self._generate_random_data(100) + return (httplib.OK, body, + self.base_headers, + httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_foo_bar_container_foo_bar_object_NOT_FOUND( + self, method, url, body, headers): + body = '' + return (httplib.NOT_FOUND, body, + self.base_headers, + httplib.responses[httplib.OK]) + + def _v1_MossoCloudFS_foo_bar_container_foo_test_stream_data( + self, method, url, body, headers): + + # test_upload_object_via_stream_success + headers = {} + headers.update(self.base_headers) + headers['etag'] = '577ef1154f3240ad5b9b413aa7346a1e' + body = 'test' + return (httplib.CREATED, + body, + headers, + httplib.responses[httplib.OK]) + +if __name__ == '__main__': + sys.exit(unittest.main())