Differential D2676 Diff 9550 swh/storage/tests/test_api_client.py

Changeset View

Standalone View

swh/storage/tests/test_api_client.py

This file was copied to swh/storage/tests/test_api_client_dicts.py.

	# Copyright (C) 2015-2020 The Software Heritage developers			# Copyright (C) 2015-2020 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 unittest.mock import patch			from unittest.mock import patch

	import pytest			import pytest

	from swh.storage.api.client import RemoteStorage
	import swh.storage.api.server as server			import swh.storage.api.server as server
	import swh.storage.storage			import swh.storage.storage
				from swh.storage import get_storage
	from swh.storage.tests.test_storage import TestStorageGeneratedData # noqa			from swh.storage.tests.test_storage import TestStorageGeneratedData # noqa
	from swh.storage.tests.test_storage import TestStorage as _TestStorage			from swh.storage.tests.test_storage import TestStorage as _TestStorage

	# tests are executed using imported classes (TestStorage and			# tests are executed using imported classes (TestStorage and
	# TestStorageGeneratedData) using overloaded swh_storage fixture			# TestStorageGeneratedData) using overloaded swh_storage fixture
	# below			# below


	@pytest.fixture			@pytest.fixture
	def app_server():			def app_server():
	storage_config = {			storage_config = {
	'cls': 'validate',
	'storage': {
	'cls': 'memory',			'cls': 'memory',
	'journal_writer': {			'journal_writer': {
	'cls': 'memory',			'cls': 'memory',
	},			},
	}			}
	}
	server.storage = swh.storage.get_storage(**storage_config)			server.storage = swh.storage.get_storage(**storage_config)
	yield server			yield server


	@pytest.fixture			@pytest.fixture
	def app(app_server):			def app(app_server):
	return app_server.app			return app_server.app


	@pytest.fixture			@pytest.fixture
	def swh_rpc_client_class():			def swh_rpc_client_class():
	return RemoteStorage			def storage_factory(**kwargs):
				storage_config = {
				'cls': 'validate',
				'storage': {
				'cls': 'remote',
				**kwargs,
				}
				}
				return get_storage(**storage_config)

				return storage_factory


	@pytest.fixture			@pytest.fixture
	def swh_storage(swh_rpc_client, app_server):			def swh_storage(swh_rpc_client, app_server):
	# This version of the swh_storage fixture uses the swh_rpc_client fixture			# This version of the swh_storage fixture uses the swh_rpc_client fixture
	# to instantiate a RemoteStorage (see swh_rpc_client_class above) that			# to instantiate a RemoteStorage (see swh_rpc_client_class above) that
	# proxies, via the swh.core RPC mechanism, the local (in memory) storage			# proxies, via the swh.core RPC mechanism, the local (in memory) storage
	# configured in the app_server fixture above.			# configured in the app_server fixture above.
	#			#
	# Also note that, for the sake of			# Also note that, for the sake of
	# making it easier to write tests, the in-memory journal writer of the			# making it easier to write tests, the in-memory journal writer of the
	# in-memory backend storage is attached to the RemoteStorage as its			# in-memory backend storage is attached to the RemoteStorage as its
	# journal_writer attribute.			# journal_writer attribute.
	storage = swh_rpc_client			storage = swh_rpc_client

	journal_writer = getattr(storage, 'journal_writer', None)			journal_writer = getattr(storage, 'journal_writer', None)
	storage.journal_writer = app_server.storage.journal_writer			storage.journal_writer = app_server.storage.journal_writer
	yield storage			yield storage
	storage.journal_writer = journal_writer			storage.journal_writer = journal_writer


	class TestStorage(_TestStorage):			class TestStorage(_TestStorage):
	def test_content_update(self, swh_storage, app_server):			def test_content_update(self, swh_storage, app_server):
	# TODO, journal_writer not supported			# TODO, journal_writer not supported
	swh_storage.journal_writer.journal = None			swh_storage.journal_writer.journal = None
	with patch.object(server.storage.journal_writer, 'journal', None):			with patch.object(server.storage.journal_writer, 'journal', None):
	super().test_content_update(swh_storage)			super().test_content_update(swh_storage)