Changeset View
Changeset View
Standalone View
Standalone View
swh/scheduler/tests/test_utils.py
# Copyright (C) 2017-2022 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 | ||||
from unittest.mock import patch | from unittest.mock import patch | ||||
import uuid | |||||
from swh.scheduler import utils | from swh.scheduler import model, utils | ||||
@patch("swh.scheduler.utils.datetime") | @patch("swh.scheduler.utils.datetime") | ||||
def test_create_oneshot_task_dict_simple(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") | ||||
▲ Show 20 Lines • Show All 57 Lines • ▼ Show 20 Lines | expected_task = { | ||||
"kwargs": {"other_stuff": "normal"}, | "kwargs": {"other_stuff": "normal"}, | ||||
}, | }, | ||||
"priority": "low", | "priority": "low", | ||||
"retries_left": 3, | "retries_left": 3, | ||||
} | } | ||||
assert 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) | ||||
def test_create_origin_task_dict(): | |||||
origin = model.ListedOrigin( | |||||
lister_id=uuid.uuid4(), | |||||
url="http://example.com/", | |||||
visit_type="git", | |||||
) | |||||
task = utils.create_origin_task_dict(origin) | |||||
assert task == { | |||||
"type": "load-git", | |||||
"arguments": {"args": [], "kwargs": {"url": "http://example.com/"}}, | |||||
} | |||||
loader_args = {"foo": "bar", "baz": {"foo": "bar"}} | |||||
origin_w_args = model.ListedOrigin( | |||||
lister_id=uuid.uuid4(), | |||||
url="http://example.com/svn/", | |||||
visit_type="svn", | |||||
extra_loader_arguments=loader_args, | |||||
) | |||||
task_w_args = utils.create_origin_task_dict(origin_w_args) | |||||
assert task_w_args == { | |||||
"type": "load-svn", | |||||
"arguments": { | |||||
"args": [], | |||||
"kwargs": {"url": "http://example.com/svn/", **loader_args}, | |||||
}, | |||||
} |