diff --git a/setup.py b/setup.py --- a/setup.py +++ b/setup.py @@ -55,6 +55,8 @@ swh-scheduler=swh.scheduler.cli:main [swh.cli.subcommands] scheduler=swh.scheduler.cli:cli + [pytest11] + pytest_swh_scheduler=swh.scheduler.pytest_plugin """, classifiers=[ "Programming Language :: Python :: 3", diff --git a/swh/scheduler/pytest_plugin.py b/swh/scheduler/pytest_plugin.py new file mode 100644 --- /dev/null +++ b/swh/scheduler/pytest_plugin.py @@ -0,0 +1,63 @@ +# Copyright (C) 2020 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 + +from datetime import timedelta +import glob +import os + +import pytest + +from swh.core.utils import numfile_sortkey as sortkey + +import swh.scheduler +from swh.scheduler import get_scheduler + + +SQL_DIR = os.path.join(os.path.dirname(swh.scheduler.__file__), "sql") +DUMP_FILES = os.path.join(SQL_DIR, "*.sql") + +# celery tasks for testing purpose; tasks themselves should be +# in swh/scheduler/tests/tasks.py +TASK_NAMES = ["ping", "multiping", "add", "error", "echo"] + + +@pytest.fixture +def swh_scheduler_config(request, postgresql): + scheduler_config = { + "db": postgresql.dsn, + } + + all_dump_files = sorted(glob.glob(DUMP_FILES), key=sortkey) + + cursor = postgresql.cursor() + for fname in all_dump_files: + with open(fname) as fobj: + cursor.execute(fobj.read()) + postgresql.commit() + + return scheduler_config + + +@pytest.fixture +def swh_scheduler(swh_scheduler_config): + scheduler = get_scheduler("local", swh_scheduler_config) + for taskname in TASK_NAMES: + scheduler.create_task_type( + { + "type": "swh-test-{}".format(taskname), + "description": "The {} testing task".format(taskname), + "backend_name": "swh.scheduler.tests.tasks.{}".format(taskname), + "default_interval": timedelta(days=1), + "min_interval": timedelta(hours=6), + "max_interval": timedelta(days=12), + } + ) + + return scheduler + + +# this alias is used to be able to easily instantiate a db-backed Scheduler +# eg. for the RPC client/server test suite. +swh_db_scheduler = swh_scheduler diff --git a/swh/scheduler/tests/__init__.py b/swh/scheduler/tests/__init__.py --- a/swh/scheduler/tests/__init__.py +++ b/swh/scheduler/tests/__init__.py @@ -1,5 +0,0 @@ -from os import path -import swh.scheduler - - -SQL_DIR = path.join(path.dirname(swh.scheduler.__file__), "sql") diff --git a/swh/scheduler/tests/conftest.py b/swh/scheduler/tests/conftest.py --- a/swh/scheduler/tests/conftest.py +++ b/swh/scheduler/tests/conftest.py @@ -5,14 +5,10 @@ import os import pytest -import glob -from datetime import datetime, timedelta, timezone +from datetime import datetime, timezone import pkg_resources from typing import List -from swh.core.utils import numfile_sortkey as sortkey -from swh.scheduler import get_scheduler -from swh.scheduler.tests import SQL_DIR from swh.scheduler.model import ListedOrigin, Lister from swh.scheduler.tests.common import LISTERS @@ -25,12 +21,6 @@ # test_cli tests depends on a en/C locale, so ensure it os.environ["LC_ALL"] = "C.UTF-8" -DUMP_FILES = os.path.join(SQL_DIR, "*.sql") - -# celery tasks for testing purpose; tasks themselves should be -# in swh/scheduler/tests/tasks.py -TASK_NAMES = ["ping", "multiping", "add", "error", "echo"] - @pytest.fixture(scope="session") def celery_enable_logging(): @@ -74,46 +64,6 @@ yield celery_session_app -@pytest.fixture -def swh_scheduler_config(request, postgresql): - scheduler_config = { - "db": postgresql.dsn, - } - - all_dump_files = sorted(glob.glob(DUMP_FILES), key=sortkey) - - cursor = postgresql.cursor() - for fname in all_dump_files: - with open(fname) as fobj: - cursor.execute(fobj.read()) - postgresql.commit() - - return scheduler_config - - -@pytest.fixture -def swh_scheduler(swh_scheduler_config): - scheduler = get_scheduler("local", swh_scheduler_config) - for taskname in TASK_NAMES: - scheduler.create_task_type( - { - "type": "swh-test-{}".format(taskname), - "description": "The {} testing task".format(taskname), - "backend_name": "swh.scheduler.tests.tasks.{}".format(taskname), - "default_interval": timedelta(days=1), - "min_interval": timedelta(hours=6), - "max_interval": timedelta(days=12), - } - ) - - return scheduler - - -# this alias is used to be able to easily instantiate a db-backed Scheduler -# eg. for the RPC client/server test suite. -swh_db_scheduler = swh_scheduler - - @pytest.fixture def stored_lister(swh_scheduler) -> Lister: """Store a lister in the scheduler and return its information"""