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 list_plugin_worker_descriptions | |||||
| assert len(list(list_plugin_worker_descriptions())) == len( | |||||
| FAKE_MODULE_ENTRY_POINTS) | |||||
| yield | |||||
| # reset monkeypatch: is that needed? | |||||
| monkeypatch.setattr( | |||||
| pkg_resources, "iter_entry_points", original_method) | |||||
| @pytest.fixture | |||||
| def local_sched_config(swh_scheduler_config): | |||||
| """Expose the local scheduler configuration | |||||
| """ | |||||
| return { | |||||
| 'scheduler': { | |||||
| 'cls': 'local', | |||||
| 'args': swh_scheduler_config | |||||
| } | |||||
| } | |||||
| @pytest.fixture | |||||
| def local_sched_configfile(local_sched_config, tmp_path): | |||||
| """Write in temporary location the local scheduler configuration | |||||
| """ | |||||
| configfile = tmp_path / 'config.yml' | |||||
| configfile.write_text(yaml.dump(local_sched_config)) | |||||
| return configfile.as_posix() | |||||
| def test_register_ttypes_all( | |||||
| mock_pkg_resources, local_sched_config, local_sched_configfile): | |||||
| """Registering all task types""" | |||||
| for command in [ | |||||
| ['--config-file', local_sched_configfile, 'task-type', 'register'], | |||||
| ['--config-file', local_sched_configfile, 'task-type', 'register', | |||||
| '-p', 'all'], | |||||
| ['--config-file', local_sched_configfile, 'task-type', 'register', | |||||
| '-p', 'lister.gnu', | |||||
| '-p', 'lister.pypi'], | |||||
| ]: | |||||
| result = CliRunner().invoke(cli, command) | |||||
| assert result.exit_code == 0, traceback.print_exception( | |||||
| *result.exc_info) | |||||
| scheduler = get_scheduler(**local_sched_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 | |||||
| def test_register_ttypes_filter( | |||||
| mock_pkg_resources, local_sched_config, local_sched_configfile): | |||||
| """Filtering on one worker should only register its associated task type | |||||
| """ | |||||
| result = CliRunner().invoke(cli, [ | |||||
| '--config-file', local_sched_configfile, | |||||
| 'task-type', 'register', '--plugins', 'lister.gnu' | |||||
| ]) | |||||
| assert result.exit_code == 0, traceback.print_exception(*result.exc_info) | |||||
| scheduler = get_scheduler(**local_sched_config['scheduler']) | |||||
| all_tasks = [ | |||||
| 'list-gnu-full', | |||||
| ] | |||||
| 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 | |||||