test_mcp_real_scenarios.py

#!/usr/bin/env python3
# 测试指令： python test_mcp_real_scenarios.py
# -*- coding: utf-8 -*-
"""
MCP服务真实场景测试文件

模拟用户真实的开发需求经过AI转换后的用例，测试MCP服务的返回结果质量。
包含多个典型开发场景，评估公共规范和匹配规范的准确性、实用性、完整性。
"""

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

# 添加项目路径
project_root = Path(__file__).parent
sys.path.insert(0, str(project_root / "src"))

try:
    from engine import RuleMatchEngine
except ImportError as e:
    print(f"导入失败: {e}")
    sys.exit(1)


@dataclass
class TestCase:
    """测试用例数据结构"""
    id: str
    name: str
    original_requirement: str  # 用户原始需求
    ai_converted_query: str    # AI转换后的查询
    expected_keywords: List[str]  # 期望匹配的关键词
    expected_type: str         # 期望的规范类型
    category: str              # 场景分类


@dataclass
class TestResult:
    """测试结果数据结构"""
    test_case: TestCase
    response: str
    response_time_ms: float
    has_common_standard: bool
    has_matched_standard: bool
    matched_keywords: List[str]
    accuracy_score: float
    usefulness_score: float
    completeness_score: float
    overall_score: float
    error_message: str = ""


class MCPServiceTester:
    """MCP服务测试器"""

    def __init__(self):
        """初始化测试器"""
        self.engine = RuleMatchEngine()
        self.test_cases = self._create_test_cases()
        self.results: List[TestResult] = []

    def _create_test_cases(self) -> List[TestCase]:
        """创建测试用例集合"""
        return [
            # React组件开发场景
            TestCase(
                id="react_login_01",
                name="React登录组件开发",
                original_requirement="我需要开发一个用户登录页面，使用React技术栈",
                ai_converted_query="react登录组件开发",
                expected_keywords=["react", "component", "登录"],
                expected_type="react+component",
                category="React开发"
            ),
            TestCase(
                id="react_form_02",
                name="React表单验证",
                original_requirement="在React项目中实现表单验证功能，包括用户输入校验",
                ai_converted_query="react表单验证",
                expected_keywords=["react", "表单", "验证"],
                expected_type="react+表单",
                category="React开发"
            ),
            TestCase(
                id="react_ts_03",
                name="React TypeScript组件",
                original_requirement="用TypeScript开发React组件，需要类型安全",
                ai_converted_query="react typescript组件",
                expected_keywords=["react", "typescript", "component"],
                expected_type="react+typescript",
                category="React开发"
            ),

            # Vue组件开发场景
            TestCase(
                id="vue_comp_04",
                name="Vue组件开发",
                original_requirement="使用Vue3开发一个数据展示组件",
                ai_converted_query="vue组件开发",
                expected_keywords=["vue", "component"],
                expected_type="vue+component",
                category="Vue开发"
            ),
            TestCase(
                id="vue_ts_05",
                name="Vue TypeScript开发",
                original_requirement="Vue3项目中使用TypeScript进行开发",
                ai_converted_query="vue typescript",
                expected_keywords=["vue", "typescript"],
                expected_type="vue+typescript",
                category="Vue开发"
            ),

            # CSS样式开发场景
            TestCase(
                id="css_module_06",
                name="CSS模块化开发",
                original_requirement="项目中使用CSS模块化方案，避免样式冲突",
                ai_converted_query="css模块化",
                expected_keywords=["css", "模块化"],
                expected_type="css+模块化",
                category="样式开发"
            ),
            TestCase(
                id="css_responsive_07",
                name="响应式CSS设计",
                original_requirement="实现移动端适配的响应式布局",
                ai_converted_query="css响应式",
                expected_keywords=["css", "响应式"],
                expected_type="css+响应式",
                category="样式开发"
            ),
            TestCase(
                id="tailwind_08",
                name="Tailwind CSS使用",
                original_requirement="项目中使用Tailwind CSS进行样式开发",
                ai_converted_query="tailwind",
                expected_keywords=["tailwind"],
                expected_type="tailwind",
                category="样式开发"
            ),

            # JavaScript/TypeScript开发场景
            TestCase(
                id="js_function_09",
                name="JavaScript函数开发",
                original_requirement="编写JavaScript工具函数，处理数据转换",
                ai_converted_query="javascript函数",
                expected_keywords=["javascript", "function"],
                expected_type="javascript",
                category="JS开发"
            ),
            TestCase(
                id="ts_interface_10",
                name="TypeScript接口定义",
                original_requirement="定义TypeScript接口和类型，确保类型安全",
                ai_converted_query="typescript接口",
                expected_keywords=["typescript"],
                expected_type="typescript",
                category="TS开发"
            ),

            # 命名规范场景
            TestCase(
                id="naming_var_11",
                name="变量命名规范",
                original_requirement="项目中的变量命名需要统一规范",
                ai_converted_query="变量命名",
                expected_keywords=["变量", "命名"],
                expected_type="变量命名",
                category="命名规范"
            ),
            TestCase(
                id="naming_comp_12",
                name="组件命名规范",
                original_requirement="React/Vue组件的命名规范要求",
                ai_converted_query="组件命名",
                expected_keywords=["组件", "命名"],
                expected_type="组件命名",
                category="命名规范"
            ),

            # 项目结构场景
            TestCase(
                id="project_struct_13",
                name="项目结构规范",
                original_requirement="前端项目的目录结构和文件组织方式",
                ai_converted_query="项目结构",
                expected_keywords=["项目", "结构"],
                expected_type="项目结构",
                category="项目规范"
            ),
            TestCase(
                id="file_naming_14",
                name="文件命名规范",
                original_requirement="项目中文件的命名规范要求",
                ai_converted_query="文件命名",
                expected_keywords=["文件", "命名"],
                expected_type="文件命名",
                category="命名规范"
            ),

            # 代码质量场景
            TestCase(
                id="eslint_15",
                name="ESLint代码检查",
                original_requirement="配置ESLint进行代码质量检查",
                ai_converted_query="eslint",
                expected_keywords=["eslint"],
                expected_type="eslint",
                category="代码质量"
            ),
            TestCase(
                id="comment_16",
                name="代码注释规范",
                original_requirement="项目中的代码注释编写规范",
                ai_converted_query="注释规范",
                expected_keywords=["注释"],
                expected_type="注释",
                category="代码质量"
            ),

            # 边界测试场景
            TestCase(
                id="unknown_tech_17",
                name="未知技术栈",
                original_requirement="使用Angular开发单页应用",
                ai_converted_query="angular开发",
                expected_keywords=[],
                expected_type="default",
                category="边界测试"
            ),
            TestCase(
                id="vague_query_18",
                name="模糊查询",
                original_requirement="前端开发最佳实践",
                ai_converted_query="前端开发",
                expected_keywords=[],
                expected_type="default",
                category="边界测试"
            )
        ]

    def run_single_test(self, test_case: TestCase) -> TestResult:
        """运行单个测试用例"""
        start_time = time.time()

        try:
            # 调用MCP服务
            response = self.engine.get_standards(test_case.ai_converted_query)
            response_time_ms = (time.time() - start_time) * 1000

            # 分析响应结果
            analysis = self._analyze_response(response, test_case)

            return TestResult(
                test_case=test_case,
                response=response,
                response_time_ms=response_time_ms,
                has_common_standard=analysis['has_common'],
                has_matched_standard=analysis['has_matched'],
                matched_keywords=analysis['matched_keywords'],
                accuracy_score=analysis['accuracy'],
                usefulness_score=analysis['usefulness'],
                completeness_score=analysis['completeness'],
                overall_score=analysis['overall']
            )

        except Exception as e:
            return TestResult(
                test_case=test_case,
                response="",
                response_time_ms=(time.time() - start_time) * 1000,
                has_common_standard=False,
                has_matched_standard=False,
                matched_keywords=[],
                accuracy_score=0.0,
                usefulness_score=0.0,
                completeness_score=0.0,
                overall_score=0.0,
                error_message=str(e)
            )

    def _analyze_response(self, response: str, test_case: TestCase) -> Dict[str, Any]:
        """分析响应结果质量"""
        # 检查是否包含公共规范（第122行的内容）
        common_standard_text = "代码风格一致，注释清晰，遵循项目规范"
        has_common = common_standard_text in response

        # 检查是否包含匹配的特定规范
        has_matched = len(response.strip()) > len(common_standard_text) + 10

        # 检查匹配的关键词
        matched_keywords = []
        response_lower = response.lower()
        for keyword in test_case.expected_keywords:
            if keyword.lower() in response_lower:
                matched_keywords.append(keyword)

        # 计算准确性分数 (0-1)
        if test_case.expected_keywords:
            accuracy = len(matched_keywords) / len(test_case.expected_keywords)
        else:
            # 对于边界测试，如果返回了默认规范则认为准确
            accuracy = 1.0 if has_common else 0.0

        # 计算实用性分数 (0-1)
        usefulness = 0.0
        if has_common:
            usefulness += 0.3  # 公共规范提供基础实用性
        if has_matched:
            usefulness += 0.7  # 匹配规范提供主要实用性

        # 计算完整性分数 (0-1)
        completeness = 0.0
        if has_common:
            completeness += 0.4  # 公共规范提供基础完整性
        if has_matched:
            completeness += 0.6  # 匹配规范提供主要完整性

        # 计算总体分数
        overall = (accuracy * 0.4 + usefulness * 0.3 + completeness * 0.3)

        return {
            'has_common': has_common,
            'has_matched': has_matched,
            'matched_keywords': matched_keywords,
            'accuracy': round(accuracy, 2),
            'usefulness': round(usefulness, 2),
            'completeness': round(completeness, 2),
            'overall': round(overall, 2)
        }

    def run_all_tests(self) -> None:
        """运行所有测试用例"""
        print("\n" + "="*80)
        print("MCP服务真实场景测试开始")
        print("="*80)

        for i, test_case in enumerate(self.test_cases, 1):
            print(f"\n[{i}/{len(self.test_cases)}] 测试: {test_case.name}")
            print(f"原始需求: {test_case.original_requirement}")
            print(f"AI查询: {test_case.ai_converted_query}")

            result = self.run_single_test(test_case)
            self.results.append(result)

            if result.error_message:
                print(f"❌ 测试失败: {result.error_message}")
            else:
                print(f"✅ 响应时间: {result.response_time_ms:.1f}ms")
                print(f"📊 评分: 准确性{result.accuracy_score} | 实用性{result.usefulness_score} | 完整性{result.completeness_score} | 总分{result.overall_score}")
                print(f"🔍 匹配关键词: {result.matched_keywords}")
                print(f"📝 响应预览: {result.response[:100]}...")

    def generate_report(self) -> Dict[str, Any]:
        """生成测试报告"""
        if not self.results:
            return {"error": "没有测试结果"}

        # 基础统计
        total_tests = len(self.results)
        successful_tests = len([r for r in self.results if not r.error_message])
        failed_tests = total_tests - successful_tests

        # 性能统计
        response_times = [r.response_time_ms for r in self.results if not r.error_message]
        avg_response_time = sum(response_times) / len(response_times) if response_times else 0
        max_response_time = max(response_times) if response_times else 0
        min_response_time = min(response_times) if response_times else 0

        # 质量统计
        accuracy_scores = [r.accuracy_score for r in self.results if not r.error_message]
        usefulness_scores = [r.usefulness_score for r in self.results if not r.error_message]
        completeness_scores = [r.completeness_score for r in self.results if not r.error_message]
        overall_scores = [r.overall_score for r in self.results if not r.error_message]

        avg_accuracy = sum(accuracy_scores) / len(accuracy_scores) if accuracy_scores else 0
        avg_usefulness = sum(usefulness_scores) / len(usefulness_scores) if usefulness_scores else 0
        avg_completeness = sum(completeness_scores) / len(completeness_scores) if completeness_scores else 0
        avg_overall = sum(overall_scores) / len(overall_scores) if overall_scores else 0

        # 功能覆盖统计
        common_standard_coverage = len([r for r in self.results if r.has_common_standard]) / total_tests
        matched_standard_coverage = len([r for r in self.results if r.has_matched_standard]) / total_tests

        # 分类统计
        category_stats = {}
        for result in self.results:
            category = result.test_case.category
            if category not in category_stats:
                category_stats[category] = {'total': 0, 'success': 0, 'avg_score': 0}

            category_stats[category]['total'] += 1
            if not result.error_message:
                category_stats[category]['success'] += 1
                category_stats[category]['avg_score'] += result.overall_score

        # 计算分类平均分
        for category, stats in category_stats.items():
            if stats['success'] > 0:
                stats['avg_score'] = round(stats['avg_score'] / stats['success'], 2)
            stats['success_rate'] = round(stats['success'] / stats['total'], 2)

        return {
            'summary': {
                'total_tests': total_tests,
                'successful_tests': successful_tests,
                'failed_tests': failed_tests,
                'success_rate': round(successful_tests / total_tests, 2)
            },
            'performance': {
                'avg_response_time_ms': round(avg_response_time, 1),
                'max_response_time_ms': round(max_response_time, 1),
                'min_response_time_ms': round(min_response_time, 1)
            },
            'quality_scores': {
                'avg_accuracy': round(avg_accuracy, 2),
                'avg_usefulness': round(avg_usefulness, 2),
                'avg_completeness': round(avg_completeness, 2),
                'avg_overall': round(avg_overall, 2)
            },
            'feature_coverage': {
                'common_standard_coverage': round(common_standard_coverage, 2),
                'matched_standard_coverage': round(matched_standard_coverage, 2)
            },
            'category_performance': category_stats,
            'recommendations': self._generate_recommendations()
        }

    def _generate_recommendations(self) -> List[str]:
        """生成改进建议"""
        recommendations = []

        if not self.results:
            return ["无法生成建议：没有测试结果"]

        # 分析成功率
        success_rate = len([r for r in self.results if not r.error_message]) / len(self.results)
        if success_rate < 0.9:
            recommendations.append(f"系统稳定性需要改进，当前成功率仅为{success_rate:.1%}")

        # 分析响应时间
        response_times = [r.response_time_ms for r in self.results if not r.error_message]
        if response_times:
            avg_time = sum(response_times) / len(response_times)
            if avg_time > 100:
                recommendations.append(f"响应时间偏慢，平均{avg_time:.1f}ms，建议优化查询性能")

        # 分析准确性
        accuracy_scores = [r.accuracy_score for r in self.results if not r.error_message]
        if accuracy_scores:
            avg_accuracy = sum(accuracy_scores) / len(accuracy_scores)
            if avg_accuracy < 0.7:
                recommendations.append(f"匹配准确性偏低，平均{avg_accuracy:.1%}，建议优化关键词匹配算法")

        # 分析公共规范覆盖
        common_coverage = len([r for r in self.results if r.has_common_standard]) / len(self.results)
        if common_coverage < 0.8:
            recommendations.append(f"公共规范覆盖率偏低，仅为{common_coverage:.1%}，建议检查公共规范配置")

        # 分析匹配规范覆盖
        matched_coverage = len([r for r in self.results if r.has_matched_standard]) / len(self.results)
        if matched_coverage < 0.6:
            recommendations.append(f"匹配规范覆盖率偏低，仅为{matched_coverage:.1%}，建议扩充规范库")

        if not recommendations:
            recommendations.append("系统表现良好，建议继续保持当前配置")

        return recommendations

    def print_detailed_report(self) -> None:
        """打印详细测试报告"""
        report = self.generate_report()

        print("\n" + "="*80)
        print("详细测试报告")
        print("="*80)

        # 总体统计
        print("\n📊 总体统计:")
        summary = report['summary']
        print(f"  总测试数: {summary['total_tests']}")
        print(f"  成功测试: {summary['successful_tests']}")
        print(f"  失败测试: {summary['failed_tests']}")
        print(f"  成功率: {summary['success_rate']:.1%}")

        # 性能统计
        print("\n⚡ 性能统计:")
        perf = report['performance']
        print(f"  平均响应时间: {perf['avg_response_time_ms']}ms")
        print(f"  最大响应时间: {perf['max_response_time_ms']}ms")
        print(f"  最小响应时间: {perf['min_response_time_ms']}ms")

        # 质量评分
        print("\n🎯 质量评分:")
        quality = report['quality_scores']
        print(f"  平均准确性: {quality['avg_accuracy']:.2f}/1.00")
        print(f"  平均实用性: {quality['avg_usefulness']:.2f}/1.00")
        print(f"  平均完整性: {quality['avg_completeness']:.2f}/1.00")
        print(f"  平均总分: {quality['avg_overall']:.2f}/1.00")

        # 功能覆盖
        print("\n🔍 功能覆盖:")
        coverage = report['feature_coverage']
        print(f"  公共规范覆盖率: {coverage['common_standard_coverage']:.1%}")
        print(f"  匹配规范覆盖率: {coverage['matched_standard_coverage']:.1%}")

        # 分类性能
        print("\n📂 分类性能:")
        for category, stats in report['category_performance'].items():
            print(f"  {category}: 成功率{stats['success_rate']:.1%}, 平均分{stats['avg_score']:.2f}")

        # 改进建议
        print("\n💡 改进建议:")
        for i, rec in enumerate(report['recommendations'], 1):
            print(f"  {i}. {rec}")

        # 失败案例分析
        failed_results = [r for r in self.results if r.error_message]
        if failed_results:
            print("\n❌ 失败案例分析:")
            for result in failed_results:
                print(f"  - {result.test_case.name}: {result.error_message}")

        # 低分案例分析
        low_score_results = [r for r in self.results if not r.error_message and r.overall_score < 0.5]
        if low_score_results:
            print("\n⚠️ 低分案例分析:")
            for result in low_score_results:
                print(f"  - {result.test_case.name}: 总分{result.overall_score:.2f}, 查询'{result.test_case.ai_converted_query}'")


def main():
    """主函数"""
    print("MCP服务真实场景测试工具")
    print("模拟用户开发需求，评估MCP服务返回结果质量")

    try:
        # 创建测试器
        tester = MCPServiceTester()

        # 运行测试
        tester.run_all_tests()

        # 生成报告
        tester.print_detailed_report()

        # 保存JSON报告
        report = tester.generate_report()
        report_file = project_root / "test_report.json"
        with open(report_file, 'w', encoding='utf-8') as f:
            json.dump(report, f, ensure_ascii=False, indent=2)

        print(f"\n📄 详细报告已保存至: {report_file}")

    except Exception as e:
        print(f"\n❌ 测试执行失败: {e}")
        import traceback
        traceback.print_exc()
        return 1

    return 0


if __name__ == "__main__":
    sys.exit(main())