Changeset View
Changeset View
Standalone View
Standalone View
swh/scheduler/tests/test_scheduler.py
| # Copyright (C) 2017-2019 The Software Heritage developers | # Copyright (C) 2017-2019 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 copy | import copy | ||||
| import datetime | import datetime | ||||
| import random | import random | ||||
| import uuid | import uuid | ||||
| from collections import defaultdict | from collections import defaultdict | ||||
| import inspect | |||||
| from typing import Any, Dict | from typing import Any, Dict | ||||
| from arrow import utcnow | from arrow import utcnow | ||||
| from swh.scheduler.interface import SchedulerInterface | |||||
| from .common import tasks_from_template, TEMPLATES, TASK_TYPES | from .common import tasks_from_template, TEMPLATES, TASK_TYPES | ||||
| def subdict(d, keys=None, excl=()): | def subdict(d, keys=None, excl=()): | ||||
| if keys is None: | if keys is None: | ||||
| keys = [k for k in d.keys()] | keys = [k for k in d.keys()] | ||||
| return {k: d[k] for k in keys if k not in excl} | return {k: d[k] for k in keys if k not in excl} | ||||
| class TestScheduler: | class TestScheduler: | ||||
| def test_interface(self, swh_scheduler): | |||||
| """Checks all methods of SchedulerInterface are implemented by this | |||||
| backend, and that they have the same signature.""" | |||||
| # Create an instance of the protocol (which cannot be instantiated | |||||
| # directly, so this creates a subclass, then instantiates it) | |||||
| interface = type("_", (SchedulerInterface,), {})() | |||||
| assert "create_task_type" in dir(interface) | |||||
| missing_methods = [] | |||||
| for meth_name in dir(interface): | |||||
| if meth_name.startswith("_"): | |||||
| continue | |||||
| interface_meth = getattr(interface, meth_name) | |||||
| try: | |||||
| concrete_meth = getattr(swh_scheduler, meth_name) | |||||
| except AttributeError: | |||||
| if not getattr(interface_meth, "deprecated_endpoint", False): | |||||
| # The backend is missing a (non-deprecated) endpoint | |||||
| missing_methods.append(meth_name) | |||||
| continue | |||||
| expected_signature = inspect.signature(interface_meth) | |||||
| actual_signature = inspect.signature(concrete_meth) | |||||
| assert expected_signature == actual_signature, meth_name | |||||
| assert missing_methods == [] | |||||
| def test_get_priority_ratios(self, swh_scheduler): | def test_get_priority_ratios(self, swh_scheduler): | ||||
| assert swh_scheduler.get_priority_ratios() == { | assert swh_scheduler.get_priority_ratios() == { | ||||
| "high": 0.5, | "high": 0.5, | ||||
| "normal": 0.3, | "normal": 0.3, | ||||
| "low": 0.2, | "low": 0.2, | ||||
| } | } | ||||
| def test_add_task_type(self, swh_scheduler): | def test_add_task_type(self, swh_scheduler): | ||||
| ▲ Show 20 Lines • Show All 551 Lines • Show Last 20 Lines | |||||