wecom_it_smart_desk/backend/tests/test_todo_integration.py

# =============================================================================
# 企微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"