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

Changeset View

Standalone View

swh/scheduler/tests/test_scheduler.py

# Copyright (C) 2017-2019 The Software Heritage developers		# Copyright (C) 2017-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 copy		import copy
import datetime		import datetime
import random		import random
import uuid		import uuid

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
		import attr
		import pytest

		from swh.scheduler.exc import StaleData
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.instance_name == lister_args.get("instance_name", "")

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

		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 test_update_lister(self, swh_scheduler):
		lister = swh_scheduler.get_or_create_lister(**LISTERS[0])

		lister.current_state = {"updated": "now"}

		updated_lister = swh_scheduler.update_lister(lister)

		assert updated_lister.updated > lister.updated
		assert updated_lister == attr.evolve(lister, updated=updated_lister.updated)

		def test_update_lister_stale(self, swh_scheduler):
		lister = swh_scheduler.get_or_create_lister(**LISTERS[0])

		swh_scheduler.update_lister(lister)

		with pytest.raises(StaleData) as exc:
		swh_scheduler.update_lister(lister)
		assert "state not updated" in exc.value.args[0]

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)