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 | from swh.model.model import Origin, Snapshot | ||||
| from swh.loader.core.loader import BaseLoader, DVCSLoader | from swh.loader.core.loader import BaseLoader, DVCSLoader | ||||
| ORIGIN = Origin(url="some-url") | ORIGIN = Origin(url="some-url") | ||||
| class DummyLoader: | class DummyLoader: | ||||
| def cleanup(self): | def cleanup(self): | ||||
| pass | pass | ||||
| def prepare(self, *args, **kwargs): | def prepare(self, *args, **kwargs): | ||||
| pass | pass | ||||
| def fetch_data(self): | def fetch_data(self): | ||||
| pass | pass | ||||
| def store_data(self): | |||||
| pass | |||||
| def get_snapshot_id(self): | def get_snapshot_id(self): | ||||
| return None | return None | ||||
| def prepare_origin_visit(self, *args, **kwargs): | def prepare_origin_visit(self, *args, **kwargs): | ||||
| self.origin = ORIGIN | self.origin = ORIGIN | ||||
| self.origin_url = ORIGIN.url | self.origin_url = ORIGIN.url | ||||
| self.visit_date = datetime.datetime.now(tz=datetime.timezone.utc) | self.visit_date = datetime.datetime.now(tz=datetime.timezone.utc) | ||||
| self.visit_type = "git" | self.visit_type = "git" | ||||
| Show All 12 Lines | def parse_config_file(self, *args, **kwargs): | ||||
| return { | return { | ||||
| "max_content_size": 100 * 1024 * 1024, | "max_content_size": 100 * 1024 * 1024, | ||||
| "storage": { | "storage": { | ||||
| "cls": "pipeline", | "cls": "pipeline", | ||||
| "steps": [{"cls": "retry",}, {"cls": "filter",}, {"cls": "memory",},], | "steps": [{"cls": "retry",}, {"cls": "filter",}, {"cls": "memory",},], | ||||
| }, | }, | ||||
| } | } | ||||
| def get_contents(self): | |||||
| return [] | |||||
| def get_directories(self): | |||||
| return [] | |||||
| def get_revisions(self): | |||||
| return [] | |||||
| def get_releases(self): | |||||
| return [] | |||||
| def get_snapshot(self): | |||||
| return Snapshot(branches={}) | |||||
| def eventful(self): | |||||
| return False | |||||
| 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 parse_config_file(self, *args, **kwargs): | def parse_config_file(self, *args, **kwargs): | ||||
| return { | return { | ||||
| Show All 13 Lines | def parse_config_file(self, *args, **kwargs): | ||||
| "release": 2, | "release": 2, | ||||
| }, | }, | ||||
| }, | }, | ||||
| {"cls": "memory",}, | {"cls": "memory",}, | ||||
| ], | ], | ||||
| }, | }, | ||||
| } | } | ||||
| def store_data(self): | |||||
| pass | |||||
| def test_base_loader(): | def test_base_loader(): | ||||
| loader = DummyBaseLoader() | loader = DummyBaseLoader() | ||||
| result = loader.load() | result = loader.load() | ||||
| assert result == {"status": "eventful"} | assert result == {"status": "eventful"} | ||||
| Show All 36 Lines | |||||