193 lines
8.5 KiB
Python
193 lines
8.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
直接测试井数据调试的脚本
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import logging
|
|
|
|
# 添加项目根目录到Python路径
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from ddms_compliance_suite.test_orchestrator import APITestOrchestrator
|
|
from ddms_compliance_suite.input_parser.parser import ParsedDMSSpec
|
|
from custom_stages.dms_crud_scenario_stage import DmsCrudScenarioStage
|
|
|
|
def test_well_data_debug():
|
|
"""直接测试井数据调试"""
|
|
|
|
# 设置日志
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
logger = logging.getLogger(__name__)
|
|
|
|
logger.info("开始直接测试井数据调试...")
|
|
|
|
try:
|
|
# 创建一个简单的API规范
|
|
simple_api_spec = {
|
|
"dms_api_list": [
|
|
{
|
|
"method": "POST",
|
|
"path": "/api/dms/well_kd_wellbore_ideas01/v1/dr_ach_survey_inc",
|
|
"title": "Create dr_ach_survey_inc",
|
|
"summary": "Create a new dr_ach_survey_inc record",
|
|
"description": "Create a new dr_ach_survey_inc record",
|
|
"operationId": "create_dr_ach_survey_inc",
|
|
"tags": ["dr_ach_survey_inc"],
|
|
"parameters": [],
|
|
"requestBody": {
|
|
"required": True,
|
|
"content": {
|
|
"application/json": {
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"version": {"type": "string", "example": "1.0.0"},
|
|
"act": {"type": "integer", "example": -1},
|
|
"data": {
|
|
"type": "array",
|
|
"items": {
|
|
"type": "object",
|
|
"properties": {
|
|
"dsid": {"type": "string", "title": "数据ID"},
|
|
"wellId": {"type": "string", "title": "井标识符"},
|
|
"wellboreId": {"type": "string", "title": "井筒唯一标识符"},
|
|
"wellCommonName": {"type": "string", "title": "井名"},
|
|
"dataRegion": {"type": "string", "title": "数据标识"},
|
|
"description": {"type": "string", "title": "测斜描述"},
|
|
"bsflag": {"type": "number", "title": "逻辑删除标识"}
|
|
},
|
|
"required": ["dsid", "dataRegion", "bsflag"]
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
},
|
|
"responses": {
|
|
"200": {
|
|
"description": "Success",
|
|
"content": {
|
|
"application/json": {
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"code": {"type": "integer"},
|
|
"message": {"type": "string"},
|
|
"data": {"type": "object"}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
},
|
|
"_test_mode": "scenario_only",
|
|
"_dms_model_pk_name": "dsid"
|
|
}
|
|
]
|
|
}
|
|
|
|
# 创建DMSEndpoint
|
|
from ddms_compliance_suite.input_parser.parser import DMSEndpoint
|
|
|
|
endpoint = DMSEndpoint(
|
|
method="POST",
|
|
path="/api/dms/well_kd_wellbore_ideas01/v1/dr_ach_survey_inc",
|
|
title="Create dr_ach_survey_inc",
|
|
request_body=simple_api_spec["dms_api_list"][0]["requestBody"],
|
|
responses=simple_api_spec["dms_api_list"][0]["responses"],
|
|
parameters=[],
|
|
test_mode="scenario_only",
|
|
model_pk_name="dsid"
|
|
)
|
|
|
|
# 创建ParsedDMSSpec
|
|
parsed_spec = ParsedDMSSpec(endpoints=[endpoint], spec=simple_api_spec)
|
|
|
|
# 创建测试编排器
|
|
orchestrator = APITestOrchestrator(
|
|
global_api_spec=parsed_spec,
|
|
base_url="https://www.dev.ideas.cnpc",
|
|
logger=logger
|
|
)
|
|
|
|
logger.info(f"测试编排器创建成功")
|
|
logger.info(f"orchestrator.well_data_manager: {getattr(orchestrator, 'well_data_manager', 'NOT_FOUND')}")
|
|
logger.info(f"type(orchestrator.well_data_manager): {type(getattr(orchestrator, 'well_data_manager', None))}")
|
|
|
|
# 创建CRUD场景
|
|
crud_stage = DmsCrudScenarioStage()
|
|
crud_stage.orchestrator = orchestrator
|
|
crud_stage.logger = logger
|
|
|
|
logger.info(f"CRUD场景创建成功")
|
|
logger.info(f"crud_stage.orchestrator: {crud_stage.orchestrator}")
|
|
logger.info(f"hasattr(crud_stage, 'orchestrator'): {hasattr(crud_stage, 'orchestrator')}")
|
|
|
|
if hasattr(crud_stage, 'orchestrator'):
|
|
logger.info(f"hasattr(crud_stage.orchestrator, 'well_data_manager'): {hasattr(crud_stage.orchestrator, 'well_data_manager')}")
|
|
if hasattr(crud_stage.orchestrator, 'well_data_manager'):
|
|
logger.info(f"crud_stage.orchestrator.well_data_manager: {crud_stage.orchestrator.well_data_manager}")
|
|
logger.info(f"type(crud_stage.orchestrator.well_data_manager): {type(crud_stage.orchestrator.well_data_manager)}")
|
|
logger.info(f"bool(crud_stage.orchestrator.well_data_manager): {bool(crud_stage.orchestrator.well_data_manager)}")
|
|
|
|
# 测试条件判断
|
|
condition_result = (hasattr(crud_stage, 'orchestrator') and
|
|
hasattr(crud_stage.orchestrator, 'well_data_manager') and
|
|
crud_stage.orchestrator.well_data_manager)
|
|
|
|
logger.info(f"完整条件判断结果: {condition_result}")
|
|
|
|
if condition_result:
|
|
logger.info("✅ 条件判断通过，井数据管理器可用")
|
|
|
|
# 测试井数据增强
|
|
test_data = {
|
|
'dsid': 'test-123',
|
|
'wellId': 'WELL001',
|
|
'wellboreId': 'WB001',
|
|
'wellCommonName': '大庆1井',
|
|
'dataRegion': '大庆油田',
|
|
'description': '测试数据',
|
|
'bsflag': 1
|
|
}
|
|
|
|
logger.info(f"原始测试数据: {test_data}")
|
|
enhanced_data = crud_stage.orchestrator.well_data_manager.enhance_data_with_well_values(test_data)
|
|
logger.info(f"增强后数据: {enhanced_data}")
|
|
|
|
else:
|
|
logger.warning("❌ 条件判断失败，井数据管理器不可用")
|
|
|
|
# 检查井数据管理器初始化
|
|
logger.info("尝试手动初始化井数据管理器...")
|
|
try:
|
|
orchestrator._initialize_well_data_manager()
|
|
logger.info(f"手动初始化后 - orchestrator.well_data_manager: {orchestrator.well_data_manager}")
|
|
logger.info(f"手动初始化后 - type: {type(orchestrator.well_data_manager)}")
|
|
logger.info(f"手动初始化后 - bool: {bool(orchestrator.well_data_manager)}")
|
|
except Exception as e:
|
|
logger.error(f"手动初始化井数据管理器失败: {e}")
|
|
|
|
logger.info("测试完成")
|
|
|
|
except Exception as e:
|
|
logger.error(f"测试失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
return True
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
success = test_well_data_debug()
|
|
sys.exit(0 if success else 1)
|
|
except Exception as e:
|
|
print(f"❌ 测试失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1)
|