LQAgentPlatform/utils_test/Sensitive_Test/test_sensitive_check.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
敏感词检查功能单元测试
测试 AIReviewEngine.check_sensitive 方法的各种场景
"""

import asyncio
import sys
import os
import time
from pathlib import Path
from typing import Dict, Any, List
from dataclasses import dataclass

# 添加项目根目录到Python路径
current_dir = Path(__file__).parent.absolute()
project_root = current_dir.parent.parent
sys.path.insert(0, str(project_root))
os.chdir(str(project_root))

from core.construction_review.component.reviewers.base_reviewer import ReviewResult
from foundation.observability.logger.loggering import review_logger as logger


class TestSensitiveCheck:
    """敏感词检查功能测试类"""

    def __init__(self):
        self.test_results = []
        logger.info("=" * 80)
        logger.info("初始化敏感词检查测试类")
        logger.info("=" * 80)

    async def check_sensitive_simplified(
        self, 
        trace_id_idx: str, 
        review_content: str, 
        review_references: str = "",
        review_location_label: str = "",
        state: Dict = None,
        stage_name: str = "测试阶段"
    ) -> Dict[str, Any]:
        """
        简化版的敏感信息检查函数（用于测试）
        
        功能说明：
        1. 使用关键词匹配进行敏感词检测
        2. 如果检测到敏感词，格式化敏感词信息并返回
        3. 如果未检测到敏感词，返回成功结果
        4. 移除了原函数中的消息推送和大模型二审功能
        
        Args:
            trace_id_idx: 追踪ID索引
            review_content: 审查内容
            review_references: 审查参考信息
            review_location_label: 审查位置标签
            state: 状态字典（简化版不使用）
            stage_name: 阶段名称
            
        Returns:
            Dict[str, Any]: 敏感信息检查结果
        """
        from core.construction_review.component.reviewers.utils import (
            check_sensitive_words_async,
            format_check_results
        )
        
        logger.info("=" * 80)
        logger.info(f"开始执行敏感词检查 - trace_id: sensitive_check{trace_id_idx}")
        logger.info(f"阶段名称: {stage_name}")
        logger.info(f"审查位置: {review_location_label}")
        logger.info(f"审查内容长度: {len(review_content)} 字符")
        logger.info(f"审查内容预览: {review_content[:100]}..." if len(review_content) > 100 else f"审查内容: {review_content}")
        logger.info("=" * 80)
        
        start_time = time.time()
        trace_id = "sensitive_check" + trace_id_idx
        
        # 第一步：使用关键词匹配式审查
        logger.info("步骤1: 开始关键词匹配检测...")
        first_results = await check_sensitive_words_async(review_content)
        detection_time = time.time() - start_time
        logger.info(f"步骤1: 关键词检测完成，耗时: {detection_time:.4f}s")
        
        # 判断是否检测到敏感词
        if first_results:
            logger.warning(f"步骤2: 检测到 {len(first_results)} 个敏感词")
            logger.warning("-" * 80)
            
            # 格式化敏感词列表
            sensitive_words_info = []
            for idx, item in enumerate(first_results, 1):
                word_info = f"敏感词: {item['word']}, 位置: {item['position']}-{item['end_position']}, 来源: {item['source']}"
                sensitive_words_info.append(word_info)
                logger.warning(f"  敏感词 #{idx}: {word_info}")
            
            formatted_sensitive_words = "\n".join(sensitive_words_info)
            logger.warning("-" * 80)
            
            # 简化版：直接返回检测结果，不调用大模型二审
            execution_time = time.time() - start_time
            
            result = ReviewResult(
                success=False,
                details={
                    "name": "sensitive_check",
                    "response": f"检测到{len(first_results)}个敏感词",
                    "sensitive_words": first_results,
                    "formatted_info": formatted_sensitive_words
                },
                error_message=None,
                execution_time=execution_time
            )
            
            logger.warning(f"步骤3: 敏感词检查完成（检测到敏感词），总耗时: {execution_time:.4f}s")
            logger.info("=" * 80)
            
            return result
            
        else:
            # 没有检测到敏感词，构造返回体
            logger.info("步骤2: 未检测到敏感词")
            
            execution_time = time.time() - start_time
            result = ReviewResult(
                success=True,
                details={"name": "sensitive_check", "response": "无明显问题"},
                error_message=None,
                execution_time=execution_time
            )
            
            # 简化版：移除消息推送功能
            logger.info(f"步骤3: 敏感词检查完成（未检测到敏感词），总耗时: {execution_time:.4f}s")
            logger.info("=" * 80)
            
            return result

    def get_test_cases(self) -> List[Dict[str, Any]]:
        """获取测试用例"""
        return [
            {
                "name": "正常文本-无敏感词",
                "trace_id_idx": "_test_001",
                "review_content": "本工程为住宅楼建设项目，采用框架结构，建筑面积约5000平方米。施工过程中应严格按照国家规范执行，确保工程质量。",
                "review_location_label": "第一章 工程概况",
                "expected_success": True,
                "description": "测试正常的施工方案文本，不包含任何敏感词"
            },
            {
                "name": "空文本",
                "trace_id_idx": "_test_002",
                "review_content": "",
                "review_location_label": "空内容测试",
                "expected_success": True,
                "description": "测试空文本的处理"
            },
            {
                "name": "短文本-无敏感词",
                "trace_id_idx": "_test_003",
                "review_content": "施工安全第一",
                "review_location_label": "安全标语",
                "expected_success": True,
                "description": "测试短文本的处理"
            },
            {
                "name": "长文本-无敏感词",
                "trace_id_idx": "_test_004",
                "review_content": """
                本施工方案编制依据包括：
                1. 《建筑工程施工质量验收统一标准》GB50300-2013
                2. 《混凝土结构工程施工质量验收规范》GB50204-2015
                3. 《建筑地基基础工程施工质量验收规范》GB50202-2018
                4. 施工图纸及相关设计文件
                5. 现场实际情况及勘察报告
                
                工程概况：
                本工程位于某市某区，为高层住宅建筑，地上30层，地下2层。
                建筑高度99.8米，总建筑面积约28000平方米。
                结构形式为框架剪力墙结构，抗震设防烈度为7度。
                """,
                "review_location_label": "第一章 编制依据与工程概况",
                "expected_success": True,
                "description": "测试包含多段落的长文本"
            },
            {
                "name": "包含特殊字符",
                "trace_id_idx": "_test_005",
                "review_content": "施工现场温度：-5℃~35℃，相对湿度：≤85%，风力：≤5级。混凝土强度等级：C30、C35。钢筋规格：Φ12、Φ16、Φ20。",
                "review_location_label": "第三章 施工条件",
                "expected_success": True,
                "description": "测试包含特殊符号和技术参数的文本"
            }
        ]

    async def run_single_test(self, test_case: Dict[str, Any]) -> bool:
        """运行单个测试用例"""
        logger.info("\n" + "█" * 80)
        logger.info(f"测试用例: {test_case['name']}")
        logger.info(f"描述: {test_case['description']}")
        logger.info("█" * 80)
        
        try:
            # 执行测试
            result = await self.check_sensitive_simplified(
                trace_id_idx=test_case['trace_id_idx'],
                review_content=test_case['review_content'],
                review_location_label=test_case['review_location_label'],
                stage_name="单元测试阶段"
            )
            
            # 验证结果
            logger.info("\n" + "-" * 80)
            logger.info("测试结果验证:")
            logger.info(f"  返回类型: {type(result)}")
            logger.info(f"  是否成功: {result.success}")
            logger.info(f"  预期成功: {test_case['expected_success']}")
            logger.info(f"  执行时间: {result.execution_time:.4f}s")
            logger.info(f"  详细信息: {result.details}")
            
            if result.error_message:
                logger.error(f"  错误信息: {result.error_message}")
            
            # 判断测试是否通过
            test_passed = result.success == test_case['expected_success']
            
            if test_passed:
                logger.info("✓ 测试通过")
            else:
                logger.error("✗ 测试失败")
                logger.error(f"  预期 success={test_case['expected_success']}, 实际 success={result.success}")
            
            logger.info("-" * 80)
            
            return test_passed
            
        except Exception as e:
            logger.error(f"✗ 测试异常: {str(e)}")
            logger.exception("详细异常信息:")
            return False

    async def run_all_tests(self):
        """运行所有测试用例"""
        logger.info("\n" + "█" * 80)
        logger.info("开始执行敏感词检查单元测试")
        logger.info("█" * 80)
        
        # 初始化敏感词检测器
        try:
            from core.construction_review.component.reviewers.utils import SensitiveWordChecker
            logger.info("\n正在初始化敏感词检测器...")
            stats = SensitiveWordChecker.initialize()
            logger.info(f"敏感词检测器初始化成功: {stats}")
        except Exception as e:
            logger.error(f"敏感词检测器初始化失败: {str(e)}")
            return False
        
        # 获取测试用例
        test_cases = self.get_test_cases()
        logger.info(f"\n共有 {len(test_cases)} 个测试用例")
        
        # 运行测试
        results = []
        for idx, test_case in enumerate(test_cases, 1):
            logger.info(f"\n{'=' * 80}")
            logger.info(f"执行测试 {idx}/{len(test_cases)}")
            logger.info(f"{'=' * 80}")
            
            passed = await self.run_single_test(test_case)
            results.append({
                'name': test_case['name'],
                'passed': passed
            })
        
        # 统计结果
        logger.info("\n" + "█" * 80)
        logger.info("测试结果汇总")
        logger.info("█" * 80)
        
        passed_count = sum(1 for r in results if r['passed'])
        total_count = len(results)
        
        logger.info(f"\n总测试数: {total_count}")
        logger.info(f"通过数量: {passed_count}")
        logger.info(f"失败数量: {total_count - passed_count}")
        logger.info(f"通过率: {passed_count/total_count*100:.2f}%")
        
        logger.info("\n详细结果:")
        for idx, result in enumerate(results, 1):
            status = "✓ 通过" if result['passed'] else "✗ 失败"
            logger.info(f"  {idx}. {result['name']}: {status}")
        
        logger.info("\n" + "█" * 80)
        
        if passed_count == total_count:
            logger.info("✓ 所有测试通过!")
            logger.info("█" * 80)
            return True
        else:
            logger.error(f"✗ 有 {total_count - passed_count} 个测试失败")
            logger.info("█" * 80)
            return False

    def run_tests_sync(self):
        """同步方式运行测试"""
        return asyncio.run(self.run_all_tests())


def main():
    """主函数"""
    logger.info("=" * 80)
    logger.info("敏感词检查功能单元测试")
    logger.info("测试文件: test_sensitive_check.py")
    logger.info("=" * 80)
    
    tester = TestSensitiveCheck()
    success = tester.run_tests_sync()
    
    sys.exit(0 if success else 1)


if __name__ == "__main__":
    main()