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 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 Origin, OriginVisit, Snapshot | from swh.model.model import Origin, OriginVisit, Snapshot | ||||
| ORIGIN = Origin(url="some-url") | ORIGIN = Origin(url="some-url") | ||||
| ▲ Show 20 Lines • Show All 53 Lines • ▼ Show 20 Lines | 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 | ||||
| def test_base_loader(swh_config): | def test_base_loader(swh_storage): | ||||
| loader = DummyBaseLoader() | 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_config): | def test_base_loader_with_config(swh_storage): | ||||
| loader = DummyBaseLoader("logger-name", DEFAULT_CONFIG) | loader = DummyBaseLoader(swh_storage, "logger-name") | ||||
| result = loader.load() | result = loader.load() | ||||
| assert result == {"status": "eventful"} | assert result == {"status": "eventful"} | ||||
| def test_dvcs_loader(swh_config): | def test_dvcs_loader(swh_storage): | ||||
| loader = DummyDVCSLoader() | 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_config): | def test_dvcs_loader_with_config(swh_storage): | ||||
| loader = DummyDVCSLoader("another-logger", DEFAULT_CONFIG) | loader = DummyDVCSLoader(swh_storage, "another-logger") | ||||
| result = loader.load() | result = loader.load() | ||||
| assert result == {"status": "eventful"} | assert result == {"status": "eventful"} | ||||
| def test_loader_logger_default_name(swh_config): | def test_loader_logger_default_name(swh_storage): | ||||
| loader = DummyBaseLoader() | loader = DummyBaseLoader(swh_storage) | ||||
| assert isinstance(loader.log, logging.Logger) | assert isinstance(loader.log, logging.Logger) | ||||
| assert loader.log.name == "swh.loader.core.tests.test_loader.DummyBaseLoader" | assert loader.log.name == "swh.loader.core.tests.test_loader.DummyBaseLoader" | ||||
| loader = DummyDVCSLoader() | loader = DummyDVCSLoader(swh_storage) | ||||
| assert isinstance(loader.log, logging.Logger) | assert isinstance(loader.log, logging.Logger) | ||||
| assert loader.log.name == "swh.loader.core.tests.test_loader.DummyDVCSLoader" | assert loader.log.name == "swh.loader.core.tests.test_loader.DummyDVCSLoader" | ||||
| def test_loader_logger_with_name(swh_config): | def test_loader_logger_with_name(swh_storage): | ||||
| loader = DummyBaseLoader("some.logger.name") | loader = DummyBaseLoader(swh_storage, "some.logger.name") | ||||
| assert isinstance(loader.log, logging.Logger) | assert isinstance(loader.log, logging.Logger) | ||||
| assert loader.log.name == "some.logger.name" | assert loader.log.name == "some.logger.name" | ||||
| def test_loader_save_data_path(swh_config, tmp_path): | def test_loader_save_data_path(swh_storage, tmp_path): | ||||
| loader = DummyBaseLoader("some.logger.name.1") | loader = DummyBaseLoader(swh_storage, "some.logger.name.1", save_data_path=tmp_path) | ||||
| url = "http://bitbucket.org/something" | url = "http://bitbucket.org/something" | ||||
| loader.origin = Origin(url=url) | loader.origin = Origin(url=url) | ||||
| loader.visit_date = datetime.datetime(year=2019, month=10, day=1) | loader.visit_date = datetime.datetime(year=2019, month=10, day=1) | ||||
| loader.config = { | |||||
| "save_data_path": tmp_path, | |||||
| } | |||||
| hash_url = hashlib.sha1(url.encode("utf-8")).hexdigest() | hash_url = hashlib.sha1(url.encode("utf-8")).hexdigest() | ||||
| expected_save_path = "%s/sha1:%s/%s/2019" % (str(tmp_path), hash_url[0:2], hash_url) | expected_save_path = "%s/sha1:%s/%s/2019" % (str(tmp_path), hash_url[0:2], hash_url) | ||||
| 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 | ||||
| Show All 23 Lines | |||||
| 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!") | ||||
| def test_dvcs_loader_exc_partial_visit(swh_config, caplog): | def test_dvcs_loader_exc_partial_visit(swh_storage, caplog): | ||||
| logger_name = "dvcsloaderexc" | logger_name = "dvcsloaderexc" | ||||
| caplog.set_level(logging.ERROR, logger=logger_name) | caplog.set_level(logging.ERROR, logger=logger_name) | ||||
| loader = DummyDVCSLoaderExc(logging_class=logger_name) | loader = DummyDVCSLoaderExc(swh_storage, logging_class=logger_name) | ||||
| # fake the loading ending up in a snapshot | # fake the loading ending up in a snapshot | ||||
| loader.loaded_snapshot_id = hash_to_bytes( | loader.loaded_snapshot_id = hash_to_bytes( | ||||
| "9e4dd2b40d1b46b70917c0949aa2195c823a648e" | "9e4dd2b40d1b46b70917c0949aa2195c823a648e" | ||||
| ) | ) | ||||
| result = loader.load() | result = loader.load() | ||||
| # loading failed | # loading failed | ||||
| assert result == {"status": "failed"} | assert result == {"status": "failed"} | ||||
| Show All 18 Lines | |||||
| class DummyDVCSLoaderStorageExc(DummyDVCSLoader): | class DummyDVCSLoaderStorageExc(DummyDVCSLoader): | ||||
| """A loader which raises an exception when loading some contents""" | """A loader which raises an exception when loading some contents""" | ||||
| def __init__(self, *args, **kwargs): | def __init__(self, *args, **kwargs): | ||||
| super().__init__(*args, **kwargs) | super().__init__(*args, **kwargs) | ||||
| self.storage = BrokenStorageProxy(self.storage) | self.storage = BrokenStorageProxy(self.storage) | ||||
| def test_dvcs_loader_storage_exc_failed_visit(swh_config, caplog): | def test_dvcs_loader_storage_exc_failed_visit(swh_storage, caplog): | ||||
| logger_name = "dvcsloaderexc" | logger_name = "dvcsloaderexc" | ||||
| caplog.set_level(logging.ERROR, logger=logger_name) | caplog.set_level(logging.ERROR, logger=logger_name) | ||||
| loader = DummyDVCSLoaderStorageExc(logging_class=logger_name) | loader = DummyDVCSLoaderStorageExc(swh_storage, 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): | class DummyDVCSLoaderNotFound(DummyDVCSLoader, BaseLoader): | ||||
| """A loader which raises a not_found exception during the prepare method call | """A loader which raises a not_found exception during the prepare method call | ||||
| """ | """ | ||||
| def prepare(*args, **kwargs): | def prepare(*args, **kwargs): | ||||
| raise NotFound("Unknown origin!") | raise NotFound("Unknown origin!") | ||||
| def load_status(self): | def load_status(self): | ||||
| return { | return { | ||||
| "status": "uneventful", | "status": "uneventful", | ||||
| } | } | ||||
| def test_loader_not_found(swh_config, caplog): | def test_loader_not_found(swh_storage, caplog): | ||||
| loader = DummyDVCSLoaderNotFound() | loader = DummyDVCSLoaderNotFound(swh_storage) | ||||
| result = loader.load() | result = loader.load() | ||||
| assert result == {"status": "uneventful"} | assert result == {"status": "uneventful"} | ||||
| _check_load_failure(caplog, loader, NotFound, "Unknown origin!", status="not_found") | _check_load_failure(caplog, loader, NotFound, "Unknown origin!", status="not_found") | ||||