Differential D1216 Diff 3901 swh/scheduler/tests/test_cli.py

Changeset View

Standalone View

swh/scheduler/tests/test_cli.py

	# Copyright (C) 2019 The Software Heritage developers			# Copyright (C) 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 datetime			import datetime
	import re			import re
	import tempfile			import tempfile
	from unittest.mock import patch			from unittest.mock import patch

	from click.testing import CliRunner			from click.testing import CliRunner
				import pytest

	from swh.scheduler.cli import cli			from swh.scheduler.cli import cli
	from swh.scheduler.utils import create_task_dict			from swh.scheduler.utils import create_task_dict


	CLI_CONFIG = '''			CLI_CONFIG = '''
	scheduler:			scheduler:
	cls: foo			cls: foo
	▲ Show 20 Lines • Show All 123 Lines • ▼ Show 20 Lines
	\[INFO\] swh.core.config -- Loading config file .*			\[INFO\] swh.core.config -- Loading config file .*
	Found 0 swh-test-ping tasks			Found 0 swh-test-ping tasks

	'''.lstrip()			'''.lstrip()
	assert result.exit_code == 0, result.output			assert result.exit_code == 0, result.output
	assert re.fullmatch(expected, result.output, re.MULTILINE), result.output			assert re.fullmatch(expected, result.output, re.MULTILINE), result.output


	def test_list_pending_tasks_one(swh_scheduler):			def test_list_pending_tasks(swh_scheduler):
	task = create_task_dict('swh-test-ping', 'oneshot', key='value')			task1 = create_task_dict('swh-test-ping', 'oneshot', key='value1')
	swh_scheduler.create_tasks([task])			task2 = create_task_dict('swh-test-ping', 'oneshot', key='value2')
				task2['next_run'] += datetime.timedelta(days=1)
				swh_scheduler.create_tasks([task1, task2])

	result = invoke(swh_scheduler, False, [			result = invoke(swh_scheduler, False, [
	'task', 'list-pending', 'swh-test-ping',			'task', 'list-pending', 'swh-test-ping',
	])			])

	expected = r'''			expected = r'''
	\[INFO\] swh.core.config -- Loading config file .*			\[INFO\] swh.core.config -- Loading config file .*
	Found 1 swh-test-ping tasks			Found 1 swh-test-ping tasks

	Task 1			Task 1
	Next run: just now \(.*\)			Next run: just now \(.*\)
	Interval: 1 day, 0:00:00			Interval: 1 day, 0:00:00
	Type: swh-test-ping			Type: swh-test-ping
	Policy: oneshot			Policy: oneshot
	Args:			Args:
	Keyword args:			Keyword args:
	key: 'value'			key: 'value1'

	'''.lstrip()			'''.lstrip()
	assert result.exit_code == 0, result.output			assert result.exit_code == 0, result.output
	assert re.fullmatch(expected, result.output, re.MULTILINE), result.output			assert re.fullmatch(expected, result.output, re.MULTILINE), result.output

				swh_scheduler.grab_ready_tasks('swh-test-ping')

	def test_list_pending_tasks_one_filter(swh_scheduler):			result = invoke(swh_scheduler, False, [
				'task', 'list-pending', 'swh-test-ping',
				])

				expected = r'''
				\[INFO\] swh.core.config -- Loading config file .*
				Found 0 swh-test-ping tasks

				'''.lstrip()
				assert result.exit_code == 0, result.output
				assert re.fullmatch(expected, result.output, re.MULTILINE), result.output


				def test_list_pending_tasks_filter(swh_scheduler):
	task = create_task_dict('swh-test-multiping', 'oneshot', key='value')			task = create_task_dict('swh-test-multiping', 'oneshot', key='value')
	swh_scheduler.create_tasks([task])			swh_scheduler.create_tasks([task])

	result = invoke(swh_scheduler, False, [			result = invoke(swh_scheduler, False, [
	'task', 'list-pending', 'swh-test-ping',			'task', 'list-pending', 'swh-test-ping',
	])			])

	expected = r'''			expected = r'''
	\[INFO\] swh.core.config -- Loading config file .*			\[INFO\] swh.core.config -- Loading config file .*
	Found 0 swh-test-ping tasks			Found 0 swh-test-ping tasks

	'''.lstrip()			'''.lstrip()
	assert result.exit_code == 0, result.output			assert result.exit_code == 0, result.output
	assert re.fullmatch(expected, result.output, re.MULTILINE), result.output			assert re.fullmatch(expected, result.output, re.MULTILINE), result.output


				def test_list_pending_tasks_filter_2(swh_scheduler):
				swh_scheduler.create_tasks([
				create_task_dict('swh-test-multiping', 'oneshot', key='value'),
				create_task_dict('swh-test-ping', 'oneshot', key='value2'),
				])

				result = invoke(swh_scheduler, False, [
				'task', 'list-pending', 'swh-test-ping',
				])

				expected = r'''
				\[INFO\] swh.core.config -- Loading config file .*
				Found 1 swh-test-ping tasks

				Task 2
				Next run: just now \(.*\)
				Interval: 1 day, 0:00:00
				Type: swh-test-ping
				Policy: oneshot
				Args:
				Keyword args:
				key: 'value2'

				'''.lstrip()
				assert result.exit_code == 0, result.output
				assert re.fullmatch(expected, result.output, re.MULTILINE), result.output


				# Fails because "task list-pending --limit 3" only returns 2 tasks, because
				# of how compute_nb_tasks_from works.
				@pytest.mark.xfail
				def test_list_pending_tasks_limit(swh_scheduler):
				swh_scheduler.create_tasks([
				create_task_dict('swh-test-ping', 'oneshot', key='value%d' % i)
				for i in range(10)
				])

				result = invoke(swh_scheduler, False, [
				'task', 'list-pending', 'swh-test-ping', '--limit', '3',
				])

				expected = r'''
				\[INFO\] swh.core.config -- Loading config file .*
				Found 2 swh-test-ping tasks

				Task 1
				Next run: just now \(.*\)
				Interval: 1 day, 0:00:00
				Type: swh-test-ping
				Policy: oneshot
				Args:
				Keyword args:
				key: 'value0'

				Task 2
				Next run: just now \(.*\)
				Interval: 1 day, 0:00:00
				Type: swh-test-ping
				Policy: oneshot
				Args:
				Keyword args:
				key: 'value1'

				Task 3
				Next run: just now \(.*\)
				Interval: 1 day, 0:00:00
				Type: swh-test-ping
				Policy: oneshot
				Args:
				Keyword args:
				key: 'value2'

				'''.lstrip()
				assert result.exit_code == 0, result.output
				assert re.fullmatch(expected, result.output, re.MULTILINE), result.output


				def test_list_pending_tasks_before(swh_scheduler):
				task1 = create_task_dict('swh-test-ping', 'oneshot', key='value')
				task2 = create_task_dict('swh-test-ping', 'oneshot', key='value2')
				task1['next_run'] += datetime.timedelta(days=3)
				task2['next_run'] += datetime.timedelta(days=1)
				swh_scheduler.create_tasks([task1, task2])

				result = invoke(swh_scheduler, False, [
				'task', 'list-pending', 'swh-test-ping', '--before',
				(datetime.date.today() + datetime.timedelta(days=2)).isoformat()
				])

				expected = r'''
				\[INFO\] swh.core.config -- Loading config file .*
				Found 1 swh-test-ping tasks

				Task 2
				Next run: in a day \(.*\)
				Interval: 1 day, 0:00:00
				Type: swh-test-ping
				Policy: oneshot
				Args:
				Keyword args:
				key: 'value2'

				'''.lstrip()
				assert result.exit_code == 0, result.output
				assert re.fullmatch(expected, result.output, re.MULTILINE), result.output


				def test_list_tasks(swh_scheduler):
				task1 = create_task_dict('swh-test-ping', 'oneshot', key='value1')
				task2 = create_task_dict('swh-test-ping', 'oneshot', key='value2')
				task1['next_run'] += datetime.timedelta(days=3, hours=2)
				swh_scheduler.create_tasks([task1, task2])

				swh_scheduler.grab_ready_tasks('swh-test-ping')

				result = invoke(swh_scheduler, False, [
				'task', 'list',
				])

				expected = r'''
				\[INFO\] swh.core.config -- Loading config file .*
				Found 2 tasks

				Task 1
				Next run: in 3 days \(.*\)
				Interval: 1 day, 0:00:00
				Type: swh-test-ping
				Policy: oneshot
				Status: next_run_not_scheduled
				Priority:\x20
				Args:
				Keyword args:
				key: 'value1'

				Task 2
				Next run: just now \(.*\)
				Interval: 1 day, 0:00:00
				Type: swh-test-ping
				Policy: oneshot
				Status: next_run_scheduled
				Priority:\x20
				Args:
				Keyword args:
				key: 'value2'

				'''.lstrip()
				assert result.exit_code == 0, result.output
				assert re.fullmatch(expected, result.output, re.MULTILINE), result.output


				def test_list_tasks_id(swh_scheduler):
				task1 = create_task_dict('swh-test-ping', 'oneshot', key='value1')
				task2 = create_task_dict('swh-test-ping', 'oneshot', key='value2')
				task3 = create_task_dict('swh-test-ping', 'oneshot', key='value3')
				swh_scheduler.create_tasks([task1, task2, task3])

				result = invoke(swh_scheduler, False, [
				'task', 'list', '--task-id', '2',
				])

				expected = r'''
				\[INFO\] swh.core.config -- Loading config file .*
				Found 1 tasks

				Task 2
				Next run: just now \(.*\)
				Interval: 1 day, 0:00:00
				Type: swh-test-ping
				Policy: oneshot
				Status: next_run_not_scheduled
				Priority:\x20
				Args:
				Keyword args:
				key: 'value2'

				'''.lstrip()
				assert result.exit_code == 0, result.output
				assert re.fullmatch(expected, result.output, re.MULTILINE), result.output


				def test_list_tasks_id_2(swh_scheduler):
				task1 = create_task_dict('swh-test-ping', 'oneshot', key='value1')
				task2 = create_task_dict('swh-test-ping', 'oneshot', key='value2')
				task3 = create_task_dict('swh-test-ping', 'oneshot', key='value3')
				swh_scheduler.create_tasks([task1, task2, task3])

				result = invoke(swh_scheduler, False, [
				'task', 'list', '--task-id', '2', '--task-id', '3'
				])

				expected = r'''
				\[INFO\] swh.core.config -- Loading config file .*
				Found 2 tasks

				Task 2
				Next run: just now \(.*\)
				Interval: 1 day, 0:00:00
				Type: swh-test-ping
				Policy: oneshot
				Status: next_run_not_scheduled
				Priority:\x20
				Args:
				Keyword args:
				key: 'value2'

				Task 3
				Next run: just now \(.*\)
				Interval: 1 day, 0:00:00
				Type: swh-test-ping
				Policy: oneshot
				Status: next_run_not_scheduled
				Priority:\x20
				Args:
				Keyword args:
				key: 'value3'

				'''.lstrip()
				assert result.exit_code == 0, result.output
				assert re.fullmatch(expected, result.output, re.MULTILINE), result.output


				def test_list_tasks_type(swh_scheduler):
				task1 = create_task_dict('swh-test-ping', 'oneshot', key='value1')
				task2 = create_task_dict('swh-test-multiping', 'oneshot', key='value2')
				task3 = create_task_dict('swh-test-ping', 'oneshot', key='value3')
				swh_scheduler.create_tasks([task1, task2, task3])

				result = invoke(swh_scheduler, False, [
				'task', 'list', '--task-type', 'swh-test-ping'
				])

				expected = r'''
				\[INFO\] swh.core.config -- Loading config file .*
				Found 2 tasks

				Task 1
				Next run: just now \(.*\)
				Interval: 1 day, 0:00:00
				Type: swh-test-ping
				Policy: oneshot
				Status: next_run_not_scheduled
				Priority:\x20
				Args:
				Keyword args:
				key: 'value1'

				Task 3
				Next run: just now \(.*\)
				Interval: 1 day, 0:00:00
				Type: swh-test-ping
				Policy: oneshot
				Status: next_run_not_scheduled
				Priority:\x20
				Args:
				Keyword args:
				key: 'value3'

				'''.lstrip()
				assert result.exit_code == 0, result.output
				assert re.fullmatch(expected, result.output, re.MULTILINE), result.output


				def test_list_tasks_limit(swh_scheduler):
				task1 = create_task_dict('swh-test-ping', 'oneshot', key='value1')
				task2 = create_task_dict('swh-test-ping', 'oneshot', key='value2')
				task3 = create_task_dict('swh-test-ping', 'oneshot', key='value3')
				swh_scheduler.create_tasks([task1, task2, task3])

				result = invoke(swh_scheduler, False, [
				'task', 'list', '--limit', '2',
				])

				expected = r'''
				\[INFO\] swh.core.config -- Loading config file .*
				Found 2 tasks

				Task 1
				Next run: just now \(.*\)
				Interval: 1 day, 0:00:00
				Type: swh-test-ping
				Policy: oneshot
				Status: next_run_not_scheduled
				Priority:\x20
				Args:
				Keyword args:
				key: 'value1'

				Task 2
				Next run: just now \(.*\)
				Interval: 1 day, 0:00:00
				Type: swh-test-ping
				Policy: oneshot
				Status: next_run_not_scheduled
				Priority:\x20
				Args:
				Keyword args:
				key: 'value2'

				'''.lstrip()
				assert result.exit_code == 0, result.output
				assert re.fullmatch(expected, result.output, re.MULTILINE), result.output


				def test_list_tasks_before(swh_scheduler):
				task1 = create_task_dict('swh-test-ping', 'oneshot', key='value1')
				task2 = create_task_dict('swh-test-ping', 'oneshot', key='value2')
				task1['next_run'] += datetime.timedelta(days=3, hours=2)
				swh_scheduler.create_tasks([task1, task2])

				swh_scheduler.grab_ready_tasks('swh-test-ping')

				result = invoke(swh_scheduler, False, [
				'task', 'list', '--before',
				(datetime.date.today() + datetime.timedelta(days=2)).isoformat()
				])

				expected = r'''
				\[INFO\] swh.core.config -- Loading config file .*
				Found 1 tasks

				Task 2
				Next run: just now \(.*\)
				Interval: 1 day, 0:00:00
				Type: swh-test-ping
				Policy: oneshot
				Status: next_run_scheduled
				Priority:\x20
				Args:
				Keyword args:
				key: 'value2'

				'''.lstrip()
				assert result.exit_code == 0, result.output
				assert re.fullmatch(expected, result.output, re.MULTILINE), result.output


				def test_list_tasks_after(swh_scheduler):
				task1 = create_task_dict('swh-test-ping', 'oneshot', key='value1')
				task2 = create_task_dict('swh-test-ping', 'oneshot', key='value2')
				task1['next_run'] += datetime.timedelta(days=3, hours=2)
				swh_scheduler.create_tasks([task1, task2])

				swh_scheduler.grab_ready_tasks('swh-test-ping')

				result = invoke(swh_scheduler, False, [
				'task', 'list', '--after',
				(datetime.date.today() + datetime.timedelta(days=2)).isoformat()
				])

				expected = r'''
				\[INFO\] swh.core.config -- Loading config file .*
				Found 1 tasks

				Task 1
				Next run: in 3 days \(.*\)
				Interval: 1 day, 0:00:00
				Type: swh-test-ping
				Policy: oneshot
				Status: next_run_not_scheduled
				Priority:\x20
				Args:
				Keyword args:
				key: 'value1'

				'''.lstrip()
				assert result.exit_code == 0, result.output
				assert re.fullmatch(expected, result.output, re.MULTILINE), result.output