Differential D3430 Diff 12160 swh/scheduler/tests/conftest.py

Changeset 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			from datetime import datetime, timezone
	from datetime import datetime, timedelta, timezone
	import pkg_resources			import pkg_resources
	from typing import List			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.model import ListedOrigin, Lister
	from swh.scheduler.tests.common import LISTERS			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
	os.environ["LC_ALL"] = "C.UTF-8"			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")			@pytest.fixture(scope="session")
	def celery_enable_logging():			def celery_enable_logging():
	return True			return True


	@pytest.fixture(scope="session")			@pytest.fixture(scope="session")
	def celery_includes():			def celery_includes():
	Show All 28 Lines
	def swh_app(celery_session_app):			def swh_app(celery_session_app):
	from swh.scheduler.celery_backend import config			from swh.scheduler.celery_backend import config

	config.app = celery_session_app			config.app = celery_session_app
	yield celery_session_app			yield celery_session_app


	@pytest.fixture			@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:			def stored_lister(swh_scheduler) -> Lister:
	"""Store a lister in the scheduler and return its information"""			"""Store a lister in the scheduler and return its information"""
	return swh_scheduler.get_or_create_lister(**LISTERS[0])			return swh_scheduler.get_or_create_lister(**LISTERS[0])


	@pytest.fixture			@pytest.fixture
	def listed_origins(stored_lister) -> List[ListedOrigin]:			def listed_origins(stored_lister) -> List[ListedOrigin]:
	"""Return a (fixed) set of 1000 listed origins"""			"""Return a (fixed) set of 1000 listed origins"""
	Show All 9 Lines