896 lines
36 KiB
Python
896 lines
36 KiB
Python
|
# =============================================================================
|
|||
|
|
# 企微IT智能服务台 — 代办事项真实数据源集成 测试
|
|||
|
|
# =============================================================================
|
|||
|
|
# 测试覆盖：
|
|||
|
|
# 1. ITSM 签名工具 (ITSMSigner)
|
|||
|
|
# 2. TodoAggregatorService — 聚合服务（缓存/并行/容错/排序/过滤/详情路由）
|
|||
|
|
# 3. ApprovalTodoService — 企微审批数据源（列表/过滤/映射/ID格式）
|
|||
|
|
# 4. ITSMService — ITSM 工单数据源（列表/详情/无凭证/ID格式）
|
|||
|
|
# 5. API 端点 — GET /todo-items, GET /todo-items/{id}, PUT status
|
|||
|
|
# 6. Schema 验证 — VALID_TODO_TYPES 移除 device
|
|||
|
|
# =============================================================================
|
|||
|
|
|
|||
|
|
import hashlib
|
|||
|
|
import json
|
|||
|
|
import logging
|
|||
|
|
from typing import Any, Dict, List, Optional
|
|||
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|||
|
|
from urllib.parse import quote_plus
|
|||
|
|
|
|||
|
|
import pytest
|
|||
|
|
import pytest_asyncio
|
|||
|
|
from httpx import ASGITransport, AsyncClient
|
|||
|
|
from pydantic import ValidationError
|
|||
|
|
|
|||
|
|
# --- 临时补丁：itsm_service.py / todo_source_service.py / approval.py 中
|
|||
|
|
# httpx.Timeout(connect=10.0, read=30.0) 在当前 httpx 版本下会抛 ValueError
|
|||
|
|
# （要求 default 或全部 4 个参数）。
|
|||
|
|
# 此补丁将部分参数的 Timeout 调用退化为 default=30s，不影响正常 Timeout 使用。
|
|||
|
|
# ⚠️ 这是源码 bug 的临时绕过，已报告工程师修复。
|
|||
|
|
import httpx as _httpx_mod
|
|||
|
|
|
|||
|
|
_UNSET = object() # 哨兵：区分"未传参"与 None
|
|||
|
|
_orig_timeout_init = _httpx_mod.Timeout.__init__
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _compat_timeout_init(
|
|||
|
|
self, timeout=_UNSET, *, connect=_UNSET, read=_UNSET, write=_UNSET, pool=_UNSET
|
|||
|
|
):
|
|||
|
|
"""兼容旧版 httpx.Timeout 部分参数调用方式。
|
|||
|
|
|
|||
|
|
当调用方只传了部分 kwargs（如 Timeout(connect=10, read=30)）而未传 default 时，
|
|||
|
|
退化为 Timeout(30.0) 以绕过新版 httpx 的校验。
|
|||
|
|
其他正常调用（Timeout(5.0) / Timeout()）原样透传。
|
|||
|
|
"""
|
|||
|
|
has_individual = any(v is not _UNSET for v in (connect, read, write, pool))
|
|||
|
|
|
|||
|
|
if has_individual and timeout is _UNSET:
|
|||
|
|
# 源码 bug 场景：Timeout(connect=10, read=30) → 退化为 default
|
|||
|
|
_orig_timeout_init(self, 30.0)
|
|||
|
|
elif timeout is _UNSET:
|
|||
|
|
_orig_timeout_init(self)
|
|||
|
|
else:
|
|||
|
|
_orig_timeout_init(self, timeout)
|
|||
|
|
|
|||
|
|
|
|||
|
|
_httpx_mod.Timeout.__init__ = _compat_timeout_init
|
|||
|
|
# --- 补丁结束 ---
|
|||
|
|
|
|||
|
|
from app.config import settings
|
|||
|
|
from app.schemas.todo_item import VALID_TODO_TYPES, TodoItemCreate
|
|||
|
|
from app.services.itsm_service import ITSMService, _itsm_priority_to_todo
|
|||
|
|
from app.services.todo_aggregator_service import CACHE_TTL, PRIORITY_ORDER, TodoAggregatorService
|
|||
|
|
from app.services.todo_source_service import ApprovalTodoService
|
|||
|
|
from app.utils.itsm_signer import ITSMSigner
|
|||
|
|
|
|||
|
|
|
|||
|
|
# =============================================================================
|
|||
|
|
# 辅助：带 keys() 支持的 Mock Redis
|
|||
|
|
# =============================================================================
|
|||
|
|
|
|||
|
|
class TestRedis:
|
|||
|
|
"""内存字典 Redis mock，支持 get/setex/delete/keys/close。"""
|
|||
|
|
|
|||
|
|
__test__ = False # 告知 pytest 不要收集此类作为测试用例
|
|||
|
|
|
|||
|
|
def __init__(self):
|
|||
|
|
self._data: Dict[str, str] = {}
|
|||
|
|
|
|||
|
|
async def get(self, key: str) -> Optional[bytes]:
|
|||
|
|
value = self._data.get(key)
|
|||
|
|
if value is not None:
|
|||
|
|
return value.encode("utf-8") if isinstance(value, str) else value
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
async def setex(self, name: str, time: int, value: str) -> None:
|
|||
|
|
self._data[name] = value
|
|||
|
|
|
|||
|
|
async def set(self, name: str, value: str, **kwargs) -> Optional[bool]:
|
|||
|
|
self._data[name] = value
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
async def delete(self, *names) -> int:
|
|||
|
|
count = 0
|
|||
|
|
for name in names:
|
|||
|
|
if name in self._data:
|
|||
|
|
del self._data[name]
|
|||
|
|
count += 1
|
|||
|
|
return count
|
|||
|
|
|
|||
|
|
async def exists(self, *keys) -> int:
|
|||
|
|
return sum(1 for k in keys if k in self._data)
|
|||
|
|
|
|||
|
|
async def keys(self, pattern: str) -> list:
|
|||
|
|
import fnmatch
|
|||
|
|
return [k for k in self._data if fnmatch.fnmatch(k, pattern)]
|
|||
|
|
|
|||
|
|
async def close(self) -> None:
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
|
|||
|
|
# =============================================================================
|
|||
|
|
# 测试数据常量
|
|||
|
|
# =============================================================================
|
|||
|
|
|
|||
|
|
AGENT_USERID = "test_agent_001"
|
|||
|
|
|
|||
|
|
# 企微审批详情 mock（当前审批人 = AGENT_USERID）
|
|||
|
|
APPROVAL_DETAIL_MINE = {
|
|||
|
|
"errcode": 0,
|
|||
|
|
"errmsg": "ok",
|
|||
|
|
"info": {
|
|||
|
|
"sp_no": "202607110001",
|
|||
|
|
"sp_name": "资产领用登记",
|
|||
|
|
"sp_status": 1,
|
|||
|
|
"template_id": "C4c8qt31AbSHwN9MuaFhYXt4Qwsx6ZLCftAFh6X1w",
|
|||
|
|
"apply_time": 1720656000,
|
|||
|
|
"applyer": {"userid": "applicant_001", "partyid": "1"},
|
|||
|
|
"sp_record": [
|
|||
|
|
{
|
|||
|
|
"status": 1,
|
|||
|
|
"type": 1,
|
|||
|
|
"approverattr": 1,
|
|||
|
|
"approver": [{"userid": AGENT_USERID, "partyid": "2"}],
|
|||
|
|
}
|
|||
|
|
],
|
|||
|
|
},
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 企微审批详情 mock（当前审批人 = 其他坐席）
|
|||
|
|
APPROVAL_DETAIL_OTHER = {
|
|||
|
|
"errcode": 0,
|
|||
|
|
"errmsg": "ok",
|
|||
|
|
"info": {
|
|||
|
|
"sp_no": "202607110002",
|
|||
|
|
"sp_name": "资产借用申请",
|
|||
|
|
"sp_status": 1,
|
|||
|
|
"template_id": "3TmACnFs8oqgYcasxVh4BfSMGNX7p9sb6ydBX77mK",
|
|||
|
|
"apply_time": 1720656000,
|
|||
|
|
"applyer": {"userid": "applicant_002", "partyid": "1"},
|
|||
|
|
"sp_record": [
|
|||
|
|
{
|
|||
|
|
"status": 1,
|
|||
|
|
"type": 1,
|
|||
|
|
"approverattr": 1,
|
|||
|
|
"approver": [{"userid": "other_agent", "partyid": "3"}],
|
|||
|
|
}
|
|||
|
|
],
|
|||
|
|
},
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# ITSM 工单详情 mock
|
|||
|
|
ITSM_DETAIL = {
|
|||
|
|
"process_instance_id": "12345",
|
|||
|
|
"title": "网络故障报修",
|
|||
|
|
"priority": "urgent",
|
|||
|
|
"status": "pending",
|
|||
|
|
"creator": "user_001",
|
|||
|
|
"executor": AGENT_USERID,
|
|||
|
|
"created_at": "2026-07-11T10:00:00Z",
|
|||
|
|
"updated_at": "2026-07-11T10:30:00Z",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# ITSM API 成功响应
|
|||
|
|
ITSM_API_RESPONSE = {
|
|||
|
|
"code": 20000,
|
|||
|
|
"message": "success",
|
|||
|
|
"data": ITSM_DETAIL,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _make_httpx_mock(response_json: dict) -> AsyncMock:
|
|||
|
|
"""创建 httpx.AsyncClient 的 mock，post 返回指定 JSON。"""
|
|||
|
|
mock_response = MagicMock()
|
|||
|
|
mock_response.json.return_value = response_json
|
|||
|
|
mock_client = AsyncMock()
|
|||
|
|
mock_client.post.return_value = mock_response
|
|||
|
|
mock_client.__aenter__.return_value = mock_client
|
|||
|
|
mock_client.__aexit__.return_value = None
|
|||
|
|
return mock_client
|
|||
|
|
|
|||
|
|
|
|||
|
|
# =============================================================================
|
|||
|
|
# 1. ITSM 签名工具测试
|
|||
|
|
# =============================================================================
|
|||
|
|
|
|||
|
|
class TestITSMSigner:
|
|||
|
|
"""ITSMSigner 签名计算和 headers 生成测试。"""
|
|||
|
|
|
|||
|
|
def test_itsm_signer_compute_signature(self):
|
|||
|
|
"""验证签名计算结果正确：sort → concat → quote_plus → sha1 → upper。"""
|
|||
|
|
app_id = "test_app_id"
|
|||
|
|
timestamp = "1700000000000"
|
|||
|
|
app_secret = "test_secret"
|
|||
|
|
biz_data = {"key": "value"}
|
|||
|
|
|
|||
|
|
result = ITSMSigner.compute_signature(app_id, timestamp, app_secret, biz_data)
|
|||
|
|
|
|||
|
|
# 1. 应为 40 字符大写 hex（SHA1 hexdigest upper）
|
|||
|
|
assert len(result) == 40
|
|||
|
|
assert result == result.upper()
|
|||
|
|
assert all(c in "0123456789ABCDEF" for c in result)
|
|||
|
|
|
|||
|
|
# 2. 独立复现算法，验证一致性
|
|||
|
|
sign_params = {
|
|||
|
|
"appSecret": app_secret,
|
|||
|
|
"appId": app_id,
|
|||
|
|
"timestamp": timestamp,
|
|||
|
|
"bizData": json.dumps(biz_data, ensure_ascii=False),
|
|||
|
|
}
|
|||
|
|
sorted_params = sorted(sign_params.items(), key=lambda x: x[0])
|
|||
|
|
canonicalized = "".join(str(v) for _, v in sorted_params)
|
|||
|
|
quoted = quote_plus(canonicalized)
|
|||
|
|
expected = hashlib.sha1(quoted.encode("utf-8")).hexdigest().upper()
|
|||
|
|
assert result == expected
|
|||
|
|
|
|||
|
|
def test_itsm_signer_compute_signature_deterministic(self):
|
|||
|
|
"""相同输入应产生相同签名（确定性）。"""
|
|||
|
|
args = ("app1", "1700000000000", "secret1", {"a": 1})
|
|||
|
|
sig1 = ITSMSigner.compute_signature(*args)
|
|||
|
|
sig2 = ITSMSigner.compute_signature(*args)
|
|||
|
|
assert sig1 == sig2
|
|||
|
|
|
|||
|
|
def test_itsm_signer_compute_signature_different_input(self):
|
|||
|
|
"""不同输入应产生不同签名。"""
|
|||
|
|
sig1 = ITSMSigner.compute_signature("app1", "ts1", "secret1", {"a": 1})
|
|||
|
|
sig2 = ITSMSigner.compute_signature("app2", "ts1", "secret1", {"a": 1})
|
|||
|
|
assert sig1 != sig2
|
|||
|
|
|
|||
|
|
def test_itsm_signer_get_headers(self):
|
|||
|
|
"""验证生成的 headers 包含 appId/timestamp/sign/Content-Type。"""
|
|||
|
|
app_id = "test_app_id"
|
|||
|
|
app_secret = "test_secret"
|
|||
|
|
biz_data = {"key": "value"}
|
|||
|
|
|
|||
|
|
headers = ITSMSigner.get_headers(app_id, app_secret, biz_data)
|
|||
|
|
|
|||
|
|
# 验证必需的 header 字段
|
|||
|
|
assert "appId" in headers
|
|||
|
|
assert "timestamp" in headers
|
|||
|
|
assert "sign" in headers
|
|||
|
|
assert "Content-Type" in headers
|
|||
|
|
|
|||
|
|
# 验证值
|
|||
|
|
assert headers["appId"] == app_id
|
|||
|
|
assert headers["Content-Type"] == "application/json"
|
|||
|
|
assert len(headers["timestamp"]) > 0 # 非空时间戳
|
|||
|
|
assert len(headers["sign"]) == 40 # SHA1 hex
|
|||
|
|
|
|||
|
|
# 验证 sign 与 compute_signature 一致
|
|||
|
|
expected_sign = ITSMSigner.compute_signature(
|
|||
|
|
app_id, headers["timestamp"], app_secret, biz_data
|
|||
|
|
)
|
|||
|
|
assert headers["sign"] == expected_sign
|
|||
|
|
|
|||
|
|
|
|||
|
|
# =============================================================================
|
|||
|
|
# 2. TodoAggregatorService 测试
|
|||
|
|
# =============================================================================
|
|||
|
|
|
|||
|
|
class TestTodoAggregatorService:
|
|||
|
|
"""聚合服务测试：缓存/并行查询/容错/排序/过滤/详情路由。"""
|
|||
|
|
|
|||
|
|
async def test_aggregator_get_todo_list_cache_hit(self):
|
|||
|
|
"""缓存命中时直接返回，不调用外部 API。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
cached = {"items": [{"id": "approval:123", "type": "approval", "priority": "high"}], "total": 1}
|
|||
|
|
cache_key = TodoAggregatorService._cache_key(AGENT_USERID, None)
|
|||
|
|
await redis.setex(cache_key, CACHE_TTL, json.dumps(cached, ensure_ascii=False))
|
|||
|
|
|
|||
|
|
aggregator = TodoAggregatorService(redis)
|
|||
|
|
|
|||
|
|
with patch("app.services.todo_aggregator_service.ApprovalTodoService") as MockAppr:
|
|||
|
|
with patch("app.services.todo_aggregator_service.ITSMService") as MockITSM:
|
|||
|
|
result = await aggregator.get_todo_list(AGENT_USERID, None)
|
|||
|
|
# 缓存命中，不应创建数据源 Service
|
|||
|
|
MockAppr.assert_not_called()
|
|||
|
|
MockITSM.assert_not_called()
|
|||
|
|
|
|||
|
|
assert result["cached"] is True
|
|||
|
|
assert result["total"] == 1
|
|||
|
|
assert result["items"][0]["id"] == "approval:123"
|
|||
|
|
|
|||
|
|
async def test_aggregator_get_todo_list_cache_miss(self):
|
|||
|
|
"""缓存未命中时并行查询两个数据源。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
aggregator = TodoAggregatorService(redis)
|
|||
|
|
|
|||
|
|
approval_items = [{"id": "approval:001", "type": "approval", "priority": "high"}]
|
|||
|
|
itsm_items = [{"id": "ticket:001", "type": "ticket", "priority": "normal"}]
|
|||
|
|
|
|||
|
|
with patch("app.services.todo_aggregator_service.ApprovalTodoService") as MockAppr:
|
|||
|
|
with patch("app.services.todo_aggregator_service.ITSMService") as MockITSM:
|
|||
|
|
mock_appr = AsyncMock()
|
|||
|
|
mock_appr.get_todo_list.return_value = approval_items
|
|||
|
|
MockAppr.return_value = mock_appr
|
|||
|
|
|
|||
|
|
mock_itsm = AsyncMock()
|
|||
|
|
mock_itsm.get_todo_list.return_value = itsm_items
|
|||
|
|
MockITSM.return_value = mock_itsm
|
|||
|
|
|
|||
|
|
result = await aggregator.get_todo_list(AGENT_USERID, None)
|
|||
|
|
|
|||
|
|
assert result["cached"] is False
|
|||
|
|
assert result["total"] == 2
|
|||
|
|
ids = [item["id"] for item in result["items"]]
|
|||
|
|
assert "approval:001" in ids
|
|||
|
|
assert "ticket:001" in ids
|
|||
|
|
|
|||
|
|
# 验证缓存已写入
|
|||
|
|
cache_key = TodoAggregatorService._cache_key(AGENT_USERID, None)
|
|||
|
|
cached_raw = await redis.get(cache_key)
|
|||
|
|
assert cached_raw is not None
|
|||
|
|
|
|||
|
|
async def test_aggregator_get_todo_list_one_source_fails(self):
|
|||
|
|
"""一个数据源失败时另一个仍正常返回（容错）。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
aggregator = TodoAggregatorService(redis)
|
|||
|
|
|
|||
|
|
itsm_items = [{"id": "ticket:001", "type": "ticket", "priority": "urgent"}]
|
|||
|
|
|
|||
|
|
with patch("app.services.todo_aggregator_service.ApprovalTodoService") as MockAppr:
|
|||
|
|
with patch("app.services.todo_aggregator_service.ITSMService") as MockITSM:
|
|||
|
|
mock_appr = AsyncMock()
|
|||
|
|
mock_appr.get_todo_list.side_effect = Exception("企微 API 不可达")
|
|||
|
|
MockAppr.return_value = mock_appr
|
|||
|
|
|
|||
|
|
mock_itsm = AsyncMock()
|
|||
|
|
mock_itsm.get_todo_list.return_value = itsm_items
|
|||
|
|
MockITSM.return_value = mock_itsm
|
|||
|
|
|
|||
|
|
result = await aggregator.get_todo_list(AGENT_USERID, None)
|
|||
|
|
|
|||
|
|
# 审批失败但工单正常返回
|
|||
|
|
assert result["total"] == 1
|
|||
|
|
assert result["items"][0]["id"] == "ticket:001"
|
|||
|
|
|
|||
|
|
async def test_aggregator_get_todo_list_priority_sort(self):
|
|||
|
|
"""返回结果按 urgent → high → normal 排序。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
aggregator = TodoAggregatorService(redis)
|
|||
|
|
|
|||
|
|
# 故意打乱顺序
|
|||
|
|
approval_items = [
|
|||
|
|
{"id": "approval:1", "type": "approval", "priority": "normal"},
|
|||
|
|
{"id": "approval:2", "type": "approval", "priority": "urgent"},
|
|||
|
|
{"id": "approval:3", "type": "approval", "priority": "high"},
|
|||
|
|
]
|
|||
|
|
itsm_items = [
|
|||
|
|
{"id": "ticket:1", "type": "ticket", "priority": "high"},
|
|||
|
|
{"id": "ticket:2", "type": "ticket", "priority": "urgent"},
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
with patch("app.services.todo_aggregator_service.ApprovalTodoService") as MockAppr:
|
|||
|
|
with patch("app.services.todo_aggregator_service.ITSMService") as MockITSM:
|
|||
|
|
mock_appr = AsyncMock()
|
|||
|
|
mock_appr.get_todo_list.return_value = approval_items
|
|||
|
|
MockAppr.return_value = mock_appr
|
|||
|
|
|
|||
|
|
mock_itsm = AsyncMock()
|
|||
|
|
mock_itsm.get_todo_list.return_value = itsm_items
|
|||
|
|
MockITSM.return_value = mock_itsm
|
|||
|
|
|
|||
|
|
result = await aggregator.get_todo_list(AGENT_USERID, None)
|
|||
|
|
|
|||
|
|
priorities = [item["priority"] for item in result["items"]]
|
|||
|
|
# urgent 应在前，normal 应在后
|
|||
|
|
urgent_idx = [i for i, p in enumerate(priorities) if p == "urgent"]
|
|||
|
|
high_idx = [i for i, p in enumerate(priorities) if p == "high"]
|
|||
|
|
normal_idx = [i for i, p in enumerate(priorities) if p == "normal"]
|
|||
|
|
|
|||
|
|
assert all(i < min(high_idx) for i in urgent_idx) if urgent_idx and high_idx else True
|
|||
|
|
assert all(i < min(normal_idx) for i in high_idx) if high_idx and normal_idx else True
|
|||
|
|
|
|||
|
|
async def test_aggregator_get_todo_list_type_filter(self):
|
|||
|
|
"""type=approval 只返回审批类型。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
aggregator = TodoAggregatorService(redis)
|
|||
|
|
|
|||
|
|
approval_items = [{"id": "approval:1", "type": "approval", "priority": "high"}]
|
|||
|
|
itsm_items = [{"id": "ticket:1", "type": "ticket", "priority": "urgent"}]
|
|||
|
|
|
|||
|
|
with patch("app.services.todo_aggregator_service.ApprovalTodoService") as MockAppr:
|
|||
|
|
with patch("app.services.todo_aggregator_service.ITSMService") as MockITSM:
|
|||
|
|
mock_appr = AsyncMock()
|
|||
|
|
mock_appr.get_todo_list.return_value = approval_items
|
|||
|
|
MockAppr.return_value = mock_appr
|
|||
|
|
|
|||
|
|
mock_itsm = AsyncMock()
|
|||
|
|
mock_itsm.get_todo_list.return_value = itsm_items
|
|||
|
|
MockITSM.return_value = mock_itsm
|
|||
|
|
|
|||
|
|
result = await aggregator.get_todo_list(AGENT_USERID, "approval")
|
|||
|
|
|
|||
|
|
assert result["total"] == 1
|
|||
|
|
assert all(item["type"] == "approval" for item in result["items"])
|
|||
|
|
|
|||
|
|
async def test_aggregator_get_todo_detail_approval(self):
|
|||
|
|
"""详情查询路由到 ApprovalTodoService。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
aggregator = TodoAggregatorService(redis)
|
|||
|
|
|
|||
|
|
with patch("app.services.todo_aggregator_service.ApprovalTodoService") as MockAppr:
|
|||
|
|
mock_appr = AsyncMock()
|
|||
|
|
mock_appr.get_todo_detail.return_value = {"id": "approval:202607110001", "type": "approval"}
|
|||
|
|
MockAppr.return_value = mock_appr
|
|||
|
|
|
|||
|
|
result = await aggregator.get_todo_detail(
|
|||
|
|
AGENT_USERID, "approval:202607110001", "approval"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
mock_appr.get_todo_detail.assert_called_once_with("202607110001")
|
|||
|
|
|
|||
|
|
assert result is not None
|
|||
|
|
assert result["id"] == "approval:202607110001"
|
|||
|
|
|
|||
|
|
async def test_aggregator_get_todo_detail_ticket(self):
|
|||
|
|
"""详情查询路由到 ITSMService。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
aggregator = TodoAggregatorService(redis)
|
|||
|
|
|
|||
|
|
with patch("app.services.todo_aggregator_service.ITSMService") as MockITSM:
|
|||
|
|
mock_itsm = AsyncMock()
|
|||
|
|
mock_itsm.get_todo_detail.return_value = {"id": "ticket:12345", "type": "ticket"}
|
|||
|
|
MockITSM.return_value = mock_itsm
|
|||
|
|
|
|||
|
|
result = await aggregator.get_todo_detail(
|
|||
|
|
AGENT_USERID, "ticket:12345", "ticket"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
mock_itsm.get_todo_detail.assert_called_once_with("12345")
|
|||
|
|
|
|||
|
|
assert result is not None
|
|||
|
|
assert result["id"] == "ticket:12345"
|
|||
|
|
|
|||
|
|
async def test_aggregator_get_todo_detail_auto_parse_type(self):
|
|||
|
|
"""未提供 todo_type 时从 item_id 自动解析类型前缀。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
aggregator = TodoAggregatorService(redis)
|
|||
|
|
|
|||
|
|
with patch("app.services.todo_aggregator_service.ApprovalTodoService") as MockAppr:
|
|||
|
|
mock_appr = AsyncMock()
|
|||
|
|
mock_appr.get_todo_detail.return_value = {"id": "approval:001"}
|
|||
|
|
MockAppr.return_value = mock_appr
|
|||
|
|
|
|||
|
|
result = await aggregator.get_todo_detail(AGENT_USERID, "approval:001")
|
|||
|
|
|
|||
|
|
assert result is not None
|
|||
|
|
mock_appr.get_todo_detail.assert_called_once_with("001")
|
|||
|
|
|
|||
|
|
async def test_aggregator_invalidate_cache(self):
|
|||
|
|
"""缓存失效后重新查询。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
# 预填充缓存
|
|||
|
|
await redis.setex("todo:cache:test_agent:all", 45, '{"items":[],"total":0}')
|
|||
|
|
await redis.setex("todo:cache:test_agent:approval", 45, '{"items":[],"total":0}')
|
|||
|
|
|
|||
|
|
aggregator = TodoAggregatorService(redis)
|
|||
|
|
await aggregator._invalidate_cache("test_agent")
|
|||
|
|
|
|||
|
|
# 验证缓存已清除
|
|||
|
|
assert await redis.get("todo:cache:test_agent:all") is None
|
|||
|
|
assert await redis.get("todo:cache:test_agent:approval") is None
|
|||
|
|
|
|||
|
|
def test_aggregator_cache_key_format(self):
|
|||
|
|
"""验证缓存 key 格式：todo:cache:{userid}:{type_or_all}。"""
|
|||
|
|
key_all = TodoAggregatorService._cache_key("user1", None)
|
|||
|
|
assert key_all == "todo:cache:user1:all"
|
|||
|
|
|
|||
|
|
key_approval = TodoAggregatorService._cache_key("user1", "approval")
|
|||
|
|
assert key_approval == "todo:cache:user1:approval"
|
|||
|
|
|
|||
|
|
key_ticket = TodoAggregatorService._cache_key("user1", "ticket")
|
|||
|
|
assert key_ticket == "todo:cache:user1:ticket"
|
|||
|
|
|
|||
|
|
|
|||
|
|
# =============================================================================
|
|||
|
|
# 3. ApprovalTodoService 测试
|
|||
|
|
# =============================================================================
|
|||
|
|
|
|||
|
|
class TestApprovalTodoService:
|
|||
|
|
"""企微审批数据源测试。"""
|
|||
|
|
|
|||
|
|
async def test_approval_get_todo_list(self):
|
|||
|
|
"""mock 企微 API 返回，验证 getapprovaldata → getapprovaldetail → filter → map 流程。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
service = ApprovalTodoService(agent_userid=AGENT_USERID, redis=redis)
|
|||
|
|
|
|||
|
|
# Mock token manager
|
|||
|
|
mock_token_mgr = AsyncMock()
|
|||
|
|
mock_token_mgr.get_token.return_value = "fake_access_token"
|
|||
|
|
mock_token_mgr.close = AsyncMock()
|
|||
|
|
|
|||
|
|
# Mock getapprovaldata response
|
|||
|
|
getapprovaldata_resp = {
|
|||
|
|
"errcode": 0,
|
|||
|
|
"data": [
|
|||
|
|
{"sp_no": "202607110001"},
|
|||
|
|
{"sp_no": "202607110002"},
|
|||
|
|
],
|
|||
|
|
"next_cursor": 0,
|
|||
|
|
}
|
|||
|
|
mock_http = _make_httpx_mock(getapprovaldata_resp)
|
|||
|
|
|
|||
|
|
with patch("app.services.todo_source_service.ApprovalTokenManager", return_value=mock_token_mgr):
|
|||
|
|
with patch("app.services.todo_source_service.httpx.AsyncClient", return_value=mock_http):
|
|||
|
|
with patch(
|
|||
|
|
"app.services.todo_source_service.get_approval_detail",
|
|||
|
|
new_callable=AsyncMock,
|
|||
|
|
side_effect=[APPROVAL_DETAIL_MINE, APPROVAL_DETAIL_OTHER],
|
|||
|
|
):
|
|||
|
|
result = await service.get_todo_list()
|
|||
|
|
|
|||
|
|
# 只有 APPROVAL_DETAIL_MINE 的当前审批人是 AGENT_USERID
|
|||
|
|
assert len(result) == 1
|
|||
|
|
assert result[0]["id"] == "approval:202607110001"
|
|||
|
|
assert result[0]["type"] == "approval"
|
|||
|
|
|
|||
|
|
async def test_approval_get_todo_list_token_fail(self):
|
|||
|
|
"""access_token 获取失败时返回空列表。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
service = ApprovalTodoService(agent_userid=AGENT_USERID, redis=redis)
|
|||
|
|
|
|||
|
|
mock_token_mgr = AsyncMock()
|
|||
|
|
mock_token_mgr.get_token.return_value = ""
|
|||
|
|
mock_token_mgr.close = AsyncMock()
|
|||
|
|
|
|||
|
|
with patch("app.services.todo_source_service.ApprovalTokenManager", return_value=mock_token_mgr):
|
|||
|
|
result = await service.get_todo_list()
|
|||
|
|
|
|||
|
|
assert result == []
|
|||
|
|
|
|||
|
|
async def test_approval_filter_by_current_approver(self):
|
|||
|
|
"""验证只返回当前审批人是当前坐席的审批单。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
service = ApprovalTodoService(agent_userid=AGENT_USERID, redis=redis)
|
|||
|
|
|
|||
|
|
details = [APPROVAL_DETAIL_MINE, APPROVAL_DETAIL_OTHER]
|
|||
|
|
filtered = service._filter_by_current_approver(details)
|
|||
|
|
|
|||
|
|
assert len(filtered) == 1
|
|||
|
|
assert filtered[0]["info"]["sp_no"] == "202607110001"
|
|||
|
|
|
|||
|
|
async def test_approval_filter_empty_list(self):
|
|||
|
|
"""空列表过滤返回空列表。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
service = ApprovalTodoService(agent_userid=AGENT_USERID, redis=redis)
|
|||
|
|
assert service._filter_by_current_approver([]) == []
|
|||
|
|
|
|||
|
|
async def test_approval_map_to_todo_item(self):
|
|||
|
|
"""验证企微审批详情正确映射为 TodoItemData 格式。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
service = ApprovalTodoService(agent_userid=AGENT_USERID, redis=redis)
|
|||
|
|
|
|||
|
|
mapped = service._map_to_todo_item(APPROVAL_DETAIL_MINE)
|
|||
|
|
|
|||
|
|
assert mapped["id"] == "approval:202607110001"
|
|||
|
|
assert mapped["type"] == "approval"
|
|||
|
|
assert mapped["title"] == "资产领用登记"
|
|||
|
|
assert mapped["priority"] == "high"
|
|||
|
|
assert mapped["status"] == "pending"
|
|||
|
|
assert mapped["assigned_agent_id"] == AGENT_USERID
|
|||
|
|
|
|||
|
|
# 验证 description 包含关键字段
|
|||
|
|
desc = mapped["description"]
|
|||
|
|
assert desc["sp_no"] == "202607110001"
|
|||
|
|
assert desc["applicant"] == "applicant_001"
|
|||
|
|
assert desc["sp_status"] == 1
|
|||
|
|
assert desc["current_approver"] == AGENT_USERID
|
|||
|
|
|
|||
|
|
async def test_approval_id_format(self):
|
|||
|
|
"""验证 ID 格式为 "approval:{sp_no}"。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
service = ApprovalTodoService(agent_userid=AGENT_USERID, redis=redis)
|
|||
|
|
|
|||
|
|
mapped = service._map_to_todo_item(APPROVAL_DETAIL_MINE)
|
|||
|
|
assert mapped["id"] == "approval:202607110001"
|
|||
|
|
assert mapped["id"].startswith("approval:")
|
|||
|
|
|
|||
|
|
async def test_approval_get_todo_detail(self):
|
|||
|
|
"""验证详情查询流程：get_token → get_approval_detail → map。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
service = ApprovalTodoService(agent_userid=AGENT_USERID, redis=redis)
|
|||
|
|
|
|||
|
|
mock_token_mgr = AsyncMock()
|
|||
|
|
mock_token_mgr.get_token.return_value = "fake_access_token"
|
|||
|
|
mock_token_mgr.close = AsyncMock()
|
|||
|
|
|
|||
|
|
with patch("app.services.todo_source_service.ApprovalTokenManager", return_value=mock_token_mgr):
|
|||
|
|
with patch(
|
|||
|
|
"app.services.todo_source_service.get_approval_detail",
|
|||
|
|
new_callable=AsyncMock,
|
|||
|
|
return_value=APPROVAL_DETAIL_MINE,
|
|||
|
|
):
|
|||
|
|
result = await service.get_todo_detail("202607110001")
|
|||
|
|
|
|||
|
|
assert result is not None
|
|||
|
|
assert result["id"] == "approval:202607110001"
|
|||
|
|
assert result["type"] == "approval"
|
|||
|
|
|
|||
|
|
|
|||
|
|
# =============================================================================
|
|||
|
|
# 4. ITSMService 测试
|
|||
|
|
# =============================================================================
|
|||
|
|
|
|||
|
|
class TestITSMService:
|
|||
|
|
"""ITSM 工单数据源测试。"""
|
|||
|
|
|
|||
|
|
async def test_itsm_get_todo_list_not_implemented(self):
|
|||
|
|
"""列表方法返回空列表（API 尚未实现）。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
with patch.object(settings, "itsm_app_id", "test_app_id"):
|
|||
|
|
with patch.object(settings, "itsm_app_secret", "test_secret"):
|
|||
|
|
with patch.object(settings, "itsm_base_url", "https://test-itsm.example.com"):
|
|||
|
|
service = ITSMService(agent_userid=AGENT_USERID, redis=redis)
|
|||
|
|
result = await service.get_todo_list()
|
|||
|
|
|
|||
|
|
assert result == []
|
|||
|
|
assert isinstance(result, list)
|
|||
|
|
|
|||
|
|
async def test_itsm_get_todo_detail(self):
|
|||
|
|
"""mock ITSM API 返回，验证详情查询和映射。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
mock_http = _make_httpx_mock(ITSM_API_RESPONSE)
|
|||
|
|
|
|||
|
|
with patch.object(settings, "itsm_app_id", "test_app_id"):
|
|||
|
|
with patch.object(settings, "itsm_app_secret", "test_secret"):
|
|||
|
|
with patch.object(settings, "itsm_base_url", "https://test-itsm.example.com"):
|
|||
|
|
service = ITSMService(agent_userid=AGENT_USERID, redis=redis)
|
|||
|
|
with patch("app.services.itsm_service.httpx.AsyncClient", return_value=mock_http):
|
|||
|
|
result = await service.get_todo_detail("12345")
|
|||
|
|
|
|||
|
|
assert result is not None
|
|||
|
|
assert result["id"] == "ticket:12345"
|
|||
|
|
assert result["type"] == "ticket"
|
|||
|
|
assert result["title"] == "网络故障报修"
|
|||
|
|
assert result["priority"] == "urgent"
|
|||
|
|
assert result["status"] == "pending"
|
|||
|
|
|
|||
|
|
async def test_itsm_no_credentials(self):
|
|||
|
|
"""itsm_app_id 为空时返回空 + 日志告警。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
with patch.object(settings, "itsm_app_id", ""):
|
|||
|
|
service = ITSMService(agent_userid=AGENT_USERID, redis=redis)
|
|||
|
|
|
|||
|
|
# 列表返回空
|
|||
|
|
list_result = await service.get_todo_list()
|
|||
|
|
assert list_result == []
|
|||
|
|
|
|||
|
|
# 详情返回 None
|
|||
|
|
detail_result = await service.get_todo_detail("12345")
|
|||
|
|
assert detail_result is None
|
|||
|
|
|
|||
|
|
async def test_itsm_id_format(self):
|
|||
|
|
"""验证 ID 格式为 "ticket:{process_instance_id}"。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
mock_http = _make_httpx_mock(ITSM_API_RESPONSE)
|
|||
|
|
|
|||
|
|
with patch.object(settings, "itsm_app_id", "test_app_id"):
|
|||
|
|
with patch.object(settings, "itsm_app_secret", "test_secret"):
|
|||
|
|
with patch.object(settings, "itsm_base_url", "https://test-itsm.example.com"):
|
|||
|
|
service = ITSMService(agent_userid=AGENT_USERID, redis=redis)
|
|||
|
|
with patch("app.services.itsm_service.httpx.AsyncClient", return_value=mock_http):
|
|||
|
|
result = await service.get_todo_detail("12345")
|
|||
|
|
|
|||
|
|
assert result is not None
|
|||
|
|
assert result["id"] == "ticket:12345"
|
|||
|
|
assert result["id"].startswith("ticket:")
|
|||
|
|
|
|||
|
|
async def test_itsm_get_todo_detail_api_error(self):
|
|||
|
|
"""ITSM API 返回错误码时返回 None。"""
|
|||
|
|
redis = TestRedis()
|
|||
|
|
error_response = {"code": 50000, "message": "internal error", "data": None}
|
|||
|
|
mock_http = _make_httpx_mock(error_response)
|
|||
|
|
|
|||
|
|
with patch.object(settings, "itsm_app_id", "test_app_id"):
|
|||
|
|
with patch.object(settings, "itsm_app_secret", "test_secret"):
|
|||
|
|
with patch.object(settings, "itsm_base_url", "https://test-itsm.example.com"):
|
|||
|
|
service = ITSMService(agent_userid=AGENT_USERID, redis=redis)
|
|||
|
|
with patch("app.services.itsm_service.httpx.AsyncClient", return_value=mock_http):
|
|||
|
|
result = await service.get_todo_detail("12345")
|
|||
|
|
|
|||
|
|
assert result is None
|
|||
|
|
|
|||
|
|
def test_itsm_priority_mapping(self):
|
|||
|
|
"""验证 ITSM 优先级映射到 urgent/high/normal。"""
|
|||
|
|
assert _itsm_priority_to_todo("urgent") == "urgent"
|
|||
|
|
assert _itsm_priority_to_todo("紧急") == "urgent"
|
|||
|
|
assert _itsm_priority_to_todo("1") == "urgent"
|
|||
|
|
assert _itsm_priority_to_todo("P0") == "urgent"
|
|||
|
|
|
|||
|
|
assert _itsm_priority_to_todo("high") == "high"
|
|||
|
|
assert _itsm_priority_to_todo("高") == "high"
|
|||
|
|
assert _itsm_priority_to_todo("2") == "high"
|
|||
|
|
|
|||
|
|
assert _itsm_priority_to_todo("normal") == "normal"
|
|||
|
|
assert _itsm_priority_to_todo(None) == "normal"
|
|||
|
|
assert _itsm_priority_to_todo("unknown") == "normal"
|
|||
|
|
|
|||
|
|
|
|||
|
|
# =============================================================================
|
|||
|
|
# 5. API 端点测试
|
|||
|
|
# =============================================================================
|
|||
|
|
|
|||
|
|
class TestTodoItemsAPI:
|
|||
|
|
"""todo-items API 端点测试。"""
|
|||
|
|
|
|||
|
|
@pytest_asyncio.fixture
|
|||
|
|
async def todo_client(self, db_session, mock_redis):
|
|||
|
|
"""创建带 mock 认证的测试客户端。"""
|
|||
|
|
from app.api.agents import get_current_agent
|
|||
|
|
from app.main import create_app
|
|||
|
|
from app.models.agent import Agent
|
|||
|
|
|
|||
|
|
app = create_app()
|
|||
|
|
|
|||
|
|
# Mock agent
|
|||
|
|
mock_agent = MagicMock()
|
|||
|
|
mock_agent.user_id = "todo_test_agent"
|
|||
|
|
mock_agent.name = "Todo Test Agent"
|
|||
|
|
|
|||
|
|
app.dependency_overrides[get_current_agent] = lambda: mock_agent
|
|||
|
|
|
|||
|
|
mock_redis_instance = AsyncMock()
|
|||
|
|
mock_redis_instance.close = AsyncMock()
|
|||
|
|
|
|||
|
|
with patch("app.api.todo_items._get_redis", return_value=mock_redis_instance):
|
|||
|
|
transport = ASGITransport(app=app)
|
|||
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|||
|
|
yield ac
|
|||
|
|
|
|||
|
|
app.dependency_overrides.clear()
|
|||
|
|
|
|||
|
|
async def test_api_list_todo_items(self, todo_client):
|
|||
|
|
"""GET /todo-items 返回正确格式。"""
|
|||
|
|
mock_result = {
|
|||
|
|
"items": [
|
|||
|
|
{"id": "approval:001", "type": "approval", "priority": "high", "title": "审批1"},
|
|||
|
|
{"id": "ticket:001", "type": "ticket", "priority": "normal", "title": "工单1"},
|
|||
|
|
],
|
|||
|
|
"total": 2,
|
|||
|
|
"cached": False,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
with patch.object(TodoAggregatorService, "get_todo_list", new_callable=AsyncMock) as mock_gl:
|
|||
|
|
mock_gl.return_value = mock_result
|
|||
|
|
response = await todo_client.get("/todo-items")
|
|||
|
|
|
|||
|
|
assert response.status_code == 200
|
|||
|
|
data = response.json()
|
|||
|
|
assert data["code"] == 0
|
|||
|
|
assert data["data"]["total"] == 2
|
|||
|
|
assert len(data["data"]["items"]) == 2
|
|||
|
|
|
|||
|
|
async def test_api_list_todo_items_with_type_filter(self, todo_client):
|
|||
|
|
"""type=approval 过滤。"""
|
|||
|
|
mock_result = {
|
|||
|
|
"items": [{"id": "approval:001", "type": "approval", "priority": "high"}],
|
|||
|
|
"total": 1,
|
|||
|
|
"cached": False,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
with patch.object(TodoAggregatorService, "get_todo_list", new_callable=AsyncMock) as mock_gl:
|
|||
|
|
mock_gl.return_value = mock_result
|
|||
|
|
response = await todo_client.get("/todo-items?type=approval")
|
|||
|
|
|
|||
|
|
assert response.status_code == 200
|
|||
|
|
data = response.json()
|
|||
|
|
assert data["data"]["total"] == 1
|
|||
|
|
assert data["data"]["items"][0]["type"] == "approval"
|
|||
|
|
|
|||
|
|
# 验证 type 参数传递正确
|
|||
|
|
call_kwargs = mock_gl.call_args
|
|||
|
|
assert call_kwargs.kwargs.get("todo_type") == "approval" or call_kwargs[1].get("todo_type") == "approval"
|
|||
|
|
|
|||
|
|
async def test_api_list_todo_items_with_force(self, todo_client):
|
|||
|
|
"""_force=1 跳过缓存。"""
|
|||
|
|
mock_result = {"items": [], "total": 0, "cached": False}
|
|||
|
|
|
|||
|
|
with patch.object(TodoAggregatorService, "_invalidate_cache", new_callable=AsyncMock) as mock_inv:
|
|||
|
|
with patch.object(TodoAggregatorService, "get_todo_list", new_callable=AsyncMock) as mock_gl:
|
|||
|
|
mock_gl.return_value = mock_result
|
|||
|
|
response = await todo_client.get("/todo-items?_force=1")
|
|||
|
|
|
|||
|
|
assert response.status_code == 200
|
|||
|
|
# 验证缓存失效被调用
|
|||
|
|
mock_inv.assert_called_once_with("todo_test_agent")
|
|||
|
|
|
|||
|
|
async def test_api_get_todo_item(self, todo_client):
|
|||
|
|
"""GET /todo-items/{id} 返回详情。"""
|
|||
|
|
mock_detail = {
|
|||
|
|
"id": "approval:202607110001",
|
|||
|
|
"type": "approval",
|
|||
|
|
"title": "资产领用登记",
|
|||
|
|
"priority": "high",
|
|||
|
|
"status": "pending",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
with patch.object(TodoAggregatorService, "get_todo_detail", new_callable=AsyncMock) as mock_gd:
|
|||
|
|
mock_gd.return_value = mock_detail
|
|||
|
|
response = await todo_client.get("/todo-items/approval:202607110001")
|
|||
|
|
|
|||
|
|
assert response.status_code == 200
|
|||
|
|
data = response.json()
|
|||
|
|
assert data["code"] == 0
|
|||
|
|
assert data["data"]["id"] == "approval:202607110001"
|
|||
|
|
assert data["data"]["type"] == "approval"
|
|||
|
|
|
|||
|
|
async def test_api_get_todo_item_not_found(self, todo_client):
|
|||
|
|
"""不存在的 ID 返回错误。"""
|
|||
|
|
with patch.object(TodoAggregatorService, "get_todo_detail", new_callable=AsyncMock) as mock_gd:
|
|||
|
|
mock_gd.return_value = None
|
|||
|
|
response = await todo_client.get("/todo-items/approval:nonexistent")
|
|||
|
|
|
|||
|
|
data = response.json()
|
|||
|
|
assert data["code"] == 1003
|
|||
|
|
|
|||
|
|
async def test_api_update_status_display_only(self, todo_client):
|
|||
|
|
"""PUT status 返回"请在原系统中操作"提示。"""
|
|||
|
|
response = await todo_client.put(
|
|||
|
|
"/todo-items/approval:202607110001/status",
|
|||
|
|
json={"status": "resolved"},
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
assert response.status_code == 200
|
|||
|
|
data = response.json()
|
|||
|
|
assert data["code"] == 0
|
|||
|
|
assert data["data"]["mode"] == "display_only"
|
|||
|
|
assert "企微审批" in data["data"]["message"]
|
|||
|
|
|
|||
|
|
async def test_api_update_status_ticket(self, todo_client):
|
|||
|
|
"""工单类型的状态更新提示 ITSM。"""
|
|||
|
|
response = await todo_client.put(
|
|||
|
|
"/todo-items/ticket:12345/status",
|
|||
|
|
json={"status": "processing"},
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
assert response.status_code == 200
|
|||
|
|
data = response.json()
|
|||
|
|
assert data["data"]["mode"] == "display_only"
|
|||
|
|
assert "ITSM" in data["data"]["message"]
|
|||
|
|
|
|||
|
|
async def test_api_update_status_invalid(self, todo_client):
|
|||
|
|
"""无效状态值返回错误。"""
|
|||
|
|
response = await todo_client.put(
|
|||
|
|
"/todo-items/approval:001/status",
|
|||
|
|
json={"status": "invalid_status"},
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
data = response.json()
|
|||
|
|
assert data["code"] == 1001
|
|||
|
|
|
|||
|
|
|
|||
|
|
# =============================================================================
|
|||
|
|
# 6. Schema 验证测试
|
|||
|
|
# =============================================================================
|
|||
|
|
|
|||
|
|
class TestTodoItemSchema:
|
|||
|
|
"""Schema 验证测试：VALID_TODO_TYPES 移除 device。"""
|
|||
|
|
|
|||
|
|
def test_schema_valid_types(self):
|
|||
|
|
"""VALID_TODO_TYPES 只包含 ticket/approval。"""
|
|||
|
|
assert VALID_TODO_TYPES == {"ticket", "approval"}
|
|||
|
|
assert "device" not in VALID_TODO_TYPES
|
|||
|
|
assert len(VALID_TODO_TYPES) == 2
|
|||
|
|
|
|||
|
|
def test_schema_reject_device_type(self):
|
|||
|
|
"""type=device 被拒绝。"""
|
|||
|
|
with pytest.raises(ValidationError) as exc_info:
|
|||
|
|
TodoItemCreate(type="device", title="测试待办")
|
|||
|
|
assert "device" in str(exc_info.value)
|
|||
|
|
|
|||
|
|
def test_schema_accept_ticket_type(self):
|
|||
|
|
"""type=ticket 被接受。"""
|
|||
|
|
item = TodoItemCreate(type="ticket", title="测试工单")
|
|||
|
|
assert item.type == "ticket"
|
|||
|
|
|
|||
|
|
def test_schema_accept_approval_type(self):
|
|||
|
|
"""type=approval 被接受。"""
|
|||
|
|
item = TodoItemCreate(type="approval", title="测试审批")
|
|||
|
|
assert item.type == "approval"
|
|||
|
|
|
|||
|
|
def test_schema_default_type(self):
|
|||
|
|
"""默认 type 为 ticket。"""
|
|||
|
|
item = TodoItemCreate(title="测试")
|
|||
|
|
assert item.type == "ticket"
|