Changeset View
Changeset View
Standalone View
Standalone View
swh/scheduler/tests/test_cli.py
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]) | |||||
def test_task_schedule_origins_with_page_token(swh_scheduler, storage): | |||||
"""Tests support of extra keyword-arguments.""" | |||||
origins = _fill_storage_with_origins(storage, 50) | |||||
result = invoke( | |||||
swh_scheduler, | |||||
False, | |||||
[ | |||||
"task", | |||||
"schedule_origins", | |||||
"swh-test-ping", | |||||
"--batch-size", | |||||
"5", | |||||
"--page-token", | |||||
"11", | |||||
], | |||||
) | |||||
# Check the output | |||||
expected = r""" | |||||
Scheduled 8 tasks \(40 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[10:]) |