Differential D100 Diff 391 swh/objstorage/tests/test_objstorage_cloud.py

Changeset View

Standalone View

swh/objstorage/tests/test_objstorage_cloud.py

This file was added.

				# Copyright (C) 2016 The Software Heritage developers
				# See the AUTHORS file at the top-level directory of this distribution
				# License: GNU General Public License version 3, or any later version
				# See top-level LICENSE file for more information

				import unittest

				from swh.objstorage.cloud.objstorage_cloud import CloudObjStorage
				from libcloud.storage.types import (ObjectDoesNotExistError,
				ContainerDoesNotExistError)
				from libcloud.common.types import InvalidCredsError

				from objstorage_testing import ObjStorageTestFixture


				API_KEY = 'API_KEY'
				API_SECRET_KEY = 'API SECRET KEY'
				CONTAINER_NAME = 'test_container'


				class MockLibcloudObject():
				""" Libcloud object mock that replicates its API """
				def __init__(self, name, content):
				self.name = name
				self.content = list(content)

				def as_stream(self):
				yield from iter(self.content)


				class MockLibcloudDriver():
				""" Mock driver that replicates the used LibCloud API """
				def __init__(self, api_key, api_secret_key):
				self.containers = {CONTAINER_NAME: {}} # Storage is initialized
				self.api_key = api_key
				self.api_secret_key = api_secret_key

				def _check_credentials(self):
				# Private method may be known as another name in Libcloud but is used
				# to replicate libcloud behavior (i.e. check credential at each
				# request)
				if self.api_key != API_KEY or self.api_secret_key != API_SECRET_KEY:
				raise InvalidCredsError()

				def get_container(self, container_name):
				try:
				return self.containers[container_name]
				except KeyError:
				raise ContainerDoesNotExistError(container_name=container_name,
				driver=self, value=None)

				def iterate_container_objects(self, container):
				self._check_credentials()
				yield from container.values()

				def get_object(self, container_name, obj_id):
				self._check_credentials()
				try:
				container = self.get_container(container_name)
				return container[obj_id]
				except KeyError:
				raise ObjectDoesNotExistError(object_name=obj_id,
				driver=self, value=None)

				def upload_object_via_stream(self, content, container, obj_id):
				self._check_credentials()
				obj = MockLibcloudObject(obj_id, content)
				container[obj_id] = obj


				class MockCloudObjStorage(CloudObjStorage):
				""" Cloud object storage that uses a mocked driver """
				def _get_driver(self, api_key, api_secret_key):
				return MockLibcloudDriver(api_key, api_secret_key)

				def _get_provider(self):
				# Implement this for the abc requirement, but behavior is defined in
				# _get_driver.
				pass


				class TestCloudObjStorage(ObjStorageTestFixture, unittest.TestCase):

				def setUp(self):
				super().setUp()
				self.storage = MockCloudObjStorage(API_KEY, API_SECRET_KEY,
				CONTAINER_NAME)