#!/usr/bin/env python3 """ OpenAPI规范验证脚本 验证API接口规范的完整性和一致性 """ import sys import json import argparse from pathlib import Path from typing import Dict, Any, List, Set import requests import logging from urllib.parse import urljoin logger = logging.getLogger(__name__) class OpenAPIValidator: """OpenAPI规范验证器""" def __init__(self, api_base_url: str = "http://localhost:8000"): self.api_base_url = api_base_url self.openapi_spec = None self.validation_results = { "overall_status": "pass", "errors": [], "warnings": [], "stats": {} } def load_openapi_spec(self) -> bool: """加载OpenAPI规范""" try: # 从FastAPI应用获取OpenAPI规范 openapi_url = urljoin(self.api_base_url, "/openapi.json") response = requests.get(openapi_url, timeout=30) response.raise_for_status() self.openapi_spec = response.json() logger.info("OpenAPI规范加载成功") return True except Exception as e: logger.error(f"加载OpenAPI规范失败: {e}") self.validation_results["errors"].append(f"无法加载OpenAPI规范: {str(e)}") return False def validate_basic_structure(self) -> bool: """验证基本结构""" logger.info("验证OpenAPI基本结构...") required_fields = ["openapi", "info", "paths"] missing_fields = [] for field in required_fields: if field not in self.openapi_spec: missing_fields.append(field) if missing_fields: error_msg = f"缺少必需字段: {', '.join(missing_fields)}" self.validation_results["errors"].append(error_msg) return False # 验证版本 if not self.openapi_spec["openapi"].startswith("3."): warning_msg = f"建议使用OpenAPI 3.x版本，当前版本: {self.openapi_spec['openapi']}" self.validation_results["warnings"].append(warning_msg) # 验证信息字段 info = self.openapi_spec["info"] required_info_fields = ["title", "version"] missing_info_fields = [field for field in required_info_fields if field not in info] if missing_info_fields: error_msg = f"info字段缺少必需字段: {', '.join(missing_info_fields)}" self.validation_results["errors"].append(error_msg) return False logger.info("基本结构验证通过") return True def validate_paths(self) -> bool: """验证路径定义""" logger.info("验证API路径定义...") paths = self.openapi_spec.get("paths", {}) if not paths: error_msg = "未定义任何API路径" self.validation_results["errors"].append(error_msg) return False path_issues = [] total_operations = 0 for path, path_item in paths.items(): # 验证路径格式 if not path.startswith("/"): path_issues.append(f"路径 '{path}' 必须以 '/' 开头") # 验证操作方法 operations = [method.upper() for method in ["get", "post", "put", "delete", "patch"] if method in path_item] total_operations += len(operations) if not operations: path_issues.append(f"路径 '{path}' 没有定义任何操作方法") continue # 验证每个操作 for method, operation in path_item.items(): if method.upper() in ["GET", "POST", "PUT", "DELETE", "PATCH"]: operation_issues = self._validate_operation(path, method, operation) path_issues.extend(operation_issues) if path_issues: self.validation_results["errors"].extend(path_issues) self.validation_results["stats"]["total_paths"] = len(paths) self.validation_results["stats"]["total_operations"] = total_operations return False logger.info(f"路径验证通过: {len(paths)} 个路径, {total_operations} 个操作") self.validation_results["stats"]["total_paths"] = len(paths) self.validation_results["stats"]["total_operations"] = total_operations return True def _validate_operation(self, path: str, method: str, operation: Dict) -> List[str]: """验证单个操作""" issues = [] # 检查必需字段 if "responses" not in operation: issues.append(f"{method.upper()} {path}: 缺少 responses 字段") return issues responses = operation["responses"] if not responses: issues.append(f"{method.upper()} {path}: responses 不能为空") return issues # 检查至少有一个成功响应 success_codes = ["200", "201", "202", "204"] has_success_response = any(code in success_codes for code in responses.keys()) if not has_success_response: issues.append(f"{method.upper()} {path}: 缺少成功响应码 (2xx)") # 检查响应定义 for status_code, response in responses.items(): if "description" not in response: issues.append(f"{method.upper()} {path} {status_code}: 缺少响应描述") # 验证响应模式 if "content" in response: content = response["content"] if not content: issues.append(f"{method.upper()} {path} {status_code}: content 不能为空") continue # 检查是否有JSON响应 if "application/json" in content: json_content = content["application/json"] if "schema" not in json_content: issues.append(f"{method.upper()} {path} {status_code}: JSON响应缺少schema定义") # 验证请求体（POST/PUT操作） if method.upper() in ["POST", "PUT"]: if "requestBody" in operation: request_body = operation["requestBody"] if "content" not in request_body: issues.append(f"{method.upper()} {path}: requestBody 缺少 content 定义") elif "application/json" in request_body.get("content", {}): json_content = request_body["content"]["application/json"] if "schema" not in json_content: issues.append(f"{method.upper()} {path}: JSON请求体缺少schema定义") return issues def validate_schemas(self) -> bool: """验证数据模式定义""" logger.info("验证数据模式定义...") components = self.openapi_spec.get("components", {}) schemas = components.get("schemas", {}) schema_issues = [] for schema_name, schema in schemas.items(): # 检查必需字段 if "type" not in schema and "allOf" not in schema and "anyOf" not in schema: schema_issues.append(f"Schema '{schema_name}': 缺少类型定义") # 验证属性定义 if "properties" in schema: properties = schema["properties"] required_fields = schema.get("required", []) for prop_name, prop_schema in properties.items(): # 检查属性类型 if "type" not in prop_schema and "allOf" not in prop_schema: schema_issues.append(f"Schema '{schema_name}' 属性 '{prop_name}': 缺少类型定义") # 检查描述 if "description" not in prop_schema: self.validation_results["warnings"].append( f"Schema '{schema_name}' 属性 '{prop_name}': 建议添加描述" ) # 检查必需字段是否在属性中定义 for required_field in required_fields: if required_field not in properties: schema_issues.append( f"Schema '{schema_name}': 必需字段 '{required_field}' 未在 properties 中定义" ) if schema_issues: self.validation_results["errors"].extend(schema_issues) self.validation_results["stats"]["total_schemas"] = len(schemas) return False logger.info(f"数据模式验证通过: {len(schemas)} 个模式定义") self.validation_results["stats"]["total_schemas"] = len(schemas) return True def validate_security_definitions(self) -> bool: """验证安全定义""" logger.info("验证安全定义...") components = self.openapi_spec.get("components", {}) security_schemes = components.get("securitySchemes", {}) if not security_schemes: self.validation_results["warnings"].append("未定义安全方案，建议添加认证机制") return True security_issues = [] for scheme_name, scheme in security_schemes.items(): if "type" not in scheme: security_issues.append(f"安全方案 '{scheme_name}': 缺少类型定义") continue scheme_type = scheme["type"] # 验证不同的认证类型 if scheme_type == "apiKey": if "name" not in scheme or "in" not in scheme: security_issues.append(f"API Key方案 '{scheme_name}': 缺少必需字段") elif scheme_type == "http": if "scheme" not in scheme: security_issues.append(f"HTTP方案 '{scheme_name}': 缺少 scheme 定义") elif scheme_type == "oauth2": if "flows" not in scheme: security_issues.append(f"OAuth2方案 '{scheme_name}': 缺少 flows 定义") if security_issues: self.validation_results["errors"].extend(security_issues) return False logger.info(f"安全定义验证通过: {len(security_schemes)} 个安全方案") return True def validate_tag_definitions(self) -> bool: """验证标签定义""" logger.info("验证标签定义...") tags = self.openapi_spec.get("tags", []) if not tags: self.validation_results["warnings"].append("未定义标签，建议添加标签分组") return True tag_issues = [] tag_names = set() for tag in tags: if "name" not in tag: tag_issues.append("标签缺少 name 字段") continue tag_name = tag["name"] if tag_name in tag_names: tag_issues.append(f"标签名称 '{tag_name}' 重复") tag_names.add(tag_name) if "description" not in tag: self.validation_results["warnings"].append(f"标签 '{tag_name}': 建议添加描述") if tag_issues: self.validation_results["errors"].extend(tag_issues) return False logger.info(f"标签定义验证通过: {len(tags)} 个标签") return True def validate_api_consistency(self) -> bool: """验证API一致性""" logger.info("验证API一致性...") consistency_issues = [] # 检查路径命名一致性 paths = self.openapi_spec.get("paths", {}) path_patterns = {} for path in paths.keys(): # 提取路径模式 parts = path.strip("/").split("/") if len(parts) >= 2: resource_type = parts[0] if resource_type not in path_patterns: path_patterns[resource_type] = [] path_patterns[resource_type].append(path) # 检查资源路径一致性 for resource_type, resource_paths in path_patterns.items(): if len(resource_paths) > 1: # 检查是否有不一致的路径模式 base_patterns = set() for path in resource_paths: # 简化路径模式检查 normalized = path.replace("/{", "/{id}") base_patterns.add(normalized) if len(base_patterns) > 2: # 允许集合和单个资源两种模式 consistency_issues.append( f"资源 '{resource_type}' 的路径模式不一致，发现 {len(base_patterns)} 种不同模式" ) # 检查响应格式一致性 operations_data = [] for path, path_item in paths.items(): for method, operation in path_item.items(): if method.upper() in ["GET", "POST", "PUT", "DELETE", "PATCH"]: operations_data.append((path, method, operation)) # 检查错误响应一致性 error_responses = {} for path, method, operation in operations_data: responses = operation.get("responses", {}) for status_code in ["400", "401", "403", "404", "500"]: if status_code in responses: if status_code not in error_responses: error_responses[status_code] = set() response_schema = str(responses[status_code].get("content", {}).get("application/json", {}).get("schema", {})) error_responses[status_code].add(response_schema) for status_code, schemas in error_responses.items(): if len(schemas) > 1: consistency_issues.append( f"错误响应 {status_code} 的格式不一致，发现 {len(schemas)} 种不同格式" ) if consistency_issues: self.validation_results["errors"].extend(consistency_issues) return False logger.info("API一致性验证通过") return True def run_validation(self) -> Dict[str, Any]: """运行完整的OpenAPI验证""" logger.info("开始OpenAPI规范验证...") # 加载规范 if not self.load_openapi_spec(): return self.validation_results # 运行各项验证 validations = [ ("基本结构", self.validate_basic_structure), ("路径定义", self.validate_paths), ("数据模式", self.validate_schemas), ("安全定义", self.validate_security_definitions), ("标签定义", self.validate_tag_definitions), ("API一致性", self.validate_api_consistency) ] passed_validations = 0 total_validations = len(validations) for validation_name, validation_func in validations: try: if validation_func(): passed_validations += 1 except Exception as e: error_msg = f"{validation_name}验证过程中发生错误: {str(e)}" self.validation_results["errors"].append(error_msg) logger.error(error_msg) # 计算整体状态 if self.validation_results["errors"]: self.validation_results["overall_status"] = "fail" self.validation_results["message"] = f"OpenAPI规范验证失败，发现 {len(self.validation_results['errors'])} 个错误" else: self.validation_results["overall_status"] = "pass" self.validation_results["message"] = "OpenAPI规范验证通过" # 添加验证统计 self.validation_results["stats"]["passed_validations"] = passed_validations self.validation_results["stats"]["total_validations"] = total_validations self.validation_results["stats"]["pass_rate"] = passed_validations / total_validations return self.validation_results def print_results(self): """打印验证结果""" print("=" * 80) print("OpenAPI规范验证结果") print("=" * 80) print(f"整体状态: {self.validation_results['overall_status'].upper()}") print(f"消息: {self.validation_results['message']}") print() # 显示统计信息 stats = self.validation_results.get("stats", {}) if stats: print("统计信息:") print("-" * 40) for key, value in stats.items(): print(f" {key}: {value}") print() # 显示错误 errors = self.validation_results.get("errors", []) if errors: print("错误:") print("-" * 40) for i, error in enumerate(errors, 1): print(f" {i}. {error}") print() # 显示警告 warnings = self.validation_results.get("warnings", []) if warnings: print("警告:") print("-" * 40) for i, warning in enumerate(warnings, 1): print(f" {i}. {warning}") print() def main(): """主函数""" parser = argparse.ArgumentParser(description='OpenAPI规范验证脚本') parser.add_argument('--url', default='http://localhost:8000', help='API基础URL') parser.add_argument('--output', '-o', help='输出结果文件') parser.add_argument('--verbose', '-v', action='store_true', help='详细输出') args = parser.parse_args() # 配置日志 log_level = logging.DEBUG if args.verbose else logging.INFO logging.basicConfig(level=log_level, format='%(levelname)s: %(message)s') # 运行验证 validator = OpenAPIValidator(args.url) results = validator.run_validation() # 打印结果 validator.print_results() # 保存结果 if args.output: with open(args.output, 'w', encoding='utf-8') as f: json.dump(results, f, ensure_ascii=False, indent=2) logger.info(f"验证结果已保存到: {args.output}") # 根据验证结果设置退出码 sys.exit(0 if results["overall_status"] == "pass" else 1) if __name__ == "__main__": main()