add:stage
This commit is contained in:
Vendored
BIN
Binary file not shown.
Vendored
BIN
Binary file not shown.
Vendored
BIN
Binary file not shown.
@@ -0,0 +1,139 @@
|
||||
from typing import Dict, Any, List, Optional
|
||||
|
||||
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, ValidationResult, APIResponseContext, APIRequestContext, TestSeverity
|
||||
|
||||
class ResponseSchemaFormatCheck(BaseAPITestCase):
|
||||
"""
|
||||
检查API响应的schema格式是否符合{"code":int or number or string,"message":"","data": any}的标准格式
|
||||
"""
|
||||
id = "TC-DMS-CORE-SCHEMA-001"
|
||||
name = "DMS核心存储服务API响应格式检查"
|
||||
description = "验证API响应的schema是否符合标准格式:{'code':int or number or string, 'message':string, 'data': any}"
|
||||
severity = TestSeverity.HIGH
|
||||
tags = ["schema", "format", "dms-core", "response"]
|
||||
|
||||
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)
|
||||
|
||||
def validate_response(self, response_context: APIResponseContext, request_context: APIRequestContext) -> List[ValidationResult]:
|
||||
"""
|
||||
仅验证API响应schema的格式,不验证实际响应
|
||||
"""
|
||||
results = []
|
||||
|
||||
# 获取规范中的响应schema
|
||||
status_codes_to_check = ["200", "201", "default"]
|
||||
|
||||
for status_code in status_codes_to_check:
|
||||
response_schema = self._get_response_schema_for_status(status_code)
|
||||
if response_schema:
|
||||
validation_results = self._validate_response_schema_format(response_schema, status_code)
|
||||
results.extend(validation_results)
|
||||
# 找到一个有效的响应schema后就不再继续检查
|
||||
break
|
||||
|
||||
# 如果没有找到任何响应schema,记录为失败
|
||||
if not results:
|
||||
results.append(self.failed(
|
||||
message="无法找到API响应的schema定义,无法验证响应格式。",
|
||||
details={"endpoint": self.endpoint_spec.get("path", "未知")}
|
||||
))
|
||||
|
||||
return results
|
||||
|
||||
def _get_response_schema_for_status(self, status_code: str) -> Optional[Dict[str, Any]]:
|
||||
"""获取指定状态码的响应schema"""
|
||||
responses = self.endpoint_spec.get("responses", {})
|
||||
|
||||
if status_code not in responses:
|
||||
return None
|
||||
|
||||
response_spec = responses[status_code]
|
||||
return self._get_resolved_response_schema(response_spec, status_code)
|
||||
|
||||
def _validate_response_schema_format(self, schema: Dict[str, Any], status_code: str) -> List[ValidationResult]:
|
||||
"""验证响应schema是否符合标准格式"""
|
||||
results = []
|
||||
|
||||
# 如果schema为空,则记录为失败
|
||||
if not schema or not isinstance(schema, dict):
|
||||
results.append(self.failed(
|
||||
message=f"响应schema不是有效的对象: {schema}",
|
||||
details={"status_code": status_code}
|
||||
))
|
||||
return results
|
||||
|
||||
# 解析schema,确保它是一个已解析的schema
|
||||
schema = self._get_resolved_schema(schema)
|
||||
|
||||
# 检查schema是否有properties
|
||||
if "properties" not in schema:
|
||||
results.append(self.failed(
|
||||
message=f"响应schema中缺少'properties'定义",
|
||||
details={"status_code": status_code, "schema": schema}
|
||||
))
|
||||
return results
|
||||
|
||||
properties = schema.get("properties", {})
|
||||
required_fields = schema.get("required", [])
|
||||
|
||||
# 检查必须的字段: code, message, data
|
||||
expected_fields = ["code", "message", "data"]
|
||||
missing_fields = [field for field in expected_fields if field not in properties]
|
||||
|
||||
if missing_fields:
|
||||
results.append(self.failed(
|
||||
message=f"响应schema中缺少必要字段: {', '.join(missing_fields)}",
|
||||
details={
|
||||
"status_code": status_code,
|
||||
"available_fields": list(properties.keys()),
|
||||
"missing_fields": missing_fields
|
||||
}
|
||||
))
|
||||
else:
|
||||
# 检查字段类型
|
||||
type_errors = []
|
||||
|
||||
# 检查code字段类型
|
||||
code_schema = properties.get("code", {})
|
||||
if not( code_schema.get("type") == "integer" or code_schema.get("type") == "number" or code_schema.get("type") == "string"):
|
||||
type_errors.append(f"'code'字段应为integer类型,实际为{code_schema.get('type')}")
|
||||
|
||||
# 检查message字段类型
|
||||
message_schema = properties.get("message", {})
|
||||
if message_schema.get("type") != "string":
|
||||
type_errors.append(f"'message'字段应为string类型,实际为{message_schema.get('type')}")
|
||||
|
||||
# # 检查data字段类型
|
||||
data_schema = properties.get("data", {})
|
||||
# if data_schema.get("type") != "object" and data_schema.get("type") is not None:
|
||||
# type_errors.append(f"'data'字段应为object类型,实际为{data_schema.get('type')}")
|
||||
|
||||
if type_errors:
|
||||
results.append(self.failed(
|
||||
message=f"响应schema中字段类型不符合要求: {'; '.join(type_errors)}",
|
||||
details={
|
||||
"status_code": status_code,
|
||||
"code_schema": code_schema,
|
||||
"message_schema": message_schema,
|
||||
"data_schema": data_schema
|
||||
}
|
||||
))
|
||||
|
||||
# 检查必填字段
|
||||
for field in expected_fields:
|
||||
if field not in required_fields:
|
||||
results.append(ValidationResult(
|
||||
passed=True, # 作为警告而非错误
|
||||
message=f"字段'{field}'在schema中未标记为必填(required)",
|
||||
details={"status_code": status_code, "required_fields": required_fields}
|
||||
))
|
||||
|
||||
# 如果没有错误,则记录为通过
|
||||
if not any(not result.passed for result in results):
|
||||
results.append(self.passed(
|
||||
message="响应schema符合标准格式: {'code':int or number or string, 'message':string, 'data': any}",
|
||||
details={"status_code": status_code}
|
||||
))
|
||||
|
||||
return results
|
||||
@@ -0,0 +1,128 @@
|
||||
import re
|
||||
from typing import Dict, Any, List, Optional
|
||||
|
||||
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, ValidationResult, APIResponseContext, APIRequestContext, TestSeverity
|
||||
|
||||
class URLVersionCheckCase(BaseAPITestCase):
|
||||
"""
|
||||
检查API URL是否包含版本号(如v1, api/v2, v3.0等)并以/api开头
|
||||
"""
|
||||
id = "TC-DMS-URL-VERSION-001"
|
||||
name = "DMS API URL版本号检查"
|
||||
description = "检查API URL是否包含标准格式的版本号,支持的格式包括:v1, api/v2, v3.0, version/1, 1.0等,并且路径需要以/api开头"
|
||||
severity = TestSeverity.MEDIUM
|
||||
tags = ["url", "version", "dms-core", "api-design"]
|
||||
|
||||
# 这个测试用例不需要发送实际请求
|
||||
skip_execution = True
|
||||
|
||||
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)
|
||||
|
||||
def validate_response(self, response_context: APIResponseContext, request_context: APIRequestContext) -> List[ValidationResult]:
|
||||
"""
|
||||
检查API URL是否以/api开头并包含版本号
|
||||
"""
|
||||
results = []
|
||||
|
||||
# 获取API路径
|
||||
path = self.endpoint_spec.get('path', '')
|
||||
if not path:
|
||||
results.append(self.failed(
|
||||
message="无法获取API路径",
|
||||
details={"endpoint_spec_keys": list(self.endpoint_spec.keys())}
|
||||
))
|
||||
return results
|
||||
|
||||
# 检查是否是系统级API(可能不需要遵循标准路径格式)
|
||||
is_system_api = re.match(r'^/(health|ping|status|metrics|system)(/|$)', path)
|
||||
|
||||
# 1. 检查路径是否以/api开头
|
||||
starts_with_api = path.startswith('/api/')
|
||||
|
||||
if not starts_with_api and not is_system_api:
|
||||
results.append(self.failed(
|
||||
message=f"API路径 '{path}' 不是以'/api/'开头",
|
||||
details={"full_path": path, "requirement": "路径必须以'/api/'开头"}
|
||||
))
|
||||
elif starts_with_api:
|
||||
results.append(self.passed(
|
||||
message=f"API路径 '{path}' 正确以'/api/'开头",
|
||||
details={"full_path": path}
|
||||
))
|
||||
|
||||
# 2. 检查路径中是否包含版本号
|
||||
version_patterns = [
|
||||
# 标准版本格式: /v1/, /v2/, /v3/ 等
|
||||
r'/v\d+/',
|
||||
# 带小数点的版本: /v1.0/, /v2.1/ 等
|
||||
r'/v\d+\.\d+/',
|
||||
# 使用 'version' 单词: /version/1/, /version/2/ 等
|
||||
r'/version/\d+/',
|
||||
# API前缀版本: /api/v1/, /api/v2/ 等
|
||||
r'/api/v\d+/',
|
||||
# 直接数字版本: /1/, /2/ (仅在特定位置)
|
||||
r'/api/\d+/',
|
||||
# 特殊格式: 如 /v1-beta/, /v2-alpha/ 等
|
||||
r'/v\d+[\-_](alpha|beta|rc\d*)/',
|
||||
# 年份版本: /2023/, /2024/ 等 (仅在特定位置)
|
||||
r'/20\d{2}/',
|
||||
]
|
||||
|
||||
# 检查是否包含版本号
|
||||
matched_pattern = None
|
||||
version_str = None
|
||||
|
||||
for pattern in version_patterns:
|
||||
match = re.search(pattern, path)
|
||||
if match:
|
||||
matched_pattern = pattern
|
||||
version_str = match.group(0).strip('/')
|
||||
break
|
||||
|
||||
if matched_pattern and version_str:
|
||||
results.append(self.passed(
|
||||
message=f"API路径 '{path}' 包含版本标识: '{version_str}'",
|
||||
details={
|
||||
"pattern_matched": matched_pattern,
|
||||
"version_string": version_str,
|
||||
"full_path": path
|
||||
}
|
||||
))
|
||||
else:
|
||||
# 特殊情况:检查是否是根API或系统级API(可能不需要版本号)
|
||||
if is_system_api:
|
||||
results.append(self.passed(
|
||||
message=f"API路径 '{path}' 是系统级API,不需要版本号",
|
||||
details={"full_path": path, "api_type": "system"}
|
||||
))
|
||||
else:
|
||||
results.append(self.failed(
|
||||
message=f"API路径 '{path}' 不包含任何已知格式的版本标识",
|
||||
details={
|
||||
"full_path": path,
|
||||
"supported_patterns": [p.replace('\\d+', 'N').replace('\\d{2}', 'NN') for p in version_patterns]
|
||||
}
|
||||
))
|
||||
|
||||
# 提供改进建议
|
||||
# 确保建议路径始终以/api开头并包含版本号
|
||||
base_path_parts = path.split('/')
|
||||
base_path_parts = [p for p in base_path_parts if p] # 移除空字符串
|
||||
|
||||
if not starts_with_api:
|
||||
# 如果不是以/api开头,建议路径应该是/api/v1/原始路径
|
||||
suggested_path = f"/api/v1/{'/'.join(base_path_parts)}"
|
||||
else:
|
||||
# 如果已经以/api开头但缺少版本号,插入v1在api之后
|
||||
suggested_path = "/api/v1"
|
||||
if len(base_path_parts) > 1: # 有api后面的部分
|
||||
suggested_path += f"/{'/'.join(base_path_parts[1:])}"
|
||||
|
||||
results.append(ValidationResult(
|
||||
passed=False,
|
||||
message=f"建议将路径修改为符合规范的格式,例如: '{suggested_path}'",
|
||||
details={"original_path": path, "suggested_path": suggested_path}
|
||||
))
|
||||
|
||||
return results
|
||||
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
Vendored
BIN
Binary file not shown.
@@ -0,0 +1,40 @@
|
||||
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, ValidationResult, APIResponseContext, APIRequestContext, TestSeverity
|
||||
import re
|
||||
from typing import Dict, Any, List, Optional
|
||||
|
||||
class ResourcePathNounCheckTestCase(BaseAPITestCase):
|
||||
id = "TC-RESTful-002"
|
||||
name = "资源路径名词检查"
|
||||
description = "验证API路径中是否使用名词而非动词来表示资源。"
|
||||
severity = TestSeverity.MEDIUM
|
||||
tags = ["normative", "restful", "url-structure"]
|
||||
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.common_verbs = {"get", "create", "update", "delete", "post", "put", "add", "remove", "set"}
|
||||
|
||||
def validate_response(self, response_context: APIResponseContext, request_context: APIRequestContext) -> List[ValidationResult]:
|
||||
path = self.endpoint_spec['path']
|
||||
path_segments = [seg for seg in path.split('/') if seg and '{' not in seg]
|
||||
|
||||
is_valid = True
|
||||
offending_verbs = []
|
||||
|
||||
for segment in path_segments:
|
||||
# 移除版本号等非资源路径部分
|
||||
if re.match(r'v\d+', segment):
|
||||
continue
|
||||
|
||||
# 检查分段是否像动词
|
||||
# 为了避免误判 (e.g., /dataset),只对完全匹配的动词进行判断
|
||||
if segment.lower() in self.common_verbs:
|
||||
is_valid = False
|
||||
offending_verbs.append(segment)
|
||||
|
||||
if not is_valid:
|
||||
message = f"路径 '{path}' 中可能包含动词: {', '.join(offending_verbs)},RESTful风格建议资源路径使用名词。"
|
||||
return [self.failed(message, details={'path': path, 'detected_verbs': offending_verbs})]
|
||||
|
||||
message = f"路径 '{path}' 符合资源名词命名规范。"
|
||||
return [self.passed(message)]
|
||||
@@ -0,0 +1,94 @@
|
||||
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, ValidationResult, APIResponseContext, APIRequestContext, TestSeverity
|
||||
import re
|
||||
from typing import Dict, Any, List, Optional
|
||||
from ddms_compliance_suite.utils import schema_utils
|
||||
|
||||
class TimeFormatCheckTestCase(BaseAPITestCase):
|
||||
id = "TC-RESTful-003"
|
||||
name = "时间字段ISO 8601格式检查"
|
||||
description = "验证返回的时间字段是否遵循 ISO 8601 格式。此检查为静态检查,会检查规范中 `format` 为 `date-time` 的字段,以及常见的时间字段名(如 createTime, update_time 等),是否包含推荐的 `pattern`。"
|
||||
severity = TestSeverity.MEDIUM
|
||||
tags = ["normative", "schema", "time-format"]
|
||||
|
||||
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)
|
||||
# 推荐的 pattern
|
||||
self.recommended_iso_8601_pattern = r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}([+-]\d{2}:\d{2}|Z)$'
|
||||
# 常见时间字段名称(小写,用于不区分大小写匹配)
|
||||
self.time_field_names = {
|
||||
"createtime", "updatetime", "starttime", "endtime", "publishtime", "timestamp",
|
||||
"created_at", "updated_at", "create_time", "update_time", "start_time", "end_time",
|
||||
"gmtcreate", "gmtmodified", "datetime", "date_time", "time"
|
||||
}
|
||||
|
||||
def validate_response(self, response_context: APIResponseContext, request_context: APIRequestContext) -> List[ValidationResult]:
|
||||
results = []
|
||||
# 此检查为静态检查,分析API规范中的响应部分
|
||||
responses = self.endpoint_spec.get('responses', {})
|
||||
for status_code, response_spec in responses.items():
|
||||
# 通常只关心成功的响应
|
||||
if not status_code.startswith('2'):
|
||||
continue
|
||||
|
||||
content = self._get_resolved_schema(response_spec.get('content', {}))
|
||||
if content:
|
||||
for media_type, media_spec in content.items():
|
||||
if 'schema' in media_spec:
|
||||
self._check_schema_properties(media_spec['schema'], results)
|
||||
|
||||
if not results:
|
||||
return [self.passed("在API规范中未找到可供静态检查的时间相关字段(如 format: date-time 或 常见时间字段名)。")]
|
||||
|
||||
return results
|
||||
|
||||
def _check_schema_properties(self, schema, results, path=""):
|
||||
if not schema or not isinstance(schema, dict):
|
||||
return
|
||||
|
||||
# 处理 allOf, oneOf, anyOf
|
||||
for keyword in ['allOf', 'oneOf', 'anyOf']:
|
||||
if keyword in schema:
|
||||
for sub_schema in schema[keyword]:
|
||||
self._check_schema_properties(sub_schema, results, path)
|
||||
|
||||
if 'properties' in schema:
|
||||
for prop_name, prop_spec in schema['properties'].items():
|
||||
prop_spec = self._get_resolved_schema(prop_spec)
|
||||
current_path = f"{path}.{prop_name}" if path else prop_name
|
||||
|
||||
# 检查是否为时间相关字段
|
||||
is_datetime_format = prop_spec.get('format') == 'date-time'
|
||||
is_common_time_name = prop_name.lower() in self.time_field_names
|
||||
|
||||
# 必须是string类型,且满足 (format是date-time) 或 (字段名在常见列表里)
|
||||
if prop_spec.get('type') == 'string' and (is_datetime_format or is_common_time_name):
|
||||
pattern = prop_spec.get('pattern')
|
||||
|
||||
# 确定字段来源以提供更清晰的消息
|
||||
source_reason = ""
|
||||
if is_datetime_format and is_common_time_name:
|
||||
source_reason = f"(format: date-time, name: '{prop_name}')"
|
||||
elif is_datetime_format:
|
||||
source_reason = f"(format: date-time)"
|
||||
else: # is_common_time_name
|
||||
source_reason = f"(name: '{prop_name}')"
|
||||
|
||||
message = f"时间字段 '{current_path}' {source_reason} "
|
||||
if not pattern:
|
||||
results.append(self.failed(
|
||||
message + f"缺少建议的 `pattern` ({self.recommended_iso_8601_pattern}) 来强制执行ISO 8601格式。",
|
||||
details={'field': current_path}
|
||||
))
|
||||
elif pattern != self.recommended_iso_8601_pattern:
|
||||
results.append(self.failed(
|
||||
message + f"其 `pattern` ('{pattern}') 与建议的模式不完全匹配。",
|
||||
details={'field': current_path, 'current_pattern': pattern, 'recommended': self.recommended_iso_8601_pattern}
|
||||
))
|
||||
else:
|
||||
results.append(self.passed(message + "已定义了建议的 `pattern` 用于格式校验。"))
|
||||
|
||||
# 递归检查
|
||||
if 'properties' in prop_spec or 'allOf' in prop_spec or 'oneOf' in prop_spec or 'anyOf' in prop_spec:
|
||||
self._check_schema_properties(prop_spec, results, current_path)
|
||||
elif prop_spec.get('type') == 'array' and 'items' in prop_spec:
|
||||
self._check_schema_properties(self._get_resolved_schema(prop_spec['items']), results, f"{current_path}[]")
|
||||
@@ -0,0 +1,59 @@
|
||||
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, ValidationResult, APIResponseContext, APIRequestContext, TestSeverity
|
||||
import re
|
||||
from typing import Dict, Any, List, Optional
|
||||
# TODO 获取资源的时候复数(get方法list)
|
||||
class ResourceCollectionPluralCheckTestCase(BaseAPITestCase):
|
||||
id = "TC-RESTful-004"
|
||||
name = "资源集合复数命名检查"
|
||||
description = "验证表示资源集合的路径是否使用复数形式。动词(如push、send等)不需要使用复数形式。"
|
||||
severity = TestSeverity.MEDIUM
|
||||
tags = ["normative", "restful", "url-structure"]
|
||||
|
||||
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)
|
||||
# 常见的API路径中的动词,这些不需要使用复数形式
|
||||
self.common_verbs = {
|
||||
"push", "send", "publish", "subscribe", "create", "update", "delete",
|
||||
"get", "set", "add", "remove", "search", "query", "find", "calculate",
|
||||
"process", "validate", "verify", "check", "analyze", "export", "import",
|
||||
"upload", "download", "sync", "login", "logout", "register", "activate",
|
||||
"deactivate", "approve", "reject", "cancel", "confirm", "notify"
|
||||
}
|
||||
|
||||
# 已知的单数形式名词,即使不以's'结尾也是正确的
|
||||
self.known_singulars = {
|
||||
"status", "gas", "analysis", "data", "info", "metadata", "media",
|
||||
"equipment", "staff", "fish", "sheep", "deer", "series", "species",
|
||||
"aircraft", "offspring", "feedback", "content", "news"
|
||||
}
|
||||
|
||||
def validate_response(self, response_context: APIResponseContext, request_context: APIRequestContext) -> List[ValidationResult]:
|
||||
path = self.endpoint_spec['path']
|
||||
method = self.endpoint_spec['method']
|
||||
|
||||
# 这个检查通常适用于返回列表的GET请求,或者创建资源的POST请求
|
||||
if method.lower() not in ['get', 'post']:
|
||||
return [self.passed(f"跳过检查:{method} 方法,不适用于资源集合复数检查。")]
|
||||
|
||||
path_segments = [seg for seg in path.strip('/').split('/') if '{' not in seg and not re.match(r'v\d+', seg)]
|
||||
|
||||
if not path_segments:
|
||||
return [self.passed("跳过检查:路径不含有效分段。")]
|
||||
|
||||
resource_segment = path_segments[-1]
|
||||
|
||||
# 检查是否为动词
|
||||
if resource_segment.lower() in self.common_verbs:
|
||||
return [self.passed(f"路径 '{path}' 的最后一个路径分段 '{resource_segment}' 是动词,不需要使用复数形式。")]
|
||||
|
||||
# 检查是否为已知的单数形式名词
|
||||
if resource_segment.lower() in self.known_singulars:
|
||||
return [self.passed(f"路径 '{path}' 的资源名 '{resource_segment}' 是已知的单数形式名词,符合规范。")]
|
||||
|
||||
# 对于其他名词,检查是否使用复数形式
|
||||
if not resource_segment.endswith('s'):
|
||||
message = f"路径 '{path}' 的最后一个路径分段 '{resource_segment}' 可能不是复数形式,建议对资源集合使用复数命名。"
|
||||
return [self.failed(message, details={'path': path, 'segment': resource_segment})]
|
||||
|
||||
message = f"路径 '{path}' 的资源集合命名 '{resource_segment}' 符合复数命名规范。"
|
||||
return [self.passed(message)]
|
||||
@@ -0,0 +1,185 @@
|
||||
import re
|
||||
from typing import Dict, Any, List, Optional
|
||||
|
||||
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, ValidationResult, APIResponseContext, APIRequestContext, TestSeverity
|
||||
from ddms_compliance_suite.utils.common_utils import is_camel_case
|
||||
from ddms_compliance_suite.utils import schema_utils
|
||||
|
||||
class CoreNamingStructureTestCase(BaseAPITestCase):
|
||||
id = "TC-RESTful-001"
|
||||
name = "核心命名与结构规范检查"
|
||||
description = "统一验证API的命名与结构是否遵循规范。包括:1)模块名全小写且用中划线连接;2)URL路径参数使用下划线命名法(snake_case);3)查询参数和请求体字段使用小驼峰命名法(camelCase);4)响应中的空数组为[]而非null;5)数组类型数据被包裹在list字段中。"
|
||||
severity = TestSeverity.HIGH
|
||||
tags = ["normative", "restful", "structure", "naming-convention"]
|
||||
|
||||
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)
|
||||
|
||||
def validate_response(self, response_context: APIResponseContext, request_context: APIRequestContext) -> List[ValidationResult]:
|
||||
results = []
|
||||
|
||||
# 静态检查,只分析API规范
|
||||
self._check_module_name(results)
|
||||
self._check_request_params_camel_case(results)
|
||||
|
||||
# 动态检查,需要分析实际响应
|
||||
# 规则 "响应中的空数组为[]而非null" 和 "数组类型数据被包裹在list字段中" 需要在实际调用后进行
|
||||
# 这里我们只对成功的响应进行检查
|
||||
if 200 <= response_context.status_code < 300:
|
||||
self._check_response_array_format(response_context, results)
|
||||
else:
|
||||
results.append(self.passed("跳过响应体检查:非成功状态码,不适用数组格式检查。"))
|
||||
|
||||
return results
|
||||
|
||||
def _check_module_name(self, results: List[ValidationResult]):
|
||||
"""检查模块名是否为全小写且用中划线连接"""
|
||||
path = self.endpoint_spec['path']
|
||||
module_name_match = re.search(r'/api/([^/]+)/', path)
|
||||
if module_name_match:
|
||||
module_name = module_name_match.group(1)
|
||||
# 模块名可以只是小写字母,但如果包含连接符,必须是中划线
|
||||
is_valid = all(c.islower() or c.isdigit() or c == '-' for c in module_name) and '_' not in module_name
|
||||
if is_valid:
|
||||
results.append(self.passed(f"模块名 '{module_name}' 格式正确 (全小写/数字/中划线)。"))
|
||||
else:
|
||||
results.append(self.failed(f"模块名 '{module_name}' 格式不正确。应为全小写字母、数字和中划线的组合。",
|
||||
details={'path': path, 'module': module_name}))
|
||||
else:
|
||||
results.append(self.failed(f"无法从路径 '{path}' 中提取模块名(格式应为 /api/module-name/...)。",
|
||||
details={'path': path}))
|
||||
|
||||
def _check_request_params_camel_case(self, results: List[ValidationResult]):
|
||||
"""检查请求参数命名规范:
|
||||
- URL路径参数应使用下划线命名法(snake_case)
|
||||
- 请求体和查询参数应使用小驼峰命名法(camelCase)
|
||||
- HTTP头部有特殊规范,不检查
|
||||
"""
|
||||
parameters = self.endpoint_spec.get('parameters', [])
|
||||
|
||||
# 定义HTTP头部例外(不检查)
|
||||
header_exceptions = ['Authorization', 'X-Tenant-ID', 'X-Data-Domain', 'tenant-id', 'Content-Type']
|
||||
|
||||
for param in parameters:
|
||||
param_name = param.get('name')
|
||||
param_in = param.get('in')
|
||||
|
||||
# 跳过HTTP头部参数
|
||||
if param_in == 'header' and param_name in header_exceptions:
|
||||
continue
|
||||
|
||||
# 路径参数使用下划线命名法(snake_case)
|
||||
if param_in == 'path':
|
||||
# 下划线命名法:全小写字母、数字和下划线,不允许连续下划线,不能以下划线开头或结尾
|
||||
is_valid_snake_case = re.match(r'^[a-z][a-z0-9_]*$', param_name) is not None and '__' not in param_name and not param_name.endswith('_')
|
||||
if not is_valid_snake_case:
|
||||
results.append(self.failed(f"路径参数 '{param_name}' 不符合下划线命名法(snake_case)规范。应为小写字母、数字和单下划线组合,不能以下划线结尾。",
|
||||
details={'parameter': param_name, 'location': param_in}))
|
||||
# 查询参数使用小驼峰命名法(camelCase)
|
||||
elif param_in == 'query':
|
||||
if not is_camel_case(param_name):
|
||||
results.append(self.failed(f"查询参数 '{param_name}' 不是小驼峰格式。",
|
||||
details={'parameter': param_name, 'location': param_in}))
|
||||
|
||||
# 检查请求体
|
||||
body_schema = self._get_resolved_request_body_schema()
|
||||
if body_schema:
|
||||
self._check_schema_properties_camel_case(body_schema, results)
|
||||
|
||||
def _check_schema_properties_camel_case(self, schema, results, path=""):
|
||||
if not schema or not isinstance(schema, dict):
|
||||
return
|
||||
|
||||
if 'properties' in schema:
|
||||
for prop_name, prop_spec in schema['properties'].items():
|
||||
if not is_camel_case(prop_name):
|
||||
full_path = f"{path}.{prop_name}" if path else prop_name
|
||||
results.append(self.failed(f"请求体字段 '{full_path}' 不是小驼峰格式。",
|
||||
details={'field': full_path}))
|
||||
|
||||
prop_spec_resolved = self._get_resolved_schema(prop_spec)
|
||||
if 'properties' in prop_spec_resolved or 'items' in prop_spec_resolved:
|
||||
self._check_schema_properties_camel_case(prop_spec_resolved, results, f"{path}.{prop_name}" if path else prop_name)
|
||||
|
||||
def _check_response_array_format(self, response_context: APIResponseContext, results: List[ValidationResult]):
|
||||
"""检查响应中的空数组和数组包裹"""
|
||||
json_content = response_context.json_content
|
||||
if json_content is None:
|
||||
# 如果响应体为空,则跳过检查
|
||||
results.append(self.passed("响应体为空,跳过数组格式检查。"))
|
||||
return
|
||||
|
||||
# 检查 "数组类型数据被包裹在list字段中"
|
||||
# 这条规则比较模糊,这里理解为:如果响应体是一个以数组为核心的列表,那么这个数组的key应该是'list'
|
||||
if isinstance(json_content, dict) and len(json_content) > 0:
|
||||
list_keys = [k for k, v in json_content.items() if isinstance(v, list)]
|
||||
if len(list_keys) == 1 and list_keys[0] != 'list':
|
||||
results.append(self.failed(f"响应中包含一个主列表,但其键名 '{list_keys[0]}' 不是 'list'。",
|
||||
details={'keys': list(json_content.keys())}))
|
||||
elif len(list_keys) > 1 and 'list' not in list_keys:
|
||||
results.append(self.failed(f"响应中包含多个列表,但没有一个的键名是 'list'。",
|
||||
details={'keys': list(json_content.keys())}))
|
||||
|
||||
# 检查 "响应中的空数组为[]而非null"
|
||||
# 查找匹配的响应定义时,需要兼容 status_code, status_code_family (e.g., 2XX), 和 default
|
||||
responses = self.endpoint_spec.get('responses', {})
|
||||
self.logger.info(f"responses: {responses}")
|
||||
status_code = response_context.status_code
|
||||
print("status_code: ", status_code)
|
||||
status_code_str = str(status_code)
|
||||
status_code_family = f"{status_code_str[0]}XX" # e.g., "2XX"
|
||||
self.logger.info(f"检查响应定义: status_code={status_code}, 可用响应定义={list(responses.keys())}")
|
||||
|
||||
# 按优先级查找响应定义:精确状态码 > 状态码族 > 默认状态码200 > default
|
||||
response_spec = None
|
||||
print("responses: ", responses)
|
||||
if status_code_str in responses:
|
||||
response_spec = responses[status_code_str]
|
||||
self.logger.info(f"找到精确状态码 {status_code_str} 的响应定义")
|
||||
elif status_code_family in responses:
|
||||
response_spec = responses[status_code_family]
|
||||
self.logger.info(f"找到状态码族 {status_code_family} 的响应定义")
|
||||
elif '200' in responses and 200 <= status_code < 300: # 对于2XX成功响应,尝试使用200定义
|
||||
response_spec = responses['200']
|
||||
self.logger.info(f"未找到状态码 {status_code_str} 的响应定义,使用默认成功状态码200的定义")
|
||||
elif 'default' in responses:
|
||||
response_spec = responses['default']
|
||||
self.logger.info(f"使用default响应定义")
|
||||
|
||||
if not response_spec:
|
||||
results.append(self.passed(f"规范中未找到响应码 {status_code} 或其类别({status_code_family}, default)的匹配定义,跳过空数组与null的检查。"))
|
||||
return
|
||||
|
||||
# 使用基类中定义好的工具函数获取响应schema
|
||||
schema = self._get_resolved_response_schema(response_spec=response_spec)
|
||||
|
||||
if not schema:
|
||||
results.append(self.passed(f"规范中响应码 {status_code} 的定义中未找到Schema,跳过空数组与null的检查。"))
|
||||
return
|
||||
|
||||
self._validate_null_for_array(json_content, schema, results, "")
|
||||
|
||||
def _validate_null_for_array(self, data: Any, schema: Dict[str, Any], results: List[ValidationResult], path: str):
|
||||
if not schema:
|
||||
return
|
||||
|
||||
schema = self._get_resolved_schema(schema)
|
||||
|
||||
if schema.get('type') == 'array' and data is None:
|
||||
results.append(self.failed(f"响应中字段 '{path}' 的值为 null,但其在规范中定义为数组,应返回 []。", details={'field': path}))
|
||||
return
|
||||
|
||||
if isinstance(data, dict) and 'properties' in schema:
|
||||
for prop_name, prop_schema in schema['properties'].items():
|
||||
if prop_name in data:
|
||||
new_path = f"{path}.{prop_name}" if path else prop_name
|
||||
self._validate_null_for_array(data[prop_name], prop_schema, results, new_path)
|
||||
|
||||
elif isinstance(data, list) and 'items' in schema:
|
||||
item_schema = schema.get('items')
|
||||
for i, item in enumerate(data):
|
||||
new_path = f"{path}[{i}]"
|
||||
self._validate_null_for_array(item, item_schema, results, new_path)
|
||||
|
||||
|
||||
|
||||
BIN
Binary file not shown.
@@ -0,0 +1,141 @@
|
||||
from ddms_compliance_suite.test_framework_core import BaseAPITestCase, ValidationResult, APIResponseContext, APIRequestContext, TestSeverity
|
||||
from typing import Dict, Any, List, Optional
|
||||
import re
|
||||
|
||||
class PaginationParamsCheckTestCase(BaseAPITestCase):
|
||||
"""
|
||||
检查API请求中是否包含标准分页参数:pageNo、pageSize和isSearchCount
|
||||
只有名称含有"查询"一类,并且不含有"详情"一类的API才应用这个验证
|
||||
"""
|
||||
id = "TC-DMS-PAGINATION-001"
|
||||
name = "分页参数检查"
|
||||
description = "检查API请求参数中是否包含标准分页参数:pageNo、pageSize和isSearchCount。只有名称含有'查询'、'列表'等并且不含有'详情'一类的API才应用此验证。"
|
||||
severity = TestSeverity.MEDIUM
|
||||
tags = ["pagination", "params", "backend-guide"]
|
||||
|
||||
# 这个测试用例不需要发送实际请求
|
||||
skip_execution = True
|
||||
|
||||
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)
|
||||
|
||||
# 定义需要验证的API名称关键词
|
||||
self.include_keywords = ["查询", "列表", "分页", "page", "list", "query", "search", "find"]
|
||||
# 定义排除的API名称关键词
|
||||
self.exclude_keywords = ["详情", "明细", "detail", "info", "get", "查看"]
|
||||
|
||||
def validate_response(self, response_context: APIResponseContext, request_context: APIRequestContext) -> List[ValidationResult]:
|
||||
"""
|
||||
检查API请求中是否包含标准分页参数
|
||||
"""
|
||||
results = []
|
||||
|
||||
# 获取API路径和方法
|
||||
path = self.endpoint_spec.get('path', '')
|
||||
method = self.endpoint_spec.get('method', '').lower()
|
||||
|
||||
# 获取API的摘要和描述信息
|
||||
summary = self.endpoint_spec.get('summary', '')
|
||||
description = self.endpoint_spec.get('description', '')
|
||||
operation_id = self.endpoint_spec.get('operationId', '')
|
||||
|
||||
# 组合所有可能包含API名称或功能描述的字段
|
||||
api_description_text = f"{summary} {description} {operation_id} {path}".lower()
|
||||
|
||||
# 检查是否包含需要验证的关键词,且不包含排除的关键词
|
||||
contains_include_keyword = any(keyword.lower() in api_description_text for keyword in self.include_keywords)
|
||||
contains_exclude_keyword = any(keyword.lower() in api_description_text for keyword in self.exclude_keywords)
|
||||
|
||||
# 如果不满足准入规则,直接返回通过
|
||||
if not contains_include_keyword or contains_exclude_keyword:
|
||||
results.append(self.passed(
|
||||
message=f"跳过检查:API不符合分页参数检查的准入规则(需包含'查询'/'列表'等关键词,且不包含'详情'等关键词)",
|
||||
details={
|
||||
"path": path,
|
||||
"method": method.upper(),
|
||||
"summary": summary,
|
||||
"contains_include_keyword": contains_include_keyword,
|
||||
"contains_exclude_keyword": contains_exclude_keyword
|
||||
}
|
||||
))
|
||||
return results
|
||||
|
||||
# 如果是GET请求或可能返回列表的请求,才进行检查
|
||||
if method not in ['get', 'post']:
|
||||
results.append(self.passed(
|
||||
message=f"跳过检查:{method.upper()} 方法,不适用于分页参数检查"
|
||||
))
|
||||
return results
|
||||
|
||||
# 初始化检查结果
|
||||
found_page_no = False
|
||||
found_page_size = False
|
||||
found_is_search_count = False
|
||||
|
||||
# 检查查询参数
|
||||
parameters = self.endpoint_spec.get('parameters', [])
|
||||
for param in parameters:
|
||||
param_name = param.get('name', '')
|
||||
param_in = param.get('in', '')
|
||||
|
||||
if param_in == 'query':
|
||||
if param_name == 'pageNo':
|
||||
found_page_no = True
|
||||
elif param_name == 'pageSize':
|
||||
found_page_size = True
|
||||
elif param_name == 'isSearchCount':
|
||||
found_is_search_count = True
|
||||
|
||||
# 检查请求体(如果是POST请求)
|
||||
if method == 'post':
|
||||
request_body = self.endpoint_spec.get('requestBody', {})
|
||||
content = request_body.get('content', {})
|
||||
|
||||
for media_type, media_content in content.items():
|
||||
if 'schema' in media_content:
|
||||
schema = self._get_resolved_schema(media_content['schema'])
|
||||
if 'properties' in schema:
|
||||
properties = schema['properties']
|
||||
|
||||
# 检查请求体属性
|
||||
if 'pageNo' in properties:
|
||||
found_page_no = True
|
||||
|
||||
if 'pageSize' in properties:
|
||||
found_page_size = True
|
||||
|
||||
if 'isSearchCount' in properties:
|
||||
found_is_search_count = True
|
||||
|
||||
# 汇总检查结果
|
||||
if found_page_no and found_page_size and found_is_search_count:
|
||||
results.append(self.passed(
|
||||
message=f"API请求包含所有标准分页参数:pageNo、pageSize和isSearchCount",
|
||||
details={"path": path, "method": method.upper()}
|
||||
))
|
||||
else:
|
||||
# 计算缺失的参数
|
||||
missing_params = []
|
||||
if not found_page_no:
|
||||
missing_params.append("pageNo")
|
||||
if not found_page_size:
|
||||
missing_params.append("pageSize")
|
||||
if not found_is_search_count:
|
||||
missing_params.append("isSearchCount")
|
||||
|
||||
if missing_params:
|
||||
results.append(self.failed(
|
||||
message=f"API请求缺少标准分页参数:{', '.join(missing_params)}",
|
||||
details={
|
||||
"path": path,
|
||||
"method": method.upper(),
|
||||
"missing_params": missing_params,
|
||||
"found_params": {
|
||||
"pageNo": found_page_no,
|
||||
"pageSize": found_page_size,
|
||||
"isSearchCount": found_is_search_count
|
||||
}
|
||||
}
|
||||
))
|
||||
|
||||
return results
|
||||
Reference in New Issue
Block a user