You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@libcloud.apache.org by to...@apache.org on 2011/03/15 20:08:48 UTC
svn commit: r1081909 [2/2] - in /incubator/libcloud/trunk: ./ libcloud/
libcloud/common/ libcloud/compute/ libcloud/compute/drivers/
libcloud/storage/ libcloud/storage/drivers/ test/ test/compute/ test/storage/
Modified: incubator/libcloud/trunk/test/storage/test_cloudfiles.py
URL: http://svn.apache.org/viewvc/incubator/libcloud/trunk/test/storage/test_cloudfiles.py?rev=1081909&r1=1081908&r2=1081909&view=diff
==============================================================================
--- incubator/libcloud/trunk/test/storage/test_cloudfiles.py (original)
+++ incubator/libcloud/trunk/test/storage/test_cloudfiles.py Tue Mar 15 19:08:47 2011
@@ -13,7 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import os
-import os.path
+import os.path # pylint: disable-msg=W0404
import random
import sys
import copy
@@ -30,17 +30,18 @@ from libcloud.storage.types import Conta
from libcloud.storage.types import ObjectDoesNotExistError
from libcloud.storage.types import ObjectHashMismatchError
from libcloud.storage.drivers.cloudfiles import CloudFilesStorageDriver
-from libcloud.storage.drivers.dummy import DummyFileObject, DummyIterator
+from libcloud.storage.drivers.dummy import DummyIterator
-from test import MockHttp, MockRawResponse
-from test.file_fixtures import StorageFileFixtures
+from test import MockHttp, MockRawResponse # pylint: disable-msg=E0611
+from test.file_fixtures import StorageFileFixtures # pylint: disable-msg=E0611
class CloudFilesTests(unittest.TestCase):
def setUp(self):
- CloudFilesStorageDriver.connectionCls.conn_classes = (None,
- CloudFilesMockHttp)
- CloudFilesStorageDriver.connectionCls.rawResponseCls = CloudFilesMockRawResponse
+ CloudFilesStorageDriver.connectionCls.conn_classes = (
+ None, CloudFilesMockHttp)
+ CloudFilesStorageDriver.connectionCls.rawResponseCls = \
+ CloudFilesMockRawResponse
CloudFilesMockHttp.type = None
CloudFilesMockRawResponse.type = None
self.driver = CloudFilesStorageDriver('dummy', 'dummy')
@@ -50,7 +51,7 @@ class CloudFilesTests(unittest.TestCase)
self._remove_test_file()
def test_get_meta_data(self):
- meta_data = self.driver.get_meta_data()
+ self.driver.get_meta_data()
def test_list_containers(self):
CloudFilesMockHttp.type = 'EMPTY'
@@ -67,7 +68,8 @@ class CloudFilesTests(unittest.TestCase)
def test_list_container_objects(self):
CloudFilesMockHttp.type = 'EMPTY'
- container = Container(name='test_container', extra={}, driver=self.driver)
+ container = Container(
+ name='test_container', extra={}, driver=self.driver)
objects = self.driver.list_container_objects(container=container)
self.assertEqual(len(objects), 0)
@@ -93,12 +95,14 @@ class CloudFilesTests(unittest.TestCase)
self.assertEqual(obj.size, 555)
self.assertEqual(obj.extra['content_type'], 'application/zip')
self.assertEqual(obj.extra['etag'], '6b21c4a111ac178feacf9ec9d0c71f17')
- self.assertEqual(obj.extra['last_modified'], 'Tue, 25 Jan 2011 22:01:49 GMT')
+ 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_create_container_success(self):
- container = self.driver.create_container(container_name='test_create_container')
+ 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)
@@ -107,28 +111,34 @@ class CloudFilesTests(unittest.TestCase)
CloudFilesMockHttp.type = 'ALREADY_EXISTS'
try:
- container = self.driver.create_container(container_name='test_create_container')
+ self.driver.create_container(
+ container_name='test_create_container')
except ContainerAlreadyExistsError:
pass
else:
- self.fail('Container already exists but an exception was not thrown')
+ self.fail(
+ 'Container already exists but an exception was not thrown')
def test_create_container_invalid_name(self):
try:
- container = self.driver.create_container(container_name='invalid//name/')
+ self.driver.create_container(container_name='invalid//name/')
except:
pass
else:
- self.fail('Invalid name was provided (name contains slashes), but exception was not thrown')
+ self.fail(
+ 'Invalid name was provided (name contains slashes)'
+ ', but exception was not thrown')
def test_create_container_invalid_name(self):
name = ''.join([ 'x' for x in range(0, 257)])
try:
- container = self.driver.create_container(container_name=name)
+ self.driver.create_container(container_name=name)
except:
pass
else:
- self.fail('Invalid name was provided (name is too long), but exception was not thrown')
+ self.fail(
+ 'Invalid name was provided (name is too long)'
+ ', but exception was not thrown')
def test_delete_container_success(self):
container = Container(name='foo_bar_container', extra={}, driver=self)
@@ -139,17 +149,18 @@ class CloudFilesTests(unittest.TestCase)
CloudFilesMockHttp.type = 'NOT_FOUND'
container = Container(name='foo_bar_container', extra={}, driver=self)
try:
- result = self.driver.delete_container(container=container)
+ self.driver.delete_container(container=container)
except ContainerDoesNotExistError:
pass
else:
- self.fail('Container does not exist but an exception was not thrown')
+ self.fail(
+ 'Container does not exist but an exception was not thrown')
def test_delete_container_not_empty(self):
CloudFilesMockHttp.type = 'NOT_EMPTY'
container = Container(name='foo_bar_container', extra={}, driver=self)
try:
- result = self.driver.delete_container(container=container)
+ self.driver.delete_container(container=container)
except ContainerIsNotEmptyError:
pass
else:
@@ -180,17 +191,22 @@ class CloudFilesTests(unittest.TestCase)
delete_on_failure=True)
self.assertFalse(result)
- def download_object_success_not_found(self):
- CloudFilesMockHttp.type = 'NOT_FOUND'
+ def test_download_object_success_not_found(self):
+ #CloudFilesMockHttp.type = 'NOT_FOUND'
+ CloudFilesMockRawResponse.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,
+ container=container,
+ meta_data=None,
driver=CloudFilesStorageDriver)
- destination_path = os.path.abspath(__file__)
+ destination_path = os.path.abspath(__file__) + '.temp'
try:
- result = self.driver.download_object(obj=obj,
- destination_path=destination_path,
- overwrite_existing=False,
- delete_on_failure=True)
+ self.driver.download_object(
+ obj=obj,
+ destination_path=destination_path,
+ overwrite_existing=False,
+ delete_on_failure=True)
except ObjectDoesNotExistError:
pass
else:
@@ -209,7 +225,8 @@ class CloudFilesTests(unittest.TestCase)
file_path = os.path.abspath(__file__)
container = Container(name='foo_bar_container', extra={}, driver=self)
object_name = 'foo_test_upload'
- obj = self.driver.upload_object(file_path=file_path, container=container,
+ obj = self.driver.upload_object(
+ file_path=file_path, container=container,
object_name=object_name)
self.assertEqual(obj.name, 'foo_test_upload')
self.assertEqual(obj.size, 1000)
@@ -228,13 +245,14 @@ class CloudFilesTests(unittest.TestCase)
container = Container(name='foo_bar_container', extra={}, driver=self)
object_name = 'foo_test_upload'
try:
- obj = self.driver.upload_object(file_path=file_path, container=container,
- object_name=object_name,
- file_hash='footest123')
+ self.driver.upload_object(file_path=file_path, container=container,
+ object_name=object_name,
+ file_hash='footest123')
except ObjectHashMismatchError:
pass
else:
- self.fail('Invalid hash was returned but an exception was not thrown')
+ self.fail(
+ 'Invalid hash was returned but an exception was not thrown')
finally:
CloudFilesStorageDriver._upload_file = old_func
@@ -248,12 +266,14 @@ class CloudFilesTests(unittest.TestCase)
container = Container(name='foo_bar_container', extra={}, driver=self)
object_name = 'foo_test_upload'
try:
- obj = self.driver.upload_object(file_path=file_path, container=container,
- object_name=object_name)
+ 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')
+ self.fail(
+ 'File content type not provided'
+ ' but an exception was not thrown')
finally:
libcloud.utils.guess_file_mime_type = old_func
@@ -273,8 +293,10 @@ class CloudFilesTests(unittest.TestCase)
container = Container(name='foo_bar_container', extra={}, driver=self)
object_name = 'foo_test_upload'
try:
- obj = self.driver.upload_object(file_path=file_path, container=container,
- object_name=object_name)
+ self.driver.upload_object(
+ file_path=file_path,
+ container=container,
+ object_name=object_name)
except LibcloudError:
pass
else:
@@ -294,8 +316,10 @@ class CloudFilesTests(unittest.TestCase)
container = Container(name='foo_bar_container', extra={}, driver=self)
object_name = 'foo_test_upload'
try:
- obj = self.driver.upload_object(file_path=file_path, container=container,
- object_name=object_name)
+ self.driver.upload_object(
+ file_path=file_path,
+ container=container,
+ object_name=object_name)
except OSError:
pass
else:
@@ -314,7 +338,7 @@ class CloudFilesTests(unittest.TestCase)
object_name = 'foo_test_stream_data'
iterator = DummyIterator(data=['2', '3', '5'])
try:
- obj = self.driver.upload_object_via_stream(container=container,
+ self.driver.upload_object_via_stream(container=container,
object_name=object_name,
iterator=iterator)
finally:
@@ -335,7 +359,7 @@ class CloudFilesTests(unittest.TestCase)
container=container, meta_data=None,
driver=CloudFilesStorageDriver)
try:
- result = self.driver.delete_object(obj=obj)
+ self.driver.delete_object(obj=obj)
except ObjectDoesNotExistError:
pass
else:
@@ -364,21 +388,30 @@ class CloudFilesMockHttp(MockHttp):
pass
def send(self, data):
- pass
+ pass
# fake auth token response
def _v1_0(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',
+ 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-cdn-management-url':
+ 'https://cdn.clouddrive.com/v1/MossoCloudFS',
'x-storage-token': 'FE011C19',
- 'x-storage-url': 'https://storage4.clouddrive.com/v1/MossoCloudFS'})
- return (httplib.NO_CONTENT, "", headers, httplib.responses[httplib.NO_CONTENT])
+ 'x-storage-url':
+ 'https://storage4.clouddrive.com/v1/MossoCloudFS'})
+ return (httplib.NO_CONTENT,
+ "",
+ headers,
+ httplib.responses[httplib.NO_CONTENT])
def _v1_MossoCloudFS_EMPTY(self, method, url, body, headers):
body = self.fixtures.load('list_containers_empty.json')
- return (httplib.OK, body, self.base_headers, httplib.responses[httplib.OK])
+ return (httplib.OK,
+ body,
+ self.base_headers,
+ httplib.responses[httplib.OK])
def _v1_MossoCloudFS(self, method, url, body, headers):
if method == 'GET':
@@ -399,7 +432,10 @@ class CloudFilesMockHttp(MockHttp):
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])
+ return (httplib.OK,
+ body,
+ self.base_headers,
+ httplib.responses[httplib.OK])
def _v1_MossoCloudFS_test_container(self, method, url, body, headers):
if method == 'GET':
@@ -417,8 +453,9 @@ class CloudFilesMockHttp(MockHttp):
})
return (status_code, body, headers, httplib.responses[httplib.OK])
- def _v1_MossoCloudFS_test_container_test_object(self, method, url, body,
- headers):
+ def _v1_MossoCloudFS_test_container_test_object(
+ self, method, url, body, headers):
+
if method == 'HEAD':
# get_object
body = self.fixtures.load('list_container_objects_empty.json')
@@ -432,7 +469,9 @@ class CloudFilesMockHttp(MockHttp):
'content-type': 'application/zip'})
return (status_code, body, headers, httplib.responses[httplib.OK])
- def _v1_MossoCloudFS_test_create_container(self, method, url, body, headers):
+ def _v1_MossoCloudFS_test_create_container(
+ self, method, url, body, headers):
+
# test_create_container_success
body = self.fixtures.load('list_container_objects_empty.json')
headers = self.base_headers
@@ -442,7 +481,9 @@ class CloudFilesMockHttp(MockHttp):
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):
+ def _v1_MossoCloudFS_test_create_container_ALREADY_EXISTS(
+ self, method, url, body, headers):
+
# test_create_container_already_exists
body = self.fixtures.load('list_container_objects_empty.json')
headers = self.base_headers
@@ -458,7 +499,9 @@ class CloudFilesMockHttp(MockHttp):
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):
+ 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')
@@ -466,7 +509,9 @@ class CloudFilesMockHttp(MockHttp):
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):
+ 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')
@@ -474,7 +519,9 @@ class CloudFilesMockHttp(MockHttp):
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):
+ 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')
@@ -482,12 +529,15 @@ class CloudFilesMockHttp(MockHttp):
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):
+ 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])
class CloudFilesMockRawResponse(MockRawResponse):
@@ -496,58 +546,85 @@ class CloudFilesMockRawResponse(MockRawR
base_headers = { 'content-type': 'application/json; charset=UTF-8'}
def __init__(self, *args, **kwargs):
- super(CloudFilesMockRawResponse, self).__init__(*args, **kwargs)
- self._data = []
- self._current_item = 0
+ super(CloudFilesMockRawResponse, self).__init__(*args, **kwargs)
+ self._data = []
+ self._current_item = 0
def next(self):
if self._current_item == len(self._data):
- raise StopIteration
+ raise StopIteration
value = self._data[self._current_item]
self._current_item += 1
return value
def _generate_random_data(self, size):
- data = []
- current_size = 0
- while current_size < size:
- value = str(random.randint(0, 9))
- value_size = len(value)
- data.append(value)
- current_size += value_size
+ data = []
+ current_size = 0
+ while current_size < size:
+ value = str(random.randint(0, 9))
+ value_size = len(value)
+ data.append(value)
+ current_size += value_size
- return data
+ return data
- def _v1_MossoCloudFS_foo_bar_container_foo_test_upload(self, method, url, body, headers):
+ def _v1_MossoCloudFS_foo_bar_container_foo_test_upload(
+ self, method, url, body, headers):
# test_object_upload_success
+
body = ''
- header = copy.deepcopy(self.base_headers)
+ headers = copy.deepcopy(self.base_headers)
+ headers.update(headers)
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):
+ def _v1_MossoCloudFS_foo_bar_container_foo_test_upload_INVALID_HASH(
+ self, method, url, body, headers):
+
# test_object_upload_invalid_hash
body = ''
headers = self.base_headers
return (httplib.UNPROCESSABLE_ENTITY, body, headers,
httplib.responses[httplib.OK])
- def _v1_MossoCloudFS_foo_bar_container_foo_bar_object(self, method, url, body, headers):
+ 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])
+ 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):
- 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])
+ 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):
- def _v1_MossoCloudFS_foo_bar_container_foo_test_stream_data(self, method, url, body, headers):
# test_upload_object_via_stream_success
body = 'test'
- return (httplib.OK, body, self.base_headers, httplib.responses[httplib.OK])
+ return (httplib.OK,
+ body,
+ self.base_headers,
+ httplib.responses[httplib.OK])
if __name__ == '__main__':
sys.exit(unittest.main())