ruoyunbai/compliance
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix:yapi

Browse Source
This commit is contained in:
gongwenxin
2025-05-26 15:38:37 +08:00
parent 4180a0ce81
commit 6dde4d73e0
29 changed files with 3552 additions and 1265 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_testcases/compliance_catalog/error_handling/__pycache__/missing_required_field_body_case.cpython-312.pyc
BIN
View File
Binary file not shown.
custom_testcases/compliance_catalog/error_handling/__pycache__/missing_required_field_query_case.cpython-312.pyc
BIN
View File
Binary file not shown.
custom_testcases/compliance_catalog/error_handling/__pycache__/type_mismatch_body_case.cpython-312.pyc
BIN
View File
Binary file not shown.
custom_testcases/compliance_catalog/error_handling/__pycache__/type_mismatch_query_param_case.cpython-312.pyc
BIN
View File
Binary file not shown.
custom_testcases/compliance_catalog/error_handling/missing_required_field_body_case.py
+8 -57
View File
@@ -11,68 +11,19 @@ class MissingRequiredFieldBodyCase(BaseAPITestCase):
tags = ["error-handling", "appendix-b", "4003", "required-fields", "request-body"]
execution_order = 210 # Before query, same as original combined
def __init__(self, endpoint_spec: Dict[str, Any], global_api_spec: Dict[str, Any], json_schema_validator: Optional[Any] = None):
super().__init__(endpoint_spec, global_api_spec, json_schema_validator)
self.logger.setLevel(logging.DEBUG) # Ensure detailed logging for this class
def __init__(self, endpoint_spec: Dict[str, Any], global_api_spec: Dict[str, Any], json_schema_validator: Optional[Any] = None, llm_service: Optional[Any] = None):
super().__init__(endpoint_spec, global_api_spec, json_schema_validator, llm_service=llm_service)
self.logger = logging.getLogger(f"testcase.{self.id}")
self.target_field_path: Optional[List[str]] = None
self.original_value_at_path: Any = None
self.removed_field_path: Optional[List[str]] = None # Path to the removed field, e.g., ['level1', 'level2_field']
self.original_body_schema: Optional[Dict[str, Any]] = None
self._try_find_removable_body_field()
def _resolve_ref_if_present(self, schema_to_resolve: Dict[str, Any]) -> Dict[str, Any]:
ref_value = None
if isinstance(schema_to_resolve, dict):
if "$ref" in schema_to_resolve:
ref_value = schema_to_resolve["$ref"]
elif "$$ref" in schema_to_resolve:
ref_value = schema_to_resolve["$$ref"]
if ref_value:
self.logger.debug(f"发现引用 '{ref_value}'，尝试解析...")
try:
actual_global_spec_dict = None
if hasattr(self.global_api_spec, 'spec') and isinstance(self.global_api_spec.spec, dict):
actual_global_spec_dict = self.global_api_spec.spec
elif isinstance(self.global_api_spec, dict):
actual_global_spec_dict = self.global_api_spec
if not actual_global_spec_dict:
self.logger.warning(f"无法从 self.global_api_spec (类型: {type(self.global_api_spec)}) 获取用于解析引用的字典。")
return schema_to_resolve
resolved_schema = None
if ref_value.startswith("#/components/schemas/"):
schema_name = ref_value.split("/")[-1]
components = actual_global_spec_dict.get("components")
if components and isinstance(components.get("schemas"), dict):
resolved_schema = components["schemas"].get(schema_name)
if resolved_schema and isinstance(resolved_schema, dict):
self.logger.info(f"成功从 #/components/schemas/ 解析引用 '{ref_value}'。")
return resolved_schema
else:
self.logger.warning(f"解析引用 '{ref_value}' (路径: #/components/schemas/) 失败：未找到或找到的不是字典: {schema_name}")
else:
self.logger.warning(f"尝试从 #/components/schemas/ 解析引用 '{ref_value}' 失败：无法找到 'components.schemas' 结构。")
# 如果从 #/components/schemas/ 未成功解析，尝试 #/definitions/
if not resolved_schema and ref_value.startswith("#/definitions/"):
schema_name = ref_value.split("/")[-1]
definitions = actual_global_spec_dict.get("definitions")
if definitions and isinstance(definitions, dict):
resolved_schema = definitions.get(schema_name)
if resolved_schema and isinstance(resolved_schema, dict):
self.logger.info(f"成功从 #/definitions/ 解析引用 '{ref_value}'。")
return resolved_schema
else:
self.logger.warning(f"解析引用 '{ref_value}' (路径: #/definitions/) 失败：未找到或找到的不是字典: {schema_name}")
else:
self.logger.warning(f"尝试从 #/definitions/ 解析引用 '{ref_value}' 失败：无法找到 'definitions' 结构。")
if not resolved_schema:
self.logger.warning(f"最终未能通过任一已知路径 (#/components/schemas/ 或 #/definitions/) 解析引用 '{ref_value}'。")
except Exception as e:
self.logger.error(f"解析引用 '{ref_value}' 时发生错误: {e}", exc_info=True)
return schema_to_resolve # 返回原始 schema 如果不是 ref 或者所有解析尝试都失败
# 根据用户进一步要求，方法体简化为直接返回，不进行任何 $ref/$ $$ref 的检查。
# self.logger.debug(f"_resolve_ref_if_present called. Returning schema as-is per new configuration.")
return schema_to_resolve
def _find_required_field_in_schema_recursive(self, current_schema: Dict[str, Any], current_path: List[str]) -> Optional[List[str]]:
"""递归查找第一个可移除的必填字段的路径。
custom_testcases/compliance_catalog/error_handling/missing_required_field_query_case.py
+15 -14
View File
@@ -10,10 +10,11 @@ class MissingRequiredFieldQueryCase(BaseAPITestCase):
tags = ["error-handling", "appendix-b", "4003", "required-fields", "query-parameters"]
execution_order = 211 # After body, before original combined one might have been
def __init__(self, endpoint_spec: Dict[str, Any], global_api_spec: Dict[str, Any], json_schema_validator: Optional[Any] = None):
super().__init__(endpoint_spec, global_api_spec, json_schema_validator)
self.removed_field_name: Optional[str] = None
def __init__(self, endpoint_spec: Dict[str, Any], global_api_spec: Dict[str, Any], json_schema_validator: Optional[Any] = None, llm_service: Optional[Any] = None):
super().__init__(endpoint_spec, global_api_spec, json_schema_validator=json_schema_validator, llm_service=llm_service)
self.target_param_name: Optional[str] = None
self._try_find_removable_query_param()
self.logger.info(f"测试用例 {self.id} ({self.name}) 已针对端点 '{self.endpoint_spec.get('method')} {self.endpoint_spec.get('path')}' 初始化。Target param to remove: {self.target_param_name}")
def _try_find_removable_query_param(self):
query_params_spec_list = self.endpoint_spec.get("parameters", [])
@@ -23,8 +24,8 @@ class MissingRequiredFieldQueryCase(BaseAPITestCase):
if isinstance(param_spec, dict) and param_spec.get("in") == "query" and param_spec.get("required") is True:
field_name = param_spec.get("name")
if field_name:
self.removed_field_name = field_name
self.logger.info(f"必填字段缺失测试的目标字段 (查询参数): '{self.removed_field_name}'")
self.target_param_name = field_name
self.logger.info(f"必填字段缺失测试的目标字段 (查询参数): '{self.target_param_name}'")
return
self.logger.info('在此端点规范中未找到可用于测试 "必填查询参数缺失" 的字段。')
@@ -34,20 +35,20 @@ class MissingRequiredFieldQueryCase(BaseAPITestCase):
return current_body
def generate_query_params(self, current_query_params: Dict[str, Any]) -> Dict[str, Any]:
if self.removed_field_name and isinstance(current_query_params, dict):
if self.removed_field_name in current_query_params:
if self.target_param_name and isinstance(current_query_params, dict):
if self.target_param_name in current_query_params:
new_params = copy.deepcopy(current_query_params)
original_value = new_params.pop(self.removed_field_name) # 移除参数
self.logger.info(f"为进行必填查询参数缺失测试，已从查询参数中移除 '{self.removed_field_name}' (原值: '{original_value}')。")
original_value = new_params.pop(self.target_param_name) # 移除参数
self.logger.info(f"为进行必填查询参数缺失测试，已从查询参数中移除 '{self.target_param_name}' (原值: '{original_value}')。")
return new_params
else:
self.logger.warning(f"计划移除的查询参数 '{self.removed_field_name}' 在当前查询参数中未找到。")
self.logger.warning(f"计划移除的查询参数 '{self.target_param_name}' 在当前查询参数中未找到。")
return current_query_params
def validate_response(self, response_context: APIResponseContext, request_context: APIRequestContext) -> List[ValidationResult]:
results = []
if not self.removed_field_name:
if not self.target_param_name:
results.append(self.passed("跳过测试：在API规范中未找到合适的必填查询参数用于移除测试。"))
self.logger.info("由于未识别到可移除的必填查询参数，跳过此测试用例。")
return results
@@ -58,7 +59,7 @@ class MissingRequiredFieldQueryCase(BaseAPITestCase):
expected_status_codes = [400, 422]
specific_error_code_from_appendix_b = "4003"
msg_prefix = f"当移除必填查询参数 '{self.removed_field_name}' 时，"
msg_prefix = f"当移除必填查询参数 '{self.target_param_name}' 时，"
if status_code in expected_status_codes:
status_msg = f"{msg_prefix}API响应了预期的错误状态码 {status_code}。"
@@ -77,8 +78,8 @@ class MissingRequiredFieldQueryCase(BaseAPITestCase):
else:
results.append(self.failed(
message=f"{msg_prefix}期望API返回状态码 {expected_status_codes} 中的一个，但实际收到 {status_code}。",
details={"status_code": status_code, "response_body": json_content, "removed_field": f"query.{self.removed_field_name}"}
details={"status_code": status_code, "response_body": json_content, "removed_field": f"query.{self.target_param_name}"}
))
self.logger.warning(f"必填查询参数缺失测试失败：期望状态码 {expected_status_codes}，实际为 {status_code}。移除的参数：'{self.removed_field_name}'")
self.logger.warning(f"必填查询参数缺失测试失败：期望状态码 {expected_status_codes}，实际为 {status_code}。移除的参数：'{self.target_param_name}'")
return results
custom_testcases/compliance_catalog/error_handling/type_mismatch_body_case.py
+9 -54
View File
@@ -11,15 +11,20 @@ class TypeMismatchBodyCase(BaseAPITestCase):
tags = ["error-handling", "appendix-b", "4001", "request-body"]
execution_order = 202 # Slightly after query param one
def __init__(self, endpoint_spec: Dict[str, Any], global_api_spec: Dict[str, Any], json_schema_validator: Optional[Any] = None):
super().__init__(endpoint_spec, global_api_spec, json_schema_validator)
def __init__(self, endpoint_spec: Dict[str, Any], global_api_spec: Dict[str, Any], json_schema_validator: Optional[Any] = None, llm_service: Optional[Any] = None):
super().__init__(endpoint_spec, global_api_spec, json_schema_validator, llm_service=llm_service)
self.logger.setLevel(logging.DEBUG)
self.target_field_path: Optional[List[str]] = None
self.original_field_type: Optional[str] = None
# Location is always 'body' for this class
self.target_field_location: str = "body"
self.target_field_schema: Optional[Dict[str, Any]] = None
self.json_schema_validator = json_schema_validator
self.original_value_at_path: Any = None
self.mismatched_value: Any = None
self._try_find_mismatch_target_in_body()
def _try_find_mismatch_target_in_body(self):
self.logger.critical(f"{self.id} __INIT__ >>> STARTED")
self.logger.debug(f"开始为端点 {self.endpoint_spec.get('method')} {self.endpoint_spec.get('path')} 初始化请求体类型不匹配测试的目标字段查找。")
@@ -64,58 +69,8 @@ class TypeMismatchBodyCase(BaseAPITestCase):
self.logger.info(f"最终，在端点 {self.endpoint_spec.get('method')} {self.endpoint_spec.get('path')} 的请求体中，均未找到可用于测试类型不匹配的字段。")
def _resolve_ref_if_present(self, schema_to_resolve: Dict[str, Any]) -> Dict[str, Any]:
ref_value = None
if isinstance(schema_to_resolve, dict):
if "$ref" in schema_to_resolve:
ref_value = schema_to_resolve["$ref"]
elif "$$ref" in schema_to_resolve:
ref_value = schema_to_resolve["$$ref"]
if ref_value:
self.logger.debug(f"发现引用 '{ref_value}'，尝试解析...")
try:
actual_global_spec_dict = None
if hasattr(self.global_api_spec, 'spec') and isinstance(self.global_api_spec.spec, dict):
actual_global_spec_dict = self.global_api_spec.spec
elif isinstance(self.global_api_spec, dict):
actual_global_spec_dict = self.global_api_spec
if not actual_global_spec_dict:
self.logger.warning(f"无法从 self.global_api_spec (类型: {type(self.global_api_spec)}) 获取用于解析引用的字典。")
return schema_to_resolve
resolved_schema = None
if ref_value.startswith("#/components/schemas/"):
schema_name = ref_value.split("/")[-1]
components = actual_global_spec_dict.get("components")
if components and isinstance(components.get("schemas"), dict):
resolved_schema = components["schemas"].get(schema_name)
if resolved_schema and isinstance(resolved_schema, dict):
self.logger.info(f"成功从 #/components/schemas/ 解析引用 '{ref_value}'。")
return resolved_schema
else:
self.logger.warning(f"解析引用 '{ref_value}' (路径: #/components/schemas/) 失败：未找到或找到的不是字典: {schema_name}")
else:
self.logger.warning(f"尝试从 #/components/schemas/ 解析引用 '{ref_value}' 失败：无法找到 'components.schemas' 结构。")
if not resolved_schema and ref_value.startswith("#/definitions/"):
schema_name = ref_value.split("/")[-1]
definitions = actual_global_spec_dict.get("definitions")
if definitions and isinstance(definitions, dict):
resolved_schema = definitions.get(schema_name)
if resolved_schema and isinstance(resolved_schema, dict):
self.logger.info(f"成功从 #/definitions/ 解析引用 '{ref_value}'。")
return resolved_schema
else:
self.logger.warning(f"解析引用 '{ref_value}' (路径: #/definitions/) 失败：未找到或找到的不是字典: {schema_name}")
else:
self.logger.warning(f"尝试从 #/definitions/ 解析引用 '{ref_value}' 失败：无法找到 'definitions' 结构。")
if not resolved_schema:
self.logger.warning(f"最终未能通过任一已知路径 (#/components/schemas/ 或 #/definitions/) 解析引用 '{ref_value}'。")
except Exception as e:
self.logger.error(f"解析引用 '{ref_value}' 时发生错误: {e}", exc_info=True)
# 根据用户进一步要求，方法体简化为直接返回，不进行任何 $ref/$ $$ref 的检查。
# self.logger.debug(f"_resolve_ref_if_present called. Returning schema as-is per new configuration.")
return schema_to_resolve
def _find_target_field_in_schema(self, schema_to_search: Dict[str, Any], base_path_for_log: str) -> bool:
custom_testcases/compliance_catalog/error_handling/type_mismatch_query_param_case.py
+63 -54
View File
@@ -11,14 +11,19 @@ class TypeMismatchQueryParamCase(BaseAPITestCase):
tags = ["error-handling", "appendix-b", "4001", "query-parameters"]
execution_order = 201 # Slightly after the combined one might have been
def __init__(self, endpoint_spec: Dict[str, Any], global_api_spec: Dict[str, Any], json_schema_validator: Optional[Any] = None):
super().__init__(endpoint_spec, global_api_spec, json_schema_validator)
def __init__(self, endpoint_spec: Dict[str, Any], global_api_spec: Dict[str, Any], json_schema_validator: Optional[Any] = None, llm_service: Optional[Any] = None):
super().__init__(endpoint_spec, global_api_spec, json_schema_validator, llm_service=llm_service)
self.logger.setLevel(logging.DEBUG)
self.target_field_path: Optional[List[str]] = None
self.original_field_type: Optional[str] = None
# Location is always 'query' for this class
self.target_field_location: str = "query"
self.target_field_schema: Optional[Dict[str, Any]] = None
self.original_value_at_path: Any = None
self.mismatched_value: Any = None
# 调用新方法来查找目标字段
self._try_find_mismatch_target_in_query()
self.logger.critical(f"{self.id} __INIT__ >>> STARTED")
self.logger.debug(f"开始为端点 {self.endpoint_spec.get('method')} {self.endpoint_spec.get('path')} 初始化查询参数类型不匹配测试的目标字段查找。")
@@ -78,59 +83,63 @@ class TypeMismatchQueryParamCase(BaseAPITestCase):
if not self.target_field_path:
self.logger.info(f"最终，在端点 {self.endpoint_spec.get('method')} {self.endpoint_spec.get('path')} 的查询参数中，均未找到可用于测试类型不匹配的字段。")
def _try_find_mismatch_target_in_query(self):
self.logger.critical(f"{self.id} _try_find_mismatch_target_in_query >>> STARTED")
self.logger.debug(f"开始为端点 {self.endpoint_spec.get('method')} {self.endpoint_spec.get('path')} 初始化查询参数类型不匹配测试的目标字段查找。")
parameters = self.endpoint_spec.get("parameters", [])
self.logger.critical(f"{self.id} _try_find_mismatch_target_in_query >>> Parameters to be processed: {parameters}")
self.logger.debug(f"传入的参数列表 (在 {self.id}中): {parameters}")
for param_spec in parameters:
if param_spec.get("in") == "query":
param_name = param_spec.get("name")
if not param_name:
self.logger.warning("发现一个没有名称的查询参数定义，已跳过。")
continue
self.logger.debug(f"检查查询参数: '{param_name}'")
param_type = param_spec.get("type")
param_schema = param_spec.get("schema")
# Scenario 1: Simple type directly in param_spec (e.g., type: string)
if param_type in ["string", "number", "integer", "boolean"]:
self.target_field_path = [param_name]
self.original_field_type = param_type
self.target_field_schema = param_spec
self.logger.info(f"目标字段(查询参数 - 简单类型): {param_name}，原始类型: {self.original_field_type}")
break
# Scenario 2: Schema defined for the query parameter (OpenAPI 3.0 style, or complex objects in query)
elif isinstance(param_schema, dict):
self.logger.debug(f"查询参数 '{param_name}' 包含嵌套 schema，尝试在其内部查找简单类型字段。")
resolved_param_schema = self._resolve_ref_if_present(param_schema)
if resolved_param_schema.get("type") == "object":
properties = resolved_param_schema.get("properties", {})
for prop_name, prop_details_orig in properties.items():
prop_details = self._resolve_ref_if_present(prop_details_orig)
if prop_details.get("type") in ["string", "number", "integer", "boolean"]:
self.target_field_path = [param_name, prop_name]
self.original_field_type = prop_details.get("type")
self.target_field_schema = prop_details
self.logger.info(f"目标字段(查询参数 - 对象属性): {param_name}.{prop_name}，原始类型: {self.original_field_type}")
break
if self.target_field_path: break
elif resolved_param_schema.get("type") in ["string", "number", "integer", "boolean"]:
self.target_field_path = [param_name]
self.original_field_type = resolved_param_schema.get("type")
self.target_field_schema = resolved_param_schema
self.logger.info(f"目标字段(查询参数 - schema为简单类型): {param_name}，原始类型: {self.original_field_type}")
break
else:
self.logger.debug(f"查询参数 '{param_name}' (type: {param_type}, schema: {param_schema}) 不是直接的简单类型，也无直接可用的对象型 schema 属性。")
if not self.target_field_path:
self.logger.info(f"最终，在端点 {self.endpoint_spec.get('method')} {self.endpoint_spec.get('path')} 的查询参数中，均未找到可用于测试类型不匹配的字段。")
def _resolve_ref_if_present(self, schema_to_resolve: Dict[str, Any]) -> Dict[str, Any]:
ref_value = None
if isinstance(schema_to_resolve, dict):
if "$ref" in schema_to_resolve:
ref_value = schema_to_resolve["$ref"]
elif "$$ref" in schema_to_resolve:
ref_value = schema_to_resolve["$$ref"]
if ref_value:
self.logger.debug(f"发现引用 '{ref_value}'，尝试解析...")
try:
actual_global_spec_dict = None
if hasattr(self.global_api_spec, 'spec') and isinstance(self.global_api_spec.spec, dict):
actual_global_spec_dict = self.global_api_spec.spec
elif isinstance(self.global_api_spec, dict):
actual_global_spec_dict = self.global_api_spec
if not actual_global_spec_dict:
self.logger.warning(f"无法从 self.global_api_spec (类型: {type(self.global_api_spec)}) 获取用于解析引用的字典。")
return schema_to_resolve
resolved_schema = None
if ref_value.startswith("#/components/schemas/"):
schema_name = ref_value.split("/")[-1]
components = actual_global_spec_dict.get("components")
if components and isinstance(components.get("schemas"), dict):
resolved_schema = components["schemas"].get(schema_name)
if resolved_schema and isinstance(resolved_schema, dict):
self.logger.info(f"成功从 #/components/schemas/ 解析引用 '{ref_value}'。")
return resolved_schema
else:
self.logger.warning(f"解析引用 '{ref_value}' (路径: #/components/schemas/) 失败：未找到或找到的不是字典: {schema_name}")
else:
self.logger.warning(f"尝试从 #/components/schemas/ 解析引用 '{ref_value}' 失败：无法找到 'components.schemas' 结构。")
if not resolved_schema and ref_value.startswith("#/definitions/"):
schema_name = ref_value.split("/")[-1]
definitions = actual_global_spec_dict.get("definitions")
if definitions and isinstance(definitions, dict):
resolved_schema = definitions.get(schema_name)
if resolved_schema and isinstance(resolved_schema, dict):
self.logger.info(f"成功从 #/definitions/ 解析引用 '{ref_value}'。")
return resolved_schema
else:
self.logger.warning(f"解析引用 '{ref_value}' (路径: #/definitions/) 失败：未找到或找到的不是字典: {schema_name}")
else:
self.logger.warning(f"尝试从 #/definitions/ 解析引用 '{ref_value}' 失败：无法找到 'definitions' 结构。")
if not resolved_schema:
self.logger.warning(f"最终未能通过任一已知路径 (#/components/schemas/ 或 #/definitions/) 解析引用 '{ref_value}'。")
except Exception as e:
self.logger.error(f"解析引用 '{ref_value}' 时发生错误: {e}", exc_info=True)
# 根据用户进一步要求，方法体简化为直接返回，不进行任何 $ref/$ $$ref 的检查。
# self.logger.debug(f"_resolve_ref_if_present called. Returning schema as-is per new configuration.")
return schema_to_resolve
# No generate_request_body, or it simply returns current_body