Differential D3469 Diff 12278 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
	from datetime import datetime, timezone			from datetime import datetime, timezone
	import pkg_resources
	from typing import List			from typing import List

				import pytest

	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

				from swh.scheduler.pytest_plugin import ( # noqa: F401 (backwards-compat imports)
				swh_scheduler_celery_app as swh_app,
				swh_scheduler_celery_worker as celery_session_worker,
				)

	# 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"


	@pytest.fixture(scope="session")
	def celery_enable_logging():
	return True


	@pytest.fixture(scope="session")
	def celery_includes():
	task_modules = [
	"swh.scheduler.tests.tasks",
	]
	for entrypoint in pkg_resources.iter_entry_points("swh.workers"):
	task_modules.extend(entrypoint.load()().get("task_modules", []))
	return task_modules


	@pytest.fixture(scope="session")
	def celery_parameters():
	return {
	"task_cls": "swh.scheduler.task:SWHTask",
	}


	@pytest.fixture(scope="session")
	def celery_config():
	return {
	"accept_content": ["application/x-msgpack", "application/json"],
	"task_serializer": "msgpack",
	"result_serializer": "json",
	}


	# use the celery_session_app fixture to monkeypatch the 'main'
	# swh.scheduler.celery_backend.config.app Celery application
	# with the test application
	@pytest.fixture(scope="session")
	def swh_app(celery_session_app):
	from swh.scheduler.celery_backend import config

	config.app = celery_session_app
	yield celery_session_app


	@pytest.fixture			@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]:
	Show All 10 Lines