wecom_it_smart_desk/backend/tests/test_exclusion.py

# -*- coding: utf-8 -*-
"""代答排除模块回归测试 — T04

测试范围:
  Matcher单元测试（12个）: keyword(5) + regex(3) + intent(2) + category(2)
  ExclusionService测试（5个）: no_rules, priority_order, first_hit_stops,
                               logs_hit, test_match_no_logging
  API测试（8个）: create, duplicate_name, get_detail, update, delete,
                  toggle, test_match, get_stats
  ai_handler集成测试（3个）: exclusion_hit, exclusion_miss, exclusion_error

测试依赖: conftest.py 提供的 client / db_session fixtures
"""

import uuid
from contextlib import ExitStack
from datetime import datetime
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.exclusion_rule import ExclusionRule
from app.models.exclusion_log import ExclusionLog
from app.models.triage_session import TriageSession


# ============================================================================
# 辅助函数
# ============================================================================

async def _login_admin(client: AsyncClient, db_session: AsyncSession) -> str:
    """创建 admin 角色用户并返回 Bearer token。"""
    from app.models.role import Role
    from app.models.user_role import UserRole

    admin_id = f"test_admin_{uuid.uuid4().hex[:8]}"

    # 确保 admin 角色存在
    stmt = select(Role).where(Role.name == "admin")
    result = await db_session.execute(stmt)
    admin_role = result.scalars().first()
    if not admin_role:
        admin_role = Role(
            name="admin", display_name="管理员",
            description="系统管理员", permissions=[],
        )
        db_session.add(admin_role)
        await db_session.flush()

    # 创建 UserRole 关联
    db_session.add(UserRole(
        employee_id=admin_id, role_id=admin_role.id,
        source="manual", assigned_by="test_fixture",
    ))
    await db_session.flush()

    # 登录
    resp = await client.post("/agents/login", json={
        "user_id": admin_id, "name": "测试管理员",
    })
    return resp.json()["data"]["token"]


async def _create_exclusion_rule(db_session: AsyncSession, **kwargs) -> ExclusionRule:
    """在数据库中创建排除规则。

    Args:
        db_session: 数据库会话
        **kwargs: 覆盖默认字段值

    Returns:
        ExclusionRule: 创建的规则对象
    """
    defaults = {
        "rule_name": f"规则-{uuid.uuid4().hex[:8]}",
        "rule_description": "测试规则",
        "priority": "P2",
        "match_type": "keyword",
        "match_condition": "密码",
        "match_scope": [],
        "action_type": "transfer_human",
        "transfer_message": "已为您转接人工坐席",
        "status": "enabled",
        "hit_count": 0,
        "created_by": "test_admin",
    }
    defaults.update(kwargs)
    rule = ExclusionRule(**defaults)
    db_session.add(rule)
    await db_session.flush()
    return rule


# ============================================================================
# Section A — Matcher 单元测试（12个）
# ============================================================================

class TestKeywordMatcher:
    """关键词匹配器测试。"""

    @pytest.mark.asyncio
    async def test_keyword_match_hit(self):
        """关键词命中。"""
        from app.services.matchers.keyword_matcher import KeywordMatcher

        matcher = KeywordMatcher()
        result = await matcher.match(
            message="我的密码过期了怎么办",
            condition="密码过期,账号锁定",
        )
        assert result.matched is True
        assert "密码过期" in result.matched_detail

    @pytest.mark.asyncio
    async def test_keyword_match_miss(self):
        """关键词未命中。"""
        from app.services.matchers.keyword_matcher import KeywordMatcher

        matcher = KeywordMatcher()
        result = await matcher.match(
            message="今天天气真好",
            condition="密码过期,账号锁定",
        )
        assert result.matched is False

    @pytest.mark.asyncio
    async def test_keyword_match_case_insensitive(self):
        """大小写不敏感匹配。"""
        from app.services.matchers.keyword_matcher import KeywordMatcher

        matcher = KeywordMatcher()
        result = await matcher.match(
            message="VPN connection failed",
            condition="vpn,password",
        )
        assert result.matched is True
        assert "vpn" in result.matched_detail.lower()

    @pytest.mark.asyncio
    async def test_keyword_match_multiple_keywords(self):
        """多关键词逗号分隔，任一命中即匹配。"""
        from app.services.matchers.keyword_matcher import KeywordMatcher

        matcher = KeywordMatcher()
        # 第三个关键词命中
        result = await matcher.match(
            message="打印机卡纸了",
            condition="密码过期,账号锁定,打印机",
        )
        assert result.matched is True
        assert "打印机" in result.matched_detail

    @pytest.mark.asyncio
    async def test_keyword_match_empty_message(self):
        """空消息返回未命中。"""
        from app.services.matchers.keyword_matcher import KeywordMatcher

        matcher = KeywordMatcher()
        result = await matcher.match(message="", condition="密码")
        assert result.matched is False

        # 空条件也应返回未命中
        result = await matcher.match(message="密码过期", condition="")
        assert result.matched is False


class TestRegexMatcher:
    """正则匹配器测试。

    注意: RegexMatcher 使用 signal.SIGALRM 做 ReDoS 超时保护，
    Windows 不支持 SIGALRM。测试通过 fixture patch signal 模块
    使正则匹配在 Windows 上正常工作。
    源码 regex_matcher.py 在 Windows 上存在兼容性问题
    （SIGALRM 不可用时直接返回未命中，应降级为无超时匹配）。
    """

    @pytest.fixture(autouse=True)
    def _patch_signal(self):
        """Windows 兼容: patch signal.SIGALRM 使正则匹配器正常工作。"""
        import signal as sig
        if not hasattr(sig, "SIGALRM"):
            with ExitStack() as stack:
                stack.enter_context(patch.object(sig, "SIGALRM", 14, create=True))
                stack.enter_context(patch.object(sig, "ITIMER_REAL", 0, create=True))
                if not hasattr(sig, "setitimer"):
                    stack.enter_context(
                        patch.object(sig, "setitimer", create=True, return_value=None)
                    )
                stack.enter_context(
                    patch.object(sig, "signal", return_value=sig.SIG_DFL)
                )
                yield
        else:
            yield

    @pytest.mark.asyncio
    async def test_regex_match_hit(self):
        """正则命中。"""
        from app.services.matchers.regex_matcher import RegexMatcher

        matcher = RegexMatcher()
        result = await matcher.match(
            message="我的密码好像过期了",
            condition="密码.*过期",
        )
        assert result.matched is True
        assert "密码" in result.matched_detail

    @pytest.mark.asyncio
    async def test_regex_match_miss(self):
        """正则未命中。"""
        from app.services.matchers.regex_matcher import RegexMatcher

        matcher = RegexMatcher()
        result = await matcher.match(
            message="今天天气真好",
            condition="密码.*过期",
        )
        assert result.matched is False

    @pytest.mark.asyncio
    async def test_regex_match_invalid_pattern(self):
        """无效正则返回未命中（不抛异常）。"""
        from app.services.matchers.regex_matcher import RegexMatcher

        matcher = RegexMatcher()
        # 无效正则括号不匹配
        result = await matcher.match(
            message="测试消息",
            condition="[unclosed",
        )
        assert result.matched is False


class TestIntentMatcher:
    """意图匹配器测试。"""

    @pytest.mark.asyncio
    async def test_intent_match_dify_unavailable(self):
        """Dify 不可用时降级返回未命中。"""
        from app.services.matchers.intent_matcher import IntentMatcher

        matcher = IntentMatcher()
        # Mock _recognize_intent 返回 None（Dify 不可用）
        with patch.object(matcher, "_recognize_intent", return_value=None):
            result = await matcher.match(
                message="我的密码忘了",
                condition="password_reset,account_unlock",
            )
        assert result.matched is False

    @pytest.mark.asyncio
    async def test_intent_match_hit(self):
        """意图命中（mock Dify 返回匹配的意图）。"""
        from app.services.matchers.intent_matcher import IntentMatcher

        matcher = IntentMatcher()
        # Mock _recognize_intent 返回 password_reset
        with patch.object(matcher, "_recognize_intent", return_value="password_reset"):
            result = await matcher.match(
                message="我的密码忘了，帮我重置一下",
                condition="password_reset,account_unlock",
            )
        assert result.matched is True
        assert "password_reset" in result.matched_detail


class TestCategoryMatcher:
    """分类匹配器测试。"""

    @pytest.mark.asyncio
    async def test_category_match_hit(self, db_session: AsyncSession):
        """分类命中 — 分诊记录的 problem_category 在排除列表中。"""
        from app.services.matchers.category_matcher import CategoryMatcher

        conv_id = f"conv-cat-{uuid.uuid4().hex[:8]}"
        # 创建分诊记录，problem_category = "Outlook"
        triage = TriageSession(
            conversation_id=conv_id,
            user_id="test_user",
            user_name="测试",
            request_title="测试",
            request_content="测试内容",
            source="wecom_h5",
            status="routed",
            urgency="low",
            problem_category="Outlook",
        )
        db_session.add(triage)
        await db_session.flush()

        matcher = CategoryMatcher()
        result = await matcher.match(
            message="测试消息",
            condition="Outlook,VPN,打印机",
            context={"conversation_id": conv_id, "db": db_session},
        )
        assert result.matched is True
        assert "Outlook" in result.matched_detail

    @pytest.mark.asyncio
    async def test_category_match_no_triage(self, db_session: AsyncSession):
        """无分诊记录返回未命中（软依赖）。"""
        from app.services.matchers.category_matcher import CategoryMatcher

        matcher = CategoryMatcher()
        # 使用不存在的 conversation_id
        result = await matcher.match(
            message="测试消息",
            condition="Outlook,VPN",
            context={"conversation_id": "non-existent-conv", "db": db_session},
        )
        assert result.matched is False


# ============================================================================
# Section B — ExclusionService 测试（5个）
# ============================================================================

class TestExclusionService:
    """ExclusionService 责任链匹配引擎测试。"""

    @pytest.mark.asyncio
    async def test_check_exclusions_no_rules(self, db_session: AsyncSession):
        """无规则时返回未命中。"""
        from app.services.exclusion_service import ExclusionService

        service = ExclusionService()
        result = await service.check_exclusions(
            db=db_session,
            message="测试消息",
            conversation_id="conv-001",
            user_id="user-001",
        )
        assert result.matched is False

    @pytest.mark.asyncio
    async def test_check_exclusions_priority_order(self, db_session: AsyncSession):
        """优先级排序 P0 > P1 — P0 规则先匹配。"""
        from app.services.exclusion_service import ExclusionService

        # 创建两条规则，P1 和 P0，消息同时包含两个关键词
        await _create_exclusion_rule(
            db_session,
            rule_name="P1规则-密码",
            priority="P1",
            match_condition="密码",
        )
        await _create_exclusion_rule(
            db_session,
            rule_name="P0规则-宕机",
            priority="P0",
            match_condition="宕机",
        )
        await db_session.flush()

        service = ExclusionService()
        result = await service.check_exclusions(
            db=db_session,
            message="系统宕机了，密码也忘了",
            conversation_id="conv-002",
            user_id="user-002",
        )
        assert result.matched is True
        # P0 规则应先匹配
        assert result.rule_name == "P0规则-宕机"

    @pytest.mark.asyncio
    async def test_check_exclusions_first_hit_stops(self, db_session: AsyncSession):
        """命中即停止 — 只记录一条日志。"""
        from app.services.exclusion_service import ExclusionService

        # 创建两条都能匹配的规则
        await _create_exclusion_rule(
            db_session,
            rule_name="规则A-密码",
            priority="P0",
            match_condition="密码",
        )
        await _create_exclusion_rule(
            db_session,
            rule_name="规则B-密码",
            priority="P1",
            match_condition="密码",
        )
        await db_session.flush()

        service = ExclusionService()
        result = await service.check_exclusions(
            db=db_session,
            message="密码过期了",
            conversation_id="conv-003",
            user_id="user-003",
        )
        assert result.matched is True
        # 只有 P0 规则应命中
        assert result.rule_name == "规则A-密码"

        # 验证只创建了一条日志
        log_result = await db_session.execute(select(ExclusionLog))
        logs = log_result.scalars().all()
        assert len(logs) == 1
        assert logs[0].rule_name == "规则A-密码"

    @pytest.mark.asyncio
    async def test_check_exclusions_logs_hit(self, db_session: AsyncSession):
        """命中记录日志 + 更新 hit_count。"""
        from app.services.exclusion_service import ExclusionService

        rule = await _create_exclusion_rule(
            db_session,
            rule_name="日志测试规则",
            match_condition="密码过期",
            hit_count=0,
        )
        await db_session.flush()

        service = ExclusionService()
        result = await service.check_exclusions(
            db=db_session,
            message="我的密码过期了",
            conversation_id="conv-004",
            user_id="user-004",
        )
        assert result.matched is True
        assert result.rule_name == "日志测试规则"

        # 验证 hit_count 已更新
        await db_session.refresh(rule)
        assert rule.hit_count == 1

        # 验证日志已创建
        log_result = await db_session.execute(
            select(ExclusionLog).where(ExclusionLog.rule_id == rule.id)
        )
        logs = log_result.scalars().all()
        assert len(logs) == 1
        assert logs[0].action_type == "transfer_human"

    @pytest.mark.asyncio
    async def test_test_match_no_logging(self, db_session: AsyncSession):
        """test_match 不记录日志、不更新 hit_count。"""
        from app.services.exclusion_service import ExclusionService

        rule = await _create_exclusion_rule(
            db_session,
            rule_name="测试匹配规则",
            match_condition="密码过期",
            hit_count=0,
        )
        await db_session.flush()

        service = ExclusionService()
        result = await service.test_match(
            db=db_session,
            message="我的密码过期了",
            rule_id=rule.id,
        )
        assert result.matched is True

        # 验证 hit_count 未更新
        await db_session.refresh(rule)
        assert rule.hit_count == 0

        # 验证无日志创建
        log_result = await db_session.execute(
            select(ExclusionLog).where(ExclusionLog.rule_id == rule.id)
        )
        logs = log_result.scalars().all()
        assert len(logs) == 0


# ============================================================================
# Section C — API 测试（8个）
# ============================================================================

class TestExclusionAPI:
    """代答排除管理 API 测试。"""

    @pytest.mark.asyncio
    async def test_create_rule_success(self, client, db_session):
        """新建规则成功。"""
        token = await _login_admin(client, db_session)

        resp = await client.post(
            "/admin/exclusion-rules",
            json={
                "rule_name": "测试规则-新建",
                "rule_description": "测试描述",
                "priority": "P1",
                "match_type": "keyword",
                "match_condition": "密码,账号",
                "match_scope": [],
                "action_type": "transfer_human",
                "transfer_message": "已转人工",
            },
            headers={"Authorization": f"Bearer {token}"},
        )

        assert resp.status_code == 200
        data = resp.json()
        assert data["code"] == 0
        rule = data["data"]
        assert rule["rule_name"] == "测试规则-新建"
        assert rule["status"] == "enabled"
        assert rule["hit_count"] == 0

    @pytest.mark.asyncio
    async def test_create_rule_duplicate_name(self, client, db_session):
        """规则名重复返回 400。"""
        token = await _login_admin(client, db_session)

        # 先创建一条规则
        await _create_exclusion_rule(db_session, rule_name="重复规则名")
        await db_session.flush()

        # 再用同名创建
        resp = await client.post(
            "/admin/exclusion-rules",
            json={
                "rule_name": "重复规则名",
                "priority": "P2",
                "match_type": "keyword",
                "match_condition": "测试",
                "action_type": "transfer_human",
            },
            headers={"Authorization": f"Bearer {token}"},
        )

        assert resp.status_code == 200
        data = resp.json()
        assert data["code"] == 400
        assert "已存在" in data["message"]

    @pytest.mark.asyncio
    async def test_get_rule_detail(self, client, db_session):
        """规则详情。"""
        token = await _login_admin(client, db_session)

        rule = await _create_exclusion_rule(
            db_session,
            rule_name="详情测试规则",
            match_type="keyword",
            match_condition="密码",
        )
        await db_session.flush()

        resp = await client.get(
            f"/admin/exclusion-rules/{rule.id}",
            headers={"Authorization": f"Bearer {token}"},
        )

        assert resp.status_code == 200
        data = resp.json()
        assert data["code"] == 0
        assert data["data"]["id"] == rule.id
        assert data["data"]["rule_name"] == "详情测试规则"

    @pytest.mark.asyncio
    async def test_update_rule(self, client, db_session):
        """编辑规则。"""
        token = await _login_admin(client, db_session)

        rule = await _create_exclusion_rule(
            db_session,
            rule_name="编辑前规则",
        )
        await db_session.flush()

        resp = await client.put(
            f"/admin/exclusion-rules/{rule.id}",
            json={
                "rule_name": "编辑后规则",
                "match_condition": "VPN,网络",
            },
            headers={"Authorization": f"Bearer {token}"},
        )

        assert resp.status_code == 200
        data = resp.json()
        assert data["code"] == 0
        assert data["data"]["rule_name"] == "编辑后规则"
        assert data["data"]["match_condition"] == "VPN,网络"

    @pytest.mark.asyncio
    async def test_delete_rule(self, client, db_session):
        """删除规则。"""
        token = await _login_admin(client, db_session)

        rule = await _create_exclusion_rule(
            db_session,
            rule_name="待删除规则",
        )
        await db_session.flush()
        rule_id = rule.id

        resp = await client.delete(
            f"/admin/exclusion-rules/{rule_id}",
            headers={"Authorization": f"Bearer {token}"},
        )

        assert resp.status_code == 200
        data = resp.json()
        assert data["code"] == 0

        # 验证已删除
        resp = await client.get(
            f"/admin/exclusion-rules/{rule_id}",
            headers={"Authorization": f"Bearer {token}"},
        )
        assert resp.json()["code"] == 404

    @pytest.mark.asyncio
    async def test_toggle_rule(self, client, db_session):
        """启用/停用规则。"""
        token = await _login_admin(client, db_session)

        rule = await _create_exclusion_rule(
            db_session,
            rule_name="切换状态规则",
            status="enabled",
        )
        await db_session.flush()

        # 停用
        resp = await client.post(
            f"/admin/exclusion-rules/{rule.id}/toggle",
            json={"status": "disabled"},
            headers={"Authorization": f"Bearer {token}"},
        )

        assert resp.status_code == 200
        data = resp.json()
        assert data["code"] == 0
        assert data["data"]["status"] == "disabled"

        # 再启用
        resp = await client.post(
            f"/admin/exclusion-rules/{rule.id}/toggle",
            json={"status": "enabled"},
            headers={"Authorization": f"Bearer {token}"},
        )

        assert resp.status_code == 200
        assert resp.json()["data"]["status"] == "enabled"

    @pytest.mark.asyncio
    async def test_test_match_api(self, client, db_session):
        """测试匹配 API。"""
        token = await _login_admin(client, db_session)

        await _create_exclusion_rule(
            db_session,
            rule_name="API测试匹配规则",
            match_type="keyword",
            match_condition="密码过期",
        )
        await db_session.flush()

        resp = await client.post(
            "/admin/exclusion-rules/test",
            json={"message": "我的密码过期了"},
            headers={"Authorization": f"Bearer {token}"},
        )

        assert resp.status_code == 200
        data = resp.json()
        assert data["code"] == 0
        assert data["data"]["matched"] is True
        assert data["data"]["rule_name"] == "API测试匹配规则"

    @pytest.mark.asyncio
    async def test_get_stats_api(self, client, db_session):
        """统计概要 API。"""
        token = await _login_admin(client, db_session)

        # 创建规则
        await _create_exclusion_rule(db_session, status="enabled")
        await _create_exclusion_rule(db_session, status="disabled")
        await db_session.flush()

        resp = await client.get(
            "/admin/exclusion-rules/stats",
            headers={"Authorization": f"Bearer {token}"},
        )

        assert resp.status_code == 200
        data = resp.json()
        assert data["code"] == 0
        stats = data["data"]
        assert "enabled_count" in stats
        assert "disabled_count" in stats
        assert "monthly_hits" in stats
        assert "monthly_transfers" in stats
        assert stats["enabled_count"] == 1
        assert stats["disabled_count"] == 1


# ============================================================================
# Section D — ai_handler 集成测试（3个）
# ============================================================================

class TestAIHandlerExclusion:
    """AI 回复处理器与代答排除的集成测试。

    验证 ai_handler.handle_message 在 AI 回复前检查排除规则：
    1. 命中排除规则 → 拦截 AI 回复，返回排除结果
    2. 未命中 → 正常调用 AI
    3. 排除检查异常 → 降级继续 AI 回复（不阻断主流程）
    """

    @pytest.mark.asyncio
    async def test_ai_handler_exclusion_hit(self, db_session: AsyncSession):
        """命中排除规则 — AI 回复前拦截，不调用 AI 服务。"""
        from app.services.ai_handler import AIHandler

        # 创建排除规则（关键词 "密码" → transfer_human）
        await _create_exclusion_rule(
            db_session,
            rule_name="密码排除规则",
            match_type="keyword",
            match_condition="密码",
            action_type="transfer_human",
            transfer_message="此问题需转人工处理",
        )
        await db_session.flush()

        # 创建 AIHandler with mock AIService
        mock_ai_service = AsyncMock()
        handler = AIHandler(ai_service=mock_ai_service)

        result = await handler.handle_message(
            content="我的密码过期了怎么办",
            conversation_id="conv-ai-001",
            user_id="user-001",
            db=db_session,
        )

        # 验证：命中排除规则，返回 excluded 类型
        assert result.reply_type == "excluded"
        assert result.should_transfer is True
        assert result.should_count is False
        # AI 服务不应被调用
        mock_ai_service.get_reply.assert_not_called()

    @pytest.mark.asyncio
    async def test_ai_handler_exclusion_miss(self, db_session: AsyncSession):
        """未命中排除规则 — 正常调用 AI 服务。"""
        from app.services.ai_handler import AIHandler

        # 创建排除规则（关键词 "密码"）
        await _create_exclusion_rule(
            db_session,
            rule_name="密码排除规则",
            match_type="keyword",
            match_condition="密码",
        )
        await db_session.flush()

        # 创建 AIHandler with mock AIService
        mock_ai_service = AsyncMock()
        mock_ai_service.get_reply.return_value = {
            "hit": True,
            "content": "建议您重启电脑试试",
            "conversation_id": "dify-conv-001",
        }
        handler = AIHandler(ai_service=mock_ai_service)

        # 消息不含 "密码" → 不命中排除规则 → 正常 AI 回复
        result = await handler.handle_message(
            content="打印机怎么连接",
            conversation_id="conv-ai-002",
            user_id="user-002",
            db=db_session,
        )

        # 验证：正常 AI 回复
        assert result.reply_type == "ai_hit"
        assert result.should_count is True
        assert "重启电脑" in result.content
        # AI 服务应被调用
        mock_ai_service.get_reply.assert_called_once()

    @pytest.mark.asyncio
    async def test_ai_handler_exclusion_error(self, db_session: AsyncSession):
        """排除检查异常不阻断主流程 — 降级继续 AI 回复。"""
        from app.services.ai_handler import AIHandler

        # 创建 AIHandler with mock AIService
        mock_ai_service = AsyncMock()
        mock_ai_service.get_reply.return_value = {
            "hit": True,
            "content": "AI降级回复",
            "conversation_id": "dify-conv-002",
        }
        handler = AIHandler(ai_service=mock_ai_service)

        # Patch check_exclusions 抛出异常
        with patch(
            "app.services.exclusion_service.get_exclusion_service"
        ) as mock_get_service:
            mock_service = AsyncMock()
            mock_service.check_exclusions.side_effect = Exception("DB connection error")
            mock_get_service.return_value = mock_service

            result = await handler.handle_message(
                content="打印机问题",
                conversation_id="conv-ai-003",
                user_id="user-003",
                db=db_session,
            )

        # 验证：排除检查异常后降级继续 AI 回复
        assert result.reply_type == "ai_hit"
        assert result.should_count is True
        # AI 服务应被调用（降级不阻断）
        mock_ai_service.get_reply.assert_called_once()