LQAgentPlatform/utils_test/AI_Review_Test/test_outline_review.py

# -*- coding: utf-8 -*-
"""
大纲审查功能单元测试
测试OutlineReviewer.outline_review方法的入参出参
"""

import pytest
import asyncio
import json
from typing import Dict, Any
import sys
import os

# 添加项目根目录到Python路径
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..')))

from core.construction_review.component.reviewers.outline_reviewer import OutlineReviewer


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

    def setup_method(self):
        """测试前置设置"""
        self.outline_reviewer = OutlineReviewer()
        self.test_trace_id = "test_outline_review_001"
        self.test_stage_name = "test_stage"
        self.test_state = {"status": "testing"}

    def load_test_data(self) -> Dict[str, Any]:
        """从测试数据文件加载大纲结构"""
        test_data_path = os.path.join(
            os.path.dirname(__file__), '..', '..', '..',
            'temp', 'document_temp', '文档切分预处理结果.json'
        )

        try:
            with open(test_data_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
                return data.get('outline', {})
        except Exception as e:
            print(f"加载测试数据失败: {e}")
            return self._get_mock_outline_data()

    def _get_mock_outline_data(self) -> Dict[str, Any]:
        """获取模拟大纲数据"""
        return {
            "chapters": [
                {
                    "index": 1,
                    "title": "第一章编制依据",
                    "page": "1",
                    "original": "第一章编制依据...................................................... 1",
                    "subsections": [
                        {
                            "title": "一、编制依据",
                            "page": "1",
                            "level": 2,
                            "original": "一、编制依据................................................................. 1"
                        },
                        {
                            "title": "二、编制原则",
                            "page": "3",
                            "level": 2,
                            "original": "二、编制原则................................................................. 3"
                        }
                    ]
                },
                {
                    "index": 2,
                    "title": "第二章工程概况",
                    "page": "5",
                    "original": "第二章工程概况...................................................... 5",
                    "subsections": [
                        {
                            "title": "一、项目背景",
                            "page": "5",
                            "level": 2,
                            "original": "一、项目背景................................................................. 5"
                        },
                        {
                            "title": "二、工程地质",
                            "page": "6",
                            "level": 2,
                            "original": "二、工程地质................................................................. 6"
                        }
                    ]
                }
            ]
        }

    def prepare_review_data(self, outline_data: Dict[str, Any]) -> Dict[str, Any]:
        """准备审查数据，模拟ai_review_engine.py中的数据处理逻辑"""
        chapters = outline_data.get('chapters', [])

        # 1. 获取整体大纲（1级大纲目录）
        overall_outline = ""
        for chapter in chapters:
            overall_outline += f"{chapter['title']} (页码: {chapter['page']})\n"

        # 2. 获取大纲各章节及其子目录的详细信息
        detailed_outline = []

        for chapter in chapters:
            # 追加章节标题和页码
            detailed_outline.append(f"\n{chapter['title']}\n")
            detailed_outline.append(f"页码: {chapter['page']}")

            # 添加子目录
            subsections = chapter.get('subsections', [])
            if subsections:
                detailed_outline.append("\n")
                for subsection in subsections:
                    indent = "  " * (subsection['level'] - 1)
                    detailed_outline.append(f"{indent}- {subsection['title']} (页码: {subsection['page']})\n")
            else:
                detailed_outline.append("\n")

        return {
            'outline_content': outline_data,
            'overall_outline': overall_outline,
            'detailed_outline': detailed_outline,
            'state': self.test_state,
            'stage_name': self.test_stage_name
        }

    @pytest.mark.asyncio
    async def test_outline_review_basic_functionality(self):
        """测试大纲审查基本功能"""
        print("\n=== 测试大纲审查基本功能 ===")

        # 加载测试数据
        outline_data = self.load_test_data()
        print(f"加载的大纲章节数量: {len(outline_data.get('chapters', []))}")

        # 准备审查数据
        review_data = self.prepare_review_data(outline_data)

        print(f"整体大纲内容预览:")
        print(f"{review_data['overall_outline'][:200]}...")

        print(f"详细大纲项目数量: {len(review_data['detailed_outline'])}")
        for i, item in enumerate(review_data['detailed_outline'][:5]):
            if item.strip():
                print(f"  {i+1}. {item.strip()[:50]}...")

        try:
            # 执行大纲审查
            print("\n开始执行大纲审查...")
            result = await self.outline_reviewer.outline_review(
                review_data, self.test_trace_id, self.test_stage_name, self.test_state
            )

            # 验证结果
            print(f"\n=== 审查结果 ===")
            print(f"审查成功状态: {result.get('success', False)}")
            print(f"执行时间: {result.get('execution_time', 0):.2f}秒")
            print(f"详细项目总数: {result.get('total_detailed_items', 0)}")

            # 阶段1结果
            stage1_result = result.get('stage1_overall_review')
            if stage1_result:
                print(f"\n--- 阶段1：整体大纲审查结果 ---")
                print(f"整体大纲审查成功: {stage1_result.get('success', False)}")
                parsed_overall = stage1_result.get('parsed_result')
                if parsed_overall and isinstance(parsed_overall, list):
                    print(f"十大类章节审查结果数量: {len(parsed_overall)}")
                    for item in parsed_overall[:3]:  # 只显示前3个
                        chapter_category = item.get('章节类别', 'N/A')
                        status = item.get('完整性状态', 'N/A')
                        # 替换特殊字符避免编码问题
                        status_safe = status.replace('✔', '[完整]').replace('⚠', '[部分]').replace('✘', '[缺失]')
                        print(f"  - {chapter_category}: {status_safe}")

            # 阶段2结果
            stage2_result = result.get('stage2_detailed_review', [])
            print(f"\n--- 阶段2：详细大纲逐项审查结果 ---")
            print(f"详细项目审查结果数量: {len(stage2_result)}")

            success_count = 0
            for i, item_result in enumerate(stage2_result[:5]):  # 只显示前5个
                item_index = item_result.get('item_index', i)
                outline_item = item_result.get('outline_item', '')[:50]
                review_result = item_result.get('review_result', {})

                print(f"  项目{item_index+1}: {outline_item}...")
                print(f"    审查成功: {review_result.get('success', False)}")
                if review_result.get('success'):
                    category = review_result.get('category', 'N/A')
                    parsed_result = review_result.get('parsed_result')
                    if parsed_result and isinstance(parsed_result, list):
                        print(f"    分类: {category}")
                        print(f"    审查项目数: {len(parsed_result)}")
                    success_count += 1
                else:
                    print(f"    错误: {review_result.get('error_message', 'N/A')}")

            print(f"\n前5个项目中成功审查数量: {success_count}")

            # 断言基本结果
            assert result.get('success', False), "大纲审查应该成功"
            assert 'stage1_overall_review' in result, "应该包含阶段1审查结果"
            assert 'stage2_detailed_review' in result, "应该包含阶段2审查结果"
            assert isinstance(result.get('stage2_detailed_review', []), list), "阶段2结果应该是列表"

            print(f"\n[测试通过] 大纲审查基本功能正常")

        except Exception as e:
            print(f"\n[测试失败] {str(e)}")
            raise

    @pytest.mark.asyncio
    async def test_outline_review_with_empty_data(self):
        """测试空数据的大纲审查"""
        print("\n=== 测试空数据的大纲审查 ===")

        # 准备空数据
        empty_review_data = {
            'outline_content': {},
            'overall_outline': '',
            'detailed_outline': [],
            'state': self.test_state,
            'stage_name': self.test_stage_name
        }

        try:
            result = await self.outline_reviewer.outline_review(
                empty_review_data, self.test_trace_id, self.test_stage_name, self.test_state
            )

            print(f"空数据审查结果: {result.get('success', False)}")
            print(f"错误信息: {result.get('error_message', 'N/A')}")

            # 空数据应该处理失败
            assert not result.get('success', False), "空数据应该审查失败"
            print(f"\n[测试通过] 空数据正确处理失败")

        except Exception as e:
            print(f"期望的异常: {str(e)}")
            print(f"\n[测试通过] 空数据正确抛出异常")

    def test_data_preparation_logic(self):
        """测试数据准备逻辑"""
        print("\n=== 测试数据准备逻辑 ===")

        # 加载测试数据
        outline_data = self.load_test_data()
        review_data = self.prepare_review_data(outline_data)

        # 验证数据格式
        assert 'overall_outline' in review_data, "应该包含整体大纲"
        assert 'detailed_outline' in review_data, "应该包含详细大纲"
        assert isinstance(review_data['detailed_outline'], list), "详细大纲应该是列表格式"

        print(f"整体大纲长度: {len(review_data['overall_outline'])}")
        print(f"详细大纲项目数: {len(review_data['detailed_outline'])}")

        # 验证整体大纲格式
        overall_lines = review_data['overall_outline'].strip().split('\n')
        print(f"整体大纲行数: {len(overall_lines)}")

        # 验证详细大纲格式
        non_empty_items = [item for item in review_data['detailed_outline'] if item.strip()]
        print(f"非空详细大纲项目数: {len(non_empty_items)}")

        print(f"\n[测试通过] 数据准备逻辑正确")


if __name__ == "__main__":
    # 运行测试
    pytest.main([__file__, "-v", "-s"])