47 lines
2.3 KiB
Python
47 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from biliup_next.app.task_control_service import TaskControlService
|
|
|
|
|
|
class TaskControlServiceTests(unittest.TestCase):
|
|
def test_run_task_delegates_to_process_task(self) -> None:
|
|
state = {"repo": object(), "settings": {"paths": {"session_dir": "/tmp/session"}}}
|
|
|
|
with patch("biliup_next.app.task_control_service.process_task", return_value={"processed": [{"task_id": "task-1"}]}) as process_mock:
|
|
result = TaskControlService(state).run_task("task-1")
|
|
|
|
self.assertEqual(result["processed"][0]["task_id"], "task-1")
|
|
process_mock.assert_called_once_with("task-1")
|
|
|
|
def test_retry_step_delegates_with_reset_step(self) -> None:
|
|
state = {"repo": object(), "settings": {"paths": {"session_dir": "/tmp/session"}}}
|
|
|
|
with patch("biliup_next.app.task_control_service.process_task", return_value={"processed": [{"step": "publish"}]}) as process_mock:
|
|
result = TaskControlService(state).retry_step("task-1", "publish")
|
|
|
|
self.assertEqual(result["processed"][0]["step"], "publish")
|
|
process_mock.assert_called_once_with("task-1", reset_step="publish")
|
|
|
|
def test_reset_to_step_combines_reset_and_run_payloads(self) -> None:
|
|
state = {"repo": object(), "settings": {"paths": {"session_dir": "/tmp/session"}}}
|
|
reset_service = SimpleNamespace(reset_to_step=lambda task_id, step_name: {"task_id": task_id, "reset_to": step_name})
|
|
|
|
with patch("biliup_next.app.task_control_service.TaskResetService", return_value=reset_service) as reset_cls:
|
|
with patch.object(reset_service, "reset_to_step", return_value={"task_id": "task-1", "reset_to": "split"}) as reset_mock:
|
|
with patch("biliup_next.app.task_control_service.process_task", return_value={"processed": [{"task_id": "task-1"}]}) as process_mock:
|
|
result = TaskControlService(state).reset_to_step("task-1", "split")
|
|
|
|
self.assertEqual(result["reset"]["reset_to"], "split")
|
|
self.assertEqual(result["run"]["processed"][0]["task_id"], "task-1")
|
|
reset_cls.assert_called_once()
|
|
reset_mock.assert_called_once_with("task-1", "split")
|
|
process_mock.assert_called_once_with("task-1")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|