Changeset View
Standalone View
swh/loader/core/tests/test_loader.py
Show All 18 Lines | from swh.model.model import ( | ||||
MetadataAuthorityType, | MetadataAuthorityType, | ||||
MetadataFetcher, | MetadataFetcher, | ||||
Origin, | Origin, | ||||
RawExtrinsicMetadata, | RawExtrinsicMetadata, | ||||
Snapshot, | Snapshot, | ||||
) | ) | ||||
import swh.storage.exc | import swh.storage.exc | ||||
ORIGIN = Origin(url="some-url") | ORIGIN = Origin(url="some-url") | ||||
olasd: Please keep these consistent (I guess by always using the kwargs form)? | |||||
PARENT_ORIGIN = Origin(url="base-origin-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(), | ||||
discovery_date=datetime.datetime.now(tz=datetime.timezone.utc), | discovery_date=datetime.datetime.now(tz=datetime.timezone.utc), | ||||
authority=METADATA_AUTHORITY, | authority=METADATA_AUTHORITY, | ||||
▲ Show 20 Lines • Show All 60 Lines • ▼ Show 20 Lines | |||||
class DummyMetadataFetcher: | class DummyMetadataFetcher: | ||||
SUPPORTED_LISTERS = {"fake-lister"} | SUPPORTED_LISTERS = {"fake-lister"} | ||||
def __init__(self, origin, credentials, lister_name, lister_instance_name): | def __init__(self, origin, credentials, lister_name, lister_instance_name): | ||||
pass | pass | ||||
def get_origin_metadata(self): | def get_origin_metadata(self): | ||||
return [REMD] | return [REMD] | ||||
def get_parent_origins(self): | |||||
Done Inline Actions...because this one doesn't match the MetadataFetcherProtocol. Maybe it would be worth adding a test that all known/registered MetadataFetchers do match the protocol? olasd: ...because this one doesn't match the `MetadataFetcherProtocol`. Maybe it would be worth adding… | |||||
Done Inline ActionsI'll just updated this one instead. And it's tested in swh-metadata-loaders. Or do you mean runtime checks? vlorentz: I'll just updated this one instead.
And it's tested in swh-metadata-loaders. Or do you mean… | |||||
Not Done Inline ActionsNo, I meant consistency checks at testing time. It's probably worth adding an ad-hoc test here for these two mock classes? olasd: No, I meant consistency checks at testing time. It's probably worth adding an ad-hoc test here… | |||||
Done Inline ActionsI can, but it wouldn't have caught this issue; isinstance doesn't check type signatures. https://docs.python.org/3/library/typing.html#typing.runtime_checkable vlorentz: I can, but it wouldn't have caught this issue; `isinstance` doesn't check type signatures. | |||||
Not Done Inline ActionsI mean, we do have such a set of explicit tests for our StorageInterface, ... etc. implementations. I agree that this may be overkill for such a small interface, though. :-) olasd: I mean, we do have such a set of explicit tests for our `StorageInterface`, ... etc. | |||||
Done Inline Actionsdone vlorentz: done | |||||
Done Inline Actionsoh, you're right. I'll consider it in swh-loader-metadata; but I don't think it's worth it to check test classes this deeply vlorentz: oh, you're right. I'll consider it in swh-loader-metadata; but I don't think it's worth it to… | |||||
return [] | |||||
class DummyMetadataFetcherWithFork: | |||||
SUPPORTED_LISTERS = {"fake-lister"} | |||||
def __init__(self, origin, credentials, lister_name, lister_instance_name): | |||||
pass | |||||
def get_origin_metadata(self): | |||||
return [REMD] | |||||
def get_parent_origins(self): | |||||
return [PARENT_ORIGIN] | |||||
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): | ||||
Show All 20 Lines | fetcher_cls.assert_called_once_with( | ||||
origin=ORIGIN, | origin=ORIGIN, | ||||
credentials={}, | credentials={}, | ||||
lister_name="fake-lister", | lister_name="fake-lister", | ||||
lister_instance_name="", | lister_instance_name="", | ||||
) | ) | ||||
assert swh_storage.raw_extrinsic_metadata_get( | assert swh_storage.raw_extrinsic_metadata_get( | ||||
ORIGIN.swhid(), METADATA_AUTHORITY | ORIGIN.swhid(), METADATA_AUTHORITY | ||||
).results == [REMD] | ).results == [REMD] | ||||
assert loader.parent_origins is None | |||||
def test_base_loader_with_unknown_lister_name(swh_storage, mocker): | def test_base_loader_with_unknown_lister_name(swh_storage, mocker): | ||||
fetcher_cls = MagicMock(wraps=DummyMetadataFetcher) | fetcher_cls = MagicMock(wraps=DummyMetadataFetcher) | ||||
fetcher_cls.SUPPORTED_LISTERS = DummyMetadataFetcher.SUPPORTED_LISTERS | fetcher_cls.SUPPORTED_LISTERS = DummyMetadataFetcher.SUPPORTED_LISTERS | ||||
mocker.patch( | mocker.patch( | ||||
"swh.loader.core.metadata_fetchers._fetchers", return_value=[fetcher_cls] | "swh.loader.core.metadata_fetchers._fetchers", return_value=[fetcher_cls] | ||||
) | ) | ||||
loader = DummyBaseLoader( | loader = DummyBaseLoader( | ||||
swh_storage, lister_name="other-lister", lister_instance_name="" | swh_storage, lister_name="other-lister", lister_instance_name="" | ||||
) | ) | ||||
result = loader.load() | result = loader.load() | ||||
assert result == {"status": "eventful"} | assert result == {"status": "eventful"} | ||||
fetcher_cls.assert_not_called() | fetcher_cls.assert_not_called() | ||||
with pytest.raises(swh.storage.exc.StorageArgumentException): | with pytest.raises(swh.storage.exc.StorageArgumentException): | ||||
swh_storage.raw_extrinsic_metadata_get(ORIGIN.swhid(), METADATA_AUTHORITY) | swh_storage.raw_extrinsic_metadata_get(ORIGIN.swhid(), METADATA_AUTHORITY) | ||||
def test_base_loader_forked_origin(swh_storage, mocker): | |||||
fetcher_cls = MagicMock(wraps=DummyMetadataFetcherWithFork) | |||||
fetcher_cls.SUPPORTED_LISTERS = DummyMetadataFetcherWithFork.SUPPORTED_LISTERS | |||||
mocker.patch( | |||||
"swh.loader.core.metadata_fetchers._fetchers", return_value=[fetcher_cls] | |||||
) | |||||
loader = DummyBaseLoader( | |||||
swh_storage, lister_name="fake-lister", lister_instance_name="" | |||||
) | |||||
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_name="", | |||||
) | |||||
assert swh_storage.raw_extrinsic_metadata_get( | |||||
ORIGIN.swhid(), METADATA_AUTHORITY | |||||
).results == [REMD] | |||||
assert loader.parent_origins == [PARENT_ORIGIN] | |||||
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 |
Please keep these consistent (I guess by always using the kwargs form)?