Differential D7632 Diff 27637 swh/loader/core/tests/test_loader.py

Changeset View

Standalone View

swh/loader/core/tests/test_loader.py

	# Copyright (C) 2018-2021 The Software Heritage developers			# Copyright (C) 2018-2021 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 datetime			import datetime
	import hashlib			import hashlib
	import logging			import logging
				from unittest.mock import MagicMock

				import pytest

	from swh.loader.core.loader import BaseLoader, DVCSLoader			from swh.loader.core.loader import BaseLoader, DVCSLoader
	from swh.loader.exception import NotFound			from swh.loader.exception import NotFound
	from swh.loader.tests import assert_last_visit_matches			from swh.loader.tests import assert_last_visit_matches
	from swh.model.hashutil import hash_to_bytes			from swh.model.hashutil import hash_to_bytes
	from swh.model.model import (			from swh.model.model import (
	MetadataAuthority,			MetadataAuthority,
	MetadataAuthorityType,			MetadataAuthorityType,
	MetadataFetcher,			MetadataFetcher,
	Origin,			Origin,
	RawExtrinsicMetadata,			RawExtrinsicMetadata,
	Snapshot,			Snapshot,
	)			)
				import swh.storage.exc

	ORIGIN = Origin(url="some-url")			ORIGIN = Origin(url="some-url")

	METADATA_AUTHORITY = MetadataAuthority(			METADATA_AUTHORITY = MetadataAuthority(
	type=MetadataAuthorityType.FORGE, url="http://example.org/"			type=MetadataAuthorityType.FORGE, url="http://example.org/"
	)			)
	REMD = RawExtrinsicMetadata(			REMD = RawExtrinsicMetadata(
	target=ORIGIN.swhid(),			target=ORIGIN.swhid(),
	▲ Show 20 Lines • Show All 54 Lines • ▼ Show 20 Lines

	class DummyBaseLoader(DummyLoader, BaseLoader):			class DummyBaseLoader(DummyLoader, BaseLoader):
	"""Buffered loader will send new data when threshold is reached"""			"""Buffered loader will send new data when threshold is reached"""

	def store_data(self):			def store_data(self):
	pass			pass


				class DummyMetadataFetcher:
				SUPPORTED_LISTERS = {"fake-lister"}

				def __init__(self, origin, credentials, lister_name, lister_instance):
				pass

				def get_origin_metadata(self):
				return [REMD]


	def test_base_loader(swh_storage):			def test_base_loader(swh_storage):
	loader = DummyBaseLoader(swh_storage)			loader = DummyBaseLoader(swh_storage)
	result = loader.load()			result = loader.load()
	assert result == {"status": "eventful"}			assert result == {"status": "eventful"}


	def test_base_loader_with_config(swh_storage):			def test_base_loader_with_config(swh_storage):
	loader = DummyBaseLoader(swh_storage, "logger-name")			loader = DummyBaseLoader(swh_storage, "logger-name")
	result = loader.load()			result = loader.load()
	assert result == {"status": "eventful"}			assert result == {"status": "eventful"}


				def test_base_loader_with_known_lister_name(swh_storage, mocker):
				fetcher_cls = MagicMock(wraps=DummyMetadataFetcher)
				fetcher_cls.SUPPORTED_LISTERS = DummyMetadataFetcher.SUPPORTED_LISTERS
				mocker.patch(
				"swh.loader.core.metadata_fetchers._fetchers", return_value=[fetcher_cls]
				)

				loader = DummyBaseLoader(swh_storage, lister_name="fake-lister", lister_instance="")
				result = loader.load()
				assert result == {"status": "eventful"}

				fetcher_cls.assert_called_once()
				fetcher_cls.assert_called_once_with(
				origin=ORIGIN, credentials={}, lister_name="fake-lister", lister_instance=""
				)
				assert swh_storage.raw_extrinsic_metadata_get(
				ORIGIN.swhid(), METADATA_AUTHORITY
				).results == [REMD]


				def test_base_loader_with_unknown_lister_name(swh_storage, mocker):
				fetcher_cls = MagicMock(wraps=DummyMetadataFetcher)
				fetcher_cls.SUPPORTED_LISTERS = DummyMetadataFetcher.SUPPORTED_LISTERS
				mocker.patch(
				"swh.loader.core.metadata_fetchers._fetchers", return_value=[fetcher_cls]
				)

				loader = DummyBaseLoader(
				swh_storage, lister_name="other-lister", lister_instance=""
				)
				result = loader.load()
				assert result == {"status": "eventful"}

				fetcher_cls.assert_not_called()
				with pytest.raises(swh.storage.exc.StorageArgumentException):
				swh_storage.raw_extrinsic_metadata_get(ORIGIN.swhid(), METADATA_AUTHORITY)


	def test_dvcs_loader(swh_storage):			def test_dvcs_loader(swh_storage):
	loader = DummyDVCSLoader(swh_storage)			loader = DummyDVCSLoader(swh_storage)
	result = loader.load()			result = loader.load()
	assert result == {"status": "eventful"}			assert result == {"status": "eventful"}


	def test_dvcs_loader_with_config(swh_storage):			def test_dvcs_loader_with_config(swh_storage):
	loader = DummyDVCSLoader(swh_storage, "another-logger")			loader = DummyDVCSLoader(swh_storage, "another-logger")
	▲ Show 20 Lines • Show All 139 Lines • Show Last 20 Lines