Fix failing tests, format responding CSV
Project: http://git-wip-us.apache.org/repos/asf/libcloud/repo Commit: http://git-wip-us.apache.org/repos/asf/libcloud/commit/5c27a927 Tree: http://git-wip-us.apache.org/repos/asf/libcloud/tree/5c27a927 Diff: http://git-wip-us.apache.org/repos/asf/libcloud/diff/5c27a927 Branch: refs/heads/trunk Commit: 5c27a9274e1d554c41b1efd8871b4941e185e387 Parents: 4d57bef Author: anthony-shaw <anthonys...@apache.org> Authored: Tue Jun 7 08:38:58 2016 +1000 Committer: anthony-shaw <anthonys...@apache.org> Committed: Tue Jun 7 08:38:58 2016 +1000 ---------------------------------------------------------------------- libcloud/common/base.py | 2 +- libcloud/compute/drivers/dimensiondata.py | 27 +-- libcloud/data/test_backblaze_b2.py | 238 +++++++++++++++++++++++ libcloud/test/compute/test_dimensiondata.py | 42 ++-- libcloud/test/storage/test_backblaze_b2.py | 238 ----------------------- 5 files changed, 279 insertions(+), 268 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/libcloud/blob/5c27a927/libcloud/common/base.py ---------------------------------------------------------------------- diff --git a/libcloud/common/base.py b/libcloud/common/base.py index 77d5152..540244d 100644 --- a/libcloud/common/base.py +++ b/libcloud/common/base.py @@ -806,7 +806,7 @@ class Connection(object): try: # @TODO: Should we just pass File object as body to request method # instead of dealing with splitting and sending the file ourselves? - if raw and method.upper() in ['POST', 'PUT']: + if raw: self.connection.putrequest(method, url, skip_host=1, skip_accept_encoding=1) http://git-wip-us.apache.org/repos/asf/libcloud/blob/5c27a927/libcloud/compute/drivers/dimensiondata.py ---------------------------------------------------------------------- diff --git a/libcloud/compute/drivers/dimensiondata.py b/libcloud/compute/drivers/dimensiondata.py index 55a3467..421ed38 100644 --- a/libcloud/compute/drivers/dimensiondata.py +++ b/libcloud/compute/drivers/dimensiondata.py @@ -44,7 +44,7 @@ from libcloud.common.dimensiondata import DimensionDataTag from libcloud.common.dimensiondata import API_ENDPOINTS, DEFAULT_REGION from libcloud.common.dimensiondata import TYPES_URN from libcloud.common.dimensiondata import SERVER_NS, NETWORK_NS, GENERAL_NS -from libcloud.utils.py3 import urlencode +from libcloud.utils.py3 import urlencode, ensure_string from libcloud.utils.xml import fixxpath, findtext, findall from libcloud.utils.py3 import basestring from libcloud.compute.types import NodeState, Provider @@ -2281,12 +2281,12 @@ class DimensionDataNodeDriver(NodeDriver): :param end_date: End date for the report :type end_date: ``str`` in format YYYY-MM-DD - :rtype: :class:`io.StringIO` + :rtype: ``list`` of ``list`` """ result = self.connection.raw_request_with_orgId_api_1( 'report/usage?startDate=%s&endDate=%s' % ( start_date, end_date)) - return result.response.body + return self._format_csv(result.response) def ex_detailed_usage_report(self, start_date, end_date): """ @@ -2298,12 +2298,12 @@ class DimensionDataNodeDriver(NodeDriver): :param end_date: End date for the report :type end_date: ``str`` in format YYYY-MM-DD - :rtype: :class:`io.StringIO` + :rtype: ``list`` of ``list`` """ result = self.connection.raw_request_with_orgId_api_1( 'report/usageDetailed?startDate=%s&endDate=%s' % ( start_date, end_date)) - return result.response.body + return self._format_csv(result.response) def ex_software_usage_report(self, start_date, end_date): """ @@ -2315,12 +2315,12 @@ class DimensionDataNodeDriver(NodeDriver): :param end_date: End date for the report :type end_date: ``str`` in format YYYY-MM-DD - :rtype: :class:`io.StringIO` + :rtype: ``list`` of ``list`` """ result = self.connection.raw_request_with_orgId_api_1( 'report/usageSoftwareUnits?startDate=%s&endDate=%s' % ( start_date, end_date)) - return result.response.body + return self._format_csv(result.response) def ex_audit_log_report(self, start_date, end_date): """ @@ -2332,12 +2332,12 @@ class DimensionDataNodeDriver(NodeDriver): :param end_date: End date for the report :type end_date: ``str`` in format YYYY-MM-DD - :rtype: :class:`io.StringIO` + :rtype: ``list`` of ``list`` """ result = self.connection.raw_request_with_orgId_api_1( 'report/usageSoftwareUnits?startDate=%s&endDate=%s' % ( start_date, end_date)) - return result.response.body + return self._format_csv(result.response) def ex_backup_usage_report(self, start_date, end_date, location): """ @@ -2353,13 +2353,18 @@ class DimensionDataNodeDriver(NodeDriver): located in this location :type location: :class:`NodeLocation` or ``str`` - :rtype: :class:`io.StringIO` + :rtype: ``list`` of ``list`` """ datacenter_id = self._location_to_location_id(location) result = self.connection.raw_request_with_orgId_api_1( 'backup/detailedUsageReport?datacenterId=%s&fromDate=%s&toDate=%s' % (datacenter_id, start_date, end_date)) - return result.response.body + return self._format_csv(result.response) + + def _format_csv(self, http_response): + text = http_response.read() + lines = str.splitlines(ensure_string(text)) + return [line.split(',') for line in lines] @staticmethod def _get_tagging_asset_type(asset): http://git-wip-us.apache.org/repos/asf/libcloud/blob/5c27a927/libcloud/data/test_backblaze_b2.py ---------------------------------------------------------------------- diff --git a/libcloud/data/test_backblaze_b2.py b/libcloud/data/test_backblaze_b2.py new file mode 100644 index 0000000..60a20f4 --- /dev/null +++ b/libcloud/data/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()) http://git-wip-us.apache.org/repos/asf/libcloud/blob/5c27a927/libcloud/test/compute/test_dimensiondata.py ---------------------------------------------------------------------- diff --git a/libcloud/test/compute/test_dimensiondata.py b/libcloud/test/compute/test_dimensiondata.py index 5958c27..d20cd0b 100644 --- a/libcloud/test/compute/test_dimensiondata.py +++ b/libcloud/test/compute/test_dimensiondata.py @@ -29,7 +29,7 @@ from libcloud.common.dimensiondata import DimensionDataTag, DimensionDataTagKey from libcloud.common.dimensiondata import TYPES_URN from libcloud.compute.drivers.dimensiondata import DimensionDataNodeDriver as DimensionData from libcloud.compute.base import Node, NodeAuthPassword, NodeLocation -from libcloud.test import MockHttp, unittest +from libcloud.test import MockHttp, unittest, MockRawResponse, StorageMockHttp from libcloud.test.compute import TestCaseMixin from libcloud.test.file_fixtures import ComputeFileFixtures from libcloud.test.secrets import DIMENSIONDATA_PARAMS @@ -40,6 +40,8 @@ class DimensionDataTests(unittest.TestCase, TestCaseMixin): def setUp(self): DimensionData.connectionCls.conn_classes = (None, DimensionDataMockHttp) + DimensionData.connectionCls.rawResponseCls = \ + DimensionDataMockRawResponse DimensionDataMockHttp.type = None self.driver = DimensionData(*DIMENSIONDATA_PARAMS) @@ -1135,15 +1137,15 @@ class DimensionDataTests(unittest.TestCase, TestCaseMixin): def test_summary_usage_report(self): report = self.driver.ex_summary_usage_report('2016-06-01', '2016-06-30') - report_content = report.readlines() + report_content = report self.assertEqual(len(report_content), 13) - self.assertEqual(len(report_content[0]), 67) + self.assertEqual(len(report_content[0]), 6) def test_detailed_usage_report(self): report = self.driver.ex_detailed_usage_report('2016-06-01', '2016-06-30') - report_content = report.readlines() + report_content = report self.assertEqual(len(report_content), 42) - self.assertEqual(len(report_content[0]), 30) + self.assertEqual(len(report_content[0]), 4) class InvalidRequestError(Exception): @@ -1151,7 +1153,23 @@ class InvalidRequestError(Exception): super(InvalidRequestError, self).__init__("Invalid Request - %s" % tag) -class DimensionDataMockHttp(MockHttp): +class DimensionDataMockRawResponse(MockRawResponse): + fixtures = ComputeFileFixtures('dimensiondata') + + def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_report_usage(self, method, url, body, headers): + body = self.fixtures.load( + 'summary_usage_report.csv' + ) + return (httplib.BAD_REQUEST, body, {}, httplib.responses[httplib.OK]) + + def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_report_usageDetailed(self, method, url, body, headers): + body = self.fixtures.load( + 'detailed_usage_report.csv' + ) + return (httplib.BAD_REQUEST, body, {}, httplib.responses[httplib.OK]) + + +class DimensionDataMockHttp(StorageMockHttp, MockHttp): fixtures = ComputeFileFixtures('dimensiondata') @@ -1317,18 +1335,6 @@ class DimensionDataMockHttp(MockHttp): ) return (httplib.BAD_REQUEST, body, {}, httplib.responses[httplib.OK]) - def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_report_usage(self, method, url, body, headers): - body = self.fixtures.load( - 'summary_usage_report.csv' - ) - return (httplib.BAD_REQUEST, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_report_usageDetailed(self, method, url, body, headers): - body = self.fixtures.load( - 'detailed_usage_report.csv' - ) - return (httplib.BAD_REQUEST, body, {}, httplib.responses[httplib.OK]) - def _caas_2_2_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server(self, method, url, body, headers): body = self.fixtures.load( 'server.xml') http://git-wip-us.apache.org/repos/asf/libcloud/blob/5c27a927/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 deleted file mode 100644 index eea9f11..0000000 --- a/libcloud/test/storage/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())
