280 lines
11 KiB
Python
280 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
测试最终井数据修复的脚本
|
|
"""
|
|
|
|
import sys
|
|
import logging
|
|
import json
|
|
|
|
def test_final_well_data_fix():
|
|
"""测试最终的井数据修复方案"""
|
|
|
|
# 设置日志
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
logger = logging.getLogger(__name__)
|
|
|
|
logger.info("开始测试最终井数据修复方案...")
|
|
|
|
# 模拟井数据管理器
|
|
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()
|
|
|
|
# 模拟CRUD场景的before_stage逻辑
|
|
class MockCrudStage:
|
|
def __init__(self):
|
|
self.orchestrator = MockOrchestrator()
|
|
self.logger = logger
|
|
|
|
# 测试1: 模拟完整的before_stage流程
|
|
logger.info("1. 测试完整的before_stage流程...")
|
|
|
|
# 模拟LLM生成的原始数据
|
|
llm_generated_data = {
|
|
'seq': 1,
|
|
'tvd': 1500.5,
|
|
'dsid': '0c61f6b0-d845-492a-a0e4-110705fac142',
|
|
'XAxis': 123456.78,
|
|
'YAxis': 987654.32,
|
|
'phase': '开发阶段',
|
|
'bsflag': 1,
|
|
'tester': '张三',
|
|
'wellId': 'WELL001', # LLM生成的模拟值
|
|
'azimuth': 45.6,
|
|
'phaseId': 'DEV',
|
|
'remarks': '测斜数据正常',
|
|
'testType': '常规测斜',
|
|
'avgDogleg': 2.5,
|
|
'checkDate': '2023-10-15T10:30:00',
|
|
'dataGroup': '测斜数据组',
|
|
'startDate': '2023-10-01',
|
|
'versionNo': '1.0',
|
|
'createDate': '2023-10-15T09:00:00',
|
|
'dataRegion': '大庆油田',
|
|
'testTypeId': 'NORMAL',
|
|
'updateDate': '2023-10-15T09:00:00',
|
|
'wellboreId': 'WB001', # LLM生成的模拟值
|
|
'checkUserId': 'USER001',
|
|
'createAppId': 'DMS系统',
|
|
'description': '测斜轨迹测斜记录',
|
|
'createUserId': 'USER001',
|
|
'updateUserId': 'USER001',
|
|
'angleDeviation': 15.3,
|
|
'closureAzimuth': 46.2,
|
|
'directionalWay': '定向钻井',
|
|
'wellCommonName': '大庆1井', # LLM生成的模拟值
|
|
'closureDistance': 1200.75,
|
|
'horiDisplacemen': 800.25,
|
|
'surveypointStep': 30,
|
|
'avgRateDeviation': 1.2,
|
|
'eastWestDisplace': 600.5,
|
|
'sourceCreateDate': '2023-10-01T08:00:00',
|
|
'surveyPointDepth': 1500,
|
|
'southNorthDisplace': 700.3,
|
|
'wellboreCommonName': '大庆1井-主井筒', # LLM生成的模拟值
|
|
'overallAngleChangeRate': 1.8,
|
|
'rateDirectionChangeMean': 0.9
|
|
}
|
|
|
|
logger.info(f"LLM生成的原始数据: {json.dumps(llm_generated_data, ensure_ascii=False, indent=2)}")
|
|
|
|
# 模拟CRUD场景的before_stage逻辑
|
|
crud_stage = MockCrudStage()
|
|
create_payload = llm_generated_data.copy()
|
|
|
|
# 🔑 关键修复：应用井数据增强到创建数据
|
|
if hasattr(crud_stage, 'orchestrator') and hasattr(crud_stage.orchestrator, 'well_data_manager') and crud_stage.orchestrator.well_data_manager:
|
|
create_payload = crud_stage.orchestrator.well_data_manager.enhance_data_with_well_values(create_payload)
|
|
logger.info(f"创建数据已通过井数据管理器增强")
|
|
|
|
# 更新负载基于创建负载，但修改描述字段
|
|
import copy
|
|
update_payload = copy.deepcopy(create_payload)
|
|
update_payload["description"] = "updated-test-entry-from-scenario"
|
|
|
|
logger.info(f"井数据增强后的创建数据: {json.dumps(create_payload, ensure_ascii=False, indent=2)}")
|
|
logger.info(f"井数据增强后的更新数据: {json.dumps(update_payload, ensure_ascii=False, indent=2)}")
|
|
|
|
# 模拟stage_context
|
|
stage_context = {
|
|
"pk_name": "dsid",
|
|
"pk_value": "0c61f6b0-d845-492a-a0e4-110705fac142",
|
|
"current_payload": create_payload,
|
|
"update_payload": update_payload
|
|
}
|
|
|
|
# 测试2: 验证请求体模板解析
|
|
logger.info("2. 测试请求体模板解析...")
|
|
|
|
# 模拟测试编排器的模板解析逻辑
|
|
def mock_resolve_template(template, context):
|
|
"""模拟测试编排器的模板解析"""
|
|
if isinstance(template, str) and template == "{{stage_context.current_payload}}":
|
|
return context["current_payload"]
|
|
elif isinstance(template, str) and template == "{{stage_context.update_payload}}":
|
|
return context["update_payload"]
|
|
elif isinstance(template, list):
|
|
return [mock_resolve_template(item, context) for item in template]
|
|
elif isinstance(template, dict):
|
|
return {k: mock_resolve_template(v, context) for k, v in template.items()}
|
|
else:
|
|
return template
|
|
|
|
# 测试创建请求体模板
|
|
create_body_template = {
|
|
"version": "1.0.0",
|
|
"act": -1,
|
|
"data": ["{{stage_context.current_payload}}"]
|
|
}
|
|
|
|
resolved_create_body = mock_resolve_template(create_body_template, stage_context)
|
|
logger.info(f"解析后的创建请求体: {json.dumps(resolved_create_body, ensure_ascii=False, indent=2)}")
|
|
|
|
# 测试更新请求体模板
|
|
update_body_template = {
|
|
"version": "1.0.0",
|
|
"act": -1,
|
|
"data": ["{{stage_context.update_payload}}"]
|
|
}
|
|
|
|
resolved_update_body = mock_resolve_template(update_body_template, stage_context)
|
|
logger.info(f"解析后的更新请求体: {json.dumps(resolved_update_body, ensure_ascii=False, indent=2)}")
|
|
|
|
# 测试3: 验证最终效果
|
|
logger.info("3. 验证最终效果...")
|
|
|
|
# 验证井数据是否被正确应用
|
|
expected_replacements = {
|
|
'wellId': 'HB00019975',
|
|
'wellboreId': 'WEBHHB100083169',
|
|
'wellCommonName': '郑4-106'
|
|
}
|
|
|
|
all_correct = True
|
|
|
|
# 验证创建请求体
|
|
create_data = resolved_create_body["data"][0]
|
|
for field, expected_value in expected_replacements.items():
|
|
if create_data.get(field) == expected_value:
|
|
logger.info(f"✅ 创建请求体 - {field} 正确替换为真实值: {expected_value}")
|
|
else:
|
|
logger.error(f"❌ 创建请求体 - {field} 未正确替换，期望: {expected_value}, 实际: {create_data.get(field)}")
|
|
all_correct = False
|
|
|
|
# 验证更新请求体
|
|
update_data = resolved_update_body["data"][0]
|
|
for field, expected_value in expected_replacements.items():
|
|
if update_data.get(field) == expected_value:
|
|
logger.info(f"✅ 更新请求体 - {field} 正确替换为真实值: {expected_value}")
|
|
else:
|
|
logger.error(f"❌ 更新请求体 - {field} 未正确替换，期望: {expected_value}, 实际: {update_data.get(field)}")
|
|
all_correct = False
|
|
|
|
# 验证DMS格式
|
|
dms_format_checks = [
|
|
("创建请求体version", resolved_create_body.get("version") == "1.0.0"),
|
|
("创建请求体act", resolved_create_body.get("act") == -1),
|
|
("创建请求体data存在", "data" in resolved_create_body),
|
|
("更新请求体version", resolved_update_body.get("version") == "1.0.0"),
|
|
("更新请求体act", resolved_update_body.get("act") == -1),
|
|
("更新请求体data存在", "data" in resolved_update_body),
|
|
("更新description字段", update_data.get("description") == "updated-test-entry-from-scenario"),
|
|
]
|
|
|
|
for check_name, result in dms_format_checks:
|
|
if result:
|
|
logger.info(f"✅ {check_name}: 通过")
|
|
else:
|
|
logger.error(f"❌ {check_name}: 失败")
|
|
all_correct = False
|
|
|
|
# 测试4: 模拟curl命令生成
|
|
logger.info("4. 模拟curl命令生成...")
|
|
|
|
create_curl = f"""curl -X POST -H "Content-Type: application/json" \\
|
|
-d '{json.dumps(resolved_create_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(resolved_update_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),
|
|
]
|
|
|
|
for check_name, result in curl_checks:
|
|
if result:
|
|
logger.info(f"✅ curl命令 - {check_name}: 通过")
|
|
else:
|
|
logger.error(f"❌ curl命令 - {check_name}: 失败")
|
|
all_correct = False
|
|
|
|
# 总结
|
|
if all_correct:
|
|
logger.info("🎉 所有测试通过！最终井数据修复方案正常工作")
|
|
return True
|
|
else:
|
|
logger.error("❌ 部分测试失败")
|
|
return False
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
success = test_final_well_data_fix()
|
|
sys.exit(0 if success else 1)
|
|
except Exception as e:
|
|
print(f"❌ 测试失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1)
|