Changeset View
Changeset View
Standalone View
Standalone View
swh/objstorage/tests/test_objstorage_cloud.py
| # Copyright (C) 2016 The Software Heritage developers | # Copyright (C) 2016 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 | ||||
| from typing import Optional | |||||
| import unittest | import unittest | ||||
| from libcloud.common.types import InvalidCredsError | from libcloud.common.types import InvalidCredsError | ||||
| from libcloud.storage.types import (ContainerDoesNotExistError, | from libcloud.storage.types import (ContainerDoesNotExistError, | ||||
| ObjectDoesNotExistError) | ObjectDoesNotExistError) | ||||
| from swh.objstorage.objstorage import decompressors | from swh.objstorage.objstorage import decompressors | ||||
| from swh.objstorage.exc import Error | from swh.objstorage.exc import Error | ||||
| ▲ Show 20 Lines • Show All 74 Lines • ▼ Show 20 Lines | class MockCloudObjStorage(CloudObjStorage): | ||||
| def _get_provider(self): | def _get_provider(self): | ||||
| # Implement this for the abc requirement, but behavior is defined in | # Implement this for the abc requirement, but behavior is defined in | ||||
| # _get_driver. | # _get_driver. | ||||
| pass | pass | ||||
| class TestCloudObjStorage(ObjStorageTestFixture, unittest.TestCase): | class TestCloudObjStorage(ObjStorageTestFixture, unittest.TestCase): | ||||
| compression = 'none' | compression = 'none' | ||||
| path_prefix: Optional[str] = None | |||||
| def setUp(self): | def setUp(self): | ||||
| super().setUp() | super().setUp() | ||||
| self.storage = MockCloudObjStorage( | self.storage = MockCloudObjStorage( | ||||
| CONTAINER_NAME, | CONTAINER_NAME, | ||||
| api_key=API_KEY, api_secret_key=API_SECRET_KEY, | api_key=API_KEY, api_secret_key=API_SECRET_KEY, | ||||
| compression=self.compression, | compression=self.compression, | ||||
| path_prefix=self.path_prefix, | |||||
| ) | ) | ||||
| def test_compression(self): | def test_compression(self): | ||||
| content, obj_id = self.hash_content(b'add_get_w_id') | content, obj_id = self.hash_content(b'add_get_w_id') | ||||
| self.storage.add(content, obj_id=obj_id) | self.storage.add(content, obj_id=obj_id) | ||||
| libcloud_object = self.storage._get_object(obj_id) | libcloud_object = self.storage._get_object(obj_id) | ||||
| raw_content = b''.join(libcloud_object.content) | raw_content = b''.join(libcloud_object.content) | ||||
| Show All 27 Lines | |||||
| class TestCloudObjStorageLzma(TestCloudObjStorage): | class TestCloudObjStorageLzma(TestCloudObjStorage): | ||||
| compression = 'lzma' | compression = 'lzma' | ||||
| class TestCloudObjStorageZlib(TestCloudObjStorage): | class TestCloudObjStorageZlib(TestCloudObjStorage): | ||||
| compression = 'zlib' | compression = 'zlib' | ||||
| class TestCloudObjStoragePrefix(TestCloudObjStorage): | |||||
| path_prefix = 'contents' | |||||
| def test_path_prefix(self): | |||||
| content, obj_id = self.hash_content(b'test content') | |||||
| self.storage.add(content, obj_id=obj_id) | |||||
| container = self.storage.driver.containers[CONTAINER_NAME] | |||||
| object_path = self.storage._object_path(obj_id) | |||||
| assert object_path.startswith(self.path_prefix + '/') | |||||
| assert object_path in container | |||||