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 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 |