llm 通过json列表判断

2025-06-19 18:33:03 +08:00
parent d3333d4f90
commit ba175cb2ae
32 changed files with 12241 additions and 8427 deletions
@@ -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