123 lines
4.0 KiB
Python
123 lines
4.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""测试命令仓库：集中维护 run_tests.py 的常用组合，支持一键执行。"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
from typing import List
|
||
|
||
RUN_TESTS_SCRIPT = os.path.join(os.path.dirname(__file__), "run_tests.py")
|
||
|
||
# 默认自动运行的预置（可根据需要修改顺序/条目）
|
||
AUTO_RUN_PRESETS = ["fetch_only"]
|
||
|
||
PRESETS = {
|
||
"fetch_only": {
|
||
"suite": ["online"],
|
||
"flow": "FETCH_ONLY",
|
||
"json_fetch_root": "tmp/json_fetch",
|
||
"keyword": "ORDERS",
|
||
"pytest_args": "-vv",
|
||
"preset_meta": "仅在线抓取阶段，输出到本地目录",
|
||
},
|
||
"ingest_local": {
|
||
"suite": ["online"],
|
||
"flow": "INGEST_ONLY",
|
||
"json_source": "tests/source-data-doc",
|
||
"keyword": "ORDERS",
|
||
"preset_meta": "从指定 JSON 目录做本地清洗入库",
|
||
},
|
||
"full_pipeline": {
|
||
"suite": ["online"],
|
||
"flow": "FULL",
|
||
"json_fetch_root": "tmp/json_fetch",
|
||
"keyword": "ORDERS",
|
||
"preset_meta": "先抓取再清洗入库的全流程",
|
||
},
|
||
}
|
||
|
||
|
||
def print_parameter_help() -> None:
|
||
print("=== 参数键说明 ===")
|
||
print("suite : 预置套件列表，如 ['online','integration']")
|
||
print("tests : 自定义 pytest 路径列表")
|
||
print("flow : PIPELINE_FLOW（FETCH_ONLY / INGEST_ONLY / FULL）")
|
||
print("json_source : JSON_SOURCE_DIR，本地清洗入库使用的 JSON 目录")
|
||
print("json_fetch_root : JSON_FETCH_ROOT，在线抓取输出根目录")
|
||
print("keyword : pytest -k 过滤关键字")
|
||
print("pytest_args : 额外 pytest 参数（字符串）")
|
||
print("env : 附加环境变量，例如 ['KEY=VALUE']")
|
||
print("preset_meta : 仅用于注释说明")
|
||
print()
|
||
|
||
|
||
def print_presets() -> None:
|
||
if not PRESETS:
|
||
print("当前未定义任何预置，请在 PRESETS 中添加。")
|
||
return
|
||
for idx, (name, payload) in enumerate(PRESETS.items(), start=1):
|
||
comment = payload.get("preset_meta", "")
|
||
print(f"{idx}. {name}")
|
||
if comment:
|
||
print(f" 说明: {comment}")
|
||
for key, value in payload.items():
|
||
if key == "preset_meta":
|
||
continue
|
||
print(f" {key}: {value}")
|
||
print()
|
||
|
||
|
||
def resolve_targets(requested: List[str] | None) -> List[str]:
|
||
if not PRESETS:
|
||
raise SystemExit("预置为空，请先在 PRESETS 中定义测试组合。")
|
||
|
||
def valid(names: List[str]) -> List[str]:
|
||
return [name for name in names if name in PRESETS]
|
||
|
||
if requested:
|
||
candidates = valid(requested)
|
||
missing = [name for name in requested if name not in PRESETS]
|
||
if missing:
|
||
print(f"警告：忽略未定义的预置 {missing}")
|
||
if candidates:
|
||
return candidates
|
||
|
||
auto = valid(AUTO_RUN_PRESETS)
|
||
if auto:
|
||
return auto
|
||
|
||
return list(PRESETS.keys())
|
||
|
||
|
||
def run_presets(preset_names: List[str], dry_run: bool) -> None:
|
||
for name in preset_names:
|
||
cmd = [sys.executable, RUN_TESTS_SCRIPT, "--preset", name]
|
||
printable = " ".join(cmd)
|
||
if dry_run:
|
||
print(f"[Dry-Run] {printable}")
|
||
else:
|
||
print(f"\n>>> 执行: {printable}")
|
||
subprocess.run(cmd, check=False)
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description="测试预置仓库（集中配置即可批量触发 run_tests）")
|
||
parser.add_argument("--preset", choices=sorted(PRESETS.keys()), nargs="+", help="指定要运行的预置命令")
|
||
parser.add_argument("--list", action="store_true", help="仅列出参数说明与所有预置")
|
||
parser.add_argument("--dry-run", action="store_true", help="仅打印命令，不执行 pytest")
|
||
args = parser.parse_args()
|
||
|
||
if args.list:
|
||
print_parameter_help()
|
||
print_presets()
|
||
return
|
||
|
||
targets = resolve_targets(args.preset)
|
||
run_presets(targets, dry_run=args.dry_run)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|