Differential D8430 Diff 30442 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
	import time			import time
	from unittest.mock import MagicMock, call			from unittest.mock import MagicMock, call

	import pytest			import pytest

	from swh.loader.core.loader import BaseLoader, DVCSLoader			from swh.loader.core.loader import (
				SENTRY_ORIGIN_URL_TAG_NAME,
				SENTRY_VISIT_TYPE_TAG_NAME,
				BaseLoader,
				DVCSLoader,
				)
	from swh.loader.core.metadata_fetchers import MetadataFetcherProtocol			from swh.loader.core.metadata_fetchers import MetadataFetcherProtocol
	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,
	▲ Show 20 Lines • Show All 451 Lines • ▼ Show 20 Lines

	def test_loader_not_found(swh_storage, caplog):			def test_loader_not_found(swh_storage, caplog):
	loader = DummyDVCSLoaderNotFound(swh_storage)			loader = DummyDVCSLoaderNotFound(swh_storage)
	result = loader.load()			result = loader.load()

	assert result == {"status": "uneventful"}			assert result == {"status": "uneventful"}

	_check_load_failure(caplog, loader, NotFound, "Unknown origin!", status="not_found")			_check_load_failure(caplog, loader, NotFound, "Unknown origin!", status="not_found")


				class DummyLoaderWithError(DummyBaseLoader):
				def prepare(self, args, *kwargs):
				raise Exception("error")


				class DummyDVCSLoaderWithError(DummyDVCSLoader, BaseLoader):
				def prepare(self, args, *kwargs):
				raise Exception("error")


				@pytest.mark.parametrize("loader_cls", [DummyLoaderWithError, DummyDVCSLoaderWithError])
				def test_loader_sentry_tags_on_error(swh_storage, sentry_events, loader_cls):
				loader = loader_cls(swh_storage)
				loader.load()
				sentry_tags = sentry_events[0]["tags"]
				assert sentry_tags.get(SENTRY_ORIGIN_URL_TAG_NAME) == ORIGIN.url
				assert sentry_tags.get(SENTRY_VISIT_TYPE_TAG_NAME) == DummyLoader.visit_type