LQAgentPlatform/utils_test/AI_Review_Test/test_outline_check.py

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

import asyncio
import sys
import os
from pathlib import Path

# 添加项目根目录到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.ai_review_engine import AIReviewEngine


class TestOutlineCheck:
    """大纲审查功能测试类"""

    def __init__(self):
        self.ai_review_engine = None

    def setup(self):
        """设置测试环境"""
        try:
            self.ai_review_engine = AIReviewEngine()
            print("AI审查引擎初始化成功")
        except Exception as e:
            print(f"AI审查引擎初始化失败: {str(e)}")

    def get_test_outline_data(self):
        """获取测试大纲数据"""
        return {
            "outline": {
                "chapters": [
                    {
                        "index": 1,
                        "title": "第一章编制依据",
                        "page": "5",
                        "original": "第一章编制依据......5",
                        "subsections": [
                            {
                                "title": "一、编制依据",
                                "page": "5",
                                "level": 2,
                                "original": "一、编制依据......5"
                            },
                            {
                                "title": "二、编制原则",
                                "page": "7",
                                "level": 2,
                                "original": "二、编制原则......7"
                            },
                            {
                                "title": "三、编制范围",
                                "page": "7",
                                "level": 2,
                                "original": "三、编制范围......7"
                            }
                        ]
                    },
                    {
                        "index": 2,
                        "title": "第二章工程概况",
                        "page": "9",
                        "original": "第二章工程概况......9",
                        "subsections": [
                            {
                                "title": "一、项目背景",
                                "page": "9",
                                "level": 2,
                                "original": "一、项目背景......9"
                            },
                            {
                                "title": "二、工程特点",
                                "page": "10",
                                "level": 2,
                                "original": "二、工程特点......10"
                            }
                        ]
                    },
                    {
                        "index": 3,
                        "title": "第三章施工计划",
                        "page": "15",
                        "original": "第三章施工计划......15",
                        "subsections": [
                            {
                                "title": "一、施工进度计划",
                                "page": "15",
                                "level": 2,
                                "original": "一、施工进度计划......15"
                            }
                        ]
                    }
                ],
                "total_chapters": 3
            }
        }

    def get_empty_outline_data(self):
        """获取空大纲测试数据"""
        return {
            "outline": {
                "chapters": [],
                "total_chapters": 0
            }
        }

    def get_malformed_outline_data(self):
        """获取格式错误的大纲测试数据"""
        return {
            "outline": {
                "chapters": [
                    {
                        "title": "第一章编制依据"
                        # 缺少必要字段
                    }
                ]
            }
        }

    async def test_outline_check_method(self):
        """测试outline_check方法的大纲提取功能"""
        print("\n=== 测试outline_check方法 ===")

        if not self.ai_review_engine:
            print("AI审查引擎未初始化，跳过测试")
            return False

        test_data = self.get_test_outline_data()
        print("输入大纲章节数量:", len(test_data.get('outline', {}).get('chapters', [])))

        try:
            # 调用outline_check方法
            result = await self.ai_review_engine.outline_check(
                trace_id_idx="test_outline_001",
                outline_content=test_data,
                state={},
                stage_name="大纲审查测试"
            )

            print("outline_check方法调用成功")
            print(f"返回结果类型: {type(result)}")

            # 检查返回结果并输出提取的大纲内容
            if result and isinstance(result, dict):
                print("返回结果的键:", list(result.keys()))

                # 输出整体大纲
                if 'overall_outline' in result:
                    print("\n=== 整体大纲提取结果 ===")
                    print(result['overall_outline'])
                    print(f"整体大纲长度: {len(result['overall_outline'])}")

                # 输出详细大纲
                if 'detailed_outline' in result:
                    print("\n=== 详细大纲提取结果 ===")
                    print(result['overall_outline'])
                    print(f"详细大纲长度: {len(result['detailed_outline'])}")

                # 输出审查结果信息
                if 'review_result' in result:
                    print(f"\n=== 审查结果信息 ===")
                    print(f"审查结果类型: {type(result['review_result'])}")
                    if result['review_result']:
                        print(f"审查结果键: {list(result['review_result'].keys()) if isinstance(result['review_result'], dict) else '非字典类型'}")
                    else:
                        print("审查结果为None")

            else:
                print("注意: 返回结果为None或非字典类型")

            print("PASS outline_check方法测试通过")
            return True

        except Exception as e:
            print(f"FAIL outline_check方法测试失败: {str(e)}")
            return False

    def test_outline_check_sync(self):
        """同步版本的outline_check测试"""
        print("\n=== 测试outline_check同步调用 ===")

        try:
            # 使用asyncio.run来运行异步方法
            result = asyncio.run(self.test_outline_check_method())
            return result
        except Exception as e:
            print(f"FAIL 同步调用测试失败: {str(e)}")
            return False

    def test_empty_outline(self):
        """测试空大纲处理"""
        print("\n=== 测试空大纲处理 ===")

        test_data = self.get_empty_outline_data()
        chapters = test_data.get('outline', {}).get('chapters', [])

        # 模拟提取逻辑
        overall_outline = ""
        for chapter in chapters:
            overall_outline += f"{chapter['title']} (页码: {chapter['page']})\n"

        detailed_outline = ""
        for chapter in chapters:
            detailed_outline += f"\n=== {chapter['title']} ===\n"
            detailed_outline += f"页码: {chapter['page']}\n"

        print("空大纲提取结果:")
        print(f"整体大纲长度: {len(overall_outline)}")
        print(f"详细大纲长度: {len(detailed_outline)}")

        success = len(overall_outline) == 0 and len(detailed_outline) == 0

        if success:
            print("PASS 空大纲处理测试通过")
        else:
            print("FAIL 空大纲处理测试失败")

        return success

    def test_malformed_outline(self):
        """测试格式错误大纲的处理"""
        print("\n=== 测试格式错误大纲处理 ===")

        test_data = self.get_malformed_outline_data()
        chapters = test_data.get('outline', {}).get('chapters', [])

        try:
            # 模拟提取逻辑，应该能处理缺失字段
            overall_outline = ""
            for chapter in chapters:
                title = chapter.get('title', '未知标题')
                page = chapter.get('page', '未知页码')
                overall_outline += f"{title} (页码: {page})\n"

            print("格式错误大纲处理结果:")
            print(overall_outline)
            print("PASS 格式错误大纲处理测试通过")
            return True

        except Exception as e:
            print(f"FAIL 格式错误大纲处理测试失败: {str(e)}")
            return False

    
    def run_all_tests(self):
        """运行所有测试"""
        print("开始大纲审查功能测试")
        print("=" * 50)

        # 设置测试环境
        self.setup()

        # 运行测试
        test_results = []

        # 主要测试：调用outline_check方法
        test_results.append(self.test_outline_check_sync())

        # 辅助测试：测试边界情况
        test_results.append(self.test_empty_outline())
        test_results.append(self.test_malformed_outline())

        # 统计测试结果
        passed = sum(test_results)
        total = len(test_results)

        print("\n" + "=" * 50)
        print(f"测试完成: {passed}/{total} 通过")

        if passed == total:
            print("PASS 所有测试通过")
        else:
            print(f"FAIL {total - passed} 个测试失败")

        return passed == total


def main():
    """主函数"""
    tester = TestOutlineCheck()
    success = tester.run_all_tests()
    sys.exit(0 if success else 1)


if __name__ == "__main__":
    main()