Changeset View
Changeset View
Standalone View
Standalone View
swh/scheduler/tests/conftest.py
| # Copyright (C) 2016-2019 The Software Heritage developers | # Copyright (C) 2016-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 os | import os | ||||
| import pytest | import pytest | ||||
| import glob | import glob | ||||
| from datetime import timedelta | from datetime import timedelta | ||||
| import pkg_resources | import pkg_resources | ||||
| from swh.core.utils import numfile_sortkey as sortkey | from swh.core.utils import numfile_sortkey as sortkey | ||||
| from swh.scheduler import get_scheduler | from swh.scheduler import get_scheduler | ||||
| from swh.scheduler.tests import SQL_DIR | from swh.scheduler.tests import SQL_DIR | ||||
| from swh.scheduler.model import Lister | |||||
| from swh.scheduler.tests.common import LISTERS | |||||
| # make sure we are not fooled by CELERY_ config environment vars | # make sure we are not fooled by CELERY_ config environment vars | ||||
| for var in [x for x in os.environ.keys() if x.startswith("CELERY")]: | for var in [x for x in os.environ.keys() if x.startswith("CELERY")]: | ||||
| os.environ.pop(var) | os.environ.pop(var) | ||||
| # test_cli tests depends on a en/C locale, so ensure it | # test_cli tests depends on a en/C locale, so ensure it | ||||
| ▲ Show 20 Lines • Show All 81 Lines • ▼ Show 20 Lines | for taskname in TASK_NAMES: | ||||
| ) | ) | ||||
| return scheduler | return scheduler | ||||
| # this alias is used to be able to easily instantiate a db-backed Scheduler | # this alias is used to be able to easily instantiate a db-backed Scheduler | ||||
| # eg. for the RPC client/server test suite. | # eg. for the RPC client/server test suite. | ||||
| swh_db_scheduler = swh_scheduler | swh_db_scheduler = swh_scheduler | ||||
| @pytest.fixture | |||||
| def stored_lister(swh_scheduler) -> Lister: | |||||
| """Store a lister in the scheduler and return its information""" | |||||
| return swh_scheduler.get_or_create_lister(**LISTERS[0]) | |||||