wecom_it_smart_desk/backend/tests/test_routing_service.py

# =============================================================================
# IT智能服务台 — 业务路由推荐功能测试
# =============================================================================
# 测试覆盖：
#   1. 关键词预过滤 routing_keyword_prefilter（命中/未命中/空值/精确匹配）
#   2. 降级兜底 _keyword_fallback_category（各业务类别/未命中/空值）
#   3. Dify 调用 detect_routing_intent（mock httpx，解析6字段）
#   4. 联系人查询 get_contact_by_category（按类别查询/未找到/停用排除）
#   5. 名片三段式发送 send_contact_card（mock db/ws，验证三消息+双通道）
#   6. 路由事件记录 record_routing_event（正常/异常不中断主流程）
#   7. 审批向后兼容（ApprovalDetectIntentResponse 默认值 + Dify 解析扩展字段）
#   8. 置信度阈值逻辑（0.85触发/0.5不触发/0.7边界）
# =============================================================================

import json
from datetime import datetime
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
import pytest_asyncio
from sqlalchemy import select

from app.config import settings
from app.models.business_contact import BusinessContact
from app.models.conversation import Conversation
from app.models.message import Message
from app.models.routing_event import RoutingEvent
from app.services.routing_service import (
    ROUTING_KEYWORD_TO_CATEGORY,
    ROUTING_PREFILTER_KEYWORDS,
    _keyword_fallback_category,
    detect_routing_intent,
    get_contact_by_category,
    record_routing_event,
    routing_keyword_prefilter,
    send_contact_card,
)


# =============================================================================
# 辅助 fixtures
# =============================================================================

@pytest_asyncio.fixture
async def seeded_contacts(db_session):
    """插入测试用业务联系人数据。"""
    contacts = [
        BusinessContact(
            name="王芳", gender="female", department="行政部",
            position="设备管理岗", responsibility="打印机/复印机/扫描仪",
            extension="8002", service_area="滨江园区 3-5楼",
            wecom_userid="WangFang", avatar_url="",
            business_category="行政", is_active=True,
        ),
        BusinessContact(
            name="陈伟", gender="male", department="行政部",
            position="行政事务岗", responsibility="办公用品/名片印刷/保洁服务",
            extension="8003", service_area="滨江园区 1-2楼",
            wecom_userid="ChenWei", avatar_url="",
            business_category="行政", is_active=True,
        ),
        BusinessContact(
            name="李娜", gender="female", department="人力资源部",
            position="员工服务岗", responsibility="工牌补办/考勤异常/入职手续",
            extension="8005", service_area="滨江园区 A栋3楼",
            wecom_userid="LiNa", avatar_url="",
            business_category="人力资源", is_active=True,
        ),
        BusinessContact(
            name="张磊", gender="male", department="人力资源部",
            position="薪酬福利岗", responsibility="社保/公积金/离职手续",
            extension="8006", service_area="滨江园区 A栋3楼",
            wecom_userid="ZhangLei", avatar_url="",
            business_category="人力资源", is_active=True,
        ),
        BusinessContact(
            name="刘洋", gender="male", department="财务部",
            position="费用报销岗", responsibility="报销/发票/借款/工资条",
            extension="8010", service_area="滨江园区 B栋4楼",
            wecom_userid="LiuYang", avatar_url="",
            business_category="财务", is_active=True,
        ),
        BusinessContact(
            name="赵敏", gender="female", department="法务部",
            position="法务顾问岗", responsibility="合同/法律咨询/知识产权",
            extension="8015", service_area="滨江园区 C栋5楼",
            wecom_userid="ZhaoMin", avatar_url="",
            business_category="法务", is_active=True,
        ),
        BusinessContact(
            name="孙强", gender="male", department="行政部",
            position="物业管理岗", responsibility="空调/电梯/门禁/停车",
            extension="8008", service_area="滨江园区 全园区",
            wecom_userid="SunQiang", avatar_url="",
            business_category="行政-物业", is_active=True,
        ),
        # 停用的联系人（不应被查询到）
        BusinessContact(
            name="停用联系人", gender="male", department="行政部",
            position="已停用岗", responsibility="测试停用",
            extension="9999", service_area="无",
            wecom_userid="Disabled", avatar_url="",
            business_category="行政", is_active=False,
        ),
    ]
    db_session.add_all(contacts)
    await db_session.flush()
    return contacts


@pytest_asyncio.fixture
async def seeded_conversation(db_session):
    """插入测试用会话数据。"""
    conv = Conversation(
        employee_id="test_emp_001",
        employee_name="测试员工",
        department="技术部",
        position="工程师",
        level="",
        status="ai_handling",
        is_vip=False,
        is_pinned=False,
        is_todo=False,
        urgency_score=1,
        tags={},
        last_message_at=datetime.now(),
        last_message_summary="测试消息",
        ai_substantive_reply_count=0,
    )
    db_session.add(conv)
    await db_session.flush()
    return conv


# =============================================================================
# 单元测试：routing_keyword_prefilter
# =============================================================================

class TestRoutingKeywordPrefilter:
    """测试路由关键词预过滤函数。"""

    def test_hit_printer(self):
        """包含 '打印机' 关键词时返回 True。"""
        assert routing_keyword_prefilter("打印机坏了") is True

    def test_hit_air_conditioner(self):
        """包含 '空调' 关键词时返回 True。"""
        assert routing_keyword_prefilter("空调不制冷了") is True

    def test_hit_badge(self):
        """包含 '工牌' 关键词时返回 True。"""
        assert routing_keyword_prefilter("工牌丢了补办找谁") is True

    def test_hit_reimbursement(self):
        """包含 '报销' 关键词时返回 True。"""
        assert routing_keyword_prefilter("报销流程是什么") is True

    def test_miss_normal_message(self):
        """普通对话消息不包含路由关键词时返回 False。"""
        assert routing_keyword_prefilter("你好") is False
        assert routing_keyword_prefilter("谢谢，问题解决了") is False

    def test_miss_empty_string(self):
        """空字符串返回 False。"""
        assert routing_keyword_prefilter("") is False

    def test_miss_none(self):
        """None 返回 False。"""
        assert routing_keyword_prefilter(None) is False

    def test_miss_it_message(self):
        """IT 相关但非路由关键词的消息返回 False。"""
        assert routing_keyword_prefilter("VPN连不上了") is False
        assert routing_keyword_prefilter("电脑蓝屏了") is False
        assert routing_keyword_prefilter("我要申请一台笔记本电脑") is False

    def test_exact_keyword_matching_print_vs_printer(self):
        """验证精确匹配：'打印' 不在关键词列表中，但 '打印机' 是。

        '我要打印一份文件' 包含 '打印' 但不包含 '打印机'，应返回 False。
        '打印机卡纸了' 包含 '打印机'，应返回 True。
        """
        # "打印" 不是关键词，只有 "打印机" 是
        assert routing_keyword_prefilter("我要打印一份文件") is False
        assert routing_keyword_prefilter("打印机卡纸了") is True

    def test_all_prefilter_keywords_work(self):
        """验证 ROUTING_PREFILTER_KEYWORDS 中的每个关键词都能被命中。"""
        for kw in ROUTING_PREFILTER_KEYWORDS:
            text = f"测试文本包含{kw}关键词"
            assert routing_keyword_prefilter(text) is True, f"关键词 '{kw}' 未被命中"


# =============================================================================
# 单元测试：_keyword_fallback_category
# =============================================================================

class TestKeywordFallbackCategory:
    """测试关键词降级兜底函数。"""

    def test_fallback_printer_to_admin(self):
        """'打印机坏了' → '行政'。"""
        assert _keyword_fallback_category("打印机坏了") == "行政"

    def test_fallback_badge_to_hr(self):
        """'工牌丢了' → '人力资源'。"""
        assert _keyword_fallback_category("工牌丢了") == "人力资源"

    def test_fallback_reimbursement_to_finance(self):
        """'报销流程是什么' → '财务'。"""
        assert _keyword_fallback_category("报销流程是什么") == "财务"

    def test_fallback_aircon_to_property(self):
        """'空调不制冷' → '行政-物业'。"""
        assert _keyword_fallback_category("空调不制冷") == "行政-物业"

    def test_fallback_elevator_to_property(self):
        """'电梯故障' → '行政-物业'。"""
        assert _keyword_fallback_category("电梯故障") == "行政-物业"

    def test_fallback_contract_to_legal(self):
        """'合同问题' → '法务'。"""
        assert _keyword_fallback_category("合同问题") == "法务"

    def test_fallback_no_match_returns_none(self):
        """无路由关键词时返回 None。"""
        assert _keyword_fallback_category("电脑蓝屏了") is None

    def test_fallback_empty_string_returns_none(self):
        """空字符串返回 None。"""
        assert _keyword_fallback_category("") is None

    def test_fallback_none_returns_none(self):
        """None 返回 None。"""
        assert _keyword_fallback_category(None) is None

    def test_fallback_normal_message_returns_none(self):
        """普通消息返回 None。"""
        assert _keyword_fallback_category("你好") is None

    def test_fallback_first_match_wins(self):
        """多个关键词命中时，返回第一个匹配的类别（字典有序遍历）。"""
        # "打印机" 在字典中排第一，映射到 "行政"
        result = _keyword_fallback_category("打印机和空调都坏了")
        assert result is not None
        assert result in ROUTING_KEYWORD_TO_CATEGORY.values()


# =============================================================================
# 单元测试：detect_routing_intent（mock httpx）
# =============================================================================

class TestDetectRoutingIntent:
    """测试 Dify 统一意图识别调用（mock httpx）。"""

    @pytest.mark.asyncio
    async def test_parse_all_six_fields(self, monkeypatch):
        """Dify 返回完整6字段时，正确解析所有字段。"""
        monkeypatch.setattr(settings, "approval_dify_base_url", "http://test-dify")
        monkeypatch.setattr(settings, "approval_dify_api_key", "test-api-key")
        monkeypatch.setattr(settings, "approval_dify_timeout", 15)

        dify_response = {
            "is_approval_request": False,
            "confidence": 0.1,
            "approval_type": None,
            "intent_type": "non_it_routing",
            "business_category": "行政",
            "routing_confidence": 0.85,
        }

        mock_response = MagicMock()
        mock_response.json.return_value = {"answer": json.dumps(dify_response)}
        mock_response.raise_for_status = MagicMock()

        mock_client = AsyncMock()
        mock_client.post.return_value = mock_response
        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
        mock_client.__aexit__ = AsyncMock(return_value=None)

        with patch("app.services.routing_service.httpx.AsyncClient", return_value=mock_client):
            result = await detect_routing_intent("打印机坏了", "test_emp_001")

        assert result["intent_type"] == "non_it_routing"
        assert result["business_category"] == "行政"
        assert result["routing_confidence"] == 0.85
        assert result["is_approval_request"] is False
        assert result["confidence"] == 0.1
        assert result["approval_type"] is None

    @pytest.mark.asyncio
    async def test_parse_approval_intent(self, monkeypatch):
        """Dify 返回审批意图时，正确解析。"""
        monkeypatch.setattr(settings, "approval_dify_base_url", "http://test-dify")
        monkeypatch.setattr(settings, "approval_dify_api_key", "test-api-key")

        dify_response = {
            "is_approval_request": True,
            "confidence": 0.95,
            "approval_type": "设备申请",
            "intent_type": "approval",
            "business_category": None,
            "routing_confidence": 0.0,
        }

        mock_response = MagicMock()
        mock_response.json.return_value = {"answer": json.dumps(dify_response)}
        mock_response.raise_for_status = MagicMock()

        mock_client = AsyncMock()
        mock_client.post.return_value = mock_response
        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
        mock_client.__aexit__ = AsyncMock(return_value=None)

        with patch("app.services.routing_service.httpx.AsyncClient", return_value=mock_client):
            result = await detect_routing_intent("我要申请笔记本电脑", "test_emp_001")

        assert result["intent_type"] == "approval"
        assert result["is_approval_request"] is True
        assert result["confidence"] == 0.95
        assert result["approval_type"] == "设备申请"
        assert result["business_category"] is None
        assert result["routing_confidence"] == 0.0

    @pytest.mark.asyncio
    async def test_parse_chitchat_intent(self, monkeypatch):
        """Dify 返回闲聊意图时，正确解析。"""
        monkeypatch.setattr(settings, "approval_dify_base_url", "http://test-dify")
        monkeypatch.setattr(settings, "approval_dify_api_key", "test-api-key")

        dify_response = {
            "is_approval_request": False,
            "confidence": 0.05,
            "approval_type": None,
            "intent_type": "chitchat",
            "business_category": None,
            "routing_confidence": 0.05,
        }

        mock_response = MagicMock()
        mock_response.json.return_value = {"answer": json.dumps(dify_response)}
        mock_response.raise_for_status = MagicMock()

        mock_client = AsyncMock()
        mock_client.post.return_value = mock_response
        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
        mock_client.__aexit__ = AsyncMock(return_value=None)

        with patch("app.services.routing_service.httpx.AsyncClient", return_value=mock_client):
            result = await detect_routing_intent("你好", "test_emp_001")

        assert result["intent_type"] == "chitchat"
        assert result["routing_confidence"] == 0.05

    @pytest.mark.asyncio
    async def test_not_configured_raises_value_error(self, monkeypatch):
        """Dify 未配置时抛出 ValueError。"""
        monkeypatch.setattr(settings, "approval_dify_base_url", "")
        monkeypatch.setattr(settings, "approval_dify_api_key", "")

        with pytest.raises(ValueError, match="Dify"):
            await detect_routing_intent("打印机坏了")

    @pytest.mark.asyncio
    async def test_http_error_raises_exception(self, monkeypatch):
        """Dify HTTP 调用失败时抛出异常。"""
        monkeypatch.setattr(settings, "approval_dify_base_url", "http://test-dify")
        monkeypatch.setattr(settings, "approval_dify_api_key", "test-api-key")

        mock_response = MagicMock()
        mock_response.raise_for_status.side_effect = Exception("HTTP 500")

        mock_client = AsyncMock()
        mock_client.post.return_value = mock_response
        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
        mock_client.__aexit__ = AsyncMock(return_value=None)

        with patch("app.services.routing_service.httpx.AsyncClient", return_value=mock_client):
            with pytest.raises(Exception):
                await detect_routing_intent("打印机坏了")

    @pytest.mark.asyncio
    async def test_invalid_json_raises_exception(self, monkeypatch):
        """Dify 返回非 JSON 时抛出异常。"""
        monkeypatch.setattr(settings, "approval_dify_base_url", "http://test-dify")
        monkeypatch.setattr(settings, "approval_dify_api_key", "test-api-key")

        mock_response = MagicMock()
        mock_response.json.return_value = {"answer": "not a json string"}
        mock_response.raise_for_status = MagicMock()

        mock_client = AsyncMock()
        mock_client.post.return_value = mock_response
        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
        mock_client.__aexit__ = AsyncMock(return_value=None)

        with patch("app.services.routing_service.httpx.AsyncClient", return_value=mock_client):
            with pytest.raises(json.JSONDecodeError):
                await detect_routing_intent("打印机坏了")

    @pytest.mark.asyncio
    async def test_missing_fields_use_defaults(self, monkeypatch):
        """Dify 返回缺少新字段时，使用默认值（向后兼容）。"""
        monkeypatch.setattr(settings, "approval_dify_base_url", "http://test-dify")
        monkeypatch.setattr(settings, "approval_dify_api_key", "test-api-key")

        # 旧版 Dify Prompt 只返回原3字段
        dify_response = {
            "is_approval_request": True,
            "confidence": 0.9,
            "approval_type": "设备申请",
        }

        mock_response = MagicMock()
        mock_response.json.return_value = {"answer": json.dumps(dify_response)}
        mock_response.raise_for_status = MagicMock()

        mock_client = AsyncMock()
        mock_client.post.return_value = mock_response
        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
        mock_client.__aexit__ = AsyncMock(return_value=None)

        with patch("app.services.routing_service.httpx.AsyncClient", return_value=mock_client):
            result = await detect_routing_intent("申请设备", "test_emp_001")

        # 新字段使用默认值
        assert result["intent_type"] == "chitchat"  # 默认值
        assert result["business_category"] is None  # 默认值
        assert result["routing_confidence"] == 0.0  # 默认值
        # 原字段正常解析
        assert result["is_approval_request"] is True
        assert result["confidence"] == 0.9

    @pytest.mark.asyncio
    async def test_employee_id_passed_to_dify(self, monkeypatch):
        """验证 employee_id 被传递给 Dify 的 user 字段。"""
        monkeypatch.setattr(settings, "approval_dify_base_url", "http://test-dify")
        monkeypatch.setattr(settings, "approval_dify_api_key", "test-api-key")

        dify_response = {
            "is_approval_request": False,
            "confidence": 0.1,
            "approval_type": None,
            "intent_type": "non_it_routing",
            "business_category": "行政",
            "routing_confidence": 0.85,
        }

        mock_response = MagicMock()
        mock_response.json.return_value = {"answer": json.dumps(dify_response)}
        mock_response.raise_for_status = MagicMock()

        mock_client = AsyncMock()
        mock_client.post.return_value = mock_response
        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
        mock_client.__aexit__ = AsyncMock(return_value=None)

        with patch("app.services.routing_service.httpx.AsyncClient", return_value=mock_client):
            await detect_routing_intent("打印机坏了", "test_emp_001")

        # 验证 post 调用的 body 中 user 字段
        call_args = mock_client.post.call_args
        body = call_args.kwargs["json"]
        assert body["user"] == "test_emp_001"

    @pytest.mark.asyncio
    async def test_empty_employee_id_uses_default(self, monkeypatch):
        """未提供 employee_id 时，Dify user 字段使用默认值。"""
        monkeypatch.setattr(settings, "approval_dify_base_url", "http://test-dify")
        monkeypatch.setattr(settings, "approval_dify_api_key", "test-api-key")

        dify_response = {
            "is_approval_request": False,
            "confidence": 0.1,
            "approval_type": None,
            "intent_type": "chitchat",
            "business_category": None,
            "routing_confidence": 0.0,
        }

        mock_response = MagicMock()
        mock_response.json.return_value = {"answer": json.dumps(dify_response)}
        mock_response.raise_for_status = MagicMock()

        mock_client = AsyncMock()
        mock_client.post.return_value = mock_response
        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
        mock_client.__aexit__ = AsyncMock(return_value=None)

        with patch("app.services.routing_service.httpx.AsyncClient", return_value=mock_client):
            await detect_routing_intent("你好")

        call_args = mock_client.post.call_args
        body = call_args.kwargs["json"]
        assert body["user"] == "routing_detection"


# =============================================================================
# 单元测试：get_contact_by_category
# =============================================================================

class TestGetContactByCategory:
    """测试按业务类别查询联系人。"""

    @pytest.mark.asyncio
    async def test_query_admin_returns_first_contact(self, db_session, seeded_contacts):
        """business_category='行政' → 返回第一个有效联系人（王芳，id最小）。"""
        result = await get_contact_by_category(db_session, "行政")
        assert result is not None
        assert result.name == "王芳"
        assert result.business_category == "行政"
        assert result.is_active is True

    @pytest.mark.asyncio
    async def test_query_hr_returns_first_contact(self, db_session, seeded_contacts):
        """business_category='人力资源' → 返回第一个有效联系人（李娜）。"""
        result = await get_contact_by_category(db_session, "人力资源")
        assert result is not None
        assert result.name == "李娜"
        assert result.business_category == "人力资源"

    @pytest.mark.asyncio
    async def test_query_finance_returns_contact(self, db_session, seeded_contacts):
        """business_category='财务' → 返回联系人（刘洋）。"""
        result = await get_contact_by_category(db_session, "财务")
        assert result is not None
        assert result.name == "刘洋"

    @pytest.mark.asyncio
    async def test_query_legal_returns_contact(self, db_session, seeded_contacts):
        """business_category='法务' → 返回联系人（赵敏）。"""
        result = await get_contact_by_category(db_session, "法务")
        assert result is not None
        assert result.name == "赵敏"

    @pytest.mark.asyncio
    async def test_query_property_returns_contact(self, db_session, seeded_contacts):
        """business_category='行政-物业' → 返回联系人（孙强）。"""
        result = await get_contact_by_category(db_session, "行政-物业")
        assert result is not None
        assert result.name == "孙强"

    @pytest.mark.asyncio
    async def test_query_nonexistent_category_returns_none(self, db_session, seeded_contacts):
        """business_category='不存在' → 返回 None。"""
        result = await get_contact_by_category(db_session, "不存在的类别")
        assert result is None

    @pytest.mark.asyncio
    async def test_inactive_contacts_excluded(self, db_session, seeded_contacts):
        """is_active=False 的联系人不返回。"""
        # 行政类别有3个联系人（王芳、陈伟 active，停用联系人 inactive）
        # 查询应返回 active 的第一个
        result = await get_contact_by_category(db_session, "行政")
        assert result is not None
        assert result.is_active is True
        assert result.name != "停用联系人"

    @pytest.mark.asyncio
    async def test_returns_first_by_id_order(self, db_session, seeded_contacts):
        """返回 id 最小的有效联系人（order_by id）。"""
        result = await get_contact_by_category(db_session, "行政")
        assert result is not None
        # 王芳 id 最小（先插入的）
        assert result.name == "王芳"


# =============================================================================
# 单元测试：send_contact_card（mock ws_manager）
# =============================================================================

class TestSendContactCard:
    """测试名片三段式发送。"""

    @pytest.mark.asyncio
    async def test_sends_three_messages(self, db_session, seeded_contacts, seeded_conversation):
        """发送名片后，数据库中新增3条消息（路由文本 + contact_card + 系统提示）。"""
        contact = seeded_contacts[0]  # 王芳
        conv = seeded_conversation

        with patch("app.services.routing_service.ws_manager") as mock_ws:
            mock_ws.broadcast_to_employees = AsyncMock()
            mock_ws.broadcast = AsyncMock()

            await send_contact_card(
                db=db_session,
                conversation=conv,
                employee_id="test_emp_001",
                contact=contact,
                reason="打印机问题属于行政设备范畴",
                business_category="行政",
                routing_confidence=0.85,
            )

        # 查询数据库中的消息
        result = await db_session.execute(
            select(Message).where(Message.conversation_id == conv.id)
        )
        messages = result.scalars().all()

        assert len(messages) == 3
        # 第1条：路由说明文本
        assert messages[0].msg_type == "text"
        assert messages[0].sender_type == "ai"
        assert "打印机" in messages[0].content
        # 第2条：contact_card 名片
        assert messages[1].msg_type == "contact_card"
        assert messages[1].sender_type == "ai"
        assert messages[1].extra_data is not None
        assert messages[1].extra_data["contact"]["name"] == "王芳"
        assert messages[1].extra_data["routing_reason"] == "打印机问题属于行政设备范畴"
        assert messages[1].extra_data["business_category"] == "行政"
        assert messages[1].extra_data["routing_confidence"] == 0.85
        # 第3条：系统提示
        assert messages[2].msg_type == "system"
        assert messages[2].sender_type == "system"

    @pytest.mark.asyncio
    async def test_ws_dual_channel_push(self, db_session, seeded_contacts, seeded_conversation):
        """每条消息都通过 WS 双通道推送（broadcast_to_employees + broadcast）。"""
        contact = seeded_contacts[0]
        conv = seeded_conversation

        with patch("app.services.routing_service.ws_manager") as mock_ws:
            mock_ws.broadcast_to_employees = AsyncMock()
            mock_ws.broadcast = AsyncMock()

            await send_contact_card(
                db=db_session,
                conversation=conv,
                employee_id="test_emp_001",
                contact=contact,
                reason="路由说明",
                business_category="行政",
                routing_confidence=0.85,
            )

        # 每条消息推送2次（H5 + 坐席），3条消息共6次
        assert mock_ws.broadcast_to_employees.call_count == 3
        assert mock_ws.broadcast.call_count == 3

    @pytest.mark.asyncio
    async def test_ws_push_contains_contact_card_type(self, db_session, seeded_contacts, seeded_conversation):
        """WS 推送中 contact_card 消息的 msg_type 为 'contact_card'。"""
        contact = seeded_contacts[0]
        conv = seeded_conversation

        with patch("app.services.routing_service.ws_manager") as mock_ws:
            mock_ws.broadcast_to_employees = AsyncMock()
            mock_ws.broadcast = AsyncMock()

            await send_contact_card(
                db=db_session,
                conversation=conv,
                employee_id="test_emp_001",
                contact=contact,
                reason="路由说明",
                business_category="行政",
                routing_confidence=0.85,
            )

        # 检查第2次推送（contact_card）的 H5 推送数据
        second_call = mock_ws.broadcast_to_employees.call_args_list[1]
        push_data = second_call.args[1]
        assert push_data["type"] == "ai_reply"
        assert push_data["data"]["msg_type"] == "contact_card"
        assert push_data["data"]["extra_data"]["contact"]["name"] == "王芳"

    @pytest.mark.asyncio
    async def test_updates_conversation_reply_count(self, db_session, seeded_contacts, seeded_conversation):
        """发送名片后会话的 ai_substantive_reply_count +1。"""
        contact = seeded_contacts[0]
        conv = seeded_conversation
        original_count = conv.ai_substantive_reply_count

        with patch("app.services.routing_service.ws_manager") as mock_ws:
            mock_ws.broadcast_to_employees = AsyncMock()
            mock_ws.broadcast = AsyncMock()

            await send_contact_card(
                db=db_session,
                conversation=conv,
                employee_id="test_emp_001",
                contact=contact,
                reason="路由说明",
                business_category="行政",
                routing_confidence=0.85,
            )

        assert conv.ai_substantive_reply_count == original_count + 1

    @pytest.mark.asyncio
    async def test_ws_broadcast_failure_does_not_raise(self, db_session, seeded_contacts, seeded_conversation):
        """坐席端 WS 广播失败时不影响主流程（仅 warning）。"""
        contact = seeded_contacts[0]
        conv = seeded_conversation

        with patch("app.services.routing_service.ws_manager") as mock_ws:
            mock_ws.broadcast_to_employees = AsyncMock()
            mock_ws.broadcast = AsyncMock(side_effect=Exception("WS 不可达"))

            # 不应抛出异常
            await send_contact_card(
                db=db_session,
                conversation=conv,
                employee_id="test_emp_001",
                contact=contact,
                reason="路由说明",
                business_category="行政",
                routing_confidence=0.85,
            )

        # 消息仍然落库
        result = await db_session.execute(
            select(Message).where(Message.conversation_id == conv.id)
        )
        messages = result.scalars().all()
        assert len(messages) == 3

    @pytest.mark.asyncio
    async def test_extra_data_structure(self, db_session, seeded_contacts, seeded_conversation):
        """contact_card 消息的 extra_data 结构符合架构设计规范。"""
        contact = seeded_contacts[0]
        conv = seeded_conversation

        with patch("app.services.routing_service.ws_manager") as mock_ws:
            mock_ws.broadcast_to_employees = AsyncMock()
            mock_ws.broadcast = AsyncMock()

            await send_contact_card(
                db=db_session,
                conversation=conv,
                employee_id="test_emp_001",
                contact=contact,
                reason="路由原因文本",
                business_category="行政",
                routing_confidence=0.85,
            )

        result = await db_session.execute(
            select(Message).where(
                Message.conversation_id == conv.id,
                Message.msg_type == "contact_card",
            )
        )
        card_msg = result.scalar_one()

        extra = card_msg.extra_data
        # 架构设计 3.3 节 extra_data 结构规范
        assert "contact" in extra
        assert "routing_reason" in extra
        assert "business_category" in extra
        assert "routing_confidence" in extra

        # contact 子结构
        contact_data = extra["contact"]
        assert contact_data["name"] == "王芳"
        assert contact_data["department"] == "行政部"
        assert contact_data["wecom_userid"] == "WangFang"
        assert contact_data["business_category"] == "行政"


# =============================================================================
# 单元测试：record_routing_event
# =============================================================================

class TestRecordRoutingEvent:
    """测试路由事件记录。"""

    @pytest.mark.asyncio
    async def test_record_event_success(self, db_session, seeded_contacts, seeded_conversation):
        """正常记录路由事件到 routing_events 表。"""
        contact = seeded_contacts[0]
        conv = seeded_conversation

        await record_routing_event(
            db=db_session,
            conversation_id=conv.id,
            employee_id="test_emp_001",
            message_content="打印机坏了",
            business_category="行政",
            routing_confidence=0.85,
            contact=contact,
        )

        result = await db_session.execute(select(RoutingEvent))
        events = result.scalars().all()

        assert len(events) == 1
        event = events[0]
        assert event.conversation_id == conv.id
        assert event.employee_id == "test_emp_001"
        assert event.message_content == "打印机坏了"
        assert event.business_category == "行政"
        assert event.routing_confidence == 0.85
        assert event.contact_id == contact.id
        assert event.contact_name == "王芳"
        assert event.is_clicked is False

    @pytest.mark.asyncio
    async def test_record_event_with_none_contact(self, db_session, seeded_conversation):
        """联系人为 None 时，contact_id=None, contact_name=''。"""
        conv = seeded_conversation

        await record_routing_event(
            db=db_session,
            conversation_id=conv.id,
            employee_id="test_emp_001",
            message_content="未知问题",
            business_category="未知类别",
            routing_confidence=0.5,
            contact=None,
        )

        result = await db_session.execute(select(RoutingEvent))
        event = result.scalar_one()
        assert event.contact_id is None
        assert event.contact_name == ""

    @pytest.mark.asyncio
    async def test_record_event_truncates_long_message(self, db_session, seeded_conversation):
        """超长消息内容截断至500字。"""
        conv = seeded_conversation
        long_message = "A" * 600

        await record_routing_event(
            db=db_session,
            conversation_id=conv.id,
            employee_id="test_emp_001",
            message_content=long_message,
            business_category="行政",
            routing_confidence=0.85,
            contact=None,
        )

        result = await db_session.execute(select(RoutingEvent))
        event = result.scalar_one()
        assert len(event.message_content) == 500

    @pytest.mark.asyncio
    async def test_record_event_failure_does_not_raise(self, db_session, seeded_conversation):
        """路由事件记录失败时不抛出异常（不影响主流程）。"""
        conv = seeded_conversation

        # 使用一个会触发异常的 mock db
        mock_db = AsyncMock()
        mock_db.add.side_effect = Exception("DB 错误")

        # 不应抛出异常
        await record_routing_event(
            db=mock_db,
            conversation_id=conv.id,
            employee_id="test_emp_001",
            message_content="打印机坏了",
            business_category="行政",
            routing_confidence=0.85,
            contact=None,
        )


# =============================================================================
# 审批向后兼容性测试
# =============================================================================

class TestApprovalBackwardCompat:
    """测试审批意图检测的向后兼容性。"""

    def test_response_has_new_fields_with_defaults(self):
        """ApprovalDetectIntentResponse 新增字段有默认值。"""
        from app.api.approval import ApprovalDetectIntentResponse

        # 只提供原4个必填字段，新字段应使用默认值
        resp = ApprovalDetectIntentResponse(
            is_approval_request=False,
            confidence=0.0,
            source="keyword_prefilter",
        )
        assert resp.intent_type == "chitchat"
        assert resp.business_category is None
        assert resp.routing_confidence == 0.0

    def test_response_accepts_new_fields(self):
        """ApprovalDetectIntentResponse 接受新字段赋值。"""
        from app.api.approval import ApprovalDetectIntentResponse

        resp = ApprovalDetectIntentResponse(
            is_approval_request=False,
            confidence=0.1,
            source="dify",
            intent_type="non_it_routing",
            business_category="行政",
            routing_confidence=0.85,
        )
        assert resp.intent_type == "non_it_routing"
        assert resp.business_category == "行政"
        assert resp.routing_confidence == 0.85

    @pytest.mark.asyncio
    async def test_dify_returns_extended_fields(self, monkeypatch):
        """_call_dify_approval_intent 正确解析 Dify 返回的扩展字段。"""
        from app.api.approval import _call_dify_approval_intent

        monkeypatch.setattr(settings, "approval_dify_base_url", "http://test-dify")
        monkeypatch.setattr(settings, "approval_dify_api_key", "test-api-key")
        monkeypatch.setattr(settings, "approval_dify_timeout", 15)

        dify_response = {
            "is_approval_request": False,
            "confidence": 0.1,
            "approval_type": None,
            "intent_type": "non_it_routing",
            "business_category": "行政",
            "routing_confidence": 0.85,
        }

        mock_response = MagicMock()
        mock_response.json.return_value = {"answer": json.dumps(dify_response)}
        mock_response.raise_for_status = MagicMock()

        mock_client = AsyncMock()
        mock_client.post.return_value = mock_response
        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
        mock_client.__aexit__ = AsyncMock(return_value=None)

        with patch("app.api.approval.httpx.AsyncClient", return_value=mock_client):
            result = await _call_dify_approval_intent("打印机坏了", "test_emp_001")

        assert result["intent_type"] == "non_it_routing"
        assert result["business_category"] == "行政"
        assert result["routing_confidence"] == 0.85
        # 原字段也正确解析
        assert result["is_approval_request"] is False
        assert result["confidence"] == 0.1
        assert result["approval_type"] is None

    @pytest.mark.asyncio
    async def test_dify_returns_old_format_compatible(self, monkeypatch):
        """Dify 返回旧格式（无新字段）时，_call_dify_approval_intent 使用默认值。"""
        from app.api.approval import _call_dify_approval_intent

        monkeypatch.setattr(settings, "approval_dify_base_url", "http://test-dify")
        monkeypatch.setattr(settings, "approval_dify_api_key", "test-api-key")

        # 旧版 Dify Prompt 只返回原3字段
        dify_response = {
            "is_approval_request": True,
            "confidence": 0.95,
            "approval_type": "设备申请",
        }

        mock_response = MagicMock()
        mock_response.json.return_value = {"answer": json.dumps(dify_response)}
        mock_response.raise_for_status = MagicMock()

        mock_client = AsyncMock()
        mock_client.post.return_value = mock_response
        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
        mock_client.__aexit__ = AsyncMock(return_value=None)

        with patch("app.api.approval.httpx.AsyncClient", return_value=mock_client):
            result = await _call_dify_approval_intent("我要申请设备", "test_emp_001")

        # 新字段使用默认值
        assert result["intent_type"] == "chitchat"
        assert result["business_category"] is None
        assert result["routing_confidence"] == 0.0
        # 原字段正常解析
        assert result["is_approval_request"] is True
        assert result["confidence"] == 0.95
        assert result["approval_type"] == "设备申请"

    @pytest.mark.asyncio
    async def test_detect_intent_returns_extended_fields(self, client, monkeypatch):
        """审批意图检测端点返回扩展字段。"""
        monkeypatch.setattr(settings, "approval_dify_base_url", "http://test-dify")
        monkeypatch.setattr(settings, "approval_dify_api_key", "test-api-key")
        monkeypatch.setattr(settings, "approval_confidence_threshold", 0.7)

        mock_dify = AsyncMock(return_value={
            "is_approval_request": False,
            "confidence": 0.1,
            "approval_type": None,
            "intent_type": "non_it_routing",
            "business_category": "行政",
            "routing_confidence": 0.85,
        })

        # 注意：审批端点使用 APPROVAL_PREFILTER_KEYWORDS 做预过滤，
        # "打印机" 不在审批关键词中，但 "电脑" 是。使用包含审批关键词的文本
        # 才能通过预过滤并触发 Dify 调用。
        with patch("app.api.approval._call_dify_approval_intent", mock_dify):
            response = await client.post(
                "/approval/detect-intent",
                json={"text": "电脑连不上打印机了"},
            )

        data = response.json()
        inner = data["data"]
        # 原字段
        assert inner["is_approval_request"] is False
        assert inner["source"] == "dify"
        # 新字段
        assert inner["intent_type"] == "non_it_routing"
        assert inner["business_category"] == "行政"
        assert inner["routing_confidence"] == 0.85

    @pytest.mark.asyncio
    async def test_detect_intent_chitchat_keyword_miss(self, client):
        """'你好' 不含审批关键词 → intent_type=chitchat, source=keyword_prefilter。"""
        response = await client.post(
            "/approval/detect-intent",
            json={"text": "你好"},
        )

        data = response.json()
        inner = data["data"]
        assert inner["is_approval_request"] is False
        assert inner["source"] == "keyword_prefilter"
        assert inner["intent_type"] == "chitchat"
        assert inner["routing_confidence"] == 0.0


# =============================================================================
# 置信度阈值逻辑测试
# =============================================================================

class TestConfidenceThreshold:
    """测试路由置信度阈值逻辑。"""

    def test_threshold_default_is_0_7(self):
        """routing_confidence_threshold 配置默认值为 0.7。"""
        # config.py 中定义的默认值
        assert settings.routing_confidence_threshold == 0.7

    def test_high_confidence_triggers_routing(self):
        """routing_confidence=0.85 ≥ 0.7 → 应触发名片推荐。"""
        confidence = 0.85
        threshold = settings.routing_confidence_threshold
        assert confidence >= threshold  # 触发条件满足

    def test_low_confidence_no_routing(self):
        """routing_confidence=0.5 < 0.7 → 不触发，走正常AI流程。"""
        confidence = 0.5
        threshold = settings.routing_confidence_threshold
        assert confidence < threshold  # 不触发

    def test_boundary_confidence_triggers(self):
        """routing_confidence=0.7 = 0.7 → 触发（边界值，>= 判断）。"""
        confidence = 0.7
        threshold = settings.routing_confidence_threshold
        assert confidence >= threshold  # 边界值触发

    def test_zero_confidence_no_routing(self):
        """routing_confidence=0.0 < 0.7 → 不触发。"""
        confidence = 0.0
        threshold = settings.routing_confidence_threshold
        assert confidence < threshold

    def test_max_confidence_triggers(self):
        """routing_confidence=1.0 ≥ 0.7 → 触发。"""
        confidence = 1.0
        threshold = settings.routing_confidence_threshold
        assert confidence >= threshold


# =============================================================================
# PRD 5.5 回归测试要点验证
# =============================================================================

class TestPRDRegression:
    """PRD 第5.5节回归测试要点验证（mock Dify 响应）。"""

    @pytest.mark.asyncio
    async def test_approval_regression(self, client, monkeypatch):
        """'我要申请一台笔记本电脑' → intent_type=approval, is_approval_request=true。"""
        monkeypatch.setattr(settings, "approval_dify_base_url", "http://test-dify")
        monkeypatch.setattr(settings, "approval_dify_api_key", "test-api-key")
        monkeypatch.setattr(settings, "approval_confidence_threshold", 0.7)

        mock_dify = AsyncMock(return_value={
            "is_approval_request": True,
            "confidence": 0.95,
            "approval_type": "设备申请",
            "intent_type": "approval",
            "business_category": None,
            "routing_confidence": 0.0,
        })

        with patch("app.api.approval._call_dify_approval_intent", mock_dify):
            response = await client.post(
                "/approval/detect-intent",
                json={"text": "我要申请一台笔记本电脑"},
            )

        data = response.json()
        inner = data["data"]
        assert inner["intent_type"] == "approval"
        assert inner["is_approval_request"] is True
        assert inner["approval_type"] == "设备申请"

    @pytest.mark.asyncio
    async def test_it_consult_regression(self, client, monkeypatch):
        """'VPN连不上了' → intent_type=it_consult, is_approval_request=false。"""
        monkeypatch.setattr(settings, "approval_dify_base_url", "http://test-dify")
        monkeypatch.setattr(settings, "approval_dify_api_key", "test-api-key")
        monkeypatch.setattr(settings, "approval_confidence_threshold", 0.7)

        mock_dify = AsyncMock(return_value={
            "is_approval_request": False,
            "confidence": 0.1,
            "approval_type": None,
            "intent_type": "it_consult",
            "business_category": None,
            "routing_confidence": 0.1,
        })

        with patch("app.api.approval._call_dify_approval_intent", mock_dify):
            response = await client.post(
                "/approval/detect-intent",
                json={"text": "VPN连不上了"},
            )

        data = response.json()
        inner = data["data"]
        assert inner["intent_type"] == "it_consult"
        assert inner["is_approval_request"] is False

    @pytest.mark.asyncio
    async def test_non_it_routing_regression(self, client, monkeypatch):
        """'电脑连不上打印机了' → intent_type=non_it_routing, business_category=行政。

        注意：PRD 5.5 原文测试用例为 "电脑连不上打印机了"（含审批关键词"电脑"），
        而非 "打印机坏了"（"打印机"不在审批预过滤关键词中，会被直接过滤）。
        """
        monkeypatch.setattr(settings, "approval_dify_base_url", "http://test-dify")
        monkeypatch.setattr(settings, "approval_dify_api_key", "test-api-key")
        monkeypatch.setattr(settings, "approval_confidence_threshold", 0.7)

        mock_dify = AsyncMock(return_value={
            "is_approval_request": False,
            "confidence": 0.1,
            "approval_type": None,
            "intent_type": "non_it_routing",
            "business_category": "行政",
            "routing_confidence": 0.85,
        })

        with patch("app.api.approval._call_dify_approval_intent", mock_dify):
            response = await client.post(
                "/approval/detect-intent",
                json={"text": "电脑连不上打印机了"},
            )

        data = response.json()
        inner = data["data"]
        assert inner["intent_type"] == "non_it_routing"
        assert inner["business_category"] == "行政"
        assert inner["routing_confidence"] >= 0.8

    @pytest.mark.asyncio
    async def test_chitchat_regression(self, client):
        """'你好' → intent_type=chitchat, routing_confidence ≤ 0.1。"""
        response = await client.post(
            "/approval/detect-intent",
            json={"text": "你好"},
        )

        data = response.json()
        inner = data["data"]
        assert inner["intent_type"] == "chitchat"
        assert inner["routing_confidence"] <= 0.1


# =============================================================================
# API 端点测试：GET /h5/routing/contact
# =============================================================================

class TestRoutingAPIContactEndpoint:
    """测试 H5 端查询联系人 API。"""

    @pytest.mark.asyncio
    async def test_get_contact_admin(self, client, db_session, seeded_contacts):
        """查询行政类别联系人 → 返回王芳信息。"""
        # 需要让 routing API 的 get_db 使用 db_session
        from app.api.routing import get_db as routing_get_db

        async def _override_db():
            yield db_session

        app = client._transport.app
        app.dependency_overrides[routing_get_db] = _override_db
        try:
            response = await client.get(
                "/h5/routing/contact",
                params={"business_category": "行政"},
            )
        finally:
            app.dependency_overrides.pop(routing_get_db, None)

        assert response.status_code == 200
        data = response.json()
        assert data["code"] == 0
        contact = data["data"]
        assert contact is not None
        assert contact["name"] == "王芳"
        assert contact["business_category"] == "行政"
        assert contact["wecom_userid"] == "WangFang"

    @pytest.mark.asyncio
    async def test_get_contact_not_found(self, client, db_session, seeded_contacts):
        """查询不存在的类别 → data 为 None。"""
        from app.api.routing import get_db as routing_get_db

        async def _override_db():
            yield db_session

        app = client._transport.app
        app.dependency_overrides[routing_get_db] = _override_db
        try:
            response = await client.get(
                "/h5/routing/contact",
                params={"business_category": "不存在的类别"},
            )
        finally:
            app.dependency_overrides.pop(routing_get_db, None)

        assert response.status_code == 200
        data = response.json()
        assert data["code"] == 0
        assert data["data"] is None


# =============================================================================
# API 端点测试：GET /routing/contacts
# =============================================================================

class TestRoutingAPIContactsListEndpoint:
    """测试坐席端联系人列表 API。"""

    @pytest.mark.asyncio
    async def test_list_all_active_contacts(self, client, db_session, seeded_contacts):
        """无筛选条件 → 返回所有启用的联系人。"""
        from app.api.routing import get_db as routing_get_db

        async def _override_db():
            yield db_session

        app = client._transport.app
        app.dependency_overrides[routing_get_db] = _override_db
        try:
            response = await client.get("/routing/contacts")
        finally:
            app.dependency_overrides.pop(routing_get_db, None)

        assert response.status_code == 200
        data = response.json()
        assert data["code"] == 0
        items = data["data"]["items"]
        # 7个启用的联系人（不含停用的）
        assert len(items) == 7

    @pytest.mark.asyncio
    async def test_list_filter_by_category(self, client, db_session, seeded_contacts):
        """按业务类别筛选 → 返回对应类别的联系人。"""
        from app.api.routing import get_db as routing_get_db

        async def _override_db():
            yield db_session

        app = client._transport.app
        app.dependency_overrides[routing_get_db] = _override_db
        try:
            response = await client.get(
                "/routing/contacts",
                params={"business_category": "人力资源"},
            )
        finally:
            app.dependency_overrides.pop(routing_get_db, None)

        data = response.json()
        items = data["data"]["items"]
        assert len(items) == 2  # 李娜 + 张磊
        for item in items:
            assert item["business_category"] == "人力资源"

    @pytest.mark.asyncio
    async def test_list_search_by_keyword(self, client, db_session, seeded_contacts):
        """按关键词搜索 → 返回匹配的联系人。"""
        from app.api.routing import get_db as routing_get_db

        async def _override_db():
            yield db_session

        app = client._transport.app
        app.dependency_overrides[routing_get_db] = _override_db
        try:
            response = await client.get(
                "/routing/contacts",
                params={"keyword": "王"},
            )
        finally:
            app.dependency_overrides.pop(routing_get_db, None)

        data = response.json()
        items = data["data"]["items"]
        assert len(items) == 1
        assert items[0]["name"] == "王芳"

    @pytest.mark.asyncio
    async def test_list_excludes_inactive(self, client, db_session, seeded_contacts):
        """列表不包含停用的联系人。"""
        from app.api.routing import get_db as routing_get_db

        async def _override_db():
            yield db_session

        app = client._transport.app
        app.dependency_overrides[routing_get_db] = _override_db
        try:
            response = await client.get("/routing/contacts")
        finally:
            app.dependency_overrides.pop(routing_get_db, None)

        data = response.json()
        items = data["data"]["items"]
        names = [item["name"] for item in items]
        assert "停用联系人" not in names


# =============================================================================
# API 端点测试：POST /conversations/{id}/send-contact-card
# =============================================================================

class TestRoutingAPISendCardEndpoint:
    """测试坐席手动发名片 API。"""

    @pytest.mark.asyncio
    async def test_send_card_success(self, client, db_session, seeded_contacts, seeded_conversation):
        """坐席手动发名片 → 成功返回。"""
        from app.api.routing import get_db as routing_get_db

        contact = seeded_contacts[0]  # 王芳
        conv = seeded_conversation

        async def _override_db():
            yield db_session

        app = client._transport.app
        app.dependency_overrides[routing_get_db] = _override_db
        try:
            with patch("app.services.routing_service.ws_manager") as mock_ws:
                mock_ws.broadcast_to_employees = AsyncMock()
                mock_ws.broadcast = AsyncMock()

                response = await client.post(
                    f"/conversations/{conv.id}/send-contact-card",
                    json={"contact_id": contact.id, "reason": "坐席手动推荐"},
                )
        finally:
            app.dependency_overrides.pop(routing_get_db, None)

        assert response.status_code == 200
        data = response.json()
        assert data["code"] == 0
        result = data["data"]
        assert result["contact_name"] == "王芳"
        assert result["business_category"] == "行政"

    @pytest.mark.asyncio
    async def test_send_card_conversation_not_found(self, client, db_session, seeded_contacts):
        """会话不存在 → data 为 None。"""
        from app.api.routing import get_db as routing_get_db

        contact = seeded_contacts[0]

        async def _override_db():
            yield db_session

        app = client._transport.app
        app.dependency_overrides[routing_get_db] = _override_db
        try:
            with patch("app.services.routing_service.ws_manager") as mock_ws:
                mock_ws.broadcast_to_employees = AsyncMock()
                mock_ws.broadcast = AsyncMock()

                response = await client.post(
                    "/conversations/nonexistent-conv-id/send-contact-card",
                    json={"contact_id": contact.id},
                )
        finally:
            app.dependency_overrides.pop(routing_get_db, None)

        data = response.json()
        assert data["code"] == 0
        assert data["data"] is None

    @pytest.mark.asyncio
    async def test_send_card_contact_not_found(self, client, db_session, seeded_conversation):
        """联系人不存在 → data 为 None。"""
        from app.api.routing import get_db as routing_get_db

        conv = seeded_conversation

        async def _override_db():
            yield db_session

        app = client._transport.app
        app.dependency_overrides[routing_get_db] = _override_db
        try:
            with patch("app.services.routing_service.ws_manager") as mock_ws:
                mock_ws.broadcast_to_employees = AsyncMock()
                mock_ws.broadcast = AsyncMock()

                response = await client.post(
                    f"/conversations/{conv.id}/send-contact-card",
                    json={"contact_id": 99999},
                )
        finally:
            app.dependency_overrides.pop(routing_get_db, None)

        data = response.json()
        assert data["code"] == 0
        assert data["data"] is None

    @pytest.mark.asyncio
    async def test_send_card_inactive_contact(self, client, db_session, seeded_contacts, seeded_conversation):
        """停用的联系人 → data 为 None。"""
        from app.api.routing import get_db as routing_get_db

        # 找到停用的联系人
        inactive_contact = next(c for c in seeded_contacts if not c.is_active)
        conv = seeded_conversation

        async def _override_db():
            yield db_session

        app = client._transport.app
        app.dependency_overrides[routing_get_db] = _override_db
        try:
            with patch("app.services.routing_service.ws_manager") as mock_ws:
                mock_ws.broadcast_to_employees = AsyncMock()
                mock_ws.broadcast = AsyncMock()

                response = await client.post(
                    f"/conversations/{conv.id}/send-contact-card",
                    json={"contact_id": inactive_contact.id},
                )
        finally:
            app.dependency_overrides.pop(routing_get_db, None)

        data = response.json()
        assert data["code"] == 0
        assert data["data"] is None

    @pytest.mark.asyncio
    async def test_send_card_default_reason(self, client, db_session, seeded_contacts, seeded_conversation):
        """未提供 reason 时使用默认路由说明文本。"""
        from app.api.routing import get_db as routing_get_db

        contact = seeded_contacts[0]
        conv = seeded_conversation

        async def _override_db():
            yield db_session

        app = client._transport.app
        app.dependency_overrides[routing_get_db] = _override_db
        try:
            with patch("app.services.routing_service.ws_manager") as mock_ws:
                mock_ws.broadcast_to_employees = AsyncMock()
                mock_ws.broadcast = AsyncMock()

                response = await client.post(
                    f"/conversations/{conv.id}/send-contact-card",
                    json={"contact_id": contact.id},  # 不提供 reason
                )
        finally:
            app.dependency_overrides.pop(routing_get_db, None)

        assert response.status_code == 200
        # 验证消息落库（使用默认 reason）
        result = await db_session.execute(
            select(Message).where(
                Message.conversation_id == conv.id,
                Message.msg_type == "text",
            )
        )
        text_msg = result.scalar_one()
        assert "行政" in text_msg.content
        assert "王芳" in text_msg.content


# =============================================================================
# 数据模型测试
# =============================================================================

class TestBusinessContactModel:
    """测试 BusinessContact 模型。"""

    def test_to_dict_returns_complete_data(self):
        """to_dict() 返回完整的联系人信息。"""
        contact = BusinessContact(
            id=1,
            name="王芳",
            gender="female",
            department="行政部",
            position="设备管理岗",
            responsibility="打印机/复印机/扫描仪",
            extension="8002",
            service_area="滨江园区 3-5楼",
            wecom_userid="WangFang",
            avatar_url=None,
            business_category="行政",
            is_active=True,
        )
        data = contact.to_dict()
        assert data["id"] == 1
        assert data["name"] == "王芳"
        assert data["gender"] == "female"
        assert data["department"] == "行政部"
        assert data["position"] == "设备管理岗"
        assert data["responsibility"] == "打印机/复印机/扫描仪"
        assert data["extension"] == "8002"
        assert data["service_area"] == "滨江园区 3-5楼"
        assert data["wecom_userid"] == "WangFang"
        assert data["avatar_url"] == ""
        assert data["business_category"] == "行政"

    def test_to_dict_handles_none_fields(self):
        """to_dict() 正确处理 None 字段（转为空字符串）。"""
        contact = BusinessContact(
            id=2,
            name="测试",
            gender="male",
            department="测试部",
            position="测试岗",
            responsibility="测试",
            extension=None,
            service_area=None,
            wecom_userid="Test",
            avatar_url=None,
            business_category="行政",
            is_active=True,
        )
        data = contact.to_dict()
        assert data["extension"] == ""
        assert data["service_area"] == ""
        assert data["avatar_url"] == ""


# =============================================================================
# 配置测试
# =============================================================================

class TestRoutingConfig:
    """测试路由推荐相关配置。"""

    def test_routing_confidence_threshold_exists(self):
        """routing_confidence_threshold 配置项存在。"""
        assert hasattr(settings, "routing_confidence_threshold")

    def test_routing_confidence_threshold_default(self):
        """routing_confidence_threshold 默认值为 0.7。"""
        assert settings.routing_confidence_threshold == 0.7

    def test_routing_keywords_cover_all_categories(self):
        """路由关键词覆盖所有5个业务类别。"""
        categories = set(ROUTING_KEYWORD_TO_CATEGORY.values())
        expected = {"行政", "人力资源", "财务", "法务", "行政-物业"}
        assert categories == expected

    def test_all_prefilter_keywords_have_category_mapping(self):
        """每个预过滤关键词都有对应的类别映射。"""
        for kw in ROUTING_PREFILTER_KEYWORDS:
            assert kw in ROUTING_KEYWORD_TO_CATEGORY, f"关键词 '{kw}' 缺少类别映射"