compliance/test_well_data_integration.py

#!/usr/bin/env python3
"""
测试井数据集成和DMS请求体格式的脚本
"""

import sys
import logging
import json
from ddms_compliance_suite.utils.data_generator import DataGenerator
from ddms_compliance_suite.utils.well_data_manager import WellDataManager

def test_well_data_integration():
    """测试井数据集成功能"""
    
    # 设置日志
    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
    logger = logging.getLogger(__name__)
    
    logger.info("开始测试井数据集成功能...")
    
    # 模拟井数据管理器（因为无法连接真实API）
    class MockWellDataManager:
        def __init__(self):
            self.well_data = [
                {"wellId": "HB00019975", "wellCommonName": "郑4-106"},
                {"wellId": "HB00019976", "wellCommonName": "郑4-107"}
            ]
            self.wellbore_data = [
                {"wellboreId": "WEBHHB100083169", "wellboreCommonName": "郑4-106主井筒"},
                {"wellboreId": "WEBHHB100083170", "wellboreCommonName": "郑4-107主井筒"}
            ]
        
        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
    
    # 创建模拟的井数据管理器
    mock_well_manager = MockWellDataManager()
    
    # 创建数据生成器
    data_generator = DataGenerator(logger_param=logger, well_data_manager=mock_well_manager)
    
    # 测试1: 测试DMS创建请求体格式
    logger.info("1. 测试DMS创建请求体格式...")
    create_schema = {
        "type": "object",
        "properties": {
            "version": {"type": "string", "example": "1.0.0"},
            "act": {"type": "integer", "example": -1},
            "data": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "wellId": {"type": "string"},
                        "wellboreId": {"type": "string"},
                        "wellCommonName": {"type": "string"},
                        "dataRegion": {"type": "string"},
                        "bsflag": {"type": "integer"}
                    }
                }
            }
        },
        "required": ["data"]
    }
    
    create_data = data_generator.generate_data_from_schema(create_schema, "create_request", "create_test")
    logger.info(f"创建请求体: {json.dumps(create_data, ensure_ascii=False, indent=2)}")
    
    # 验证创建请求体格式
    assert "version" in create_data, "创建请求体缺少version字段"
    assert "act" in create_data, "创建请求体缺少act字段"
    assert "data" in create_data, "创建请求体缺少data字段"
    assert create_data["act"] == -1, f"创建请求体act字段值错误，期望-1，实际{create_data['act']}"
    
    # 验证井数据是否被正确应用
    if create_data["data"] and len(create_data["data"]) > 0:
        first_item = create_data["data"][0]
        if "wellId" in first_item:
            assert first_item["wellId"] == "HB00019975", f"wellId未使用真实值，实际值: {first_item['wellId']}"
            logger.info("✅ wellId使用了真实值")
        if "wellCommonName" in first_item:
            assert first_item["wellCommonName"] == "郑4-106", f"wellCommonName未使用真实值，实际值: {first_item['wellCommonName']}"
            logger.info("✅ wellCommonName使用了真实值")
    
    # 测试2: 测试DMS删除请求体格式
    logger.info("2. 测试DMS删除请求体格式...")
    delete_schema = {
        "type": "object",
        "properties": {
            "version": {"type": "string", "example": "1.0.0"},
            "data": {
                "type": "array",
                "items": {"type": "string"}
            }
        },
        "required": ["data"]
    }
    
    delete_data = data_generator.generate_data_from_schema(delete_schema, "delete_request", "delete_test")
    logger.info(f"删除请求体: {json.dumps(delete_data, ensure_ascii=False, indent=2)}")
    
    # 验证删除请求体格式
    assert "version" in delete_data, "删除请求体缺少version字段"
    assert "data" in delete_data, "删除请求体缺少data字段"
    assert delete_data["version"] == "1.0.0", f"删除请求体version字段值错误，期望1.0.0，实际{delete_data['version']}"
    
    # 测试3: 测试查询请求体格式
    logger.info("3. 测试查询请求体格式...")
    list_schema = {
        "type": "object",
        "properties": {
            "version": {"type": "string", "example": "1.0.0"},
            "isSearchCount": {"type": "boolean", "example": True},
            "query": {
                "type": "object",
                "properties": {
                    "fields": {"type": "array", "items": {"type": "string"}},
                    "filter": {"type": "object"}
                }
            }
        }
    }
    
    list_data = data_generator.generate_data_from_schema(list_schema, "list_request", "list_test")
    logger.info(f"查询请求体: {json.dumps(list_data, ensure_ascii=False, indent=2)}")
    
    # 验证查询请求体格式
    assert "version" in list_data, "查询请求体缺少version字段"
    assert list_data["version"] == "1.0.0", f"查询请求体version字段值错误，期望1.0.0，实际{list_data['version']}"
    
    logger.info("🎉 所有测试通过！")
    return True

if __name__ == "__main__":
    try:
        success = test_well_data_integration()
        sys.exit(0 if success else 1)
    except Exception as e:
        print(f"❌ 测试失败: {e}")
        import traceback
        traceback.print_exc()
        sys.exit(1)