Differential D5039 Diff 18043 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 swh.loader.core.loader import DEFAULT_CONFIG, BaseLoader, DVCSLoader		from swh.loader.core.loader import DEFAULT_CONFIG, BaseLoader, DVCSLoader
		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 Origin, OriginVisit, Snapshot		from swh.model.model import Origin, OriginVisit, Snapshot

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


class DummyLoader:		class DummyLoader:
▲ Show 20 Lines • Show All 111 Lines • ▼ Show 20 Lines	def test_loader_save_data_path(swh_config, tmp_path):

save_path = loader.get_save_data_path()		save_path = loader.get_save_data_path()
assert save_path == expected_save_path		assert save_path == expected_save_path


def _check_load_failure(caplog, loader, exc_class, exc_text, status="partial"):		def _check_load_failure(caplog, loader, exc_class, exc_text, status="partial"):
"""Check whether a failed load properly logged its exception, and that the		"""Check whether a failed load properly logged its exception, and that the
snapshot didn't get referenced in storage"""		snapshot didn't get referenced in storage"""
		assert isinstance(loader, DVCSLoader) # was implicit so far
for record in caplog.records:		for record in caplog.records:
if record.levelname != "ERROR":		if record.levelname != "ERROR":
continue		continue
assert "Loading failure" in record.message		assert "Loading failure" in record.message
assert record.exc_info		assert record.exc_info
exc = record.exc_info[1]		exc = record.exc_info[1]
assert isinstance(exc, exc_class)		assert isinstance(exc, exc_class)
assert exc_text in exc.args[0]		assert exc_text in exc.args[0]
▲ Show 20 Lines • Show All 62 Lines • ▼ Show 20 Lines	def test_dvcs_loader_storage_exc_failed_visit(swh_config, caplog):
loader = DummyDVCSLoaderStorageExc(logging_class=logger_name)		loader = DummyDVCSLoaderStorageExc(logging_class=logger_name)
result = loader.load()		result = loader.load()

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

_check_load_failure(		_check_load_failure(
caplog, loader, RuntimeError, "Failed to add snapshot!", status="failed"		caplog, loader, RuntimeError, "Failed to add snapshot!", status="failed"
)		)


		class DummyDVCSLoaderNotFound(DummyDVCSLoader, BaseLoader):
		"""A loader which raises a not_found exception during the prepare method call

		"""

		def prepare(args, *kwargs):
		raise NotFound("Unknown origin!")

		def load_status(self):
		return {
		"status": "uneventful",
		anlambertUnsubmitted Not Done Inline Actions You can remove that if you return the failed status after catching the NotFound exception. anlambert: You can remove that if you return the failed status after catching the NotFound exception.
		ardumontAuthorUnsubmitted Done Inline Actions Yes, i mislead myself here ¯\_(ツ)_/¯ (and thus you as well... sorry) It should have been uneventful to be consistent with the actual runtime behavior. ardumont: Yes, i mislead myself here ¯\_(ツ)_/¯ (and thus you as well... sorry) It should have been…
		anlambertUnsubmitted Not Done Inline Actions Is the `load_status` override really needed here ? This should already return the `uneventful` status if I am not wrong. anlambert: Is the `load_status` override really needed here ? This should already return the `uneventful`…
		ardumontAuthorUnsubmitted Done Inline Actions unsure, i think i added because otherwise, it'd be "eventful" (default implem ¯\_(ツ)_/¯). I'll check ;) ardumont: unsure, i think i added because otherwise, it'd be "eventful" (default implem ¯\_(ツ)_/¯). I'll…
		}


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

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

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