| # 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 json |
| import tempfile |
| from io import BytesIO |
| |
| from libcloud.test import generate_random_data # pylint: disable-msg=E0611 |
| from libcloud.test import unittest |
| from libcloud.utils.py3 import b, httplib, parse_qs, urlparse, basestring |
| from libcloud.common.types import LibcloudError, InvalidCredsError |
| from libcloud.storage.base import Object, Container |
| from libcloud.test.secrets import STORAGE_AZURE_BLOBS_PARAMS, STORAGE_AZURITE_BLOBS_PARAMS |
| from libcloud.storage.types import ( |
| ObjectDoesNotExistError, |
| ObjectHashMismatchError, |
| ContainerIsNotEmptyError, |
| InvalidContainerNameError, |
| ContainerDoesNotExistError, |
| ContainerAlreadyExistsError, |
| ) |
| from libcloud.test.storage.base import BaseRangeDownloadMockHttp |
| from libcloud.test.file_fixtures import StorageFileFixtures # pylint: disable-msg=E0611 |
| from libcloud.storage.drivers.azure_blobs import ( |
| AZURE_UPLOAD_CHUNK_SIZE, |
| AzureBlobsStorageDriver, |
| AzureBlobsActiveDirectoryConnection, |
| ) |
| |
| |
| class AzureBlobsMockHttp(BaseRangeDownloadMockHttp, unittest.TestCase): |
| |
| fixtures = StorageFileFixtures("azure_blobs") |
| base_headers = {} |
| |
| # Note: using this method to get the oauth key for azure ad authentication |
| def __getattr__(self, n): |
| def fn(method, url, body, headers): |
| fixture = self.fixtures.load(n + ".json") |
| |
| if method in ("POST", "PUT"): |
| try: |
| body = json.loads(body) |
| fixture_tmp = json.loads(fixture) |
| fixture_tmp = self._update(fixture_tmp, body) |
| fixture = json.dumps(fixture_tmp) |
| except ValueError: |
| pass |
| |
| return (httplib.OK, fixture, headers, httplib.responses[httplib.OK]) |
| |
| return fn |
| |
| def _UNAUTHORIZED(self, method, url, body, headers): |
| return ( |
| httplib.UNAUTHORIZED, |
| "", |
| self.base_headers, |
| httplib.responses[httplib.UNAUTHORIZED], |
| ) |
| |
| 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): |
| query_string = urlparse.urlsplit(url).query |
| query = parse_qs(query_string) |
| |
| if "marker" not in query: |
| body = self.fixtures.load("list_containers_1.xml") |
| else: |
| body = self.fixtures.load("list_containers_2.xml") |
| |
| return (httplib.OK, body, self.base_headers, httplib.responses[httplib.OK]) |
| |
| def _test_container_EMPTY(self, method, url, body, headers): |
| if method == "DELETE": |
| body = "" |
| return ( |
| httplib.ACCEPTED, |
| body, |
| self.base_headers, |
| httplib.responses[httplib.ACCEPTED], |
| ) |
| |
| else: |
| body = self.fixtures.load("list_objects_empty.xml") |
| return (httplib.OK, body, self.base_headers, httplib.responses[httplib.OK]) |
| |
| def _new__container_INVALID_NAME(self, method, url, body, headers): |
| return ( |
| httplib.BAD_REQUEST, |
| body, |
| self.base_headers, |
| httplib.responses[httplib.BAD_REQUEST], |
| ) |
| |
| def _test_container(self, method, url, body, headers): |
| query_string = urlparse.urlsplit(url).query |
| query = parse_qs(query_string) |
| |
| if "marker" not in query: |
| body = self.fixtures.load("list_objects_1.xml") |
| else: |
| body = self.fixtures.load("list_objects_2.xml") |
| |
| return (httplib.OK, body, self.base_headers, httplib.responses[httplib.OK]) |
| |
| def _test_container100(self, method, url, body, headers): |
| body = "" |
| |
| if method != "HEAD": |
| return ( |
| httplib.BAD_REQUEST, |
| body, |
| self.base_headers, |
| httplib.responses[httplib.BAD_REQUEST], |
| ) |
| |
| return ( |
| httplib.NOT_FOUND, |
| body, |
| self.base_headers, |
| httplib.responses[httplib.NOT_FOUND], |
| ) |
| |
| def _test_container200(self, method, url, body, headers): |
| body = "" |
| |
| if method != "HEAD": |
| return ( |
| httplib.BAD_REQUEST, |
| body, |
| self.base_headers, |
| httplib.responses[httplib.BAD_REQUEST], |
| ) |
| |
| headers = {} |
| |
| headers["etag"] = "0x8CFB877BB56A6FB" |
| headers["last-modified"] = "Fri, 04 Jan 2013 09:48:06 GMT" |
| headers["x-ms-lease-status"] = "unlocked" |
| headers["x-ms-lease-state"] = "available" |
| headers["x-ms-meta-meta1"] = "value1" |
| |
| return (httplib.OK, body, headers, httplib.responses[httplib.OK]) |
| |
| def _test_container200_test(self, method, url, body, headers): |
| body = "" |
| |
| if method != "HEAD": |
| return ( |
| httplib.BAD_REQUEST, |
| body, |
| self.base_headers, |
| httplib.responses[httplib.BAD_REQUEST], |
| ) |
| |
| headers = {} |
| |
| headers["etag"] = "0x8CFB877BB56A6FB" |
| headers["last-modified"] = "Fri, 04 Jan 2013 09:48:06 GMT" |
| headers["content-length"] = "12345" |
| headers["content-type"] = "application/zip" |
| headers["x-ms-blob-type"] = "Block" |
| headers["x-ms-lease-status"] = "unlocked" |
| headers["x-ms-lease-state"] = "available" |
| headers["x-ms-meta-rabbits"] = "monkeys" |
| |
| return (httplib.OK, body, 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"', |
| "x-amz-meta-rabbits": "monkeys", |
| "content-length": "12345", |
| "last-modified": "Thu, 13 Sep 2012 07:13:22 GMT", |
| } |
| |
| return (httplib.OK, 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.CONFLICT]) |
| |
| def _new_container(self, method, url, body, headers): |
| # test_create_container, test_delete_container |
| |
| headers = {} |
| |
| if method == "PUT": |
| status = httplib.CREATED |
| |
| headers["etag"] = "0x8CFB877BB56A6FB" |
| headers["last-modified"] = "Fri, 04 Jan 2013 09:48:06 GMT" |
| headers["x-ms-lease-status"] = "unlocked" |
| headers["x-ms-lease-state"] = "available" |
| headers["x-ms-meta-meta1"] = "value1" |
| |
| elif method == "DELETE": |
| status = httplib.NO_CONTENT |
| |
| return (status, body, headers, httplib.responses[status]) |
| |
| def _new_container_DOESNT_EXIST(self, method, url, body, headers): |
| # test_delete_container |
| return (httplib.NOT_FOUND, body, headers, httplib.responses[httplib.NOT_FOUND]) |
| |
| 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.NOT_FOUND]) |
| |
| 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.NOT_FOUND]) |
| |
| def _foo_bar_container_foo_bar_object_DELETE(self, method, url, body, headers): |
| # test_delete_object |
| return (httplib.ACCEPTED, body, headers, httplib.responses[httplib.ACCEPTED]) |
| |
| def _foo_bar_container_foo_test_upload(self, method, url, body, headers): |
| self._assert_content_length_header_is_string(headers=headers) |
| |
| query_string = urlparse.urlsplit(url).query |
| query = parse_qs(query_string) |
| comp = query.get("comp", []) |
| |
| headers = {} |
| body = "" |
| |
| if "blocklist" in comp or not comp: |
| headers["etag"] = '"0x8CFB877BB56A6FB"' |
| headers["content-md5"] = "d4fe4c9829f7ca1cc89db7ad670d2bbd" |
| elif "block" in comp: |
| headers["content-md5"] = "lvcfx/bOJvndpRlrdKU1YQ==" |
| else: |
| raise NotImplementedError("Unknown request comp: {}".format(comp)) |
| |
| return (httplib.CREATED, body, headers, httplib.responses[httplib.CREATED]) |
| |
| def _foo_bar_container_foo_test_upload_block(self, method, url, body, headers): |
| # test_upload_object_success |
| self._assert_content_length_header_is_string(headers=headers) |
| |
| body = "" |
| headers = {} |
| headers["etag"] = "0x8CFB877BB56A6FB" |
| return (httplib.CREATED, body, headers, httplib.responses[httplib.CREATED]) |
| |
| def _foo_bar_container_foo_test_upload_blocklist(self, method, url, body, headers): |
| # test_upload_object_success |
| self._assert_content_length_header_is_string(headers=headers) |
| |
| body = "" |
| headers = {} |
| headers["etag"] = "0x8CFB877BB56A6FB" |
| headers["content-md5"] = "d4fe4c9829f7ca1cc89db7ad670d2bbd" |
| |
| return (httplib.CREATED, body, headers, httplib.responses[httplib.CREATED]) |
| |
| def _foo_bar_container_foo_test_upload_lease(self, method, url, body, headers): |
| # test_upload_object_success |
| self._assert_content_length_header_is_string(headers=headers) |
| |
| action = headers["x-ms-lease-action"] |
| rheaders = {"x-ms-lease-id": "someleaseid"} |
| body = "" |
| |
| if action == "acquire": |
| return (httplib.CREATED, body, rheaders, httplib.responses[httplib.CREATED]) |
| |
| else: |
| if headers.get("x-ms-lease-id", None) != "someleaseid": |
| return ( |
| httplib.BAD_REQUEST, |
| body, |
| rheaders, |
| httplib.responses[httplib.BAD_REQUEST], |
| ) |
| |
| return (httplib.OK, body, headers, httplib.responses[httplib.CREATED]) |
| |
| def _foo_bar_container_foo_test_upload_INVALID_HASH(self, method, url, body, headers): |
| # test_upload_object_invalid_hash1 |
| self._assert_content_length_header_is_string(headers=headers) |
| |
| body = "" |
| headers = {} |
| headers["etag"] = "0x8CFB877BB56A6FB" |
| headers["content-md5"] = "d4fe4c9829f7ca1cc89db7ad670d2bbd" |
| |
| return (httplib.CREATED, body, headers, httplib.responses[httplib.CREATED]) |
| |
| def _foo_bar_container_foo_bar_object(self, method, url, body, headers): |
| # test_upload_object_invalid_file_size |
| self._assert_content_length_header_is_string(headers=headers) |
| |
| body = generate_random_data(1000) |
| return (httplib.OK, body, headers, httplib.responses[httplib.OK]) |
| |
| def _foo_bar_container_foo_bar_object_range(self, method, url, body, headers): |
| # test_download_object_range_success |
| body = "0123456789123456789" |
| |
| self.assertTrue("x-ms-range" in headers) |
| self.assertEqual(headers["x-ms-range"], "bytes=5-6") |
| |
| start_bytes, end_bytes = self._get_start_and_end_bytes_from_range_str( |
| headers["x-ms-range"], body |
| ) |
| |
| return ( |
| httplib.PARTIAL_CONTENT, |
| body[start_bytes : end_bytes + 1], |
| headers, |
| httplib.responses[httplib.PARTIAL_CONTENT], |
| ) |
| |
| def _foo_bar_container_foo_bar_object_range_stream(self, method, url, body, headers): |
| # test_download_object_range_as_stream_success |
| body = "0123456789123456789" |
| |
| self.assertTrue("x-ms-range" in headers) |
| self.assertEqual(headers["x-ms-range"], "bytes=4-5") |
| |
| start_bytes, end_bytes = self._get_start_and_end_bytes_from_range_str( |
| headers["x-ms-range"], body |
| ) |
| |
| return ( |
| httplib.PARTIAL_CONTENT, |
| body[start_bytes : end_bytes + 1], |
| headers, |
| httplib.responses[httplib.PARTIAL_CONTENT], |
| ) |
| |
| def _foo_bar_container_foo_bar_object_INVALID_SIZE(self, method, url, body, headers): |
| # test_upload_object_invalid_file_size |
| self._assert_content_length_header_is_string(headers=headers) |
| |
| body = "" |
| return (httplib.OK, body, headers, httplib.responses[httplib.OK]) |
| |
| def _assert_content_length_header_is_string(self, headers): |
| if "Content-Length" in headers: |
| self.assertTrue(isinstance(headers["Content-Length"], basestring)) |
| |
| |
| class AzuriteBlobsMockHttp(AzureBlobsMockHttp): |
| fixtures = StorageFileFixtures("azurite_blobs") |
| |
| def _get_method_name(self, *args, **kwargs): |
| method_name = super()._get_method_name(*args, **kwargs) |
| |
| if method_name.startswith("_account"): |
| method_name = method_name[8:] |
| |
| return method_name |
| |
| |
| class AzureBlobsTests(unittest.TestCase): |
| driver_type = AzureBlobsStorageDriver |
| driver_args = STORAGE_AZURE_BLOBS_PARAMS |
| mock_response_klass = AzureBlobsMockHttp |
| |
| @classmethod |
| def create_driver(self): |
| return self.driver_type(*self.driver_args) |
| |
| def setUp(self): |
| self.driver_type.connectionCls.conn_class = self.mock_response_klass |
| self.mock_response_klass.type = None |
| self.driver = self.create_driver() |
| |
| def tearDown(self): |
| self._remove_test_file() |
| |
| def _remove_test_file(self): |
| file_path = os.path.abspath(__file__) + ".temp" |
| |
| try: |
| os.unlink(file_path) |
| except OSError: |
| pass |
| |
| def test_invalid_credentials(self): |
| self.mock_response_klass.type = "UNAUTHORIZED" |
| try: |
| self.driver.list_containers() |
| except InvalidCredsError as e: |
| self.assertEqual(True, isinstance(e, InvalidCredsError)) |
| else: |
| self.fail("Exception was not thrown") |
| |
| def test_list_containers_empty(self): |
| self.mock_response_klass.type = "list_containers_EMPTY" |
| containers = self.driver.list_containers() |
| self.assertEqual(len(containers), 0) |
| |
| def test_list_containers_success(self): |
| self.mock_response_klass.type = "list_containers" |
| AzureBlobsStorageDriver.RESPONSES_PER_REQUEST = 2 |
| containers = self.driver.list_containers() |
| self.assertEqual(len(containers), 4) |
| |
| self.assertTrue("last_modified" in containers[1].extra) |
| self.assertTrue("url" in containers[1].extra) |
| self.assertTrue("etag" in containers[1].extra) |
| self.assertTrue("lease" in containers[1].extra) |
| self.assertTrue("meta_data" in containers[1].extra) |
| self.assertEqual(containers[1].extra["etag"], "0x8CFBAB7B5B82D8E") |
| |
| def test_list_container_objects_empty(self): |
| 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): |
| self.mock_response_klass.type = None |
| AzureBlobsStorageDriver.RESPONSES_PER_REQUEST = 2 |
| |
| container = Container(name="test_container", extra={}, driver=self.driver) |
| |
| objects = self.driver.list_container_objects(container=container) |
| self.assertEqual(len(objects), 4) |
| |
| obj = objects[1] |
| self.assertEqual(obj.name, "object2.txt") |
| self.assertEqual(obj.hash, "0x8CFB90F1BA8CD8F") |
| self.assertEqual(obj.size, 1048576) |
| self.assertEqual(obj.container.name, "test_container") |
| self.assertTrue("meta1" in obj.meta_data) |
| self.assertTrue("meta2" in obj.meta_data) |
| self.assertTrue("last_modified" in obj.extra) |
| self.assertTrue("content_type" in obj.extra) |
| self.assertTrue("content_encoding" in obj.extra) |
| self.assertTrue("content_language" in obj.extra) |
| |
| def test_list_container_objects_with_prefix(self): |
| self.mock_response_klass.type = None |
| AzureBlobsStorageDriver.RESPONSES_PER_REQUEST = 2 |
| |
| container = Container(name="test_container", extra={}, driver=self.driver) |
| objects = self.driver.list_container_objects(container=container, prefix="test_prefix") |
| self.assertEqual(len(objects), 4) |
| |
| obj = objects[1] |
| self.assertEqual(obj.name, "object2.txt") |
| self.assertEqual(obj.hash, "0x8CFB90F1BA8CD8F") |
| self.assertEqual(obj.size, 1048576) |
| self.assertEqual(obj.container.name, "test_container") |
| self.assertTrue("meta1" in obj.meta_data) |
| self.assertTrue("meta2" in obj.meta_data) |
| self.assertTrue("last_modified" in obj.extra) |
| self.assertTrue("content_type" in obj.extra) |
| self.assertTrue("content_encoding" in obj.extra) |
| self.assertTrue("content_language" in obj.extra) |
| |
| def test_get_container_doesnt_exist(self): |
| self.mock_response_klass.type = None |
| try: |
| self.driver.get_container(container_name="test_container100") |
| except ContainerDoesNotExistError: |
| pass |
| else: |
| self.fail("Exception was not thrown") |
| |
| def test_get_container_success(self): |
| self.mock_response_klass.type = None |
| container = self.driver.get_container(container_name="test_container200") |
| |
| self.assertTrue(container.name, "test_container200") |
| self.assertTrue(container.extra["etag"], "0x8CFB877BB56A6FB") |
| self.assertTrue(container.extra["last_modified"], "Fri, 04 Jan 2013 09:48:06 GMT") |
| self.assertTrue(container.extra["lease"]["status"], "unlocked") |
| self.assertTrue(container.extra["lease"]["state"], "available") |
| self.assertTrue(container.extra["meta_data"]["meta1"], "value1") |
| |
| if self.driver.secure: |
| expected_url = "https://account.blob.core.windows.net/test_container200" |
| else: |
| expected_url = "http://localhost/account/test_container200" |
| |
| self.assertEqual(container.extra["url"], expected_url) |
| |
| def test_get_object_cdn_url(self): |
| obj = self.driver.get_object(container_name="test_container200", object_name="test") |
| |
| url = urlparse.urlparse(self.driver.get_object_cdn_url(obj)) |
| query = urlparse.parse_qs(url.query) |
| |
| self.assertEqual(len(query["sig"]), 1) |
| self.assertGreater(len(query["sig"][0]), 0) |
| |
| def test_get_object_container_doesnt_exist(self): |
| # This method makes two requests which makes mocking the response a bit |
| # trickier |
| self.mock_response_klass.type = None |
| try: |
| self.driver.get_object(container_name="test_container100", object_name="test") |
| except ContainerDoesNotExistError: |
| pass |
| else: |
| self.fail("Exception was not thrown") |
| |
| def test_get_object_success(self): |
| # This method makes two requests which makes mocking the response a bit |
| # trickier |
| self.mock_response_klass.type = None |
| obj = self.driver.get_object(container_name="test_container200", object_name="test") |
| |
| self.assertEqual(obj.name, "test") |
| self.assertEqual(obj.container.name, "test_container200") |
| self.assertEqual(obj.size, 12345) |
| self.assertEqual(obj.hash, "0x8CFB877BB56A6FB") |
| self.assertEqual(obj.extra["last_modified"], "Fri, 04 Jan 2013 09:48:06 GMT") |
| self.assertEqual(obj.extra["content_type"], "application/zip") |
| self.assertEqual(obj.meta_data["rabbits"], "monkeys") |
| |
| def test_create_container_invalid_name(self): |
| # invalid container name |
| self.mock_response_klass.type = "INVALID_NAME" |
| try: |
| self.driver.create_container(container_name="new--container") |
| except InvalidContainerNameError: |
| pass |
| else: |
| self.fail("Exception was not thrown") |
| |
| def test_create_container_already_exists(self): |
| # container with this name already exists |
| self.mock_response_klass.type = "ALREADY_EXISTS" |
| try: |
| self.driver.create_container(container_name="new-container") |
| except ContainerAlreadyExistsError: |
| pass |
| else: |
| self.fail("Exception was not thrown") |
| |
| def test_create_container_success(self): |
| # success |
| self.mock_response_klass.type = None |
| name = "new-container" |
| container = self.driver.create_container(container_name=name) |
| self.assertEqual(container.name, name) |
| |
| def test_delete_container_doesnt_exist(self): |
| container = Container(name="new_container", extra=None, driver=self.driver) |
| self.mock_response_klass.type = "DOESNT_EXIST" |
| try: |
| self.driver.delete_container(container=container) |
| except ContainerDoesNotExistError: |
| pass |
| else: |
| self.fail("Exception was not thrown") |
| |
| def test_delete_container_not_empty(self): |
| self.mock_response_klass.type = None |
| AzureBlobsStorageDriver.RESPONSES_PER_REQUEST = 2 |
| |
| container = Container(name="test_container", extra={}, driver=self.driver) |
| |
| try: |
| self.driver.delete_container(container=container) |
| except ContainerIsNotEmptyError: |
| pass |
| else: |
| self.fail("Exception was not thrown") |
| |
| def test_delete_container_success(self): |
| self.mock_response_klass.type = "EMPTY" |
| AzureBlobsStorageDriver.RESPONSES_PER_REQUEST = 2 |
| |
| container = Container(name="test_container", extra={}, driver=self.driver) |
| |
| self.assertTrue(self.driver.delete_container(container=container)) |
| |
| def test_delete_container_not_found(self): |
| self.mock_response_klass.type = "NOT_FOUND" |
| container = Container(name="foo_bar_container", extra={}, driver=self.driver) |
| try: |
| self.driver.delete_container(container=container) |
| except ContainerDoesNotExistError: |
| pass |
| else: |
| self.fail("Container does not exist but an exception was not" + "thrown") |
| |
| def test_download_object_success(self): |
| container = Container(name="foo_bar_container", extra={}, driver=self.driver) |
| obj = Object( |
| name="foo_bar_object", |
| size=1000, |
| hash=None, |
| extra={}, |
| container=container, |
| meta_data=None, |
| driver=self.driver_type, |
| ) |
| destination_path = os.path.abspath(__file__) + ".temp" |
| result = self.driver.download_object( |
| obj=obj, |
| destination_path=destination_path, |
| overwrite_existing=False, |
| delete_on_failure=True, |
| ) |
| self.assertTrue(result) |
| |
| def test_download_object_invalid_file_size(self): |
| self.mock_response_klass.type = "INVALID_SIZE" |
| container = Container(name="foo_bar_container", extra={}, driver=self.driver) |
| obj = Object( |
| name="foo_bar_object", |
| size=1000, |
| hash=None, |
| extra={}, |
| container=container, |
| meta_data=None, |
| driver=self.driver_type, |
| ) |
| destination_path = os.path.abspath(__file__) + ".temp" |
| result = self.driver.download_object( |
| obj=obj, |
| destination_path=destination_path, |
| overwrite_existing=False, |
| delete_on_failure=True, |
| ) |
| self.assertFalse(result) |
| |
| def test_download_object_invalid_file_already_exists(self): |
| self.mock_response_klass.type = "INVALID_SIZE" |
| container = Container(name="foo_bar_container", extra={}, driver=self.driver) |
| obj = Object( |
| name="foo_bar_object", |
| size=1000, |
| hash=None, |
| extra={}, |
| container=container, |
| meta_data=None, |
| driver=self.driver_type, |
| ) |
| destination_path = os.path.abspath(__file__) |
| try: |
| self.driver.download_object( |
| obj=obj, |
| destination_path=destination_path, |
| overwrite_existing=False, |
| delete_on_failure=True, |
| ) |
| except LibcloudError: |
| pass |
| else: |
| self.fail("Exception was not thrown") |
| |
| def test_download_object_as_stream_success(self): |
| container = Container(name="foo_bar_container", extra={}, driver=self.driver) |
| |
| obj = Object( |
| name="foo_bar_object", |
| size=1000, |
| hash=None, |
| extra={}, |
| container=container, |
| meta_data=None, |
| driver=self.driver_type, |
| ) |
| |
| stream = self.driver.download_object_as_stream(obj=obj, chunk_size=None) |
| |
| consumed_stream = "".join(chunk.decode("utf-8") for chunk in stream) |
| self.assertEqual(len(consumed_stream), obj.size) |
| |
| def test_download_object_range_success(self): |
| container = Container(name="foo_bar_container", extra={}, driver=self.driver) |
| obj = Object( |
| name="foo_bar_object_range", |
| size=1000, |
| hash=None, |
| extra={}, |
| container=container, |
| meta_data=None, |
| driver=self.driver_type, |
| ) |
| destination_path = os.path.abspath(__file__) + ".temp" |
| result = self.driver.download_object_range( |
| obj=obj, |
| start_bytes=5, |
| end_bytes=7, |
| destination_path=destination_path, |
| overwrite_existing=False, |
| delete_on_failure=True, |
| ) |
| self.assertTrue(result) |
| |
| with open(destination_path) as fp: |
| content = fp.read() |
| |
| self.assertEqual(content, "56") |
| |
| def test_download_object_range_as_stream_success(self): |
| container = Container(name="foo_bar_container", extra={}, driver=self.driver) |
| |
| obj = Object( |
| name="foo_bar_object_range_stream", |
| size=2, |
| hash=None, |
| extra={}, |
| container=container, |
| meta_data=None, |
| driver=self.driver_type, |
| ) |
| |
| stream = self.driver.download_object_range_as_stream( |
| obj=obj, start_bytes=4, end_bytes=6, chunk_size=None |
| ) |
| |
| consumed_stream = "".join(chunk.decode("utf-8") for chunk in stream) |
| self.assertEqual(consumed_stream, "45") |
| self.assertEqual(len(consumed_stream), obj.size) |
| |
| def test_upload_object_invalid_md5(self): |
| # Invalid md5 is returned by azure |
| self.mock_response_klass.type = "INVALID_HASH" |
| |
| container = Container(name="foo_bar_container", extra={}, driver=self.driver) |
| object_name = "foo_test_upload" |
| file_path = os.path.abspath(__file__) |
| try: |
| self.driver.upload_object( |
| file_path=file_path, |
| container=container, |
| object_name=object_name, |
| verify_hash=True, |
| ) |
| except ObjectHashMismatchError: |
| pass |
| else: |
| self.fail("Invalid hash was returned but an exception was not thrown") |
| |
| def test_upload_small_block_object_success(self): |
| file_path = os.path.abspath(__file__) |
| file_size = os.stat(file_path).st_size |
| |
| container = Container(name="foo_bar_container", extra={}, driver=self.driver) |
| object_name = "foo_test_upload" |
| extra = {"meta_data": {"some-value": "foobar"}} |
| obj = self.driver.upload_object( |
| file_path=file_path, |
| container=container, |
| object_name=object_name, |
| extra=extra, |
| verify_hash=False, |
| ) |
| |
| self.assertEqual(obj.name, "foo_test_upload") |
| self.assertEqual(obj.size, file_size) |
| self.assertTrue("some-value" in obj.meta_data) |
| |
| def test_upload_big_block_object_success(self): |
| _, file_path = tempfile.mkstemp(suffix=".jpg") |
| file_size = AZURE_UPLOAD_CHUNK_SIZE + 1 |
| |
| with open(file_path, "w") as file_hdl: |
| file_hdl.write("0" * file_size) |
| |
| container = Container(name="foo_bar_container", extra={}, driver=self.driver) |
| object_name = "foo_test_upload" |
| extra = {"meta_data": {"some-value": "foobar"}} |
| obj = self.driver.upload_object( |
| file_path=file_path, |
| container=container, |
| object_name=object_name, |
| extra=extra, |
| verify_hash=False, |
| ) |
| |
| self.assertEqual(obj.name, "foo_test_upload") |
| self.assertEqual(obj.size, file_size) |
| self.assertTrue("some-value" in obj.meta_data) |
| |
| os.remove(file_path) |
| |
| def test_upload_small_block_object_success_with_lease(self): |
| self.mock_response_klass.use_param = "comp" |
| file_path = os.path.abspath(__file__) |
| file_size = os.stat(file_path).st_size |
| |
| container = Container(name="foo_bar_container", extra={}, driver=self.driver) |
| object_name = "foo_test_upload" |
| extra = {"meta_data": {"some-value": "foobar"}} |
| obj = self.driver.upload_object( |
| file_path=file_path, |
| container=container, |
| object_name=object_name, |
| extra=extra, |
| verify_hash=False, |
| ex_use_lease=True, |
| ) |
| |
| self.assertEqual(obj.name, "foo_test_upload") |
| self.assertEqual(obj.size, file_size) |
| self.assertTrue("some-value" in obj.meta_data) |
| self.mock_response_klass.use_param = None |
| |
| def test_upload_big_block_object_success_with_lease(self): |
| self.mock_response_klass.use_param = "comp" |
| _, file_path = tempfile.mkstemp(suffix=".jpg") |
| file_size = AZURE_UPLOAD_CHUNK_SIZE * 2 |
| |
| with open(file_path, "w") as file_hdl: |
| file_hdl.write("0" * file_size) |
| |
| container = Container(name="foo_bar_container", extra={}, driver=self.driver) |
| object_name = "foo_test_upload" |
| extra = {"meta_data": {"some-value": "foobar"}} |
| obj = self.driver.upload_object( |
| file_path=file_path, |
| container=container, |
| object_name=object_name, |
| extra=extra, |
| verify_hash=False, |
| ex_use_lease=False, |
| ) |
| |
| self.assertEqual(obj.name, "foo_test_upload") |
| self.assertEqual(obj.size, file_size) |
| self.assertTrue("some-value" in obj.meta_data) |
| |
| os.remove(file_path) |
| self.mock_response_klass.use_param = None |
| |
| def test_upload_blob_object_via_stream(self): |
| self.mock_response_klass.use_param = "comp" |
| container = Container(name="foo_bar_container", extra={}, driver=self.driver) |
| |
| object_name = "foo_test_upload" |
| iterator = BytesIO(b("345")) |
| extra = {"content_type": "text/plain"} |
| obj = self.driver.upload_object_via_stream( |
| container=container, object_name=object_name, iterator=iterator, extra=extra |
| ) |
| |
| self.assertEqual(obj.name, object_name) |
| self.assertEqual(obj.size, 3) |
| self.mock_response_klass.use_param = None |
| |
| def test_upload_blob_object_via_stream_from_iterable(self): |
| self.mock_response_klass.use_param = "comp" |
| container = Container(name="foo_bar_container", extra={}, driver=self.driver) |
| |
| object_name = "foo_test_upload" |
| iterator = iter([b("34"), b("5")]) |
| extra = {"content_type": "text/plain"} |
| obj = self.driver.upload_object_via_stream( |
| container=container, object_name=object_name, iterator=iterator, extra=extra |
| ) |
| |
| self.assertEqual(obj.name, object_name) |
| self.assertEqual(obj.size, 3) |
| self.mock_response_klass.use_param = None |
| |
| def test_upload_blob_object_via_stream_with_lease(self): |
| self.mock_response_klass.use_param = "comp" |
| container = Container(name="foo_bar_container", extra={}, driver=self.driver) |
| |
| object_name = "foo_test_upload" |
| iterator = BytesIO(b("345")) |
| extra = {"content_type": "text/plain"} |
| obj = self.driver.upload_object_via_stream( |
| container=container, |
| object_name=object_name, |
| iterator=iterator, |
| extra=extra, |
| ex_use_lease=True, |
| ) |
| |
| self.assertEqual(obj.name, object_name) |
| self.assertEqual(obj.size, 3) |
| self.mock_response_klass.use_param = None |
| |
| def test_delete_object_not_found(self): |
| self.mock_response_klass.type = "NOT_FOUND" |
| container = Container(name="foo_bar_container", extra={}, driver=self.driver) |
| obj = Object( |
| name="foo_bar_object", |
| size=1234, |
| hash=None, |
| extra=None, |
| meta_data=None, |
| container=container, |
| driver=self.driver, |
| ) |
| try: |
| self.driver.delete_object(obj=obj) |
| except ObjectDoesNotExistError: |
| pass |
| else: |
| self.fail("Exception was not thrown") |
| |
| def test_delete_object_success(self): |
| self.mock_response_klass.type = "DELETE" |
| container = Container(name="foo_bar_container", extra={}, driver=self.driver) |
| obj = Object( |
| name="foo_bar_object", |
| size=1234, |
| hash=None, |
| extra=None, |
| meta_data=None, |
| container=container, |
| driver=self.driver, |
| ) |
| |
| result = self.driver.delete_object(obj=obj) |
| self.assertTrue(result) |
| |
| def test_storage_driver_host(self): |
| # Non regression tests for issue LIBCLOUD-399 dealing with the bad |
| # management of the connectionCls.host class attribute |
| driver1 = self.driver_type("fakeaccount1", "deadbeafcafebabe==") |
| driver2 = self.driver_type("fakeaccount2", "deadbeafcafebabe==") |
| driver3 = self.driver_type("fakeaccount3", "deadbeafcafebabe==", host="test.foo.bar.com") |
| |
| host1 = driver1.connection.host |
| host2 = driver2.connection.host |
| host3 = driver3.connection.host |
| |
| self.assertEqual(host1, "fakeaccount1.blob.core.windows.net") |
| self.assertEqual(host2, "fakeaccount2.blob.core.windows.net") |
| self.assertEqual(host3, "test.foo.bar.com") |
| |
| def test_normalize_http_headers(self): |
| driver = self.driver_type("fakeaccount1", "deadbeafcafebabe==") |
| |
| headers = driver._fix_headers( |
| { |
| # should be normalized to include x-ms-blob prefix |
| "Content-Encoding": "gzip", |
| "content-language": "en-us", |
| # should be passed through |
| "x-foo": "bar", |
| } |
| ) |
| |
| self.assertEqual( |
| headers, |
| { |
| "x-ms-blob-content-encoding": "gzip", |
| "x-ms-blob-content-language": "en-us", |
| "x-foo": "bar", |
| }, |
| ) |
| |
| def test_storage_driver_host_govcloud(self): |
| driver1 = self.driver_type( |
| "fakeaccount1", "deadbeafcafebabe==", host="blob.core.usgovcloudapi.net" |
| ) |
| driver2 = self.driver_type( |
| "fakeaccount2", |
| "deadbeafcafebabe==", |
| host="fakeaccount2.blob.core.usgovcloudapi.net", |
| ) |
| |
| host1 = driver1.connection.host |
| host2 = driver2.connection.host |
| account_prefix_1 = driver1.connection.account_prefix |
| account_prefix_2 = driver2.connection.account_prefix |
| |
| self.assertEqual(host1, "fakeaccount1.blob.core.usgovcloudapi.net") |
| self.assertEqual(host2, "fakeaccount2.blob.core.usgovcloudapi.net") |
| self.assertIsNone(account_prefix_1) |
| self.assertIsNone(account_prefix_2) |
| |
| def test_storage_driver_host_azurite(self): |
| driver = self.driver_type( |
| "fakeaccount1", |
| "deadbeafcafebabe==", |
| host="localhost", |
| port=10000, |
| secure=False, |
| ) |
| |
| host = driver.connection.host |
| account_prefix = driver.connection.account_prefix |
| |
| self.assertEqual(host, "localhost") |
| self.assertEqual(account_prefix, "fakeaccount1") |
| |
| def test_storage_driver_azure_ad(self): |
| AzureBlobsActiveDirectoryConnection.conn_class = AzureBlobsMockHttp |
| driver = self.driver_type( |
| key="fakeaccount1", |
| secret="DEKjfhdakkdjfhei~", |
| tenant_id="77777777-7777-7777-7777-777777777777", |
| identity="55555555-5555-5555-5555-555555555555", |
| auth_type="azureAd", |
| secure=True, |
| ) |
| host = driver.connection.host |
| |
| self.assertEqual(host, "fakeaccount1.blob.core.windows.net") |
| |
| def test_get_azure_ad_object_success(self): |
| AzureBlobsActiveDirectoryConnection.conn_class = AzureBlobsMockHttp |
| driver = self.driver_type( |
| key="fakeaccount1", |
| secret="DEKjfhdakkdjfhei~", |
| tenant_id="77777777-7777-7777-7777-777777777777", |
| identity="55555555-5555-5555-5555-555555555555", |
| auth_type="azureAd", |
| secure=True, |
| ) |
| self.mock_response_klass.type = None |
| container = driver.get_container(container_name="test_container200") |
| |
| self.assertTrue(container.name, "test_container200") |
| self.assertTrue(container.extra["etag"], "0x8CFB877BB56A6FB") |
| self.assertTrue(container.extra["last_modified"], "Fri, 04 Jan 2013 09:48:06 GMT") |
| self.assertTrue(container.extra["lease"]["status"], "unlocked") |
| self.assertTrue(container.extra["lease"]["state"], "available") |
| self.assertTrue(container.extra["meta_data"]["meta1"], "value1") |
| |
| |
| class AzuriteBlobsTests(AzureBlobsTests): |
| driver_args = STORAGE_AZURITE_BLOBS_PARAMS |
| mock_response_klass = AzuriteBlobsMockHttp |
| |
| |
| if __name__ == "__main__": |
| sys.exit(unittest.main()) |
