Differential D3289 Diff 11682 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			import glob
	from datetime import timedelta			from datetime import datetime, timedelta, timezone
	import pkg_resources			import pkg_resources
				from typing import List

	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.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)


	▲ Show 20 Lines • Show All 88 Lines • ▼ Show 20 Lines
	# 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			@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
				def listed_origins(stored_lister) -> List[ListedOrigin]:
				"""Return a (fixed) set of 1000 listed origins"""
				return [
				ListedOrigin(
				lister_id=stored_lister.id,
				url=f"https://example.com/{i:04d}.git",
				visit_type="git",
				last_update=datetime(2020, 6, 15, 16, 0, 0, i, tzinfo=timezone.utc),
				)
				for i in range(1000)
				]