wecom_it_smart_desk/backend/tests/test_approval_state_machine.py

# =============================================================================
# 企微IT智能服务台 — 审批状态机测试（D7 / P0-6）
# =============================================================================
# 说明：测试五态流转 + 合法转换校验 + 非法转换拦截。
#   状态机路径：
#     pending → approved → applied → graph_synced
#     pending → queued → approved → applied → graph_synced
#     pending → rejected
#     pending → expired
#     (改写) → pending
# =============================================================================

import pytest

from app.schemas.enums import (
    SuggestionStatusEnum,
    is_valid_transition,
)


class TestApprovalStateMachine:
    """审批状态机 — 合法转换测试。"""

    # ── 合法转换 ──

    def test_pending_to_approved_valid(self):
        """pending → approved 合法。"""
        assert is_valid_transition(
            SuggestionStatusEnum.pending, SuggestionStatusEnum.approved
        ) is True

    def test_pending_to_queued_valid(self):
        """pending → queued 合法。"""
        assert is_valid_transition(
            SuggestionStatusEnum.pending, SuggestionStatusEnum.queued
        ) is True

    def test_pending_to_rejected_valid(self):
        """pending → rejected 合法。"""
        assert is_valid_transition(
            SuggestionStatusEnum.pending, SuggestionStatusEnum.rejected
        ) is True

    def test_pending_to_expired_valid(self):
        """pending → expired 合法。"""
        assert is_valid_transition(
            SuggestionStatusEnum.pending, SuggestionStatusEnum.expired
        ) is True

    def test_queued_to_approved_valid(self):
        """queued → approved 合法。"""
        assert is_valid_transition(
            SuggestionStatusEnum.queued, SuggestionStatusEnum.approved
        ) is True

    def test_approved_to_applied_valid(self):
        """approved → applied 合法。"""
        assert is_valid_transition(
            SuggestionStatusEnum.approved, SuggestionStatusEnum.applied
        ) is True

    def test_applied_to_graph_synced_valid(self):
        """applied → graph_synced 合法（终态）。"""
        assert is_valid_transition(
            SuggestionStatusEnum.applied, SuggestionStatusEnum.graph_synced
        ) is True

    # ── 非法转换 ──

    def test_graph_synced_is_terminal(self):
        """graph_synced 是终态，不可再转换。"""
        assert is_valid_transition(
            SuggestionStatusEnum.graph_synced, SuggestionStatusEnum.pending
        ) is False
        assert is_valid_transition(
            SuggestionStatusEnum.graph_synced, SuggestionStatusEnum.applied
        ) is False

    def test_rejected_is_terminal(self):
        """rejected 是终态，不可再转换。"""
        assert is_valid_transition(
            SuggestionStatusEnum.rejected, SuggestionStatusEnum.pending
        ) is False
        assert is_valid_transition(
            SuggestionStatusEnum.rejected, SuggestionStatusEnum.approved
        ) is False

    def test_expired_is_terminal(self):
        """expired 是终态，不可再转换。"""
        assert is_valid_transition(
            SuggestionStatusEnum.expired, SuggestionStatusEnum.pending
        ) is False

    def test_pending_direct_to_graph_synced_invalid(self):
        """pending 不能直接跳到 graph_synced（必经 approved→applied）。"""
        assert is_valid_transition(
            SuggestionStatusEnum.pending, SuggestionStatusEnum.graph_synced
        ) is False

    def test_queued_direct_to_applied_invalid(self):
        """queued 不能直接跳到 applied（必经 approved）。"""
        assert is_valid_transition(
            SuggestionStatusEnum.queued, SuggestionStatusEnum.applied
        ) is False

    def test_applied_cannot_go_back_to_pending(self):
        """applied 不能回退到 pending。"""
        assert is_valid_transition(
            SuggestionStatusEnum.applied, SuggestionStatusEnum.pending
        ) is False


class TestApprovalServiceFlows:
    """审批服务层流程测试（需要数据库）。"""

    @pytest.mark.asyncio
    async def test_approve_suggestion_not_found(self, db_session):
        """验证审批不存在的建议返回 None。"""
        from app.services.knowledge_iteration_service import KnowledgeIterationService

        service = KnowledgeIterationService()
        result = await service.approve_suggestion(
            db_session, "nonexistent-id", "reviewer-001"
        )
        assert result is None

    @pytest.mark.asyncio
    async def test_reject_suggestion_not_found(self, db_session):
        """验证拒绝不存在的建议返回 None。"""
        from app.services.knowledge_iteration_service import KnowledgeIterationService

        service = KnowledgeIterationService()
        result = await service.reject_suggestion(
            db_session, "nonexistent-id", "reviewer-001", "不需要"
        )
        assert result is None

    @pytest.mark.asyncio
    async def test_rewrite_suggestion_not_found(self, db_session):
        """验证改写不存在的建议返回 None。"""
        from app.services.knowledge_iteration_service import KnowledgeIterationService

        service = KnowledgeIterationService()
        result = await service.rewrite_suggestion(
            db_session, "nonexistent-id", "reviewer-001", {"title": "新标题"}
        )
        assert result is None

    @pytest.mark.asyncio
    async def test_rewrite_resets_to_pending(self, db_session):
        """验证改写后状态重置为 pending。"""
        from app.services.knowledge_iteration_service import KnowledgeIterationService
        from app.models.knowledge_suggestion import KnowledgeSuggestion

        # 创建测试建议
        suggestion = KnowledgeSuggestion(
            suggestion_type="new_faq",
            status="approved",
            title="测试标题",
            content="测试内容",
            category="网络",
            tags=[],
            source_type="conversation",
            source_data=["conv-001"],
            reason="测试",
            confidence=0.8,
            audience="employee_quick_reply",
        )
        db_session.add(suggestion)
        await db_session.commit()
        await db_session.refresh(suggestion)

        # 改写
        service = KnowledgeIterationService()
        result = await service.rewrite_suggestion(
            db_session, suggestion.id, "reviewer-001", {"title": "改写后的标题"}
        )

        assert result is not None
        assert result.status == "pending"
        assert result.title == "改写后的标题"

    @pytest.mark.asyncio
    async def test_approve_suggestion_full_flow(self, db_session):
        """验证完整审批流程：pending→approved→applied。"""
        from app.services.knowledge_iteration_service import KnowledgeIterationService
        from app.models.knowledge_suggestion import KnowledgeSuggestion

        # 创建测试建议
        suggestion = KnowledgeSuggestion(
            suggestion_type="new_faq",
            status="pending",
            title="VPN连接测试",
            content="1.检查网络 2.重启客户端",
            category="网络",
            tags=["VPN"],
            source_type="conversation",
            source_data=["conv-002"],
            reason="测试流程",
            confidence=0.86,
            audience="employee_quick_reply",
            issue="VPN问题",
            action="VPN连接修复",
            relation_type="LEADS_TO",
        )
        db_session.add(suggestion)
        await db_session.commit()
        await db_session.refresh(suggestion)

        # 审批通过（不传 neo4j_client，只到 applied）
        service = KnowledgeIterationService()
        result = await service.approve_suggestion(
            db_session, suggestion.id, "reviewer-001", neo4j_client=None
        )

        assert result is not None
        assert result.status in ("applied", "graph_synced")
        assert result.reviewer_id == "reviewer-001"
        assert result.reviewed_at is not None

    @pytest.mark.asyncio
    async def test_queue_suggestion_flow(self, db_session):
        """验证入队列流程：pending→queued。"""
        from app.services.knowledge_iteration_service import KnowledgeIterationService
        from app.models.knowledge_suggestion import KnowledgeSuggestion

        suggestion = KnowledgeSuggestion(
            suggestion_type="new_faq",
            status="pending",
            title="队列测试",
            content="测试内容",
            category="软件",
            tags=[],
            source_type="conversation",
            source_data=["conv-003"],
            reason="测试入队列",
            confidence=0.5,
            audience="employee_quick_reply",
        )
        db_session.add(suggestion)
        await db_session.commit()
        await db_session.refresh(suggestion)

        service = KnowledgeIterationService()
        result = await service.queue_suggestion(db_session, suggestion.id)

        assert result is not None
        assert result.status == "queued"
        assert result.queued_at is not None