v0.2

ddms_compliance_suite/utils/common_utils.py

@@ -0,0 +1,51 @@
+# -*- coding: utf-8 -*-
+import logging
+import re
+from typing import Dict, Any
+
+logger = logging.getLogger(__name__)
+
+def format_url_with_path_params(path_template: str, path_params: Dict[str, Any]) -> str:
+    """
+    使用提供的路径参数格式化URL路径模板。
+    例如, path_template='/users/{userId}/items/{itemId}' 
+    path_params={'userId': 123, 'itemId': 'abc'} 
+    -> '/users/123/items/abc'
+
+    Args:
+        path_template: 包含占位符的URL路径，例如 /resource/{id}。
+        path_params: 包含占位符名称及其值的字典。
+
+    Returns:
+        格式化后的URL路径。
+    """
+    url = path_template
+    try:
+        # 优先使用 .format(**path_params) 如果所有占位符都能匹配
+        # 这要求 path_params 中的键与模板中的占位符完全对应
+        # url = path_template.format(**path_params) # 更简洁，但如果参数不完全匹配会报错
+
+        # 使用正则表达式逐个替换更安全，可以处理部分参数或额外参数的情况
+        for param_name, param_value in path_params.items():
+            placeholder = f"{{{param_name}}}"
+            if placeholder in url:
+                url = url.replace(placeholder, str(param_value))
+            else:
+                # Log if a path param was provided but not found in template. Could be optional.
+                logger.debug(f"Path parameter '{param_name}' provided but placeholder '{placeholder}' not found in template '{path_template}'.")
+        
+        # 检查是否还有未替换的占位符 (可选，但推荐)
+        remaining_placeholders = re.findall(r"({[^{}]+?})", url)
+        if remaining_placeholders:
+            logger.warning(f"URL '{url}' 中仍有未替换的路径参数占位符: {remaining_placeholders}。原始模板: '{path_template}', 提供参数: {path_params}")
+
+    except KeyError as e:
+        logger.error(f"格式化URL路径 '{path_template}' 失败：路径参数 '{e}' 未在提供的 path_params 中找到。可用参数: {list(path_params.keys())}")
+        # 根据需要，这里可以选择是返回原始模板还是抛出异常
+        # return path_template 
+        raise ValueError(f"Missing path parameter {e} for URL template {path_template}") from e
+    except Exception as e:
+        logger.error(f"格式化URL路径 '{path_template}' 时发生未知错误: {e}")
+        raise # 或者返回原始模板
+
+    return url 

ddms_compliance_suite/utils/schema_utils.py

@@ -271,4 +271,342 @@ def util_remove_value_at_path(
         return data_container, None, False
     
     logger.error(f"[Util] util_remove_value_at_path 未能在循环内按预期返回。路径: {'.'.join(map(str,path))}")
-    return data_container, None, False 
+    return data_container, None, False 
+
+def util_set_value_at_path(
+    data_container: Any, 
+    path: List[Union[str, int]], 
+    new_value: Any,
+    # logger_param: Optional[logging.Logger] = None
+) -> Tuple[Any, bool]:
+    """
+    (框架辅助方法) 在嵌套的字典/列表中为指定路径设置新值。
+    如果路径中的某些部分不存在，会尝试创建它们 (字典会创建，列表会尝试填充到指定索引，但需谨慎)。
+    返回 (修改后的容器, 是否成功)。
+    """
+    # effective_logger = logger_param or logger
+    if not path:
+        logger.error("[Util] util_set_value_at_path: 路径不能为空。")
+        # 如果路径为空，是否应该用 new_value 替换整个 data_container？
+        # 当前行为：返回原始容器和失败状态，因为路径通常指向容器内部。
+        # 如果要支持替换整个容器，需要明确此行为。
+        if data_container is None and new_value is not None: # 特殊情况：如果原始容器是None，且路径为空，则新值成为容器
+             logger.info("[Util] util_set_value_at_path: 路径为空，原始容器为None，新值将作为新容器返回。")
+             return new_value, True
+        elif data_container is not None and new_value is None and not path: # 路径为空，新值为None，则清空容器
+             logger.info("[Util] util_set_value_at_path: 路径为空，新值为None，容器将被清空 (返回None)。")
+             return None, True
+        # 对于路径为空且两者都不是None的情况，目前返回失败，因为通常期望有路径。
+        # 或者可以考虑直接返回 new_value，意味着整个对象被替换。
+        # logger.info(f"[Util] util_set_value_at_path: Path is empty. Replacing entire container.")
+        # return new_value, True # 备选行为：替换整个对象
+        return data_container, False
+
+    # 深拷贝以避免修改原始输入，除非原始输入是None
+    container_copy = copy.deepcopy(data_container) if data_container is not None else None
+
+    current_level = container_copy
+    
+    try:
+        for i, key_or_index in enumerate(path):
+            is_last_element = (i == len(path) - 1)
+
+            if is_last_element:
+                if isinstance(key_or_index, str):
+                    if not isinstance(current_level, dict):
+                        # 如果当前层级不是字典 (例如是None，或是被意外替换为其他类型)，无法设置键值对
+                        logger.error(f"[Util] util_set_value_at_path: 路径的最后一部分 '{key_or_index}' (string key) 期望父级是字典，但找到 {type(current_level)}。路径: {'.'.join(map(str,path))}")
+                        # 尝试强制转换为字典？这可能不是预期行为。
+                        # 如果 current_level 是 None 且它是根 (container_copy is None)，则初始化 container_copy
+                        if current_level is None and i == 0: # 路径只有一级，且容器本身是None
+                            container_copy = {}
+                            current_level = container_copy
+                        else: # 更深层级的None或类型错误
+                            return data_container, False 
+                    current_level[key_or_index] = new_value
+                    logger.info(f"[Util] 在路径 {'.'.join(map(str,path))} (键 '{key_or_index}') 处设置值为 '{new_value}'")
+                    return container_copy, True
+                elif isinstance(key_or_index, int): # key_or_index is an integer (list index)
+                    if not isinstance(current_level, list):
+                        logger.error(f"[Util] util_set_value_at_path: 路径的最后一部分索引 '{key_or_index}' 期望父级是列表，但找到 {type(current_level)}。路径: {'.'.join(map(str,path))}")
+                        if current_level is None and i == 0: # 路径只有一级，且容器本身是None
+                            container_copy = [None] * (key_or_index + 1) # 创建足够长度的列表
+                            current_level = container_copy
+                        else:
+                             return data_container, False
+                    elif isinstance(key_or_index, int):
+                        # 确保列表足够长以容纳索引
+                        while len(current_level) <= key_or_index:
+                            current_level.append(None) # 用 None 填充直到达到所需长度
+                        current_level[key_or_index] = new_value
+                        logger.info(f"[Util] 在路径 {'.'.join(map(str,path))} (索引 '{key_or_index}') 处设置值为 '{new_value}'")
+                        return container_copy, True
+                else:
+                    logger.error(f"[Util] util_set_value_at_path: 路径的最后一部分 '{key_or_index}' 类型未知。路径: {'.'.join(map(str,path))}")
+                    return data_container, False
+            else: # Not the last element, traverse deeper
+                next_key_or_index_is_int = isinstance(path[i+1], int)
+
+                if isinstance(key_or_index, str): # Current path part is a dictionary key
+                    if not isinstance(current_level, dict):
+                        # 如果在根级别且容器是None，则初始化为字典
+                        if current_level is None and i == 0:
+                            container_copy = {}
+                            current_level = container_copy
+                        else:
+                            logger.error(f"[Util] util_set_value_at_path: 路径期望字典，但在 '{key_or_index}' 处找到 {type(current_level)}。路径: {'.'.join(map(str,path[:i+1]))}")
+                            return data_container, False
+                    if key_or_index not in current_level or current_level[key_or_index] is None or \
+                       (next_key_or_index_is_int and not isinstance(current_level[key_or_index], list)) or \
+                       (not next_key_or_index_is_int and not isinstance(current_level[key_or_index], dict)):
+                        # 如果键不存在，或值为None，或类型与下一路径部分不匹配，则创建/重置
+                        logger.debug(f"[Util] util_set_value_at_path: 在路径 '{key_or_index}' 处创建/重置结构。下一个是索引: {next_key_or_index_is_int}")
+                        current_level[key_or_index] = [] if next_key_or_index_is_int else {}
+                    current_level = current_level[key_or_index]
+                elif isinstance(key_or_index, int): # Current path part is a list index
+                    if not isinstance(current_level, list):
+                        if current_level is None and i == 0:
+                            container_copy = []
+                            current_level = container_copy
+                        else:
+                            logger.error(f"[Util] util_set_value_at_path: 路径期望列表以应用索引 '{key_or_index}'，但找到 {type(current_level)}。路径: {'.'.join(map(str,path[:i+1]))}")
+                            return data_container, False
+                    elif isinstance(key_or_index, int):
+                        # 确保列表足够长以容纳索引，并确保该索引处的元素是正确的类型 (list/dict)
+                        while len(current_level) <= key_or_index:
+                            current_level.append(None) # 用 None 填充
+                        if current_level[key_or_index] is None or \
+                           (next_key_or_index_is_int and not isinstance(current_level[key_or_index], list)) or \
+                           (not next_key_or_index_is_int and not isinstance(current_level[key_or_index], dict)):
+                            logger.debug(f"[Util] util_set_value_at_path: 在列表索引 '{key_or_index}' 处创建/重置结构。下一个是索引: {next_key_or_index_is_int}")
+                            current_level[key_or_index] = [] if next_key_or_index_is_int else {}
+                    current_level = current_level[key_or_index]
+                else:
+                    logger.error(f"[Util] util_set_value_at_path: 路径部分 '{key_or_index}' 类型未知 ({type(key_or_index)})。路径: {'.'.join(map(str,path[:i+1]))}")
+                    return data_container, False
+    except Exception as e:
+        logger.error(f"[Util] 在准备设置字段路径 {'.'.join(map(str,path))} 的值时发生错误: {e}", exc_info=True)
+        return data_container, False
+
+    # Should not be reached if logic is correct, path must have at least one element by initial check.
+    logger.error(f"[Util] util_set_value_at_path 未能在循环内按预期返回。路径: {'.'.join(map(str,path))}")
+    return data_container, False
+
+def generate_mismatched_value(
+    original_type: Optional[str], 
+    original_value: Any, 
+    field_schema: Optional[Dict[str, Any]],
+    logger_param: Optional[logging.Logger] = None
+) -> Any:
+    """
+    (框架辅助方法) 根据原始数据类型、原始值和字段 schema 生成一个类型不匹配的值。
+    主要用于类型不匹配的测试用例。
+    Args:
+        original_type: 字段的原始 OpenAPI 类型 (e.g., "string", "integer").
+        original_value: 字段的原始值 (当前未直接用于生成逻辑，但可供未来扩展).
+        field_schema: 字段的 schema 定义，用于检查如 "enum" 之类的约束。
+        logger_param: 可选的 logger 实例。
+    Returns:
+        一个与 original_type 不匹配的值。
+    """
+    effective_logger = logger_param or logger
+
+    # 优先考虑 schema 中的 enum，选择一个不在 enum 中且类型不匹配的值
+    if field_schema and "enum" in field_schema and isinstance(field_schema["enum"], list):
+        enum_values = field_schema["enum"]
+        if original_type == "string":
+            if 123 not in enum_values: return 123
+            if False not in enum_values: return False
+            # 如果数字和布尔都在枚举中，尝试一个与已知枚举值不同的字符串
+            # (虽然这仍然是字符串类型，但目的是为了触发非枚举值的验证)
+            # 或者，如果目的是严格类型不匹配，这里应该返回非字符串。
+            # 当前逻辑倾向于返回一个肯定非字符串的值。
+        elif original_type == "integer":
+            if "not-an-integer" not in enum_values: return "not-an-integer"
+            if 3.14 not in enum_values: return 3.14 
+        elif original_type == "number": # Includes float/double
+            if "not-a-number" not in enum_values: return "not-a-number"
+        elif original_type == "boolean":
+            if "not-a-boolean" not in enum_values: return "not-a-boolean"
+            if 1 not in enum_values: return 1 
+        # 如果枚举覆盖了所有简单备选，则回退到下面的通用逻辑
+
+    # 通用类型不匹配逻辑 (当 enum 不存在或 enum 检查未返回时)
+    if original_type == "string":
+        return 12345 # Number instead of string
+    elif original_type == "integer":
+        return "not-an-integer" # String instead of integer
+    elif original_type == "number": # Includes float/double
+        return "not-a-number" # String instead of number
+    elif original_type == "boolean":
+        return "not-a-boolean" # String instead of boolean
+    elif original_type == "array":
+        return {"value": "not-an-array"} # Object instead of array
+    elif original_type == "object":
+        return ["not", "an", "object"] # Array instead of object
+    
+    effective_logger.warning(f"generate_mismatched_value: 原始类型 '{original_type}' 未知或无法生成不匹配值。将返回固定字符串 'mismatch_test_default'。")
+    return "mismatch_test_default" # Fallback for unknown types
+
+def build_object_schema_for_params(params_spec_list: List[Dict[str, Any]], model_name_base: str, logger_param: Optional[logging.Logger] = None) -> Tuple[Optional[Dict[str, Any]], str]:
+    """
+    从参数规范列表构建一个对象的JSON schema，主要用于请求体、查询参数或头部的聚合。
+    Args:
+        params_spec_list: 参数规范的列表 (例如，OpenAPI参数对象列表)。
+        model_name_base: 用于生成动态模型名称的基础字符串。
+        logger_param: 可选的 logger 实例。
+    Returns:
+        一个元组，包含 (构建的JSON object schema 或 None, 模型名称字符串)。
+    """
+    effective_logger = logger_param or logger # Use passed logger or module logger
+    
+    if not params_spec_list:
+        effective_logger.debug(f"参数列表为空，无需为 '{model_name_base}' 构建对象 schema。")
+        return None, f"{model_name_base}EmptyParams"
+
+    properties = {}
+    required_fields = []
+
+    for param_spec in params_spec_list:
+        param_name = param_spec.get("name")
+        if not param_name:
+            effective_logger.warning(f"参数规范缺少 'name' 字段，已跳过: {param_spec}")
+            continue
+
+        # 从参数规范中提取 schema (OpenAPI 3.x)
+        param_schema = param_spec.get("schema") 
+        if not param_schema:
+            # 尝试兼容 OpenAPI 2.0 (Swagger) 的情况，其中类型信息直接在参数级别
+            # 例如: type, format, items, default, enum 等
+            # https://swagger.io/specification/v2/#parameterObject
+            # 注意: 这种兼容性可能不完整，因为很多属性需要映射
+            compatible_schema = {}
+            if "type" in param_spec:
+                compatible_schema["type"] = param_spec["type"]
+            if "format" in param_spec:
+                compatible_schema["format"] = param_spec["format"]
+            if "items" in param_spec: # for array types
+                compatible_schema["items"] = param_spec["items"]
+            if "default" in param_spec:
+                compatible_schema["default"] = param_spec["default"]
+            if "enum" in param_spec:
+                compatible_schema["enum"] = param_spec["enum"]
+            # 其他如 description, example 等也可以考虑加入
+            if compatible_schema: # 如果至少收集到了一些类型信息
+                param_schema = compatible_schema
+                effective_logger.debug(f"参数 '{param_name}' 没有 'schema' 字段，但从顶级字段构建了兼容 schema: {param_schema}")
+            else:
+                effective_logger.warning(f"参数 '{param_name}' 缺少 'schema' 字段且无法构建兼容schema，已跳过。规范: {param_spec}")
+                continue
+        
+        properties[param_name] = param_schema
+        if param_spec.get("required", False):
+            required_fields.append(param_name)
+
+    if not properties:
+        effective_logger.debug(f"未能从参数列表为 '{model_name_base}' 提取任何属性。")
+        return None, f"{model_name_base}NoProps"
+
+    final_schema: Dict[str, Any] = {
+        "type": "object",
+        "properties": properties
+    }
+    if required_fields:
+        final_schema["required"] = required_fields
+    
+    # 生成一个稍微独特的名字，以防多个操作有相同的 param_type
+    # 例如 OperationIdQueryRequest, OperationIdHeaderRequest
+    model_name = f"{model_name_base.replace(' ', '')}Params"
+    effective_logger.debug(f"为 '{model_name_base}' 构建的对象 schema: {final_schema}, 模型名: {model_name}")
+    return final_schema, model_name
+
+def find_first_simple_type_field_recursive(
+    current_schema: Dict[str, Any], 
+    current_path: Optional[List[Union[str, int]]] = None,
+    # full_api_spec_for_refs: Optional[Dict[str, Any]] = None, # Schema is expected to be pre-resolved
+    logger_param: Optional[logging.Logger] = None
+) -> Optional[Tuple[List[Union[str, int]], str, Dict[str, Any]]]:
+    """
+    递归地在给定的 schema 中查找第一个简单类型的字段 (string, integer, number, boolean)。
+    这包括查找嵌套在对象或数组中的简单类型字段。
+
+    Args:
+        current_schema: 当前正在搜索的 schema 部分 (应为字典)。
+        current_path: 到达当前 schema 的路径列表 (用于构建完整路径)。
+        logger_param: 可选的 logger 实例。
+
+    Returns:
+        一个元组 (field_path, field_type, field_schema) 如果找到，否则为 None。
+        field_path 是一个列表，表示从根 schema 到找到的字段的路径。
+        field_type 是字段的原始类型字符串 (e.g., "string")。
+        field_schema 是该字段自身的 schema 定义。
+    """
+    effective_logger = logger_param or logger # Use module logger if specific one not provided
+    path_so_far = current_path if current_path is not None else []
+
+    if not isinstance(current_schema, dict):
+        effective_logger.debug(f"Schema at path {'.'.join(map(str, path_so_far))} is not a dict, cannot search further.")
+        return None
+
+    schema_type = current_schema.get("type")
+    # effective_logger.debug(f"Searching in path: {'.'.join(map(str, path_so_far))}, Schema Type: '{schema_type}'")
+
+    if schema_type == "object":
+        properties = current_schema.get("properties", {})
+        for name, prop_schema in properties.items():
+            if not isinstance(prop_schema, dict):
+                effective_logger.debug(f"Property '{name}' at path {'.'.join(map(str, path_so_far + [name]))} has non-dict schema. Skipping.")
+                continue
+            
+            prop_type = prop_schema.get("type")
+            if prop_type in ["string", "integer", "number", "boolean"]:
+                field_path = path_so_far + [name]
+                effective_logger.info(f"Found simple type field: Path={'.'.join(map(str, field_path))}, Type={prop_type}")
+                return field_path, prop_type, prop_schema
+            elif prop_type == "object":
+                found_in_nested_object = find_first_simple_type_field_recursive(
+                    prop_schema, 
+                    path_so_far + [name],
+                    logger_param=effective_logger
+                )
+                if found_in_nested_object:
+                    return found_in_nested_object
+            elif prop_type == "array":
+                items_schema = prop_schema.get("items")
+                if isinstance(items_schema, dict):
+                    # Look for simple type or object within array items
+                    item_type = items_schema.get("type")
+                    if item_type in ["string", "integer", "number", "boolean"]:
+                        field_path = path_so_far + [name, 0] # Target first item of the array
+                        effective_logger.info(f"Found simple type field in array item: Path={'.'.join(map(str, field_path))}, Type={item_type}")
+                        return field_path, item_type, items_schema
+                    elif item_type == "object":
+                        # Path to the first item of the array, then recurse into that item's object schema
+                        found_in_array_item_object = find_first_simple_type_field_recursive(
+                            items_schema, 
+                            path_so_far + [name, 0],
+                            logger_param=effective_logger
+                        )
+                        if found_in_array_item_object:
+                            return found_in_array_item_object
+    
+    elif schema_type == "array": # If the current_schema itself is an array (e.g., root schema is an array)
+        items_schema = current_schema.get("items")
+        if isinstance(items_schema, dict):
+            item_type = items_schema.get("type")
+            if item_type in ["string", "integer", "number", "boolean"]:
+                field_path = path_so_far + [0] # Target first item of this root/current array
+                effective_logger.info(f"Found simple type field in root/current array item: Path={'.'.join(map(str, field_path))}, Type={item_type}")
+                return field_path, item_type, items_schema
+            elif item_type == "object":
+                # Path to the first item of this root/current array, then recurse
+                found_in_root_array_item_object = find_first_simple_type_field_recursive(
+                    items_schema, 
+                    path_so_far + [0],
+                    logger_param=effective_logger
+                )
+                if found_in_root_array_item_object:
+                    return found_in_root_array_item_object
+
+    # effective_logger.debug(f"No simple type field found at path {'.'.join(map(str, path_so_far))}")
+    return None