张翔
275 lines
9.6 KiB
Python
275 lines
9.6 KiB
Python
#!/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()
|