72 lines
2.7 KiB
Python
72 lines
2.7 KiB
Python
"""
|
|
Tests for sanguo_orchestrator.task module
|
|
Tests Task state transitions and validation
|
|
"""
|
|
import pytest
|
|
from sanguo_orchestrator.task import TaskState, Task
|
|
|
|
|
|
class TestTaskState:
|
|
"""Test TaskState enum values"""
|
|
|
|
def test_task_state_enum_values(self):
|
|
"""Test TaskState has correct enum values"""
|
|
assert TaskState.PENDING.value == "pending"
|
|
assert TaskState.RUNNING.value == "running"
|
|
assert TaskState.DONE.value == "done"
|
|
assert TaskState.FAILED.value == "failed"
|
|
|
|
|
|
class TestTask:
|
|
"""Test Task dataclass and state transitions"""
|
|
|
|
def test_task_initial_state(self):
|
|
"""Test Task starts with PENDING state"""
|
|
task = Task(task_id="test_1", task_type="test")
|
|
assert task.task_id == "test_1"
|
|
assert task.task_type == "test"
|
|
assert task.status == TaskState.PENDING
|
|
assert task.result_id is None
|
|
assert task.error_msg is None
|
|
|
|
def test_task_start_from_pending(self):
|
|
"""Test start() transitions PENDING to RUNNING"""
|
|
task = Task(task_id="test_1", task_type="test")
|
|
task.start()
|
|
assert task.status == TaskState.RUNNING
|
|
|
|
def test_task_start_from_running_raises_error(self):
|
|
"""Test start() from RUNNING raises ValueError"""
|
|
task = Task(task_id="test_1", task_type="test")
|
|
task.start()
|
|
with pytest.raises(ValueError, match="不能从 RUNNING 启动"):
|
|
task.start()
|
|
|
|
def test_task_start_from_done_raises_error(self):
|
|
"""Test start() from DONE raises ValueError"""
|
|
task = Task(task_id="test_1", task_type="test")
|
|
task.status = TaskState.DONE
|
|
with pytest.raises(ValueError, match="不能从 DONE 启动"):
|
|
task.start()
|
|
|
|
def test_task_start_from_failed_raises_error(self):
|
|
"""Test start() from FAILED raises ValueError"""
|
|
task = Task(task_id="test_1", task_type="test")
|
|
task.status = TaskState.FAILED
|
|
with pytest.raises(ValueError, match="不能从 FAILED 启动"):
|
|
task.start()
|
|
|
|
def test_task_complete_transitions_to_done(self):
|
|
"""Test complete() transitions to DONE and sets result_id"""
|
|
task = Task(task_id="test_1", task_type="test")
|
|
task.complete(result_id=12345)
|
|
assert task.status == TaskState.DONE
|
|
assert task.result_id == 12345
|
|
|
|
def test_task_fail_transitions_to_failed(self):
|
|
"""Test fail() transitions to FAILED and sets error_msg"""
|
|
task = Task(task_id="test_1", task_type="test")
|
|
task.fail(error_msg="Test error")
|
|
assert task.status == TaskState.FAILED
|
|
assert task.error_msg == "Test error"
|
|
assert task.result_id is None |