diff --git a/tests/unit/test_state_machine.py b/tests/unit/test_state_machine.py new file mode 100644 index 0000000..0be7567 --- /dev/null +++ b/tests/unit/test_state_machine.py @@ -0,0 +1,301 @@ +"""P0 单元测试:状态机转换 VALID_TRANSITIONS 全覆盖。""" + +import pytest + +pytestmark = pytest.mark.unit + +from src.blackboard.db import VALID_TRANSITIONS + +ALL_STATES = set(VALID_TRANSITIONS.keys()) + + +class TestCompleteness: + """完整性校验:测试覆盖的状态集合 == VALID_TRANSITIONS 的 key 集合。""" + + def test_all_states_are_tested(self): + tested = { + "pending", "claimed", "working", "paused", "review", + "blocked", "failed", "escalated", "waiting_human", + "done", "reviewing", "cancelled", + } + assert tested == ALL_STATES + + +class TestPendingTransitions: + """pending 状态的合法/非法转换。""" + + @pytest.mark.parametrize("target", ["claimed", "paused", "blocked", "cancelled"]) + def test_valid_from_pending(self, target): + assert target in VALID_TRANSITIONS["pending"] + + def test_invalid_from_pending_to_working(self): + assert "working" not in VALID_TRANSITIONS["pending"] + + def test_invalid_from_pending_to_done(self): + assert "done" not in VALID_TRANSITIONS["pending"] + + def test_invalid_from_pending_to_review(self): + assert "review" not in VALID_TRANSITIONS["pending"] + + def test_invalid_from_pending_to_failed(self): + assert "failed" not in VALID_TRANSITIONS["pending"] + + def test_invalid_from_pending_to_escalated(self): + assert "escalated" not in VALID_TRANSITIONS["pending"] + + def test_invalid_from_pending_to_waiting_human(self): + assert "waiting_human" not in VALID_TRANSITIONS["pending"] + + +class TestClaimedTransitions: + """claimed 状态的合法/非法转换。""" + + @pytest.mark.parametrize("target", ["working", "paused", "pending", "cancelled"]) + def test_valid_from_claimed(self, target): + assert target in VALID_TRANSITIONS["claimed"] + + def test_invalid_from_claimed_to_done(self): + assert "done" not in VALID_TRANSITIONS["claimed"] + + def test_invalid_from_claimed_to_review(self): + assert "review" not in VALID_TRANSITIONS["claimed"] + + def test_invalid_from_claimed_to_failed(self): + assert "failed" not in VALID_TRANSITIONS["claimed"] + + def test_invalid_from_claimed_to_blocked(self): + assert "blocked" not in VALID_TRANSITIONS["claimed"] + + def test_invalid_from_claimed_to_escalated(self): + assert "escalated" not in VALID_TRANSITIONS["claimed"] + + +class TestWorkingTransitions: + """working 状态的合法/非法转换。""" + + @pytest.mark.parametrize( + "target", + ["review", "done", "blocked", "failed", "paused", "escalated", "waiting_human", "cancelled", "pending"], + ) + def test_valid_from_working(self, target): + assert target in VALID_TRANSITIONS["working"] + + def test_working_to_pending_is_valid(self): + """重点:working → pending,Mail spawn 失败回退。""" + assert "pending" in VALID_TRANSITIONS["working"] + + def test_invalid_from_working_to_claimed(self): + assert "claimed" not in VALID_TRANSITIONS["working"] + + +class TestPausedTransitions: + """paused 状态的合法/非法转换——可恢复到多个状态。""" + + @pytest.mark.parametrize("target", ["working", "claimed", "review", "escalated", "waiting_human", "cancelled"]) + def test_valid_from_paused(self, target): + assert target in VALID_TRANSITIONS["paused"] + + def test_paused_can_resume_to_multiple_states(self): + """重点:paused 可以恢复到 working / claimed / review。""" + resume_targets = {"working", "claimed", "review"} + assert resume_targets.issubset(VALID_TRANSITIONS["paused"]) + + def test_invalid_from_paused_to_done(self): + assert "done" not in VALID_TRANSITIONS["paused"] + + def test_invalid_from_paused_to_pending(self): + assert "pending" not in VALID_TRANSITIONS["paused"] + + def test_invalid_from_paused_to_blocked(self): + assert "blocked" not in VALID_TRANSITIONS["paused"] + + def test_invalid_from_paused_to_failed(self): + assert "failed" not in VALID_TRANSITIONS["paused"] + + +class TestReviewTransitions: + """review 状态的合法/非法转换。""" + + @pytest.mark.parametrize("target", ["done", "pending", "failed", "paused", "escalated", "waiting_human", "cancelled"]) + def test_valid_from_review(self, target): + assert target in VALID_TRANSITIONS["review"] + + def test_invalid_from_review_to_working(self): + assert "working" not in VALID_TRANSITIONS["review"] + + def test_invalid_from_review_to_claimed(self): + assert "claimed" not in VALID_TRANSITIONS["review"] + + def test_invalid_from_review_to_blocked(self): + assert "blocked" not in VALID_TRANSITIONS["review"] + + +class TestBlockedTransitions: + """blocked 状态的合法/非法转换。""" + + @pytest.mark.parametrize("target", ["pending", "escalated", "cancelled"]) + def test_valid_from_blocked(self, target): + assert target in VALID_TRANSITIONS["blocked"] + + def test_invalid_from_blocked_to_working(self): + assert "working" not in VALID_TRANSITIONS["blocked"] + + def test_invalid_from_blocked_to_done(self): + assert "done" not in VALID_TRANSITIONS["blocked"] + + def test_invalid_from_blocked_to_review(self): + assert "review" not in VALID_TRANSITIONS["blocked"] + + def test_invalid_from_blocked_to_paused(self): + assert "paused" not in VALID_TRANSITIONS["blocked"] + + def test_invalid_from_blocked_to_failed(self): + assert "failed" not in VALID_TRANSITIONS["blocked"] + + +class TestFailedTransitions: + """failed 状态的合法/非法转换。""" + + @pytest.mark.parametrize("target", ["pending", "escalated", "cancelled"]) + def test_valid_from_failed(self, target): + assert target in VALID_TRANSITIONS["failed"] + + def test_invalid_from_failed_to_working(self): + assert "working" not in VALID_TRANSITIONS["failed"] + + def test_invalid_from_failed_to_done(self): + assert "done" not in VALID_TRANSITIONS["failed"] + + def test_invalid_from_failed_to_review(self): + assert "review" not in VALID_TRANSITIONS["failed"] + + def test_invalid_from_failed_to_paused(self): + assert "paused" not in VALID_TRANSITIONS["failed"] + + def test_invalid_from_failed_to_claimed(self): + assert "claimed" not in VALID_TRANSITIONS["failed"] + + +class TestEscalatedTransitions: + """escalated 状态的合法/非法转换。""" + + @pytest.mark.parametrize("target", ["working", "pending", "paused", "cancelled"]) + def test_valid_from_escalated(self, target): + assert target in VALID_TRANSITIONS["escalated"] + + def test_invalid_from_escalated_to_done(self): + assert "done" not in VALID_TRANSITIONS["escalated"] + + def test_invalid_from_escalated_to_review(self): + assert "review" not in VALID_TRANSITIONS["escalated"] + + def test_invalid_from_escalated_to_blocked(self): + assert "blocked" not in VALID_TRANSITIONS["escalated"] + + def test_invalid_from_escalated_to_failed(self): + assert "failed" not in VALID_TRANSITIONS["escalated"] + + def test_invalid_from_escalated_to_claimed(self): + assert "claimed" not in VALID_TRANSITIONS["escalated"] + + +class TestWaitingHumanTransitions: + """waiting_human 状态的合法/非法转换。""" + + @pytest.mark.parametrize("target", ["working", "done", "paused", "cancelled"]) + def test_valid_from_waiting_human(self, target): + assert target in VALID_TRANSITIONS["waiting_human"] + + def test_invalid_from_waiting_human_to_pending(self): + assert "pending" not in VALID_TRANSITIONS["waiting_human"] + + def test_invalid_from_waiting_human_to_review(self): + assert "review" not in VALID_TRANSITIONS["waiting_human"] + + def test_invalid_from_waiting_human_to_claimed(self): + assert "claimed" not in VALID_TRANSITIONS["waiting_human"] + + def test_invalid_from_waiting_human_to_blocked(self): + assert "blocked" not in VALID_TRANSITIONS["waiting_human"] + + def test_invalid_from_waiting_human_to_failed(self): + assert "failed" not in VALID_TRANSITIONS["waiting_human"] + + +class TestDoneTransitions: + """done 状态的合法/非法转换——只有 cancelled 和 reviewing。""" + + @pytest.mark.parametrize("target", ["cancelled", "reviewing"]) + def test_valid_from_done(self, target): + assert target in VALID_TRANSITIONS["done"] + + def test_done_to_reviewing_is_valid(self): + """重点:done → reviewing,审查回溯。""" + assert "reviewing" in VALID_TRANSITIONS["done"] + + def test_invalid_from_done_to_working(self): + assert "working" not in VALID_TRANSITIONS["done"] + + def test_invalid_from_done_to_pending(self): + assert "pending" not in VALID_TRANSITIONS["done"] + + def test_invalid_from_done_to_review(self): + assert "review" not in VALID_TRANSITIONS["done"] + + def test_invalid_from_done_to_paused(self): + assert "paused" not in VALID_TRANSITIONS["done"] + + def test_invalid_from_done_to_blocked(self): + assert "blocked" not in VALID_TRANSITIONS["done"] + + +class TestReviewingTransitions: + """reviewing 状态的合法/非法转换。""" + + @pytest.mark.parametrize("target", ["done", "working", "cancelled"]) + def test_valid_from_reviewing(self, target): + assert target in VALID_TRANSITIONS["reviewing"] + + def test_invalid_from_reviewing_to_pending(self): + assert "pending" not in VALID_TRANSITIONS["reviewing"] + + def test_invalid_from_reviewing_to_review(self): + assert "review" not in VALID_TRANSITIONS["reviewing"] + + def test_invalid_from_reviewing_to_paused(self): + assert "paused" not in VALID_TRANSITIONS["reviewing"] + + def test_invalid_from_reviewing_to_blocked(self): + assert "blocked" not in VALID_TRANSITIONS["reviewing"] + + def test_invalid_from_reviewing_to_failed(self): + assert "failed" not in VALID_TRANSITIONS["reviewing"] + + +class TestCancelledTransitions: + """cancelled 状态的合法/非法转换——唯一出路是回到 pending。""" + + def test_valid_from_cancelled_to_pending(self): + """重点:cancelled → pending,唯一能从 cancelled 出来的路径。""" + assert "pending" in VALID_TRANSITIONS["cancelled"] + + def test_cancelled_only_has_pending(self): + assert VALID_TRANSITIONS["cancelled"] == {"pending"} + + def test_invalid_from_cancelled_to_working(self): + assert "working" not in VALID_TRANSITIONS["cancelled"] + + def test_invalid_from_cancelled_to_done(self): + assert "done" not in VALID_TRANSITIONS["cancelled"] + + def test_invalid_from_cancelled_to_review(self): + assert "review" not in VALID_TRANSITIONS["cancelled"] + + def test_invalid_from_cancelled_to_failed(self): + assert "failed" not in VALID_TRANSITIONS["cancelled"] + + def test_invalid_from_cancelled_to_paused(self): + assert "paused" not in VALID_TRANSITIONS["cancelled"] + + def test_invalid_from_cancelled_to_claimed(self): + assert "claimed" not in VALID_TRANSITIONS["cancelled"]