Changeset View
Changeset View
Standalone View
Standalone View
swh/scheduler/tests/test_cli_task_type.py
- This file was added.
| # Copyright (C) 2019 The Software Heritage developers | |||||
| # See the AUTHORS file at the top-level directory of this distribution | |||||
| # License: GNU General Public License version 3, or any later version | |||||
| # See top-level LICENSE file for more information | |||||
| import pytest | |||||
| import traceback | |||||
| import yaml | |||||
| import pkg_resources | |||||
| from click.testing import CliRunner | |||||
| from swh.scheduler import get_scheduler | |||||
| from swh.scheduler.cli import cli | |||||
| FAKE_MODULE_ENTRY_POINTS = { | |||||
| 'lister.gnu=swh.lister.gnu:register', | |||||
| 'lister.pypi=swh.lister.pypi:register', | |||||
| } | |||||
| @pytest.fixture | |||||
| def mock_pkg_resources(monkeypatch): | |||||
| """Monkey patch swh.scheduler's mock_pkg_resources.iter_entry_point call | |||||
| """ | |||||
| def fake_iter_entry_points(*args, **kwargs): | |||||
| """Substitute fake function to return a fixed set of entrypoints | |||||
| """ | |||||
| from pkg_resources import EntryPoint, Distribution | |||||
| d = Distribution() | |||||
| return [EntryPoint.parse(entry, dist=d) | |||||
| for entry in FAKE_MODULE_ENTRY_POINTS] | |||||
| original_method = pkg_resources.iter_entry_points | |||||
| monkeypatch.setattr( | |||||
| pkg_resources, "iter_entry_points", fake_iter_entry_points) | |||||
| # Now making sure the monkeypatching step is working | |||||
| from swh.scheduler.cli.task_type import load_worker_modules | |||||
| assert len(list(load_worker_modules())) == len(FAKE_MODULE_ENTRY_POINTS) | |||||
| yield | |||||
| # reset monkeypatch: is that needed? | |||||
| monkeypatch.setattr( | |||||
| pkg_resources, "iter_entry_points", original_method) | |||||
| def test_register_ttypes(mock_pkg_resources, swh_scheduler_config, tmp_path): | |||||
| configfile = tmp_path / 'config.yml' | |||||
| config = { | |||||
| 'scheduler': { | |||||
| 'cls': 'local', | |||||
| 'args': swh_scheduler_config | |||||
| } | |||||
| } | |||||
| configfile.write_text(yaml.dump(config)) | |||||
| runner = CliRunner() | |||||
| result = runner.invoke(cli, [ | |||||
| '--config-file', configfile.as_posix(), | |||||
| 'task-type', 'register', | |||||
| ]) | |||||
| assert result.exit_code == 0, traceback.print_exception(*result.exc_info) | |||||
| scheduler = get_scheduler(**config['scheduler']) | |||||
| all_tasks = [ | |||||
| 'list-gnu-full', | |||||
| 'list-pypi', | |||||
| ] | |||||
| for task in all_tasks: | |||||
| task_type_desc = scheduler.get_task_type(task) | |||||
| assert task_type_desc | |||||
| assert task_type_desc['type'] == task | |||||
| assert task_type_desc['backoff_factor'] == 1 | |||||