添加error 测试用例，但是测试用例太复杂，还需要优化框架
This commit is contained in:
gongwenxin
@@ -0,0 +1 @@
|
||||
# This file marks the compliance_catalog directory as a Python package.
|
||||
@@ -0,0 +1 @@
|
||||
# This file marks the error_handling directory as a Python package.
|
||||
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
+296
@@ -0,0 +1,296 @@
|
||||
from typing import Dict, Any, Optional, List
|
||||
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, TestSeverity, ValidationResult, APIRequestContext, APIResponseContext
|
||||
import copy
|
||||
import logging
|
||||
|
||||
class MissingRequiredFieldBodyCase(BaseAPITestCase):
|
||||
id = "TC-ERROR-4003-BODY"
|
||||
name = "Error Code 4003 - Missing Required Request Body Field Validation"
|
||||
description = "测试当请求体中缺少API规范定义的必填字段时，API是否按预期返回类似4003的错误（或通用400错误）。"
|
||||
severity = TestSeverity.HIGH
|
||||
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
|
||||
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 或者所有解析尝试都失败
|
||||
|
||||
def _find_required_field_in_schema_recursive(self, current_schema: Dict[str, Any], current_path: List[str]) -> Optional[List[str]]:
|
||||
"""递归查找第一个可移除的必填字段的路径。
|
||||
现在也会查找数组内对象中必填的字段。"""
|
||||
resolved_schema = self._resolve_ref_if_present(current_schema)
|
||||
|
||||
if not isinstance(resolved_schema, dict) or resolved_schema.get("type") != "object":
|
||||
# If not an object schema, cannot have 'required' or 'properties' in the way we expect.
|
||||
return None
|
||||
|
||||
required_fields_at_current_level = resolved_schema.get("required", [])
|
||||
properties = resolved_schema.get("properties", {})
|
||||
self.logger.debug(f"递归查找路径: {current_path}, 当前层级必填字段: {required_fields_at_current_level}, 属性: {list(properties.keys())}")
|
||||
|
||||
# 策略1: 查找当前层级直接声明的必填字段 (简单类型或复杂类型均可)
|
||||
if required_fields_at_current_level and properties:
|
||||
for field_name in required_fields_at_current_level:
|
||||
if field_name in properties:
|
||||
# 任何在 'required' 数组中列出的字段，无论其类型，都可以作为目标
|
||||
# (例如，移除一个必填的整个对象或数组也是一种有效的测试场景)
|
||||
self.logger.info(f"策略1: 在路径 {'.'.join(current_path) if current_path else 'root'} 找到可直接移除的必填字段: '{field_name}'")
|
||||
return current_path + [field_name]
|
||||
|
||||
# 策略2: 如果当前层级没有直接的必填字段可移除，则查找数组属性，看其内部item是否有必填字段
|
||||
# 这种情况下，数组本身可能不是必填的，但如果提供了数组，其item需要满足条件
|
||||
if properties: # 确保有属性可迭代
|
||||
for prop_name, prop_schema_orig in properties.items():
|
||||
prop_schema = self._resolve_ref_if_present(prop_schema_orig) # 解析属性自身的schema (可能也是ref)
|
||||
if isinstance(prop_schema, dict) and prop_schema.get("type") == "array":
|
||||
items_schema_orig = prop_schema.get("items")
|
||||
if isinstance(items_schema_orig, dict):
|
||||
items_schema = self._resolve_ref_if_present(items_schema_orig) # 解析 items 的 schema
|
||||
if isinstance(items_schema, dict) and items_schema.get("type") == "object":
|
||||
item_required_fields = items_schema.get("required", [])
|
||||
item_properties = items_schema.get("properties", {})
|
||||
if item_required_fields and item_properties:
|
||||
first_required_field_in_item = None
|
||||
for req_item_field in item_required_fields:
|
||||
if req_item_field in item_properties: # 确保该必填字段在属性中定义
|
||||
first_required_field_in_item = req_item_field
|
||||
break
|
||||
|
||||
if first_required_field_in_item:
|
||||
self.logger.info(f"策略2: 在数组属性 '{prop_name}' (路径 {'.'.join(current_path) if current_path else 'root'}) 的元素内找到必填字段: '{first_required_field_in_item}'. 将尝试移除路径: {current_path + [prop_name, 0, first_required_field_in_item]}")
|
||||
# 将路径指向数组的第一个元素 (index 0) 内的那个必填字段
|
||||
return current_path + [prop_name, 0, first_required_field_in_item]
|
||||
|
||||
# 策略3: （可选，如果需要更深层次的普通对象递归）
|
||||
# 如果以上策略都未找到，并且希望深入到非必填的子对象中查找，可以启用以下逻辑。
|
||||
# 但这通常不用于"顶层必填字段缺失"的测试目的，除非测试用例目标是验证任意深度的必填。
|
||||
# for prop_name, prop_schema_orig_for_recurse in properties.items():
|
||||
# prop_schema_for_recurse = self._resolve_ref_if_present(prop_schema_orig_for_recurse)
|
||||
# if isinstance(prop_schema_for_recurse, dict) and prop_schema_for_recurse.get("type") == "object":
|
||||
# # 确保不陷入无限循环，例如，如果一个对象属性是可选的但其内部有必填字段
|
||||
# # 这里需要小心，因为我们可能已经检查过当前级别的required字段
|
||||
# # 主要用于当某个对象不是顶层必填，但如果提供了它，它内部又有必填项的场景
|
||||
# # 但这与当前测试用例的 primary goal 可能不完全一致
|
||||
# self.logger.debug(f"策略3: 尝试递归进入对象属性 '{prop_name}' (路径 {'.'.join(current_path)}) (此对象本身在当前层级非必填或已检查)")
|
||||
# found_path_deeper = self._find_required_field_in_schema_recursive(prop_schema_for_recurse, current_path + [prop_name])
|
||||
# if found_path_deeper:
|
||||
# # 确保返回的路径确实比当前路径深，并且该深层路径的父级（即prop_name）不是当前层级已知的必填字段
|
||||
# # (以避免重复发现已被策略1覆盖的场景)
|
||||
# # if prop_name not in required_fields_at_current_level:
|
||||
# self.logger.info(f"策略3: 递归在对象属性 '{prop_name}' (路径 {'.'.join(current_path)}) 中找到必填字段路径: {found_path_deeper}")
|
||||
# return found_path_deeper
|
||||
|
||||
self.logger.debug(f"在路径 {'.'.join(current_path) if current_path else 'root'} 未通过任何策略找到可移除的必填字段。")
|
||||
return None
|
||||
def _try_find_removable_body_field(self):
|
||||
body_schema_to_check: Optional[Dict[str, Any]] = None
|
||||
request_body_spec = self.endpoint_spec.get("requestBody")
|
||||
if request_body_spec and isinstance(request_body_spec, dict):
|
||||
content = request_body_spec.get("content", {})
|
||||
json_schema_entry = content.get("application/json")
|
||||
if json_schema_entry and isinstance(json_schema_entry, dict) and isinstance(json_schema_entry.get("schema"), dict):
|
||||
body_schema_to_check = json_schema_entry["schema"]
|
||||
|
||||
if not body_schema_to_check:
|
||||
parameters = self.endpoint_spec.get("parameters", [])
|
||||
if isinstance(parameters, list):
|
||||
for param in parameters:
|
||||
if isinstance(param, dict) and param.get("in") == "body":
|
||||
if isinstance(param.get("schema"), dict):
|
||||
body_schema_to_check = param["schema"]
|
||||
break
|
||||
|
||||
if body_schema_to_check:
|
||||
self.original_body_schema = copy.deepcopy(body_schema_to_check)
|
||||
self.removed_field_path = self._find_required_field_in_schema_recursive(self.original_body_schema, [])
|
||||
if self.removed_field_path:
|
||||
self.logger.info(f"必填字段缺失测试的目标字段 (请求体): '{'.'.join(map(str, self.removed_field_path))}'")
|
||||
self.field_to_remove_details = {
|
||||
"path": self.removed_field_path,
|
||||
# ... existing code ...
|
||||
}
|
||||
else:
|
||||
self.logger.info('在请求体 schema 中未找到可用于测试 "必填字段缺失" 的字段。')
|
||||
else:
|
||||
self.logger.info('此端点规范中未定义请求体 schema。')
|
||||
|
||||
def generate_query_params(self, current_query_params: Dict[str, Any]) -> Dict[str, Any]:
|
||||
self.logger.debug(f"{self.id} is focused on request body, generate_query_params will not modify query parameters.")
|
||||
return current_query_params
|
||||
|
||||
def generate_request_body(self, current_body: Optional[Any]) -> Optional[Any]:
|
||||
if not self.removed_field_path:
|
||||
self.logger.debug("No field path identified for removal in request body.")
|
||||
return current_body
|
||||
|
||||
if current_body is None:
|
||||
self.logger.debug("current_body is None. Orchestrator should ideally provide a base body. Attempting to build minimal structure for removal.")
|
||||
new_body = {}
|
||||
else:
|
||||
new_body = copy.deepcopy(current_body)
|
||||
|
||||
temp_obj_ref = new_body
|
||||
|
||||
try:
|
||||
for i, key_or_index in enumerate(self.removed_field_path):
|
||||
is_last_element = (i == len(self.removed_field_path) - 1)
|
||||
|
||||
if is_last_element:
|
||||
if isinstance(key_or_index, str): # Key for a dictionary (field name)
|
||||
if isinstance(temp_obj_ref, dict) and key_or_index in temp_obj_ref:
|
||||
original_value = temp_obj_ref.pop(key_or_index)
|
||||
self.logger.info(f"为进行必填字段缺失测试，已从请求体中移除字段路径 '{'.'.join(map(str,self.removed_field_path))}' (原值: '{original_value}')。")
|
||||
return new_body
|
||||
elif isinstance(temp_obj_ref, dict): # Key not in dict, but it's a dict
|
||||
self.logger.warning(f"计划移除的请求体字段路径的最后一部分 '{key_or_index}' (string key) 在对象中未找到，但该对象是字典。可能该字段本就是可选的或不存在于提供的current_body。路径: {'.'.join(map(str,self.removed_field_path))}")
|
||||
return new_body
|
||||
else: # temp_obj_ref is not a dict
|
||||
self.logger.warning(f"计划移除的请求体字段路径的最后一部分 '{key_or_index}' (string key) 期望父级是字典，但找到 {type(temp_obj_ref)}。路径: {'.'.join(map(str,self.removed_field_path))}")
|
||||
return current_body
|
||||
else: # Last element of path is an index - this should not happen as we remove a *field name*
|
||||
self.logger.error(f"路径的最后一部分 '{key_or_index}' 预期为字符串字段名，但类型为 {type(key_or_index)}. Path: {'.'.join(map(str,self.removed_field_path))}")
|
||||
return current_body
|
||||
else: # Not the last element, so we are traversing or building the structure
|
||||
next_key_or_index = self.removed_field_path[i+1]
|
||||
|
||||
if isinstance(key_or_index, str): # Current path part is a dictionary key
|
||||
if not isinstance(temp_obj_ref, dict):
|
||||
self.logger.warning(f"路径期望字典，但在 '{key_or_index}' (父级)处找到 {type(temp_obj_ref)}. Path: {'.'.join(map(str,self.removed_field_path))}. 如果current_body为None，则尝试创建字典。")
|
||||
if temp_obj_ref is new_body and not new_body :
|
||||
temp_obj_ref = {}
|
||||
else:
|
||||
return current_body
|
||||
|
||||
if isinstance(next_key_or_index, int):
|
||||
if key_or_index not in temp_obj_ref or not isinstance(temp_obj_ref.get(key_or_index), list):
|
||||
self.logger.debug(f"路径 '{key_or_index}' 需要是列表 (为索引 {next_key_or_index} 做准备)，但未找到或类型不符。将创建空列表。")
|
||||
temp_obj_ref[key_or_index] = []
|
||||
temp_obj_ref = temp_obj_ref[key_or_index]
|
||||
else:
|
||||
if key_or_index not in temp_obj_ref or not isinstance(temp_obj_ref.get(key_or_index), dict):
|
||||
self.logger.debug(f"路径 '{key_or_index}' 需要是字典 (为键 '{next_key_or_index}' 做准备)，但未找到或类型不符。将创建空字典。")
|
||||
temp_obj_ref[key_or_index] = {}
|
||||
temp_obj_ref = temp_obj_ref[key_or_index]
|
||||
|
||||
elif isinstance(key_or_index, int):
|
||||
if not isinstance(temp_obj_ref, list):
|
||||
self.logger.error(f"路径期望列表以应用索引 '{key_or_index}'，但找到 {type(temp_obj_ref)}. Path: {'.'.join(map(str,self.removed_field_path))}")
|
||||
return current_body
|
||||
|
||||
while len(temp_obj_ref) <= key_or_index:
|
||||
self.logger.debug(f"数组在索引 {key_or_index} 处需要元素，将添加空字典作为占位符（因为后续预期是字段名）。")
|
||||
temp_obj_ref.append({})
|
||||
|
||||
if isinstance(next_key_or_index, str):
|
||||
if not isinstance(temp_obj_ref[key_or_index], dict):
|
||||
self.logger.debug(f"数组项 at index {key_or_index} 需要是字典 (为键 '{next_key_or_index}' 做准备)。如果它是其他类型，将被替换为空字典。")
|
||||
temp_obj_ref[key_or_index] = {}
|
||||
|
||||
temp_obj_ref = temp_obj_ref[key_or_index]
|
||||
|
||||
else:
|
||||
self.logger.error(f"路径部分 '{key_or_index}' 类型未知 ({type(key_or_index)}). Path: {'.'.join(map(str,self.removed_field_path))}")
|
||||
return current_body
|
||||
except Exception as e: # Ensuring the try has an except
|
||||
self.logger.error(f"在准备移除字段路径 '{'.'.join(map(str,self.removed_field_path))}' 时发生错误: {e}", exc_info=True)
|
||||
return current_body
|
||||
|
||||
self.logger.error(f"generate_request_body 未能在循环内按预期返回。路径: {'.'.join(map(str,self.removed_field_path))}")
|
||||
return current_body
|
||||
|
||||
def validate_response(self, response_context: APIResponseContext, request_context: APIRequestContext) -> List[ValidationResult]:
|
||||
results = []
|
||||
if not self.removed_field_path:
|
||||
results.append(self.passed("跳过测试：在API规范中未找到合适的必填请求体字段用于移除测试。"))
|
||||
self.logger.info("由于未识别到可移除的必填请求体字段，跳过此测试用例。")
|
||||
return results
|
||||
|
||||
status_code = response_context.status_code
|
||||
json_content = response_context.json_content
|
||||
expected_status_codes = [400, 422]
|
||||
specific_error_code_from_appendix_b = "4003"
|
||||
removed_field_str = '.'.join(map(str, self.removed_field_path))
|
||||
|
||||
msg_prefix = f"当移除必填请求体字段 '{removed_field_str}' 时，"
|
||||
|
||||
if status_code in expected_status_codes:
|
||||
status_msg = f"{msg_prefix}API响应了预期的错误状态码 {status_code}。"
|
||||
if json_content and isinstance(json_content, dict) and str(json_content.get("code")) == specific_error_code_from_appendix_b:
|
||||
results.append(self.passed(f"{status_msg} 且响应体中包含特定的错误码 '{specific_error_code_from_appendix_b}'。"))
|
||||
self.logger.info(f"正确接收到状态码 {status_code} 和错误码 '{specific_error_code_from_appendix_b}'。")
|
||||
elif json_content and isinstance(json_content, dict) and "code" in json_content:
|
||||
results.append(ValidationResult(passed=True,
|
||||
message=f"{status_msg} 响应体中的错误码为 '{json_content.get('code')}' (期望或类似 '{specific_error_code_from_appendix_b}')。",
|
||||
details=json_content
|
||||
))
|
||||
self.logger.warning(f"接收到状态码 {status_code}，但错误码是 '{json_content.get('code')}' 而不是期望的 '{specific_error_code_from_appendix_b}'。此结果仍标记为通过，因状态码正确。")
|
||||
else:
|
||||
results.append(self.passed(f"{status_msg} 但响应体中未找到特定的错误码字段或响应体结构不符合预期。"))
|
||||
self.logger.info(f"正确接收到状态码 {status_code}，但在响应体中未找到错误码字段或预期结构。")
|
||||
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"body.{removed_field_str}"}
|
||||
))
|
||||
self.logger.warning(f"必填请求体字段缺失测试失败：期望状态码 {expected_status_codes}，实际为 {status_code}。移除的字段：'body.{removed_field_str}'")
|
||||
|
||||
return results
|
||||
+84
@@ -0,0 +1,84 @@
|
||||
from typing import Dict, Any, Optional, List
|
||||
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, TestSeverity, ValidationResult, APIRequestContext, APIResponseContext
|
||||
import copy
|
||||
|
||||
class MissingRequiredFieldQueryCase(BaseAPITestCase):
|
||||
id = "TC-ERROR-4003-QUERY"
|
||||
name = "Error Code 4003 - Missing Required Query Parameter Validation"
|
||||
description = "测试当请求中缺少API规范定义的必填查询参数时，API是否按预期返回类似4003的错误（或通用400错误）。"
|
||||
severity = TestSeverity.HIGH
|
||||
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
|
||||
self._try_find_removable_query_param()
|
||||
|
||||
def _try_find_removable_query_param(self):
|
||||
query_params_spec_list = self.endpoint_spec.get("parameters", [])
|
||||
if query_params_spec_list:
|
||||
self.logger.debug(f"检查查询参数的必填字段，总共 {len(query_params_spec_list)} 个参数定义。")
|
||||
for param_spec in query_params_spec_list:
|
||||
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}'")
|
||||
return
|
||||
self.logger.info('在此端点规范中未找到可用于测试 "必填查询参数缺失" 的字段。')
|
||||
|
||||
def generate_request_body(self, current_body: Optional[Any]) -> Optional[Any]:
|
||||
# This test case focuses on query parameters, so it does not modify the request body.
|
||||
self.logger.debug(f"{self.id} is focused on query parameters, generate_request_body will not modify the request body.")
|
||||
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:
|
||||
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}')。")
|
||||
return new_params
|
||||
else:
|
||||
self.logger.warning(f"计划移除的查询参数 '{self.removed_field_name}' 在当前查询参数中未找到。")
|
||||
return current_query_params
|
||||
|
||||
def validate_response(self, response_context: APIResponseContext, request_context: APIRequestContext) -> List[ValidationResult]:
|
||||
results = []
|
||||
|
||||
if not self.removed_field_name:
|
||||
results.append(self.passed("跳过测试：在API规范中未找到合适的必填查询参数用于移除测试。"))
|
||||
self.logger.info("由于未识别到可移除的必填查询参数，跳过此测试用例。")
|
||||
return results
|
||||
|
||||
status_code = response_context.status_code
|
||||
json_content = response_context.json_content
|
||||
|
||||
expected_status_codes = [400, 422]
|
||||
specific_error_code_from_appendix_b = "4003"
|
||||
|
||||
msg_prefix = f"当移除必填查询参数 '{self.removed_field_name}' 时，"
|
||||
|
||||
if status_code in expected_status_codes:
|
||||
status_msg = f"{msg_prefix}API响应了预期的错误状态码 {status_code}。"
|
||||
if json_content and isinstance(json_content, dict) and str(json_content.get("code")) == specific_error_code_from_appendix_b:
|
||||
results.append(self.passed(f"{status_msg} 且响应体中包含特定的错误码 '{specific_error_code_from_appendix_b}'。"))
|
||||
self.logger.info(f"正确接收到状态码 {status_code} 和错误码 '{specific_error_code_from_appendix_b}'。")
|
||||
elif json_content and isinstance(json_content, dict) and "code" in json_content:
|
||||
results.append(ValidationResult(passed=True,
|
||||
message=f"{status_msg} 响应体中的错误码为 '{json_content.get('code')}' (期望或类似 '{specific_error_code_from_appendix_b}')。",
|
||||
details=json_content
|
||||
))
|
||||
self.logger.warning(f"接收到状态码 {status_code}，但错误码是 '{json_content.get('code')}' 而不是期望的 '{specific_error_code_from_appendix_b}'。此结果仍标记为通过，因状态码正确。")
|
||||
else:
|
||||
results.append(self.passed(f"{status_msg} 但响应体中未找到特定的错误码字段或响应体结构不符合预期。"))
|
||||
self.logger.info(f"正确接收到状态码 {status_code}，但在响应体中未找到错误码字段或预期结构。")
|
||||
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}"}
|
||||
))
|
||||
self.logger.warning(f"必填查询参数缺失测试失败：期望状态码 {expected_status_codes}，实际为 {status_code}。移除的参数：'{self.removed_field_name}'")
|
||||
|
||||
return results
|
||||
Reference in New Issue
Block a user