mvp

custom_testcases/井筒/RESTful接口设计要求/tc_normative_002.py

@@ -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)] 

custom_testcases/井筒/RESTful接口设计要求/tc_normative_003.py

@@ -0,0 +1,79 @@
+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 = "验证返回的时间字段是否遵循 YYYY-MM-DDTHH:MM:SS+08:00 的ISO 8601格式。此检查为静态检查，验证规范中`string`类型且`format`为`date-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)$'
+
+    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规范中未找到可供静态检查的时间相关字段（类型为string且格式为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
+                
+                # 检查类型为string且格式为date-time的字段
+                if prop_spec.get('type') == 'string' and prop_spec.get('format') == 'date-time':
+                    pattern = prop_spec.get('pattern')
+                    message = f"时间字段 '{current_path}' (format: date-time) "
+                    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}[]")
+
+    def _get_resolved_schema(self, schema_or_ref):
+        if '$ref' in schema_or_ref:
+            return schema_utils.util_resolve_ref(schema_or_ref['$ref'], self.global_api_spec)
+        return schema_or_ref 

custom_testcases/井筒/RESTful接口设计要求/tc_normative_004.py

@@ -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)] 

custom_testcases/井筒/RESTful接口设计要求/tc_normative_combined_001.py

@@ -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)
+
+
+