- 使用 PAYLOAD=$(cat <<ENDPAYLOAD) 替代 cat > file <<EOF - 确保环境变量在 heredoc 中正确展开 - 添加测试脚本验证环境变量展开 - 修复构建详情链接和消息内容缺失问题
This commit is contained in:
张翔
@@ -0,0 +1,274 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user