#!/usr/bin/env python3 """ 测试CRUD场景预构建请求体的脚本 """ import sys import logging import json def test_crud_prebuilt_requests(): """测试CRUD场景预构建请求体功能""" # 设置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) logger.info("开始测试CRUD场景预构建请求体功能...") # 模拟井数据管理器 class MockWellDataManager: def __init__(self): self.well_data = [ {"wellId": "HB00019975", "wellCommonName": "郑4-106"}, ] self.wellbore_data = [ {"wellboreId": "WEBHHB100083169", "wellboreCommonName": "郑4-106主井筒"}, ] def is_well_related_field(self, field_name): return field_name in ['wellId', 'wellboreId', 'wellCommonName'] def get_well_value_for_field(self, field_name): if field_name == 'wellId': return self.well_data[0]['wellId'] elif field_name == 'wellboreId': return self.wellbore_data[0]['wellboreId'] elif field_name == 'wellCommonName': return self.well_data[0]['wellCommonName'] return None def enhance_data_with_well_values(self, data): if not isinstance(data, dict): return data enhanced_data = data.copy() for field_name, value in data.items(): if self.is_well_related_field(field_name): real_value = self.get_well_value_for_field(field_name) if real_value is not None: enhanced_data[field_name] = real_value logger.info(f"🔄 替换字段 '{field_name}': {value} -> {real_value}") return enhanced_data # 模拟编排器 class MockOrchestrator: def __init__(self): self.well_data_manager = MockWellDataManager() # 测试1: 模拟CRUD场景的预构建请求体逻辑 logger.info("1. 测试CRUD场景预构建请求体逻辑...") # 模拟LLM生成的原始数据 llm_generated_data = { 'seq': 1, 'tvd': 1500.5, 'dsid': 'd8682d8c-86c2-47b4-b28a-90f3408487de', 'XAxis': 123456.78, 'YAxis': 987654.32, 'phase': '开发阶段', 'bsflag': 1, 'tester': '张三', 'wellId': 'WELL001', # LLM生成的模拟值 'azimuth': 45.6, 'phaseId': 'PHASE001', 'remarks': '测斜数据正常', 'testType': '常规测斜', 'avgDogleg': 3.2, 'checkDate': '2023-05-15T10:30:00', 'dataGroup': '测斜数据组', 'startDate': '2023-05-10', 'versionNo': '1.0.0', 'createDate': '2023-05-15T09:00:00', 'dataRegion': '大庆油田', 'testTypeId': 'SURVEY001', 'updateDate': '2023-05-15T09:00:00', 'wellboreId': 'WELLBORE001', # LLM生成的模拟值 'checkUserId': 'USER001', 'createAppId': 'DMS系统', 'description': '测斜轨迹测斜记录', 'createUserId': 'USER001', 'updateUserId': 'USER001', 'angleDeviation': 12.3, 'closureAzimuth': 46.2, 'directionalWay': '定向钻井', 'wellCommonName': '大庆1号井', # LLM生成的模拟值 'closureDistance': 1200.5, 'horiDisplacemen': 800.75, 'surveypointStep': 30, 'avgRateDeviation': 1.5, 'eastWestDisplace': 600.25, 'sourceCreateDate': '2023-05-10T08:00:00', 'surveyPointDepth': 1500, 'southNorthDisplace': 700.5, 'wellboreCommonName': '大庆1号井筒', # LLM生成的模拟值 'overallAngleChangeRate': 2.8, 'rateDirectionChangeMean': 1.2 } logger.info(f"LLM生成的原始数据: {json.dumps(llm_generated_data, ensure_ascii=False, indent=2)}") # 模拟井数据增强 mock_orchestrator = MockOrchestrator() create_payload = llm_generated_data.copy() # 应用井数据增强 if mock_orchestrator.well_data_manager: create_payload = mock_orchestrator.well_data_manager.enhance_data_with_well_values(create_payload) logger.info(f"LLM生成数据已通过井数据管理器增强") logger.info(f"增强后的创建数据: {json.dumps(create_payload, ensure_ascii=False, indent=2)}") # 生成更新数据 import copy update_payload = copy.deepcopy(create_payload) update_payload["description"] = "updated-test-entry-from-scenario" logger.info(f"更新数据: {json.dumps(update_payload, ensure_ascii=False, indent=2)}") # 预构建请求体 create_request_body = { "version": "1.0.0", "act": -1, "data": [create_payload] } update_request_body = { "version": "1.0.0", "act": -1, "data": [update_payload] } logger.info(f"预构建创建请求体: {json.dumps(create_request_body, ensure_ascii=False, indent=2)}") logger.info(f"预构建更新请求体: {json.dumps(update_request_body, ensure_ascii=False, indent=2)}") # 测试2: 验证预构建请求体的正确性 logger.info("2. 验证预构建请求体的正确性...") # 验证创建请求体 create_checks = [ ("version字段", create_request_body.get("version") == "1.0.0"), ("act字段", create_request_body.get("act") == -1), ("data字段存在", "data" in create_request_body), ("data是数组", isinstance(create_request_body.get("data"), list)), ("data数组有内容", len(create_request_body.get("data", [])) > 0), ("wellId被替换", create_request_body["data"][0].get("wellId") == "HB00019975"), ("wellboreId被替换", create_request_body["data"][0].get("wellboreId") == "WEBHHB100083169"), ("wellCommonName被替换", create_request_body["data"][0].get("wellCommonName") == "郑4-106"), ("非井字段保持", create_request_body["data"][0].get("dataRegion") == "大庆油田"), ] all_create_passed = True for check_name, result in create_checks: if result: logger.info(f"✅ 创建请求体 - {check_name}: 通过") else: logger.error(f"❌ 创建请求体 - {check_name}: 失败") all_create_passed = False # 验证更新请求体 update_checks = [ ("version字段", update_request_body.get("version") == "1.0.0"), ("act字段", update_request_body.get("act") == -1), ("data字段存在", "data" in update_request_body), ("data是数组", isinstance(update_request_body.get("data"), list)), ("data数组有内容", len(update_request_body.get("data", [])) > 0), ("wellId被替换", update_request_body["data"][0].get("wellId") == "HB00019975"), ("wellboreId被替换", update_request_body["data"][0].get("wellboreId") == "WEBHHB100083169"), ("wellCommonName被替换", update_request_body["data"][0].get("wellCommonName") == "郑4-106"), ("description被更新", update_request_body["data"][0].get("description") == "updated-test-entry-from-scenario"), ("非井字段保持", update_request_body["data"][0].get("dataRegion") == "大庆油田"), ] all_update_passed = True for check_name, result in update_checks: if result: logger.info(f"✅ 更新请求体 - {check_name}: 通过") else: logger.error(f"❌ 更新请求体 - {check_name}: 失败") all_update_passed = False # 测试3: 模拟curl命令生成 logger.info("3. 模拟curl命令生成...") create_curl = f"""curl -X POST -H "Content-Type: application/json" \\ -d '{json.dumps(create_request_body, ensure_ascii=False)}' \\ --insecure \\ https://www.dev.ideas.cnpc/api/dms/well_kd_wellbore_ideas01/v1/dr_ach_survey_inc""" update_curl = f"""curl -X PUT -H "Content-Type: application/json" \\ -d '{json.dumps(update_request_body, ensure_ascii=False)}' \\ --insecure \\ https://www.dev.ideas.cnpc/api/dms/well_kd_wellbore_ideas01/v1/dr_ach_survey_inc""" logger.info(f"创建curl命令:\n{create_curl}") logger.info(f"更新curl命令:\n{update_curl}") # 验证curl命令中的井数据 curl_checks = [ ("创建curl包含真实wellId", "HB00019975" in create_curl), ("创建curl包含真实wellboreId", "WEBHHB100083169" in create_curl), ("创建curl包含真实wellCommonName", "郑4-106" in create_curl), ("创建curl不包含模拟wellId", "WELL001" not in create_curl), ("更新curl包含真实wellId", "HB00019975" in update_curl), ("更新curl包含真实wellboreId", "WEBHHB100083169" in update_curl), ("更新curl包含真实wellCommonName", "郑4-106" in update_curl), ("更新curl不包含模拟wellId", "WELL001" not in update_curl), ] all_curl_passed = True for check_name, result in curl_checks: if result: logger.info(f"✅ curl命令 - {check_name}: 通过") else: logger.error(f"❌ curl命令 - {check_name}: 失败") all_curl_passed = False # 总结 if all_create_passed and all_update_passed and all_curl_passed: logger.info("🎉 所有测试通过！CRUD场景预构建请求体功能正常工作") return True else: logger.error("❌ 部分测试失败") return False if __name__ == "__main__": try: success = test_crud_prebuilt_requests() sys.exit(0 if success else 1) except Exception as e: print(f"❌ 测试失败: {e}") import traceback traceback.print_exc() sys.exit(1)