146 lines
4.2 KiB
Python
146 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding:utf-8 -*-
|
|
# 主执行文件
|
|
|
|
import os
|
|
import sys
|
|
import argparse
|
|
import logging
|
|
import json
|
|
from typing import Dict, List, Any
|
|
|
|
from common.logger import get_logger, LoggerManager
|
|
from common.config import ConfigManager, DEFAULT_REPO_CONFIGS
|
|
from repodata.checker import RepoChecker
|
|
|
|
# 设置基本日志级别
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = get_logger("main")
|
|
|
|
def parse_arguments():
|
|
parser = argparse.ArgumentParser(description='仓库测试检查工具')
|
|
|
|
parser.add_argument('--config', type=str, help='配置文件路径')
|
|
parser.add_argument('--output-dir', type=str, default='./results', help='输出目录')
|
|
parser.add_argument('--max-workers', type=int, default=5, help='最大并行检查线程数')
|
|
parser.add_argument('--product', type=str, help='仅检查指定产品')
|
|
parser.add_argument('--arch', type=str, help='仅检查指定架构')
|
|
parser.add_argument('--verbose', action='store_true', help='显示详细日志')
|
|
|
|
return parser.parse_args()
|
|
|
|
def load_config(config_path: str) -> List[Dict[str, Any]]:
|
|
"""
|
|
加载配置文件
|
|
|
|
Args:
|
|
config_path: 配置文件路径
|
|
|
|
Returns:
|
|
仓库配置列表
|
|
"""
|
|
# 检查文件是否存在
|
|
if not os.path.exists(config_path):
|
|
logger.error(f"配置文件不存在: {config_path}")
|
|
return DEFAULT_REPO_CONFIGS
|
|
|
|
# 根据文件扩展名决定加载方式
|
|
ext = os.path.splitext(config_path)[1].lower()
|
|
|
|
if ext == '.json':
|
|
config = ConfigManager.load_json_config(config_path)
|
|
elif ext in ['.yaml', '.yml']:
|
|
config = ConfigManager.load_yaml_config(config_path)
|
|
else:
|
|
logger.error(f"不支持的配置文件格式: {ext}")
|
|
return DEFAULT_REPO_CONFIGS
|
|
|
|
# 检查配置是否有效
|
|
if not config or not isinstance(config, list):
|
|
logger.error("无效的配置格式，使用默认配置")
|
|
return DEFAULT_REPO_CONFIGS
|
|
|
|
return config
|
|
|
|
def filter_configs(configs: List[Dict[str, Any]], product: str = None, arch: str = None) -> List[Dict[str, Any]]:
|
|
"""
|
|
根据条件过滤仓库配置
|
|
|
|
Args:
|
|
configs: 原始配置列表
|
|
product: 产品名称过滤
|
|
arch: 架构过滤
|
|
|
|
Returns:
|
|
过滤后的配置列表
|
|
"""
|
|
if not product and not arch:
|
|
return configs
|
|
|
|
filtered_configs = []
|
|
|
|
for config in configs:
|
|
# 如果指定了产品，但不匹配，则跳过
|
|
if product and config["product"] != product:
|
|
continue
|
|
|
|
# 如果指定了架构，需要检查
|
|
if arch:
|
|
# 克隆配置但只保留指定架构
|
|
if arch in config["architectures"]:
|
|
new_config = config.copy()
|
|
new_config["architectures"] = [arch]
|
|
filtered_configs.append(new_config)
|
|
else:
|
|
# 没有指定架构，保留所有
|
|
filtered_configs.append(config)
|
|
|
|
return filtered_configs
|
|
|
|
def setup_logging(verbose: bool):
|
|
"""设置日志级别"""
|
|
level = logging.DEBUG if verbose else logging.INFO
|
|
logging.getLogger().setLevel(level)
|
|
|
|
def main():
|
|
"""主函数"""
|
|
# 解析命令行参数
|
|
args = parse_arguments()
|
|
|
|
# 设置日志级别
|
|
setup_logging(args.verbose)
|
|
|
|
# 加载配置
|
|
if args.config:
|
|
logger.info(f"加载配置文件: {args.config}")
|
|
repo_configs = load_config(args.config)
|
|
else:
|
|
logger.info("使用默认配置")
|
|
repo_configs = DEFAULT_REPO_CONFIGS
|
|
|
|
# 过滤配置
|
|
filtered_configs = filter_configs(repo_configs, args.product, args.arch)
|
|
|
|
if not filtered_configs:
|
|
logger.error("没有匹配的仓库配置")
|
|
return 1
|
|
|
|
logger.info(f"将检查 {len(filtered_configs)} 个仓库配置")
|
|
|
|
# 创建输出目录
|
|
if not os.path.exists(args.output_dir):
|
|
os.makedirs(args.output_dir)
|
|
|
|
# 创建检查器并执行检查
|
|
checker = RepoChecker(args.output_dir)
|
|
results = checker.check_all_repositories(filtered_configs, args.max_workers)
|
|
|
|
# 检查是否有失败
|
|
has_failures = any(not result.get("passed", False) for result in results.values())
|
|
|
|
# 返回状态码
|
|
return 1 if has_failures else 0
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|