Differential D3271 Diff 11607 swh/scheduler/tests/test_scheduler.py

Changeset View

Standalone View

swh/scheduler/tests/test_scheduler.py

Show All 10 Lines
from collections import defaultdict		from collections import defaultdict
import inspect		import inspect
from typing import Any, Dict		from typing import Any, Dict

from arrow import utcnow		from arrow import utcnow

from swh.scheduler.interface import SchedulerInterface		from swh.scheduler.interface import SchedulerInterface

from .common import tasks_from_template, TEMPLATES, TASK_TYPES		from .common import tasks_from_template, TEMPLATES, TASK_TYPES, LISTERS


def subdict(d, keys=None, excl=()):		def subdict(d, keys=None, excl=()):
if keys is None:		if keys is None:
keys = [k for k in d.keys()]		keys = [k for k in d.keys()]
return {k: d[k] for k in keys if k not in excl}		return {k: d[k] for k in keys if k not in excl}


▲ Show 20 Lines • Show All 579 Lines • ▼ Show 20 Lines	def test_get_task_runs_with_executed(self, swh_scheduler):
"backend_id": btask["backend_id"],		"backend_id": btask["backend_id"],
"scheduled": btask["scheduled"],		"scheduled": btask["scheduled"],
"started": ts,		"started": ts,
"ended": ts2,		"ended": ts2,
"metadata": {"something": "stupid", "other": "stuff"},		"metadata": {"something": "stupid", "other": "stuff"},
"status": "eventful",		"status": "eventful",
}		}

		def test_get_or_create_lister(self, swh_scheduler):
		db_listers = []
		for lister_args in LISTERS:
		db_listers.append(swh_scheduler.get_or_create_lister(**lister_args))

		for lister, lister_args in zip(db_listers, LISTERS):
		assert lister.name == lister_args["name"]
		assert lister.arguments == lister_args.get("arguments", {})

		lister_get_again = swh_scheduler.get_or_create_lister(
		lister.name, lister.arguments
		)

		ardumontUnsubmitted Done Inline Actions why not directly assert lister == lister_get_again ? ardumont: why not directly ``` assert lister == lister_get_again ``` ?
		assert lister == lister_get_again

def _create_task_types(self, scheduler):		def _create_task_types(self, scheduler):
for tt in TASK_TYPES.values():		for tt in TASK_TYPES.values():
scheduler.create_task_type(tt)		scheduler.create_task_type(tt)