Changeset View
Changeset View
Standalone View
Standalone View
swh/scheduler/tests/test_utils.py
| # Copyright (C) 2017-2018 The Software Heritage developers | # Copyright (C) 2017-2022 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 | ||||
| from datetime import timezone | from datetime import timezone | ||||
| import unittest | |||||
| from unittest.mock import patch | from unittest.mock import patch | ||||
| from swh.scheduler import utils | from swh.scheduler import utils | ||||
| class UtilsTest(unittest.TestCase): | |||||
| @patch("swh.scheduler.utils.datetime") | @patch("swh.scheduler.utils.datetime") | ||||
| def test_create_oneshot_task_dict_simple(self, mock_datetime): | def test_create_oneshot_task_dict_simple(mock_datetime): | ||||
| mock_datetime.now.return_value = "some-date" | mock_datetime.now.return_value = "some-date" | ||||
| actual_task = utils.create_oneshot_task_dict("some-task-type") | actual_task = utils.create_oneshot_task_dict("some-task-type") | ||||
| expected_task = { | expected_task = { | ||||
| "policy": "oneshot", | "policy": "oneshot", | ||||
| "type": "some-task-type", | "type": "some-task-type", | ||||
| "next_run": "some-date", | "next_run": "some-date", | ||||
| "arguments": { | "arguments": { | ||||
| "args": [], | "args": [], | ||||
| "kwargs": {}, | "kwargs": {}, | ||||
| }, | }, | ||||
| } | } | ||||
| self.assertEqual(actual_task, expected_task) | assert actual_task == expected_task | ||||
| mock_datetime.now.assert_called_once_with(tz=timezone.utc) | mock_datetime.now.assert_called_once_with(tz=timezone.utc) | ||||
| @patch("swh.scheduler.utils.datetime") | @patch("swh.scheduler.utils.datetime") | ||||
| def test_create_oneshot_task_dict_other_call(self, mock_datetime): | def test_create_oneshot_task_dict_other_call(mock_datetime): | ||||
| mock_datetime.now.return_value = "some-other-date" | mock_datetime.now.return_value = "some-other-date" | ||||
| actual_task = utils.create_oneshot_task_dict( | actual_task = utils.create_oneshot_task_dict( | ||||
| "some-task-type", "arg0", "arg1", priority="high", other_stuff="normal" | "some-task-type", "arg0", "arg1", priority="high", other_stuff="normal" | ||||
| ) | ) | ||||
| expected_task = { | expected_task = { | ||||
| "policy": "oneshot", | "policy": "oneshot", | ||||
| "type": "some-task-type", | "type": "some-task-type", | ||||
| "next_run": "some-other-date", | "next_run": "some-other-date", | ||||
| "arguments": { | "arguments": { | ||||
| "args": ("arg0", "arg1"), | "args": ("arg0", "arg1"), | ||||
| "kwargs": {"other_stuff": "normal"}, | "kwargs": {"other_stuff": "normal"}, | ||||
| }, | }, | ||||
| "priority": "high", | "priority": "high", | ||||
| } | } | ||||
| self.assertEqual(actual_task, expected_task) | assert actual_task == expected_task | ||||
| mock_datetime.now.assert_called_once_with(tz=timezone.utc) | mock_datetime.now.assert_called_once_with(tz=timezone.utc) | ||||
| @patch("swh.scheduler.utils.datetime") | @patch("swh.scheduler.utils.datetime") | ||||
| def test_create_task_dict(self, mock_datetime): | def test_create_task_dict(mock_datetime): | ||||
| mock_datetime.now.return_value = "date" | mock_datetime.now.return_value = "date" | ||||
| actual_task = utils.create_task_dict( | actual_task = utils.create_task_dict( | ||||
| "task-type", | "task-type", | ||||
| "recurring", | "recurring", | ||||
| "arg0", | "arg0", | ||||
| "arg1", | "arg1", | ||||
| priority="low", | priority="low", | ||||
| other_stuff="normal", | other_stuff="normal", | ||||
| retries_left=3, | retries_left=3, | ||||
| ) | ) | ||||
| expected_task = { | expected_task = { | ||||
| "policy": "recurring", | "policy": "recurring", | ||||
| "type": "task-type", | "type": "task-type", | ||||
| "next_run": "date", | "next_run": "date", | ||||
| "arguments": { | "arguments": { | ||||
| "args": ("arg0", "arg1"), | "args": ("arg0", "arg1"), | ||||
| "kwargs": {"other_stuff": "normal"}, | "kwargs": {"other_stuff": "normal"}, | ||||
| }, | }, | ||||
| "priority": "low", | "priority": "low", | ||||
| "retries_left": 3, | "retries_left": 3, | ||||
| } | } | ||||
| self.assertEqual(actual_task, expected_task) | assert actual_task == expected_task | ||||
| mock_datetime.now.assert_called_once_with(tz=timezone.utc) | mock_datetime.now.assert_called_once_with(tz=timezone.utc) | ||||