You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@libcloud.apache.org by an...@apache.org on 2016/06/07 00:46:47 UTC
[26/28] libcloud git commit: undone file move
undone file move
Project: http://git-wip-us.apache.org/repos/asf/libcloud/repo
Commit: http://git-wip-us.apache.org/repos/asf/libcloud/commit/0106662b
Tree: http://git-wip-us.apache.org/repos/asf/libcloud/tree/0106662b
Diff: http://git-wip-us.apache.org/repos/asf/libcloud/diff/0106662b
Branch: refs/heads/trunk
Commit: 0106662b15ffa3d0f9d7f597cba13e4a35b22293
Parents: e04f62c
Author: anthony-shaw <an...@apache.org>
Authored: Tue Jun 7 09:40:39 2016 +1000
Committer: anthony-shaw <an...@apache.org>
Committed: Tue Jun 7 09:40:39 2016 +1000
----------------------------------------------------------------------
libcloud/data/test_backblaze_b2.py | 238 ------------------------
libcloud/test/storage/test_backblaze_b2.py | 238 ++++++++++++++++++++++++
2 files changed, 238 insertions(+), 238 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/libcloud/blob/0106662b/libcloud/data/test_backblaze_b2.py
----------------------------------------------------------------------
diff --git a/libcloud/data/test_backblaze_b2.py b/libcloud/data/test_backblaze_b2.py
deleted file mode 100644
index 60a20f4..0000000
--- a/libcloud/data/test_backblaze_b2.py
+++ /dev/null
@@ -1,238 +0,0 @@
-# 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.
-
-import os
-import sys
-import tempfile
-
-import mock
-
-from libcloud.storage.drivers.backblaze_b2 import BackblazeB2StorageDriver
-from libcloud.utils.py3 import httplib
-from libcloud.test import unittest
-from libcloud.test import StorageMockHttp
-from libcloud.test import MockRawResponse
-from libcloud.test import MockHttpTestCase
-from libcloud.test.file_fixtures import StorageFileFixtures
-
-
-class MockAuthConn(mock.Mock):
- account_id = 'abcdefgh'
-
-
-class BackblazeB2StorageDriverTestCase(unittest.TestCase):
- driver_klass = BackblazeB2StorageDriver
- driver_args = ('a', 'b')
-
- def setUp(self):
- self.driver_klass.connectionCls.authCls = MockAuthConn()
- self.driver_klass.connectionCls.conn_classes = (
- None, BackblazeB2MockHttp)
- self.driver_klass.connectionCls.rawResponseCls = \
- BackblazeB2MockRawResponse
- BackblazeB2MockHttp.type = None
- BackblazeB2MockRawResponse.type = None
- self.driver = self.driver_klass(*self.driver_args)
-
- def test_list_containers(self):
- containers = self.driver.list_containers()
- self.assertEqual(len(containers), 3)
- self.assertEqual(containers[0].name, 'test00001')
- self.assertEqual(containers[0].extra['id'], '481c37de2e1ab3bf5e150710')
- self.assertEqual(containers[0].extra['bucketType'], 'allPrivate')
-
- def test_list_container_objects(self):
- container = self.driver.list_containers()[0]
- objects = self.driver.list_container_objects(container=container)
- self.assertEqual(len(objects), 4)
- self.assertEqual(objects[0].name, '2.txt')
- self.assertEqual(objects[0].size, 2)
- self.assertEqual(objects[0].extra['fileId'], 'abcd')
- self.assertEqual(objects[0].extra['uploadTimestamp'], 1450545966000)
-
- def test_get_container(self):
- container = self.driver.get_container('test00001')
- self.assertEqual(container.name, 'test00001')
- self.assertEqual(container.extra['id'], '481c37de2e1ab3bf5e150710')
- self.assertEqual(container.extra['bucketType'], 'allPrivate')
-
- def test_get_object(self):
- obj = self.driver.get_object('test00001', '2.txt')
- self.assertEqual(obj.name, '2.txt')
- self.assertEqual(obj.size, 2)
- self.assertEqual(obj.extra['fileId'], 'abcd')
- self.assertEqual(obj.extra['uploadTimestamp'], 1450545966000)
-
- def test_create_container(self):
- container = self.driver.create_container(container_name='test0005')
- self.assertEqual(container.name, 'test0005')
- self.assertEqual(container.extra['id'], '681c87aebeaa530f5e250710')
- self.assertEqual(container.extra['bucketType'], 'allPrivate')
-
- def test_delete_container(self):
- container = self.driver.list_containers()[0]
- result = self.driver.delete_container(container=container)
- self.assertTrue(result)
-
- def test_download_object(self):
- container = self.driver.list_containers()[0]
- obj = self.driver.list_container_objects(container=container)[0]
- _, destination_path = tempfile.mkstemp()
- result = self.driver.download_object(obj=obj, destination_path=destination_path,
- overwrite_existing=True)
- self.assertTrue(result)
-
- def test_download_object_as_stream(self):
- container = self.driver.list_containers()[0]
- obj = self.driver.list_container_objects(container=container)[0]
- result = self.driver.download_object_as_stream(obj=obj)
- result = ''.join([x.decode('utf-8') for x in list(result)])
- self.assertEqual(result, 'ab')
-
- def test_upload_object(self):
- file_path = os.path.abspath(__file__)
- container = self.driver.list_containers()[0]
- obj = self.driver.upload_object(file_path=file_path, container=container,
- object_name='test0007.txt')
- self.assertEqual(obj.name, 'test0007.txt')
- self.assertEqual(obj.size, 24)
- self.assertEqual(obj.extra['fileId'], 'abcde')
-
- def test_upload_object_via_stream(self):
- container = self.driver.list_containers()[0]
- file_path = os.path.abspath(__file__)
- file = open(file_path, 'rb')
- iterator = iter(file)
- obj = self.driver.upload_object_via_stream(iterator=iterator,
- container=container,
- object_name='test0007.txt')
- self.assertEqual(obj.name, 'test0007.txt')
- self.assertEqual(obj.size, 24)
- self.assertEqual(obj.extra['fileId'], 'abcde')
-
- def test_delete_object(self):
- container = self.driver.list_containers()[0]
- obj = self.driver.list_container_objects(container=container)[0]
- result = self.driver.delete_object(obj=obj)
- self.assertTrue(result)
-
- def test_ex_hide_object(self):
- container = self.driver.list_containers()[0]
- container_id = container.extra['id']
- obj = self.driver.ex_hide_object(container_id=container_id,
- object_name='2.txt')
- self.assertEqual(obj.name, '2.txt')
-
- def test_ex_list_object_versions(self):
- container = self.driver.list_containers()[0]
- container_id = container.extra['id']
- objects = self.driver.ex_list_object_versions(container_id=container_id)
- self.assertEqual(len(objects), 9)
-
- def test_ex_get_upload_data(self):
- container = self.driver.list_containers()[0]
- container_id = container.extra['id']
- data = self.driver.ex_get_upload_data(container_id=container_id)
- self.assertEqual(data['authorizationToken'], 'nope')
- self.assertEqual(data['bucketId'], '481c37de2e1ab3bf5e150710')
- self.assertEqual(data['uploadUrl'], 'https://podxxx.backblaze.com/b2api/v1/b2_upload_file/abcd/defg')
-
- def test_ex_get_upload_url(self):
- container = self.driver.list_containers()[0]
- container_id = container.extra['id']
- url = self.driver.ex_get_upload_url(container_id=container_id)
- self.assertEqual(url, 'https://podxxx.backblaze.com/b2api/v1/b2_upload_file/abcd/defg')
-
-
-class BackblazeB2MockHttp(StorageMockHttp, MockHttpTestCase):
- fixtures = StorageFileFixtures('backblaze_b2')
-
- def _b2api_v1_b2_list_buckets(self, method, url, body, headers):
- if method == 'GET':
- body = self.fixtures.load('b2_list_buckets.json')
- else:
- raise AssertionError('Unsupported method')
- return (httplib.OK, body, {}, httplib.responses[httplib.OK])
-
- def _b2api_v1_b2_list_file_names(self, method, url, body, headers):
- if method == 'GET':
- body = self.fixtures.load('b2_list_file_names.json')
- else:
- raise AssertionError('Unsupported method')
- return (httplib.OK, body, {}, httplib.responses[httplib.OK])
-
- def _b2api_v1_b2_create_bucket(self, method, url, body, headers):
- if method == 'POST':
- body = self.fixtures.load('b2_create_bucket.json')
- else:
- raise AssertionError('Unsupported method')
- return (httplib.OK, body, {}, httplib.responses[httplib.OK])
-
- def _b2api_v1_b2_delete_bucket(self, method, url, body, headers):
- if method == 'POST':
- body = self.fixtures.load('b2_delete_bucket.json')
- else:
- raise AssertionError('Unsupported method')
- return (httplib.OK, body, {}, httplib.responses[httplib.OK])
-
- def _b2api_v1_b2_delete_file_version(self, method, url, body, headers):
- if method == 'POST':
- body = self.fixtures.load('b2_delete_file_version.json')
- else:
- raise AssertionError('Unsupported method')
- return (httplib.OK, body, {}, httplib.responses[httplib.OK])
-
- def _b2api_v1_b2_get_upload_url(self, method, url, body, headers):
- # test_upload_object
- if method == 'GET':
- body = self.fixtures.load('b2_get_upload_url.json')
- else:
- raise AssertionError('Unsupported method')
- return (httplib.OK, body, {}, httplib.responses[httplib.OK])
-
- def _b2api_v1_b2_upload_file_abcd_defg(self, method, url, body, headers):
- # test_upload_object
- if method == 'POST':
- body = self.fixtures.load('b2_upload_file.json')
- else:
- raise AssertionError('Unsupported method')
- return (httplib.OK, body, {}, httplib.responses[httplib.OK])
-
- def _b2api_v1_b2_list_file_versions(self, method, url, body, headers):
- if method == 'GET':
- body = self.fixtures.load('b2_list_file_versions.json')
- else:
- raise AssertionError('Unsupported method')
- return (httplib.OK, body, {}, httplib.responses[httplib.OK])
-
- def _b2api_v1_b2_hide_file(self, method, url, body, headers):
- if method == 'POST':
- body = self.fixtures.load('b2_hide_file.json')
- else:
- raise AssertionError('Unsupported method')
- return (httplib.OK, body, {}, httplib.responses[httplib.OK])
-
-
-class BackblazeB2MockRawResponse(MockRawResponse):
- def _file_test00001_2_txt(self, method, url, body, headers):
- # test_download_object
- if method == 'GET':
- body = 'ab'
- else:
- raise AssertionError('Unsupported method')
- return (httplib.OK, body, {}, httplib.responses[httplib.OK])
-
-if __name__ == '__main__':
- sys.exit(unittest.main())
http://git-wip-us.apache.org/repos/asf/libcloud/blob/0106662b/libcloud/test/storage/test_backblaze_b2.py
----------------------------------------------------------------------
diff --git a/libcloud/test/storage/test_backblaze_b2.py b/libcloud/test/storage/test_backblaze_b2.py
new file mode 100644
index 0000000..60a20f4
--- /dev/null
+++ b/libcloud/test/storage/test_backblaze_b2.py
@@ -0,0 +1,238 @@
+# 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.
+
+import os
+import sys
+import tempfile
+
+import mock
+
+from libcloud.storage.drivers.backblaze_b2 import BackblazeB2StorageDriver
+from libcloud.utils.py3 import httplib
+from libcloud.test import unittest
+from libcloud.test import StorageMockHttp
+from libcloud.test import MockRawResponse
+from libcloud.test import MockHttpTestCase
+from libcloud.test.file_fixtures import StorageFileFixtures
+
+
+class MockAuthConn(mock.Mock):
+ account_id = 'abcdefgh'
+
+
+class BackblazeB2StorageDriverTestCase(unittest.TestCase):
+ driver_klass = BackblazeB2StorageDriver
+ driver_args = ('a', 'b')
+
+ def setUp(self):
+ self.driver_klass.connectionCls.authCls = MockAuthConn()
+ self.driver_klass.connectionCls.conn_classes = (
+ None, BackblazeB2MockHttp)
+ self.driver_klass.connectionCls.rawResponseCls = \
+ BackblazeB2MockRawResponse
+ BackblazeB2MockHttp.type = None
+ BackblazeB2MockRawResponse.type = None
+ self.driver = self.driver_klass(*self.driver_args)
+
+ def test_list_containers(self):
+ containers = self.driver.list_containers()
+ self.assertEqual(len(containers), 3)
+ self.assertEqual(containers[0].name, 'test00001')
+ self.assertEqual(containers[0].extra['id'], '481c37de2e1ab3bf5e150710')
+ self.assertEqual(containers[0].extra['bucketType'], 'allPrivate')
+
+ def test_list_container_objects(self):
+ container = self.driver.list_containers()[0]
+ objects = self.driver.list_container_objects(container=container)
+ self.assertEqual(len(objects), 4)
+ self.assertEqual(objects[0].name, '2.txt')
+ self.assertEqual(objects[0].size, 2)
+ self.assertEqual(objects[0].extra['fileId'], 'abcd')
+ self.assertEqual(objects[0].extra['uploadTimestamp'], 1450545966000)
+
+ def test_get_container(self):
+ container = self.driver.get_container('test00001')
+ self.assertEqual(container.name, 'test00001')
+ self.assertEqual(container.extra['id'], '481c37de2e1ab3bf5e150710')
+ self.assertEqual(container.extra['bucketType'], 'allPrivate')
+
+ def test_get_object(self):
+ obj = self.driver.get_object('test00001', '2.txt')
+ self.assertEqual(obj.name, '2.txt')
+ self.assertEqual(obj.size, 2)
+ self.assertEqual(obj.extra['fileId'], 'abcd')
+ self.assertEqual(obj.extra['uploadTimestamp'], 1450545966000)
+
+ def test_create_container(self):
+ container = self.driver.create_container(container_name='test0005')
+ self.assertEqual(container.name, 'test0005')
+ self.assertEqual(container.extra['id'], '681c87aebeaa530f5e250710')
+ self.assertEqual(container.extra['bucketType'], 'allPrivate')
+
+ def test_delete_container(self):
+ container = self.driver.list_containers()[0]
+ result = self.driver.delete_container(container=container)
+ self.assertTrue(result)
+
+ def test_download_object(self):
+ container = self.driver.list_containers()[0]
+ obj = self.driver.list_container_objects(container=container)[0]
+ _, destination_path = tempfile.mkstemp()
+ result = self.driver.download_object(obj=obj, destination_path=destination_path,
+ overwrite_existing=True)
+ self.assertTrue(result)
+
+ def test_download_object_as_stream(self):
+ container = self.driver.list_containers()[0]
+ obj = self.driver.list_container_objects(container=container)[0]
+ result = self.driver.download_object_as_stream(obj=obj)
+ result = ''.join([x.decode('utf-8') for x in list(result)])
+ self.assertEqual(result, 'ab')
+
+ def test_upload_object(self):
+ file_path = os.path.abspath(__file__)
+ container = self.driver.list_containers()[0]
+ obj = self.driver.upload_object(file_path=file_path, container=container,
+ object_name='test0007.txt')
+ self.assertEqual(obj.name, 'test0007.txt')
+ self.assertEqual(obj.size, 24)
+ self.assertEqual(obj.extra['fileId'], 'abcde')
+
+ def test_upload_object_via_stream(self):
+ container = self.driver.list_containers()[0]
+ file_path = os.path.abspath(__file__)
+ file = open(file_path, 'rb')
+ iterator = iter(file)
+ obj = self.driver.upload_object_via_stream(iterator=iterator,
+ container=container,
+ object_name='test0007.txt')
+ self.assertEqual(obj.name, 'test0007.txt')
+ self.assertEqual(obj.size, 24)
+ self.assertEqual(obj.extra['fileId'], 'abcde')
+
+ def test_delete_object(self):
+ container = self.driver.list_containers()[0]
+ obj = self.driver.list_container_objects(container=container)[0]
+ result = self.driver.delete_object(obj=obj)
+ self.assertTrue(result)
+
+ def test_ex_hide_object(self):
+ container = self.driver.list_containers()[0]
+ container_id = container.extra['id']
+ obj = self.driver.ex_hide_object(container_id=container_id,
+ object_name='2.txt')
+ self.assertEqual(obj.name, '2.txt')
+
+ def test_ex_list_object_versions(self):
+ container = self.driver.list_containers()[0]
+ container_id = container.extra['id']
+ objects = self.driver.ex_list_object_versions(container_id=container_id)
+ self.assertEqual(len(objects), 9)
+
+ def test_ex_get_upload_data(self):
+ container = self.driver.list_containers()[0]
+ container_id = container.extra['id']
+ data = self.driver.ex_get_upload_data(container_id=container_id)
+ self.assertEqual(data['authorizationToken'], 'nope')
+ self.assertEqual(data['bucketId'], '481c37de2e1ab3bf5e150710')
+ self.assertEqual(data['uploadUrl'], 'https://podxxx.backblaze.com/b2api/v1/b2_upload_file/abcd/defg')
+
+ def test_ex_get_upload_url(self):
+ container = self.driver.list_containers()[0]
+ container_id = container.extra['id']
+ url = self.driver.ex_get_upload_url(container_id=container_id)
+ self.assertEqual(url, 'https://podxxx.backblaze.com/b2api/v1/b2_upload_file/abcd/defg')
+
+
+class BackblazeB2MockHttp(StorageMockHttp, MockHttpTestCase):
+ fixtures = StorageFileFixtures('backblaze_b2')
+
+ def _b2api_v1_b2_list_buckets(self, method, url, body, headers):
+ if method == 'GET':
+ body = self.fixtures.load('b2_list_buckets.json')
+ else:
+ raise AssertionError('Unsupported method')
+ return (httplib.OK, body, {}, httplib.responses[httplib.OK])
+
+ def _b2api_v1_b2_list_file_names(self, method, url, body, headers):
+ if method == 'GET':
+ body = self.fixtures.load('b2_list_file_names.json')
+ else:
+ raise AssertionError('Unsupported method')
+ return (httplib.OK, body, {}, httplib.responses[httplib.OK])
+
+ def _b2api_v1_b2_create_bucket(self, method, url, body, headers):
+ if method == 'POST':
+ body = self.fixtures.load('b2_create_bucket.json')
+ else:
+ raise AssertionError('Unsupported method')
+ return (httplib.OK, body, {}, httplib.responses[httplib.OK])
+
+ def _b2api_v1_b2_delete_bucket(self, method, url, body, headers):
+ if method == 'POST':
+ body = self.fixtures.load('b2_delete_bucket.json')
+ else:
+ raise AssertionError('Unsupported method')
+ return (httplib.OK, body, {}, httplib.responses[httplib.OK])
+
+ def _b2api_v1_b2_delete_file_version(self, method, url, body, headers):
+ if method == 'POST':
+ body = self.fixtures.load('b2_delete_file_version.json')
+ else:
+ raise AssertionError('Unsupported method')
+ return (httplib.OK, body, {}, httplib.responses[httplib.OK])
+
+ def _b2api_v1_b2_get_upload_url(self, method, url, body, headers):
+ # test_upload_object
+ if method == 'GET':
+ body = self.fixtures.load('b2_get_upload_url.json')
+ else:
+ raise AssertionError('Unsupported method')
+ return (httplib.OK, body, {}, httplib.responses[httplib.OK])
+
+ def _b2api_v1_b2_upload_file_abcd_defg(self, method, url, body, headers):
+ # test_upload_object
+ if method == 'POST':
+ body = self.fixtures.load('b2_upload_file.json')
+ else:
+ raise AssertionError('Unsupported method')
+ return (httplib.OK, body, {}, httplib.responses[httplib.OK])
+
+ def _b2api_v1_b2_list_file_versions(self, method, url, body, headers):
+ if method == 'GET':
+ body = self.fixtures.load('b2_list_file_versions.json')
+ else:
+ raise AssertionError('Unsupported method')
+ return (httplib.OK, body, {}, httplib.responses[httplib.OK])
+
+ def _b2api_v1_b2_hide_file(self, method, url, body, headers):
+ if method == 'POST':
+ body = self.fixtures.load('b2_hide_file.json')
+ else:
+ raise AssertionError('Unsupported method')
+ return (httplib.OK, body, {}, httplib.responses[httplib.OK])
+
+
+class BackblazeB2MockRawResponse(MockRawResponse):
+ def _file_test00001_2_txt(self, method, url, body, headers):
+ # test_download_object
+ if method == 'GET':
+ body = 'ab'
+ else:
+ raise AssertionError('Unsupported method')
+ return (httplib.OK, body, {}, httplib.responses[httplib.OK])
+
+if __name__ == '__main__':
+ sys.exit(unittest.main())