Changeset View
Changeset View
Standalone 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": "failed", | |||||
} | |||||
anlambert: You can remove that if you return the failed status after catching the NotFound exception. | |||||
ardumontAuthorUnsubmitted Done Inline ActionsYes, i mislead myself here ¯\_(ツ)_/¯ 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 ActionsIs 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 Actionsunsure, i think i added because otherwise, it'd be "eventful" (default implem ¯\_(ツ)_/¯). 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": "failed"} | |||||
_check_load_failure(caplog, loader, NotFound, "Unknown origin!", status="not_found") |
You can remove that if you return the failed status after catching the NotFound exception.