wecom_it_smart_desk/backend/tests/test_automation_approval.py

# =============================================================================
# 阶段5 自动化 - 审批单服务单元测试
# =============================================================================
# 测试范围：approval.py - 审批单创建/流转/审计
# =============================================================================

import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from datetime import datetime, timezone

from app.services.automation.approval import ApprovalService
from app.models.automation import ApprovalTicket


class TestApprovalService:
    """审批单服务测试用例"""

    @pytest.mark.asyncio
    async def test_ensure_ticket_creates_new_ticket(self, db_session):
        """测试 ensure_ticket 创建新审批单"""
        # Arrange
        approval_svc = ApprovalService(db_session)

        # 创建 mock action
        action = MagicMock()
        action.id = "action_001"
        action.session_id = "session_001"

        # Act
        ticket = await approval_svc.ensure_ticket(
            action=action,
            channel="agent",
            reason="测试审批单"
        )

        # Assert
        assert ticket is not None
        assert ticket.action_id == "action_001"
        assert ticket.session_id == "session_001"
        assert ticket.channel == "agent"
        assert ticket.status == "pending"
        assert ticket.reason == "测试审批单"

    @pytest.mark.asyncio
    async def test_ensure_ticket_idempotent(self, db_session):
        """测试 ensure_ticket 幂等性 - 相同动作只创建一张待决审批单"""
        # Arrange
        approval_svc = ApprovalService(db_session)

        action = MagicMock()
        action.id = "action_002"
        action.session_id = "session_002"

        # Act - 第一次调用
        ticket1 = await approval_svc.ensure_ticket(
            action=action,
            channel="agent",
            reason="第一次创建"
        )

        # Act - 第二次调用（幂等）
        ticket2 = await approval_svc.ensure_ticket(
            action=action,
            channel="h5",
            reason="第二次创建（应返回已存在的）"
        )

        # Assert - 两次返回同一张审批单
        assert ticket1.id == ticket2.id
        assert ticket1.status == "pending"
        # 保持原有信息不变
        assert ticket1.reason == "第一次创建"

    @pytest.mark.asyncio
    async def test_decide_approve(self, db_session):
        """测试 decide 审批通过"""
        # Arrange
        approval_svc = ApprovalService(db_session)

        # 先创建审批单
        action = MagicMock()
        action.id = "action_003"
        action.session_id = "session_003"

        ticket = await approval_svc.ensure_ticket(
            action=action,
            channel="agent",
            reason="测试审批"
        )

        # Act - 审批通过
        result = await approval_svc.decide(
            ticket_id=ticket.id,
            decision="approve",
            note="同意执行",
            approver_id="agent_001"
        )

        # Assert
        assert result.status == "approved"
        assert result.decision_note == "同意执行"
        assert result.approver_id == "agent_001"
        assert result.decided_at is not None

    @pytest.mark.asyncio
    async def test_decide_reject(self, db_session):
        """测试 decide 审批驳回"""
        # Arrange
        approval_svc = ApprovalService(db_session)

        action = MagicMock()
        action.id = "action_004"
        action.session_id = "session_004"

        ticket = await approval_svc.ensure_ticket(
            action=action,
            channel="agent",
            reason="测试审批"
        )

        # Act - 审批驳回
        result = await approval_svc.decide(
            ticket_id=ticket.id,
            decision="reject",
            note="风险太高，驳回",
            approver_id="agent_002"
        )

        # Assert
        assert result.status == "rejected"
        assert result.decision_note == "风险太高，驳回"
        assert result.approver_id == "agent_002"

    @pytest.mark.asyncio
    async def test_decide_nonexistent_ticket(self, db_session):
        """测试 decide 对不存在的审批单抛出异常"""
        # Arrange
        approval_svc = ApprovalService(db_session)

        # Act & Assert
        with pytest.raises(ValueError, match="审批单不存在"):
            await approval_svc.decide(
                ticket_id="nonexistent_id",
                decision="approve",
                note="test",
                approver_id="agent_001"
            )

    @pytest.mark.asyncio
    async def test_decide_already_decided(self, db_session):
        """测试 decide 对已决审批单不重复处理"""
        # Arrange
        approval_svc = ApprovalService(db_session)

        action = MagicMock()
        action.id = "action_005"
        action.session_id = "session_005"

        ticket = await approval_svc.ensure_ticket(
            action=action,
            channel="agent",
            reason="测试"
        )

        # 第一次审批通过
        await approval_svc.decide(
            ticket_id=ticket.id,
            decision="approve",
            note="通过",
            approver_id="agent_001"
        )

        # 第二次尝试审批（应该直接返回，不处理）
        result = await approval_svc.decide(
            ticket_id=ticket.id,
            decision="reject",
            note="试图驳回",
            approver_id="agent_002"
        )

        # Assert - 状态应保持为 approved，不受第二次影响
        assert result.status == "approved"
        assert result.approver_id == "agent_001"  # 保持原审批人

    @pytest.mark.asyncio
    async def test_get_pending_for_action(self, db_session):
        """测试 get_pending_for_action 查询待决审批单"""
        # Arrange
        approval_svc = ApprovalService(db_session)

        action = MagicMock()
        action.id = "action_006"
        action.session_id = "session_006"

        # 创建审批单
        await approval_svc.ensure_ticket(
            action=action,
            channel="agent",
            reason="待决审批单"
        )

        # Act
        pending = await approval_svc.get_pending_for_action(action.id)

        # Assert
        assert pending is not None
        assert pending.action_id == action.id
        assert pending.status == "pending"