compliance/memory-bank/api_spec_parser_design.md
gongwenxin df90a5377f mvp
2025-06-16 14:49:49 +08:00

13 KiB
Raw Blame History

API规范解析框架设计

背景与需求

当前系统需要处理多种API规范格式包括YAPI、Swagger/OpenAPI 2.0和OpenAPI 3.0等。目前的实现在BaseAPITestCase类中包含了针对不同格式的特定处理逻辑,这导致:

  1. 代码重复和维护困难
  2. 处理逻辑分散在多个地方
  3. 添加新格式支持需要修改多处代码
  4. 测试用例需要了解底层规范格式的细节

我们需要一个统一的解析框架将不同格式的API规范转换为一致的内部表示使测试用例能够以统一的方式访问API规范信息而不必关心原始格式的差异。

设计目标

  1. 统一接口提供一套统一的接口来访问API规范信息无论原始格式如何
  2. 可扩展性易于添加新的API规范格式支持
  3. 完全解析:在解析阶段处理所有的引用和格式特定的细节
  4. 一致性:确保不同格式的规范被转换为相同的内部表示
  5. 性能优化:减少重复解析和处理

架构设计

1. 核心组件

1.1 统一解析器接口 (APISpecParser)

from abc import ABC, abstractmethod
from typing import Dict, Any, Optional

class APISpecParser(ABC):
    """API规范解析器的抽象基类"""
    
    @abstractmethod
    def parse(self, spec_content: Dict[str, Any]) -> Dict[str, Any]:
        """
        解析API规范内容返回统一格式的内部表示
        
        Args:
            spec_content: 原始API规范内容
            
        Returns:
            统一格式的API规范内部表示
        """
        pass
        
    @abstractmethod
    def detect_format(self, spec_content: Dict[str, Any]) -> str:
        """
        检测API规范的格式
        
        Args:
            spec_content: 原始API规范内容
            
        Returns:
            规范格式的标识符,如 'yapi', 'openapi2', 'openapi3'
        """
        pass

1.2 格式特定的解析器实现

class YAPIParser(APISpecParser):
    """YAPI格式解析器"""
    
    def parse(self, spec_content: Dict[str, Any]) -> Dict[str, Any]:
        # YAPI特定的解析逻辑
        # 转换为统一的内部表示
        pass
        
    def detect_format(self, spec_content: Dict[str, Any]) -> str:
        # 检测是否为YAPI格式
        if self._is_yapi_format(spec_content):
            return 'yapi'
        return ''
        
    def _is_yapi_format(self, spec_content: Dict[str, Any]) -> bool:
        # 判断是否为YAPI格式的逻辑
        pass

class OpenAPI2Parser(APISpecParser):
    """OpenAPI 2.0 (Swagger)格式解析器"""
    
    def parse(self, spec_content: Dict[str, Any]) -> Dict[str, Any]:
        # OpenAPI 2.0特定的解析逻辑
        pass
        
    def detect_format(self, spec_content: Dict[str, Any]) -> str:
        # 检测是否为OpenAPI 2.0格式
        if self._is_openapi2_format(spec_content):
            return 'openapi2'
        return ''
        
    def _is_openapi2_format(self, spec_content: Dict[str, Any]) -> bool:
        # 判断是否为OpenAPI 2.0格式的逻辑
        pass

class OpenAPI3Parser(APISpecParser):
    """OpenAPI 3.0格式解析器"""
    
    def parse(self, spec_content: Dict[str, Any]) -> Dict[str, Any]:
        # OpenAPI 3.0特定的解析逻辑
        pass
        
    def detect_format(self, spec_content: Dict[str, Any]) -> str:
        # 检测是否为OpenAPI 3.0格式
        if self._is_openapi3_format(spec_content):
            return 'openapi3'
        return ''
        
    def _is_openapi3_format(self, spec_content: Dict[str, Any]) -> bool:
        # 判断是否为OpenAPI 3.0格式的逻辑
        pass

1.3 解析器工厂 (APISpecParserFactory)

class APISpecParserFactory:
    """API规范解析器工厂用于创建适合特定规范格式的解析器"""
    
    def __init__(self):
        self.parsers = [
            YAPIParser(),
            OpenAPI2Parser(),
            OpenAPI3Parser()
        ]
        
    def get_parser(self, spec_content: Dict[str, Any]) -> Optional[APISpecParser]:
        """
        根据规范内容自动选择合适的解析器
        
        Args:
            spec_content: 原始API规范内容
            
        Returns:
            适合处理该规范的解析器实例如果没有找到则返回None
        """
        for parser in self.parsers:
            format_type = parser.detect_format(spec_content)
            if format_type:
                return parser
                
        return None
        
    def register_parser(self, parser: APISpecParser):
        """
        注册新的解析器
        
        Args:
            parser: 解析器实例
        """
        self.parsers.append(parser)

1.4 统一API规范管理器 (UnifiedAPISpecManager)

class UnifiedAPISpecManager:
    """统一API规范管理器负责解析和提供统一的API规范访问接口"""
    
    def __init__(self):
        self.parser_factory = APISpecParserFactory()
        self.cached_specs = {}  # 缓存已解析的规范
        
    def parse_spec(self, spec_content: Dict[str, Any], spec_id: str = None) -> Dict[str, Any]:
        """
        解析API规范返回统一格式的内部表示
        
        Args:
            spec_content: 原始API规范内容
            spec_id: 规范的唯一标识符,用于缓存
            
        Returns:
            统一格式的API规范内部表示
        """
        if spec_id and spec_id in self.cached_specs:
            return self.cached_specs[spec_id]
            
        parser = self.parser_factory.get_parser(spec_content)
        if not parser:
            raise ValueError("无法识别的API规范格式")
            
        parsed_spec = parser.parse(spec_content)
        
        if spec_id:
            self.cached_specs[spec_id] = parsed_spec
            
        return parsed_spec
        
    def register_custom_parser(self, parser: APISpecParser):
        """
        注册自定义解析器
        
        Args:
            parser: 自定义解析器实例
        """
        self.parser_factory.register_parser(parser)

2. 统一内部表示格式

所有解析器都应该将原始规范转换为一个统一的内部表示格式,该格式应该包含以下核心元素:

{
    "info": {
        "title": "API标题",
        "version": "API版本",
        "description": "API描述"
    },
    "paths": {
        "/path/to/resource": {
            "get": {
                "summary": "操作摘要",
                "description": "操作描述",
                "parameters": [...],  # 统一格式的参数列表
                "requestBody": {...},  # 统一格式的请求体定义
                "responses": {
                    "200": {
                        "description": "成功响应",
                        "content": {
                            "application/json": {
                                "schema": {...}  # 统一格式的响应schema
                            }
                        }
                    }
                }
            }
        }
    },
    "components": {
        "schemas": {...},  # 统一格式的schema定义
        "parameters": {...},  # 统一格式的参数定义
        "responses": {...}  # 统一格式的响应定义
    }
}

3. 解析过程中的标准化处理

在解析过程中,需要进行以下标准化处理:

  1. 路径标准化:确保所有路径格式一致
  2. 参数标准化:将不同格式的参数定义转换为统一格式
  3. 响应标准化根据HTTP方法添加适当的默认状态码
  4. Schema标准化:解析所有的$ref引用
  5. 数据类型标准化:确保数据类型表示一致

4. 与测试框架的集成

4.1 更新 BaseAPITestCase

class BaseAPITestCase:
    # ... 现有代码 ...
    
    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):
        """
        初始化测试用例。
        Args:
            endpoint_spec: 当前被测API端点的详细定义 (已转换为统一格式)。
            global_api_spec: 完整的API规范文档 (已转换为统一格式)。
            json_schema_validator: APITestOrchestrator 传入的 JSONSchemaValidator 实例 (可选)。
            llm_service: APITestOrchestrator 传入的 LLMService 实例 (可选)。
        """
        self.endpoint_spec = endpoint_spec
        self.global_api_spec = global_api_spec
        self.logger = logging.getLogger(f"testcase.{self.id}")
        self.json_schema_validator = json_schema_validator
        self.llm_service = llm_service
        self.logger.debug(f"Test case '{self.id}' initialized for endpoint: {self.endpoint_spec.get('method', '')} {self.endpoint_spec.get('path', '')}")
    
    # 简化的方法,不再需要处理不同格式的差异
    def _get_request_body_schema(self) -> Optional[Dict[str, Any]]:
        """获取请求体schema"""
        request_body = self.endpoint_spec.get("requestBody", {})
        return request_body.get("schema")
        
    def _get_response_schema(self, status_code: str) -> Optional[Dict[str, Any]]:
        """获取指定状态码的响应schema"""
        responses = self.endpoint_spec.get("responses", {})
        response = responses.get(status_code)
        if response:
            return response.get("schema")
        return None

4.2 更新 APITestOrchestrator

class APITestOrchestrator:
    # ... 现有代码 ...
    
    def __init__(self, config: Dict[str, Any]):
        # ... 现有代码 ...
        self.spec_manager = UnifiedAPISpecManager()
        
    def load_api_spec(self, spec_file_path: str) -> Dict[str, Any]:
        """
        加载并解析API规范文件
        
        Args:
            spec_file_path: API规范文件路径
            
        Returns:
            统一格式的API规范内部表示
        """
        with open(spec_file_path, 'r', encoding='utf-8') as f:
            spec_content = json.load(f)
            
        return self.spec_manager.parse_spec(spec_content, spec_id=spec_file_path)
        
    def execute_tests(self, api_spec: Dict[str, Any], base_url: str):
        """
        执行测试用例
        
        Args:
            api_spec: 统一格式的API规范内部表示
            base_url: API基础URL
        """
        # ... 现有代码 ...
        
        for path, path_item in api_spec["paths"].items():
            for method, operation in path_item.items():
                # 使用统一格式的operation创建测试用例
                self._execute_tests_for_endpoint(operation, path, method, base_url)

实现计划

阶段1基础框架搭建

  1. 创建核心接口和基类

    • APISpecParser 抽象基类
    • APISpecParserFactory 工厂类
    • UnifiedAPISpecManager 管理器类
  2. 实现格式检测逻辑

    • 为YAPI、OpenAPI 2.0和OpenAPI 3.0实现格式检测方法

阶段2解析器实现

  1. 实现YAPI解析器

    • 分析YAPI特有的结构
    • 实现转换为统一格式的逻辑
  2. 实现OpenAPI 2.0解析器

    • 分析Swagger特有的结构
    • 实现转换为统一格式的逻辑
  3. 实现OpenAPI 3.0解析器

    • 分析OpenAPI 3.0特有的结构
    • 实现转换为统一格式的逻辑

阶段3标准化处理

  1. 实现路径标准化
  2. 实现参数标准化
  3. 实现响应标准化
  4. 实现Schema标准化
  5. 实现数据类型标准化

阶段4框架集成

  1. 更新BaseAPITestCase

    • 简化API规范访问方法
    • 移除格式特定的处理逻辑
  2. 更新APITestOrchestrator

    • 集成UnifiedAPISpecManager
    • 使用统一格式的API规范
  3. 更新测试用例

    • 修改现有测试用例以使用新的API

阶段5测试与验证

  1. 编写单元测试

    • 测试各个解析器
    • 测试标准化处理
  2. 编写集成测试

    • 测试完整的解析流程
    • 测试与测试框架的集成
  3. 性能测试

    • 测试解析大型API规范的性能
    • 测试缓存机制的有效性

扩展性考虑

添加新格式支持

要添加对新的API规范格式的支持只需

  1. 创建一个新的解析器类,继承自APISpecParser
  2. 实现parsedetect_format方法
  3. 通过UnifiedAPISpecManager.register_custom_parser注册新的解析器

自定义处理逻辑

对于特定的标准化需求,可以:

  1. 创建专门的处理器类
  2. 在解析过程中调用这些处理器
  3. 通过配置控制处理器的行为

结论

通过实现这个统一的API规范解析框架我们可以

  1. 使测试用例代码更加简洁、可维护
  2. 轻松支持新的API规范格式
  3. 确保所有测试用例使用一致的API规范表示
  4. 提高解析性能和可靠性

这个框架将为DDMS合规性测试工具提供一个坚实的基础使其能够适应各种API规范格式并且易于扩展和维护。