#!/usr/bin/env python3 """ Woodpecker CI/CD 配置验证脚本 系统性地测试和验收 .woodpecker.yml 配置 """ import yaml import sys from pathlib import Path from typing import Dict, List, Any class WoodpeckerValidator: """Woodpecker CI 配置验证器""" def __init__(self, config_path: str): self.config_path = Path(config_path) self.config = None self.errors = [] self.warnings = [] self.info = [] def load_config(self) -> bool: """加载配置文件""" try: with open(self.config_path, 'r', encoding='utf-8') as f: self.config = yaml.safe_load(f) self.info.append("✅ 配置文件加载成功") return True except Exception as e: self.errors.append(f"❌ 配置文件加载失败: {e}") return False def validate_structure(self) -> bool: """验证配置结构""" required_keys = ['steps', 'services', 'workspace', 'clone'] for key in required_keys: if key not in self.config: self.errors.append(f"❌ 缺少必需的配置项: {key}") else: self.info.append(f"✅ 找到配置项: {key}") return len(self.errors) == 0 def validate_branch_triggers(self) -> Dict[str, List[str]]: """验证分支触发条件""" branch_mapping = { 'feature': ['feature/**'], 'dev': ['dev'], 'release': ['release', 'release/**'], 'main': [] # main 不应该触发任何步骤 } step_branches = {} for step_name, step_config in self.config.get('steps', {}).items(): if not isinstance(step_config, dict): continue when_config = step_config.get('when', {}) if not when_config: self.warnings.append(f"⚠️ 步骤 '{step_name}' 没有触发条件") continue if isinstance(when_config, list): branches = [] for condition in when_config: if isinstance(condition, dict) and 'branch' in condition: branches.extend(condition['branch']) elif isinstance(when_config, dict): branches = when_config.get('branch', []) else: branches = [] if branches: step_branches[step_name] = branches self.info.append(f"✅ 步骤 '{step_name}' 触发分支: {branches}") return step_branches def validate_test_strategy(self) -> Dict[str, List[str]]: """验证测试策略分层""" expected_tests = { 'feature/**': ['lint', 'type-check', 'security-scan', 'unit-tests', 'e2e-smoke'], 'dev': ['lint', 'type-check', 'security-scan', 'unit-tests', 'e2e-standard'], 'release': ['lint', 'type-check', 'security-scan', 'unit-tests', 'e2e-standard', 'e2e-deep', 'e2e-performance', 'e2e-accessibility', 'e2e-visual', 'build-image', 'deploy-production', 'archive-to-main'] } test_coverage = {} for branch, expected_steps in expected_tests.items(): test_coverage[branch] = [] for step_name in expected_steps: if step_name in self.config.get('steps', {}): test_coverage[branch].append(step_name) self.info.append(f"✅ 分支 '{branch}' 包含步骤: {step_name}") else: self.errors.append(f"❌ 分支 '{branch}' 缺少步骤: {step_name}") return test_coverage def validate_archive_logic(self) -> bool: """验证归档逻辑""" archive_step = self.config.get('steps', {}).get('archive-to-main', {}) if not archive_step: self.errors.append("❌ 缺少 archive-to-main 步骤") return False commands = archive_step.get('commands', []) has_dynamic_branch = False for cmd in commands: if isinstance(cmd, str) and 'CURRENT_BRANCH="${CI_COMMIT_BRANCH}"' in cmd: has_dynamic_branch = True self.info.append("✅ 归档步骤使用动态分支变量") break if not has_dynamic_branch: self.warnings.append("⚠️ 归档步骤可能未使用动态分支变量") when_config = archive_step.get('when', {}) branches = when_config.get('branch', []) if 'release' in branches and 'release/**' in branches: self.info.append("✅ 归档步骤支持 release 和 release/** 分支") else: self.errors.append("❌ 归档步骤分支配置不完整") return len(self.errors) == 0 def validate_deployment_safety(self) -> bool: """验证部署安全性""" deploy_step = self.config.get('steps', {}).get('deploy-production', {}) if not deploy_step: self.errors.append("❌ 缺少 deploy-production 步骤") return False commands = deploy_step.get('commands', []) has_rollback = False has_health_check = False for cmd in commands: if isinstance(cmd, str): if 'rolling back' in cmd.lower() or 'rollback' in cmd.lower(): has_rollback = True self.info.append("✅ 部署步骤包含回滚机制") if 'health check' in cmd.lower(): has_health_check = True self.info.append("✅ 部署步骤包含健康检查") if not has_rollback: self.warnings.append("⚠️ 部署步骤可能缺少回滚机制") if not has_health_check: self.warnings.append("⚠️ 部署步骤可能缺少健康检查") secrets = deploy_step.get('environment', {}) if 'SSH_PRIVATE_KEY' in secrets and 'REGISTRY_PASSWORD' in secrets: self.info.append("✅ 部署步骤使用 Secret 管理敏感信息") else: self.errors.append("❌ 部署步骤未正确配置 Secrets") return len(self.errors) == 0 def validate_docker_build(self) -> bool: """验证 Docker 构建配置""" build_step = self.config.get('steps', {}).get('build-image', {}) if not build_step: self.errors.append("❌ 缺少 build-image 步骤") return False commands = build_step.get('commands', []) has_tagging = False for cmd in commands: if isinstance(cmd, str) and 'docker tag' in cmd: has_tagging = True self.info.append("✅ Docker 构建步骤包含镜像标签") break if not has_tagging: self.warnings.append("⚠️ Docker 构建步骤可能缺少镜像标签") volumes = build_step.get('volumes', []) if any('/var/run/docker.sock' in str(v) for v in volumes): self.info.append("✅ Docker 构建步骤挂载了 Docker socket") else: self.warnings.append("⚠️ Docker 构建步骤可能未挂载 Docker socket") return len(self.errors) == 0 def validate_services(self) -> bool: """验证服务配置""" services = self.config.get('services', {}) if 'docker' not in services: self.warnings.append("⚠️ 缺少 Docker 服务配置") else: self.info.append("✅ Docker 服务配置正确") return True def generate_report(self) -> str: """生成测试报告""" report = [] report.append("\n" + "="*60) report.append("Woodpecker CI/CD 配置验证报告") report.append("="*60) report.append(f"\n📋 配置文件: {self.config_path}") report.append(f"📊 总步骤数: {len(self.config.get('steps', {}))}") report.append("\n✅ 通过的检查:") for msg in self.info: report.append(f" {msg}") report.append("\n⚠️ 警告:") for msg in self.warnings: report.append(f" {msg}") report.append("\n❌ 错误:") for msg in self.errors: report.append(f" {msg}") report.append("\n" + "="*60) if self.errors: report.append("❌ 验证失败 - 发现 {} 个错误".format(len(self.errors))) return "\n".join(report) else: report.append("✅ 验证通过 - 配置文件符合要求") return "\n".join(report) def main(): """主函数""" config_path = ".woodpecker.yml" if not Path(config_path).exists(): print(f"❌ 配置文件不存在: {config_path}") sys.exit(1) validator = WoodpeckerValidator(config_path) print("🔍 开始验证 Woodpecker CI/CD 配置...") if not validator.load_config(): print(validator.generate_report()) sys.exit(1) validator.validate_structure() validator.validate_branch_triggers() validator.validate_test_strategy() validator.validate_archive_logic() validator.validate_deployment_safety() validator.validate_docker_build() validator.validate_services() print(validator.generate_report()) if validator.errors: sys.exit(1) else: sys.exit(0) if __name__ == "__main__": main()