Simon
250 lines
9.1 KiB
Python
250 lines
9.1 KiB
Python
# =============================================================================
|
||
# 企微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
|