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 | ||||
| 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.model import hashutil | |||||
| from swh.objstorage.objstorage import decompressors | from swh.objstorage.objstorage import decompressors | ||||
| from swh.objstorage.exc import Error | from swh.objstorage.exc import Error | ||||
| from swh.objstorage.backends.libcloud import CloudObjStorage | from swh.objstorage.backends.libcloud import CloudObjStorage | ||||
| from .objstorage_testing import ObjStorageTestFixture | from .objstorage_testing import ObjStorageTestFixture | ||||
| API_KEY = 'API_KEY' | API_KEY = 'API_KEY' | ||||
| API_SECRET_KEY = 'API SECRET KEY' | API_SECRET_KEY = 'API SECRET KEY' | ||||
| ▲ Show 20 Lines • Show All 80 Lines • ▼ Show 20 Lines | def setUp(self): | ||||
| 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, | ||||
| ) | ) | ||||
| 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) | ||||
| data = self.storage.driver.containers[CONTAINER_NAME] | |||||
| obj_id = hashutil.hash_to_hex(obj_id) | libcloud_object = self.storage._get_object(obj_id) | ||||
| raw_content = b''.join(data[obj_id].content) | raw_content = b''.join(libcloud_object.content) | ||||
| d = decompressors[self.compression]() | d = decompressors[self.compression]() | ||||
| assert d.decompress(raw_content) == content | assert d.decompress(raw_content) == content | ||||
| assert d.unused_data == b'' | assert d.unused_data == b'' | ||||
| def test_trailing_data_on_stored_blob(self): | def test_trailing_data_on_stored_blob(self): | ||||
| content, obj_id = self.hash_content(b'test content without garbage') | content, obj_id = self.hash_content(b'test content without garbage') | ||||
| self.storage.add(content, obj_id=obj_id) | self.storage.add(content, obj_id=obj_id) | ||||
| data = self.storage.driver.containers[CONTAINER_NAME] | libcloud_object = self.storage._get_object(obj_id) | ||||
| obj_id = hashutil.hash_to_hex(obj_id) | libcloud_object.content.append(b'trailing garbage') | ||||
| data[obj_id].content.append(b'trailing garbage') | |||||
| if self.compression == 'none': | if self.compression == 'none': | ||||
| with self.assertRaises(Error) as e: | with self.assertRaises(Error) as e: | ||||
| self.storage.check(obj_id) | self.storage.check(obj_id) | ||||
| else: | else: | ||||
| with self.assertRaises(Error) as e: | with self.assertRaises(Error) as e: | ||||
| self.storage.get(obj_id) | self.storage.get(obj_id) | ||||
| assert 'trailing data' in e.exception.args[0] | assert 'trailing data' in e.exception.args[0] | ||||
| Show All 16 Lines | |||||