libcloud-notifications mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From anthonys...@apache.org
Subject [21/28] libcloud git commit: Fix failing tests, format responding CSV
Date Tue, 07 Jun 2016 00:46:42 GMT
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 <anthonyshaw@apache.org>
Authored: Tue Jun 7 08:38:58 2016 +1000
Committer: anthony-shaw <anthonyshaw@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())


Mime
View raw message