Changeset View
Changeset View
Standalone View
Standalone View
swh/objstorage/tests/test_objstorage_azure.py
| # Copyright (C) 2016-2018 The Software Heritage developers | # Copyright (C) 2016-2018 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| import unittest | import unittest | ||||
| from collections import defaultdict | from collections import defaultdict | ||||
| from unittest.mock import patch | from unittest.mock import patch | ||||
| from typing import Any, Dict | from typing import Any, Dict, Optional | ||||
| from azure.common import AzureMissingResourceHttpError | from azure.common import AzureMissingResourceHttpError | ||||
| from swh.model.hashutil import hash_to_hex | from swh.model.hashutil import hash_to_hex | ||||
| from swh.objstorage import get_objstorage | from swh.objstorage import get_objstorage | ||||
| from swh.objstorage.objstorage import decompressors | |||||
| from .objstorage_testing import ObjStorageTestFixture | from .objstorage_testing import ObjStorageTestFixture | ||||
| class MockBlob(): | class MockBlob(): | ||||
| """ Libcloud object mock that replicates its API """ | """ Libcloud object mock that replicates its API """ | ||||
| def __init__(self, name, content): | def __init__(self, name, content): | ||||
| self.name = name | self.name = name | ||||
| Show All 38 Lines | class MockBlockBlobService(): | ||||
| def list_blobs(self, container_name, marker=None, maxresults=None): | def list_blobs(self, container_name, marker=None, maxresults=None): | ||||
| for blob_name, content in sorted(self._data[container_name].items()): | for blob_name, content in sorted(self._data[container_name].items()): | ||||
| if marker is None or blob_name > marker: | if marker is None or blob_name > marker: | ||||
| yield MockBlob(name=blob_name, content=content) | yield MockBlob(name=blob_name, content=content) | ||||
| class TestAzureCloudObjStorage(ObjStorageTestFixture, unittest.TestCase): | class TestAzureCloudObjStorage(ObjStorageTestFixture, unittest.TestCase): | ||||
| compression = None # type: Optional[str] | |||||
| def setUp(self): | def setUp(self): | ||||
| super().setUp() | super().setUp() | ||||
| patcher = patch( | patcher = patch( | ||||
| 'swh.objstorage.backends.azure.BlockBlobService', | 'swh.objstorage.backends.azure.BlockBlobService', | ||||
| MockBlockBlobService, | MockBlockBlobService, | ||||
| ) | ) | ||||
| patcher.start() | patcher.start() | ||||
| self.addCleanup(patcher.stop) | self.addCleanup(patcher.stop) | ||||
| self.storage = get_objstorage('azure', { | self.storage = get_objstorage('azure', { | ||||
| 'account_name': 'account-name', | 'account_name': 'account-name', | ||||
| 'api_secret_key': 'api-secret-key', | 'api_secret_key': 'api-secret-key', | ||||
| 'container_name': 'container-name', | 'container_name': 'container-name', | ||||
| 'compression': self.compression, | |||||
| }) | }) | ||||
| def test_compression(self): | |||||
| content, obj_id = self.hash_content(b'test content is compressed') | |||||
| self.storage.add(content, obj_id=obj_id) | |||||
| blob_service, container = self.storage.get_blob_service(obj_id) | |||||
| internal_id = self.storage._internal_id(obj_id) | |||||
| raw_blob = blob_service.get_blob_to_bytes(container, internal_id) | |||||
| d = decompressors[self.compression]() | |||||
| assert d.decompress(raw_blob.content) == content | |||||
| assert d.unused_data == b'' | |||||
| class TestAzureCloudObjStorageGzip(TestAzureCloudObjStorage): | |||||
| compression = 'gzip' | |||||
| class TestAzureCloudObjStorageZlib(TestAzureCloudObjStorage): | |||||
| compression = 'zlib' | |||||
| class TestAzureCloudObjStorageLzma(TestAzureCloudObjStorage): | |||||
| compression = 'lzma' | |||||
| class TestAzureCloudObjStorageBz2(TestAzureCloudObjStorage): | |||||
| compression = 'bz2' | |||||
| class TestPrefixedAzureCloudObjStorage(ObjStorageTestFixture, | class TestPrefixedAzureCloudObjStorage(ObjStorageTestFixture, | ||||
| unittest.TestCase): | unittest.TestCase): | ||||
| def setUp(self): | def setUp(self): | ||||
| super().setUp() | super().setUp() | ||||
| patcher = patch( | patcher = patch( | ||||
| 'swh.objstorage.backends.azure.BlockBlobService', | 'swh.objstorage.backends.azure.BlockBlobService', | ||||
| MockBlockBlobService, | MockBlockBlobService, | ||||
| ▲ Show 20 Lines • Show All 43 Lines • Show Last 20 Lines | |||||