Changeset View

Standalone View

Show First 20 Lines • Show All 701 Lines • ▼ Show 20 Lines	""".lstrip()

# Check tasks		# Check tasks
tasks = swh_scheduler.search_tasks()		tasks = swh_scheduler.search_tasks()
_assert_origin_tasks_contraints(tasks, 2, 20, origins)		_assert_origin_tasks_contraints(tasks, 2, 20, origins)
assert all(		assert all(
task["arguments"]["kwargs"] == {"key1": "value1", "key2": "value2"}		task["arguments"]["kwargs"] == {"key1": "value1", "key2": "value2"}
for task in tasks		for task in tasks
)		)


		def test_task_schedule_origins_with_limit(swh_scheduler, storage):
		"""Tests support of extra keyword-arguments."""
		origins = _fill_storage_with_origins(storage, 50)

		limit = 20
		result = invoke(
		swh_scheduler,
		False,
		[
		"task",
		"schedule_origins",
		"swh-test-ping",
		"--batch-size",
		"5",
		"--limit",
		limit,
		],
		)

		# Check the output
		expected = r"""
		Scheduled 4 tasks \(20 origins\).
		Done.
		""".lstrip()
		assert result.exit_code == 0, result.output
		assert re.fullmatch(expected, result.output, re.MULTILINE), repr(result.output)

		# Check tasks
		tasks = swh_scheduler.search_tasks()
		_assert_origin_tasks_contraints(tasks, 10, 5, origins[:limit])