DataOps-platform/tests/test_approve_order.py

"""
测试 approve_order 功能

测试场景：
1. 验证 extract_output_domain_and_logic 方法能正确提取输出域和处理逻辑
2. 验证 generate_order_resources 方法能正确创建资源
3. 验证 approve_order 完整流程
"""

import json
from unittest.mock import MagicMock, patch

import pytest


class TestExtractOutputDomainAndLogic:
    """测试 extract_output_domain_and_logic 方法"""

    @patch("app.core.data_service.data_product_service.current_app")
    @patch("app.core.data_service.data_product_service.OpenAI")
    def test_extract_success(self, mock_openai, mock_app):
        """测试成功提取输出域和处理逻辑"""
        from app.core.data_service.data_product_service import DataOrderService

        # 模拟配置
        mock_app.config.get.side_effect = lambda key: {
            "LLM_API_KEY": "test-key",
            "LLM_BASE_URL": "http://test-url",
            "LLM_MODEL_NAME": "test-model",
        }.get(key)

        # 模拟 LLM 响应
        mock_response = MagicMock()
        mock_response.choices = [MagicMock()]
        mock_response.choices[0].message.content = json.dumps(
            {
                "output_domain": {
                    "name_zh": "会员消费分析报表",
                    "name_en": "member_consumption_analysis",
                    "describe": "汇总会员消费数据的分析报表",
                },
                "processing_logic": "1. 从会员表提取会员ID、姓名；2. 从消费记录表提取消费金额；3. 按会员汇总消费总额",
            }
        )
        mock_openai.return_value.chat.completions.create.return_value = mock_response

        # 执行测试
        result = DataOrderService.extract_output_domain_and_logic(
            description="我需要一个会员消费分析报表，统计每个会员的消费总额",
            input_domains=[{"name_zh": "会员表"}, {"name_zh": "消费记录表"}],
        )

        # 验证结果
        assert "output_domain" in result
        assert result["output_domain"]["name_zh"] == "会员消费分析报表"
        assert result["output_domain"]["name_en"] == "member_consumption_analysis"
        assert "processing_logic" in result
        assert "消费" in result["processing_logic"]

    @patch("app.core.data_service.data_product_service.current_app")
    @patch("app.core.data_service.data_product_service.OpenAI")
    def test_extract_with_markdown_response(self, mock_openai, mock_app):
        """测试 LLM 返回带 markdown 代码块的响应"""
        from app.core.data_service.data_product_service import DataOrderService

        mock_app.config.get.side_effect = lambda key: {
            "LLM_API_KEY": "test-key",
            "LLM_BASE_URL": "http://test-url",
            "LLM_MODEL_NAME": "test-model",
        }.get(key)

        # 模拟带 markdown 代码块的响应
        mock_response = MagicMock()
        mock_response.choices = [MagicMock()]
        mock_response.choices[0].message.content = """```json
{
    "output_domain": {
        "name_zh": "销售报表",
        "name_en": "sales_report",
        "describe": "销售数据汇总"
    },
    "processing_logic": "汇总销售数据"
}
```"""
        mock_openai.return_value.chat.completions.create.return_value = mock_response

        result = DataOrderService.extract_output_domain_and_logic(
            description="生成销售报表"
        )

        assert result["output_domain"]["name_zh"] == "销售报表"

    @patch("app.core.data_service.data_product_service.current_app")
    @patch("app.core.data_service.data_product_service.OpenAI")
    def test_extract_fallback_on_error(self, mock_openai, mock_app):
        """测试 LLM 调用失败时的回退逻辑"""
        from app.core.data_service.data_product_service import DataOrderService

        mock_app.config.get.side_effect = lambda key: {
            "LLM_API_KEY": "test-key",
            "LLM_BASE_URL": "http://test-url",
            "LLM_MODEL_NAME": "test-model",
        }.get(key)

        # 模拟 LLM 调用异常
        mock_openai.return_value.chat.completions.create.side_effect = Exception(
            "LLM 服务不可用"
        )

        result = DataOrderService.extract_output_domain_and_logic(
            description="测试描述内容"
        )

        # 验证回退到默认值
        assert result["output_domain"]["name_zh"] == "数据产品"
        assert result["output_domain"]["name_en"] == "data_product"
        assert result["processing_logic"] == "测试描述内容"
        assert "error" in result


class TestGenerateOrderResources:
    """测试 generate_order_resources 方法"""

    @patch("app.core.data_service.data_product_service.db")
    @patch("app.core.data_service.data_product_service.neo4j_driver")
    @patch.object(
        __import__(
            "app.core.data_service.data_product_service", fromlist=["DataOrderService"]
        ).DataOrderService,
        "extract_output_domain_and_logic",
    )
    def test_generate_resources_creates_all_components(
        self, mock_extract, mock_neo4j, mock_db
    ):
        """测试 generate_order_resources 创建所有必要的组件"""
        from app.core.data_service.data_product_service import DataOrderService
        from app.models.data_product import DataOrder

        # 模拟 LLM 提取结果
        mock_extract.return_value = {
            "output_domain": {
                "name_zh": "测试数据产品",
                "name_en": "test_data_product",
                "describe": "测试描述",
            },
            "processing_logic": "测试处理逻辑",
        }

        # 模拟订单对象
        mock_order = MagicMock(spec=DataOrder)
        mock_order.id = 1
        mock_order.order_no = "DO20240101001"
        mock_order.title = "测试订单"
        mock_order.description = "测试描述"
        mock_order.extraction_purpose = "测试用途"
        mock_order.extracted_fields = ["字段1", "字段2"]
        mock_order.graph_analysis = {
            "matched_domains": [
                {"id": 100, "name_zh": "源表1"},
                {"id": 101, "name_zh": "源表2"},
            ]
        }

        # 模拟 Neo4j session
        mock_session = MagicMock()
        mock_neo4j.get_session.return_value.__enter__ = MagicMock(
            return_value=mock_session
        )
        mock_neo4j.get_session.return_value.__exit__ = MagicMock(return_value=False)

        # 模拟创建 BusinessDomain 返回 ID
        mock_bd_result = MagicMock()
        mock_bd_result.__getitem__ = lambda self, key: 200 if key == "bd_id" else None
        mock_session.run.return_value.single.side_effect = [
            mock_bd_result,  # 创建 BusinessDomain
            MagicMock(
                __getitem__=lambda self, key: 300 if key == "df_id" else None
            ),  # 创建 DataFlow
        ]

        # 模拟 task_list 插入
        mock_db.session.execute.return_value.fetchone.return_value = (1,)

        # 执行测试
        result = DataOrderService.generate_order_resources(mock_order)

        # 验证结果
        assert result["target_business_domain_id"] == 200
        assert result["dataflow_id"] == 300
        assert result["input_domain_ids"] == [100, 101]
        assert "task_id" in result

        # 验证 LLM 提取被调用
        mock_extract.assert_called_once()


class TestApproveOrderFlow:
    """测试完整的 approve_order 流程"""

    def test_approve_order_status_validation(self):
        """测试订单状态验证"""
        # 这个测试需要在实际环境中运行
        # 这里只提供测试框架
        pass

    def test_approve_order_success_flow(self):
        """测试审批成功的完整流程"""
        # 这个测试需要在实际环境中运行
        # 这里只提供测试框架
        pass


if __name__ == "__main__":
    pytest.main([__file__, "-v"])