Author: tomaz
Date: Thu Oct 6 11:15:30 2011
New Revision: 1179575
URL: http://svn.apache.org/viewvc?rev=1179575&view=rev
Log:
Refactor S3 storage tests so other test classes can inherit from it.
Modified:
libcloud/trunk/test/storage/test_s3.py
Modified: libcloud/trunk/test/storage/test_s3.py
URL:
http://svn.apache.org/viewvc/libcloud/trunk/test/storage/test_s3.py?rev=1179575&r1=1179574&r2=1179575&view=diff
==============================================================================
--- libcloud/trunk/test/storage/test_s3.py (original)
+++ libcloud/trunk/test/storage/test_s3.py Thu Oct 6 11:15:30 2011
@@ -34,15 +34,214 @@ from libcloud.storage.drivers.dummy impo
from test import StorageMockHttp, MockRawResponse # pylint: disable-msg=E0611
from test.file_fixtures import StorageFileFixtures # pylint: disable-msg=E0611
+from test.secrets import STORAGE_S3_PARAMS
+
+class S3MockHttp(StorageMockHttp):
+
+ fixtures = StorageFileFixtures('s3')
+ base_headers = {}
+
+ def _UNAUTHORIZED(self, method, url, body, headers):
+ return (httplib.UNAUTHORIZED,
+ '',
+ self.base_headers,
+ httplib.responses[httplib.OK])
+
+ def _DIFFERENT_REGION(self, method, url, body, headers):
+ return (httplib.MOVED_PERMANENTLY,
+ '',
+ self.base_headers,
+ httplib.responses[httplib.OK])
+
+ def _list_containers_EMPTY(self, method, url, body, headers):
+ body = self.fixtures.load('list_containers_empty.xml')
+ return (httplib.OK,
+ body,
+ self.base_headers,
+ httplib.responses[httplib.OK])
+
+ def _list_containers(self, method, url, body, headers):
+ body = self.fixtures.load('list_containers.xml')
+ return (httplib.OK,
+ body,
+ self.base_headers,
+ httplib.responses[httplib.OK])
+
+ def _test_container_EMPTY(self, method, url, body, headers):
+ body = self.fixtures.load('list_container_objects_empty.xml')
+ return (httplib.OK,
+ body,
+ self.base_headers,
+ httplib.responses[httplib.OK])
+
+ def _test_container(self, method, url, body, headers):
+ body = self.fixtures.load('list_container_objects.xml')
+ return (httplib.OK,
+ body,
+ self.base_headers,
+ httplib.responses[httplib.OK])
+
+ def _test_container_ITERATOR(self, method, url, body, headers):
+ if url.find('3.zip') == -1:
+ # First part of the response (first 3 objects)
+ body =
self.fixtures.load('list_container_objects_not_exhausted1.xml')
+ else:
+ body =
self.fixtures.load('list_container_objects_not_exhausted2.xml')
+
+ return (httplib.OK,
+ body,
+ self.base_headers,
+ httplib.responses[httplib.OK])
+
+
+ def _test2_test_list_containers(self, method, url, body, headers):
+ # test_get_object
+ body = self.fixtures.load('list_containers.xml')
+ headers = { 'content-type': 'application/zip',
+ 'etag': '"e31208wqsdoj329jd"',
+ 'content-length': 12345,
+ }
+
+ return (httplib.OK,
+ body,
+ headers,
+ httplib.responses[httplib.OK])
+
+ def _new_container_INVALID_NAME(self, method, url, body, headers):
+ # test_create_container
+ return (httplib.BAD_REQUEST,
+ body,
+ headers,
+ httplib.responses[httplib.OK])
+
+ def _new_container_ALREADY_EXISTS(self, method, url, body, headers):
+ # test_create_container
+ return (httplib.CONFLICT,
+ body,
+ headers,
+ httplib.responses[httplib.OK])
+
+ def _new_container(self, method, url, body, headers):
+ # test_create_container, test_delete_container
+
+ if method == 'PUT':
+ status = httplib.OK
+ elif method == 'DELETE':
+ status = httplib.NO_CONTENT
+
+ return (status,
+ body,
+ headers,
+ httplib.responses[httplib.OK])
+
+ def _new_container_DOESNT_EXIST(self, method, url, body, headers):
+ # test_delete_container
+ return (httplib.NOT_FOUND,
+ body,
+ headers,
+ httplib.responses[httplib.OK])
+
+ def _new_container_NOT_EMPTY(self, method, url, body, headers):
+ # test_delete_container
+ return (httplib.CONFLICT,
+ body,
+ headers,
+ httplib.responses[httplib.OK])
+
+ def _foo_bar_container(self, method, url, body, headers):
+ # test_delete_container
+ return (httplib.NO_CONTENT,
+ body,
+ headers,
+ httplib.responses[httplib.OK])
+
+ def _foo_bar_container_NOT_FOUND(self, method, url, body, headers):
+ # test_delete_container_not_found
+ return (httplib.NOT_FOUND,
+ body,
+ headers,
+ httplib.responses[httplib.OK])
+
+ def _foo_bar_container_foo_bar_object_NOT_FOUND(self, method, url, body,
headers):
+ # test_delete_object_not_found
+ return (httplib.NOT_FOUND,
+ body,
+ headers,
+ httplib.responses[httplib.OK])
+
+ def _foo_bar_container_foo_bar_object(self, method, url, body, headers):
+ # test_delete_object
+ return (httplib.NO_CONTENT,
+ body,
+ headers,
+ httplib.responses[httplib.OK])
+
+class S3MockRawResponse(MockRawResponse):
+
+ fixtures = StorageFileFixtures('s3')
+
+ def _foo_bar_container_foo_bar_object(self, method, url, body, headers):
+ # test_download_object_success
+ body = ''
+ self._data = self._generate_random_data(1000)
+ return (httplib.OK,
+ body,
+ headers,
+ httplib.responses[httplib.OK])
+
+ def _foo_bar_container_foo_test_upload_INVALID_HASH1(self, method, url,
body, headers):
+ body = ''
+ headers = {}
+ headers['etag'] = '"foobar"'
+ # test_upload_object_invalid_hash1
+ return (httplib.OK,
+ body,
+ headers,
+ httplib.responses[httplib.OK])
+
+ def _foo_bar_container_foo_test_upload_INVALID_HASH2(self, method, url,
body, headers):
+ # test_upload_object_invalid_hash2
+ body = ''
+ headers = { 'etag': '"hash343hhash89h932439jsaa89"'}
+ return (httplib.OK,
+ body,
+ headers,
+ httplib.responses[httplib.OK])
+
+ def _foo_bar_container_foo_test_upload(self, method, url, body, headers):
+ # test_upload_object_success
+ body = ''
+ headers = { 'etag': '"0cc175b9c0f1b6a831c399e269772661"'}
+ return (httplib.OK,
+ body,
+ headers,
+ httplib.responses[httplib.OK])
+
+ def _foo_bar_container_foo_bar_object_INVALID_SIZE(self, method, url,
body, headers):
+ # test_upload_object_invalid_file_size
+ body = ''
+ return (httplib.OK,
+ body,
+ headers,
+ httplib.responses[httplib.OK])
+
class S3Tests(unittest.TestCase):
+ driver_type = S3StorageDriver
+ driver_args = STORAGE_S3_PARAMS
+ mock_response_klass = S3MockHttp
+ mock_raw_response_klass = S3MockRawResponse
+
+ @classmethod
+ def create_driver(self):
+ return self.driver_type(*self.driver_args)
def setUp(self):
- S3StorageDriver.connectionCls.conn_classes = (None, S3MockHttp)
- S3StorageDriver.connectionCls.rawResponseCls = S3MockRawResponse
- S3MockHttp.type = None
- S3MockRawResponse.type = None
- self.driver = S3StorageDriver('dummy', 'dummy')
+ self.driver_type.connectionCls.conn_classes = (None,
self.mock_response_klass)
+ self.driver_type.connectionCls.rawResponseCls =
self.mock_raw_response_klass
+ self.mock_response_klass.type = None
+ self.mock_raw_response_klass.type = None
+ self.driver = self.create_driver()
def tearDown(self):
self._remove_test_file()
@@ -56,7 +255,7 @@ class S3Tests(unittest.TestCase):
pass
def test_invalid_credentials(self):
- S3MockHttp.type = 'UNAUTHORIZED'
+ self.mock_response_klass.type = 'UNAUTHORIZED'
try:
self.driver.list_containers()
except InvalidCredsError, e:
@@ -65,7 +264,7 @@ class S3Tests(unittest.TestCase):
self.fail('Exception was not thrown')
def test_bucket_is_located_in_different_region(self):
- S3MockHttp.type = 'DIFFERENT_REGION'
+ self.mock_response_klass.type = 'DIFFERENT_REGION'
try:
self.driver.list_containers()
except LibcloudError:
@@ -74,26 +273,26 @@ class S3Tests(unittest.TestCase):
self.fail('Exception was not thrown')
def test_list_containers_empty(self):
- S3MockHttp.type = 'list_containers_EMPTY'
+ self.mock_response_klass.type = 'list_containers_EMPTY'
containers = self.driver.list_containers()
self.assertEqual(len(containers), 0)
def test_list_containers_success(self):
- S3MockHttp.type = 'list_containers'
+ self.mock_response_klass.type = 'list_containers'
containers = self.driver.list_containers()
self.assertEqual(len(containers), 2)
self.assertTrue('creation_date' in containers[1].extra)
def test_list_container_objects_empty(self):
- S3MockHttp.type = 'EMPTY'
+ self.mock_response_klass.type = 'EMPTY'
container = Container(name='test_container', extra={},
driver=self.driver)
objects = self.driver.list_container_objects(container=container)
self.assertEqual(len(objects), 0)
def test_list_container_objects_success(self):
- S3MockHttp.type = None
+ self.mock_response_klass.type = None
container = Container(name='test_container', extra={},
driver=self.driver)
objects = self.driver.list_container_objects(container=container)
@@ -106,7 +305,7 @@ class S3Tests(unittest.TestCase):
self.assertTrue('owner' in obj.meta_data)
def test_list_container_objects_iterator_has_more(self):
- S3MockHttp.type = 'ITERATOR'
+ self.mock_response_klass.type = 'ITERATOR'
container = Container(name='test_container', extra={},
driver=self.driver)
objects = self.driver.list_container_objects(container=container)
@@ -120,7 +319,7 @@ class S3Tests(unittest.TestCase):
self.assertEqual(len(objects), 5)
def test_get_container_doesnt_exist(self):
- S3MockHttp.type = 'list_containers'
+ self.mock_response_klass.type = 'list_containers'
try:
self.driver.get_container(container_name='container1')
except ContainerDoesNotExistError:
@@ -129,14 +328,14 @@ class S3Tests(unittest.TestCase):
self.fail('Exception was not thrown')
def test_get_container_success(self):
- S3MockHttp.type = 'list_containers'
+ self.mock_response_klass.type = 'list_containers'
container = self.driver.get_container(container_name='test1')
self.assertTrue(container.name, 'test1')
def test_get_object_container_doesnt_exist(self):
# This method makes two requests which makes mocking the response a bit
# trickier
- S3MockHttp.type = 'list_containers'
+ self.mock_response_klass.type = 'list_containers'
try:
self.driver.get_object(container_name='test-inexistent',
object_name='test')
@@ -148,7 +347,7 @@ class S3Tests(unittest.TestCase):
def test_get_object_success(self):
# This method makes two requests which makes mocking the response a bit
# trickier
- S3MockHttp.type = 'list_containers'
+ self.mock_response_klass.type = 'list_containers'
obj = self.driver.get_object(container_name='test2',
object_name='test')
@@ -159,7 +358,7 @@ class S3Tests(unittest.TestCase):
def test_create_container_invalid_name(self):
# invalid container name
- S3MockHttp.type = 'INVALID_NAME'
+ self.mock_response_klass.type = 'INVALID_NAME'
try:
self.driver.create_container(container_name='new_container')
except InvalidContainerNameError:
@@ -169,7 +368,7 @@ class S3Tests(unittest.TestCase):
def test_create_container_already_exists(self):
# container with this name already exists
- S3MockHttp.type = 'ALREADY_EXISTS'
+ self.mock_response_klass.type = 'ALREADY_EXISTS'
try:
self.driver.create_container(container_name='new-container')
except InvalidContainerNameError:
@@ -179,13 +378,13 @@ class S3Tests(unittest.TestCase):
def test_create_container_success(self):
# success
- S3MockHttp.type = None
+ self.mock_response_klass.type = None
container =
self.driver.create_container(container_name='new_container')
self.assertEqual(container.name, 'new_container')
def test_delete_container_doesnt_exist(self):
container = Container(name='new_container', extra=None, driver=self)
- S3MockHttp.type = 'DOESNT_EXIST'
+ self.mock_response_klass.type = 'DOESNT_EXIST'
try:
self.driver.delete_container(container=container)
except ContainerDoesNotExistError:
@@ -195,7 +394,7 @@ class S3Tests(unittest.TestCase):
def test_delete_container_not_empty(self):
container = Container(name='new_container', extra=None, driver=self)
- S3MockHttp.type = 'NOT_EMPTY'
+ self.mock_response_klass.type = 'NOT_EMPTY'
try:
self.driver.delete_container(container=container)
except ContainerIsNotEmptyError:
@@ -204,11 +403,11 @@ class S3Tests(unittest.TestCase):
self.fail('Exception was not thrown')
# success
- S3MockHttp.type = None
+ self.mock_response_klass.type = None
self.assertTrue(self.driver.delete_container(container=container))
def test_delete_container_not_found(self):
- S3MockHttp.type = 'NOT_FOUND'
+ self.mock_response_klass.type = 'NOT_FOUND'
container = Container(name='foo_bar_container', extra={}, driver=self)
try:
self.driver.delete_container(container=container)
@@ -218,7 +417,7 @@ class S3Tests(unittest.TestCase):
self.fail('Container does not exist but an exception was not
thrown')
def test_delete_container_success(self):
- S3MockHttp.type = None
+ self.mock_response_klass.type = None
container = Container(name='new_container', extra=None, driver=self)
self.assertTrue(self.driver.delete_container(container=container))
@@ -235,7 +434,7 @@ class S3Tests(unittest.TestCase):
self.assertTrue(result)
def test_download_object_invalid_file_size(self):
- S3MockRawResponse.type = 'INVALID_SIZE'
+ self.mock_raw_response_klass.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,
@@ -248,7 +447,7 @@ class S3Tests(unittest.TestCase):
self.assertFalse(result)
def test_download_object_invalid_file_already_exists(self):
- S3MockRawResponse.type = 'INVALID_SIZE'
+ self.mock_raw_response_klass.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,
@@ -298,7 +497,7 @@ class S3Tests(unittest.TestCase):
calculate_hash=True):
return True, 'hash343hhash89h932439jsaa89', 1000
- S3MockRawResponse.type = 'INVALID_HASH1'
+ self.mock_raw_response_klass.type = 'INVALID_HASH1'
old_func = S3StorageDriver._upload_file
S3StorageDriver._upload_file = upload_file
@@ -324,7 +523,7 @@ class S3Tests(unittest.TestCase):
calculate_hash=True):
return True, '0cc175b9c0f1b6a831c399e269772661', 1000
- S3MockRawResponse.type = 'INVALID_HASH2'
+ self.mock_raw_response_klass.type = 'INVALID_HASH2'
old_func = S3StorageDriver._upload_file
S3StorageDriver._upload_file = upload_file
@@ -377,7 +576,7 @@ class S3Tests(unittest.TestCase):
self.fail('Exception was not thrown')
def test_delete_object_not_found(self):
- S3MockHttp.type = 'NOT_FOUND'
+ self.mock_response_klass.type = 'NOT_FOUND'
container = Container(name='foo_bar_container', extra={}, driver=self)
obj = Object(name='foo_bar_object', size=1234, hash=None, extra=None,
meta_data=None, container=container, driver=self.driver)
@@ -397,225 +596,16 @@ class S3Tests(unittest.TestCase):
self.assertTrue(result)
class S3USWestTests(S3Tests):
- def setUp(self):
- S3USWestStorageDriver.connectionCls.conn_classes = (None, S3MockHttp)
- S3USWestStorageDriver.connectionCls.rawResponseCls = S3MockRawResponse
- S3MockHttp.type = None
- S3MockRawResponse.type = None
- self.driver = S3USWestStorageDriver('dummy', 'dummy')
+ driver_type = S3USWestStorageDriver
class S3EUWestTests(S3Tests):
- def setUp(self):
- S3EUWestStorageDriver.connectionCls.conn_classes = (None, S3MockHttp)
- S3EUWestStorageDriver.connectionCls.rawResponseCls = S3MockRawResponse
- S3MockHttp.type = None
- S3MockRawResponse.type = None
- self.driver = S3EUWestStorageDriver('dummy', 'dummy')
+ driver_type = S3EUWestStorageDriver
class S3APSETests(S3Tests):
- def setUp(self):
- S3APSEStorageDriver.connectionCls.conn_classes = (None, S3MockHttp)
- S3APSEStorageDriver.connectionCls.rawResponseCls = S3MockRawResponse
- S3MockHttp.type = None
- S3MockRawResponse.type = None
- self.driver = S3APSEStorageDriver('dummy', 'dummy')
+ driver_type = S3APSEStorageDriver
class S3APNETests(S3Tests):
- def setUp(self):
- S3APNEStorageDriver.connectionCls.conn_classes = (None, S3MockHttp)
- S3APNEStorageDriver.connectionCls.rawResponseCls = S3MockRawResponse
- S3MockHttp.type = None
- S3MockRawResponse.type = None
- self.driver = S3APNEStorageDriver('dummy', 'dummy')
-
-class S3MockHttp(StorageMockHttp):
-
- fixtures = StorageFileFixtures('s3')
- base_headers = {}
-
- def _UNAUTHORIZED(self, method, url, body, headers):
- return (httplib.UNAUTHORIZED,
- '',
- self.base_headers,
- httplib.responses[httplib.OK])
-
- def _DIFFERENT_REGION(self, method, url, body, headers):
- return (httplib.MOVED_PERMANENTLY,
- '',
- self.base_headers,
- httplib.responses[httplib.OK])
-
- def _list_containers_EMPTY(self, method, url, body, headers):
- body = self.fixtures.load('list_containers_empty.xml')
- return (httplib.OK,
- body,
- self.base_headers,
- httplib.responses[httplib.OK])
-
- def _list_containers(self, method, url, body, headers):
- body = self.fixtures.load('list_containers.xml')
- return (httplib.OK,
- body,
- self.base_headers,
- httplib.responses[httplib.OK])
-
- def _test_container_EMPTY(self, method, url, body, headers):
- body = self.fixtures.load('list_container_objects_empty.xml')
- return (httplib.OK,
- body,
- self.base_headers,
- httplib.responses[httplib.OK])
-
- def _test_container(self, method, url, body, headers):
- body = self.fixtures.load('list_container_objects.xml')
- return (httplib.OK,
- body,
- self.base_headers,
- httplib.responses[httplib.OK])
-
- def _test_container_ITERATOR(self, method, url, body, headers):
- if url.find('3.zip') == -1:
- # First part of the response (first 3 objects)
- body =
self.fixtures.load('list_container_objects_not_exhausted1.xml')
- else:
- body =
self.fixtures.load('list_container_objects_not_exhausted2.xml')
-
- return (httplib.OK,
- body,
- self.base_headers,
- httplib.responses[httplib.OK])
-
-
- def _test2_test_list_containers(self, method, url, body, headers):
- # test_get_object
- body = self.fixtures.load('list_containers.xml')
- headers = { 'content-type': 'application/zip',
- 'etag': '"e31208wqsdoj329jd"',
- 'content-length': 12345,
- }
-
- return (httplib.OK,
- body,
- headers,
- httplib.responses[httplib.OK])
-
- def _new_container_INVALID_NAME(self, method, url, body, headers):
- # test_create_container
- return (httplib.BAD_REQUEST,
- body,
- headers,
- httplib.responses[httplib.OK])
-
- def _new_container_ALREADY_EXISTS(self, method, url, body, headers):
- # test_create_container
- return (httplib.CONFLICT,
- body,
- headers,
- httplib.responses[httplib.OK])
-
- def _new_container(self, method, url, body, headers):
- # test_create_container, test_delete_container
-
- if method == 'PUT':
- status = httplib.OK
- elif method == 'DELETE':
- status = httplib.NO_CONTENT
-
- return (status,
- body,
- headers,
- httplib.responses[httplib.OK])
-
- def _new_container_DOESNT_EXIST(self, method, url, body, headers):
- # test_delete_container
- return (httplib.NOT_FOUND,
- body,
- headers,
- httplib.responses[httplib.OK])
-
- def _new_container_NOT_EMPTY(self, method, url, body, headers):
- # test_delete_container
- return (httplib.CONFLICT,
- body,
- headers,
- httplib.responses[httplib.OK])
-
- def _foo_bar_container(self, method, url, body, headers):
- # test_delete_container
- return (httplib.NO_CONTENT,
- body,
- headers,
- httplib.responses[httplib.OK])
-
- def _foo_bar_container_NOT_FOUND(self, method, url, body, headers):
- # test_delete_container_not_found
- return (httplib.NOT_FOUND,
- body,
- headers,
- httplib.responses[httplib.OK])
-
- def _foo_bar_container_foo_bar_object_NOT_FOUND(self, method, url, body,
headers):
- # test_delete_object_not_found
- return (httplib.NOT_FOUND,
- body,
- headers,
- httplib.responses[httplib.OK])
-
- def _foo_bar_container_foo_bar_object(self, method, url, body, headers):
- # test_delete_object
- return (httplib.NO_CONTENT,
- body,
- headers,
- httplib.responses[httplib.OK])
-
-class S3MockRawResponse(MockRawResponse):
-
- fixtures = StorageFileFixtures('s3')
-
- def _foo_bar_container_foo_bar_object(self, method, url, body, headers):
- # test_download_object_success
- body = ''
- self._data = self._generate_random_data(1000)
- return (httplib.OK,
- body,
- headers,
- httplib.responses[httplib.OK])
-
- def _foo_bar_container_foo_test_upload_INVALID_HASH1(self, method, url,
body, headers):
- body = ''
- headers = {}
- headers['etag'] = '"foobar"'
- # test_upload_object_invalid_hash1
- return (httplib.OK,
- body,
- headers,
- httplib.responses[httplib.OK])
-
- def _foo_bar_container_foo_test_upload_INVALID_HASH2(self, method, url,
body, headers):
- # test_upload_object_invalid_hash2
- body = ''
- headers = { 'etag': '"hash343hhash89h932439jsaa89"'}
- return (httplib.OK,
- body,
- headers,
- httplib.responses[httplib.OK])
-
- def _foo_bar_container_foo_test_upload(self, method, url, body, headers):
- # test_upload_object_success
- body = ''
- headers = { 'etag': '"0cc175b9c0f1b6a831c399e269772661"'}
- return (httplib.OK,
- body,
- headers,
- httplib.responses[httplib.OK])
-
- def _foo_bar_container_foo_bar_object_INVALID_SIZE(self, method, url,
body, headers):
- # test_upload_object_invalid_file_size
- body = ''
- return (httplib.OK,
- body,
- headers,
- httplib.responses[httplib.OK])
+ driver_tyoe = S3APNEStorageDriver
if __name__ == '__main__':
sys.exit(unittest.main())
