""" 测试 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"])