Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/core/tests/test_loader.py
| # Copyright (C) 2018-2020 The Software Heritage developers | # Copyright (C) 2018-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 | ||||
| import datetime | import datetime | ||||
| import hashlib | import hashlib | ||||
| import logging | import logging | ||||
| from swh.model.model import Origin, OriginVisit, Snapshot | from swh.model.model import Origin, OriginVisit, Snapshot | ||||
| from swh.loader.core.loader import BaseLoader, DVCSLoader | from swh.loader.core.loader import BaseLoader, DVCSLoader | ||||
| from swh.loader.tests.common import assert_last_visit_matches | |||||
| ORIGIN = Origin(url="some-url") | ORIGIN = Origin(url="some-url") | ||||
| class DummyLoader: | class DummyLoader: | ||||
| def cleanup(self): | def cleanup(self): | ||||
| pass | pass | ||||
| ▲ Show 20 Lines • Show All 146 Lines • ▼ Show 20 Lines | def _check_load_failure(caplog, loader, exc_class, exc_text): | ||||
| # Check that the get_snapshot operation would have succeeded | # Check that the get_snapshot operation would have succeeded | ||||
| assert loader.get_snapshot() is not None | assert loader.get_snapshot() is not None | ||||
| # But that the snapshot didn't get loaded | # But that the snapshot didn't get loaded | ||||
| assert loader.loaded_snapshot_id is None | assert loader.loaded_snapshot_id is None | ||||
| # And confirm that the visit doesn't reference a snapshot | # And confirm that the visit doesn't reference a snapshot | ||||
| visit = loader.storage.origin_visit_get_latest(ORIGIN.url) | visit = assert_last_visit_matches(loader.storage, ORIGIN.url, status="partial") | ||||
| assert visit["status"] == "partial" | assert visit.snapshot is None | ||||
| assert visit["snapshot"] is None | |||||
| class DummyDVCSLoaderExc(DummyDVCSLoader): | class DummyDVCSLoaderExc(DummyDVCSLoader): | ||||
| """A loader which raises an exception when loading some contents""" | """A loader which raises an exception when loading some contents""" | ||||
| def get_contents(self): | def get_contents(self): | ||||
| raise RuntimeError("Failed to get contents!") | raise RuntimeError("Failed to get contents!") | ||||
| ▲ Show 20 Lines • Show All 42 Lines • Show Last 20 Lines | |||||