Differential D2972 Diff 10711 swh/storage/tests/test_api_client_dicts.py

Changeset View

Standalone View

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			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.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
				from swh.storage.tests.conftest import BWCompatInMemoryJournalWriter
				from swh.storage.tests.test_api_client import swh_storage # noqa

	# 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",			"cls": "validate",
	"storage": {"cls": "memory", "journal_writer": {"cls": "memory",},},			"storage": {"cls": "memory", "journal_writer": {"cls": "memory",},},
	}			}
				with patch(
				"swh.journal.writer.inmemory.InMemoryJournalWriter",
				return_value=BWCompatInMemoryJournalWriter(),
				):
	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			return RemoteStorage


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

	@pytest.mark.skip("non-applicable test")			@pytest.mark.skip("non-applicable test")
	def test_content_add_from_lazy_content(self):			def test_content_add_from_lazy_content(self):
	pass			pass