30 lines
1.3 KiB
Python
30 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
from .models import TaskState
|
|
|
|
|
|
ALLOWED_TRANSITIONS: dict[TaskState, set[TaskState]] = {
|
|
TaskState.DISCOVERED: {TaskState.CLAIMED, TaskState.CANCELLED},
|
|
TaskState.CLAIMED: {TaskState.PLANNING, TaskState.FAILED, TaskState.BLOCKED, TaskState.CANCELLED},
|
|
TaskState.PLANNING: {TaskState.IMPLEMENTING, TaskState.FAILED, TaskState.BLOCKED, TaskState.CANCELLED},
|
|
TaskState.IMPLEMENTING: {TaskState.TESTING, TaskState.FAILED, TaskState.BLOCKED, TaskState.CANCELLED},
|
|
TaskState.TESTING: {
|
|
TaskState.PR_OPENED,
|
|
TaskState.HUMAN_REVIEW_READY,
|
|
TaskState.FAILED,
|
|
TaskState.BLOCKED,
|
|
TaskState.CANCELLED,
|
|
},
|
|
TaskState.PR_OPENED: {TaskState.REVIEWING, TaskState.FAILED, TaskState.CANCELLED},
|
|
TaskState.REVIEWING: {TaskState.HUMAN_REVIEW_READY, TaskState.FAILED, TaskState.CANCELLED},
|
|
TaskState.HUMAN_REVIEW_READY: {TaskState.DISCOVERED},
|
|
TaskState.BLOCKED: {TaskState.DISCOVERED, TaskState.CANCELLED},
|
|
TaskState.FAILED: {TaskState.DISCOVERED, TaskState.CANCELLED},
|
|
TaskState.CANCELLED: {TaskState.DISCOVERED},
|
|
}
|
|
|
|
|
|
def validate_transition(from_state: TaskState, to_state: TaskState) -> None:
|
|
if to_state not in ALLOWED_TRANSITIONS[from_state]:
|
|
raise ValueError(f"invalid task transition: {from_state.value} -> {to_state.value}")
|