外部算子仓可插拔测试用例框架

[TecJesh]

外部算子仓可插拔测试用例 — 设计文档

关联模块：pipeline/test.py、utils/config.py、flow.py

---

1. 目标

在现有 pytest_ut 测试流程基础上，新增外部算子仓可插拔测试用例能力。允许用户通过配置文件指定外部 GitHub 仓库及其测试用例，工作流自动克隆仓库、按算子仓为单位顺序执行测试，失败时触发 AI 修复，通过后继续下一个算子仓。

---

2. 设计决策

2.0 控制开关：enabled

提供一个独立的启用/禁用开关，与运行模式解耦：

层级	配置方式	说明
环境变量	TA_EXTERNAL_TEST_ENABLED=true	全局开关，优先级最高
CLI 参数	--external-test / --no-external-test	命令行传入，覆盖环境变量
YAML 配置	enabled: true	配置文件内开关，默认值

生效逻辑（优先级从高到低）：

1. CLI 参数（如果传入）
2. 环境变量 TA_EXTERNAL_TEST_ENABLED
3. YAML 配置文件中的 enabled 字段
4. 默认值：false（默认不运行外部测试，需用户主动开启）

关闭时（enabled=false）：整个外部测试阶段完全跳过，不加载配置、不克隆仓库、不执行任何测试。等价于 mode=off，但更易于理解和配置。

设计理由：

• enabled 和 mode 职责分离：enabled 控制"要不要跑"，mode 控制"怎么跑"
• 默认关闭（false），避免用户意外拉取外部仓库
• 提供多层配置入口，适配不同使用场景（CI/CD 用环境变量、本地调试用 CLI、持久化配置用 YAML）

2.1 配置文件格式：YAML

方案	优点	缺点	结论
YAML	可读性强、支持注释、Python 原生解析（PyYAML）、结构清晰	需额外依赖	✅ 选用
JSON	无需依赖	不支持注释、括号嵌套深、手写不友好	❌
TOML	Python 生态常用	嵌套数组语法不直观	❌
Python 脚本	灵活	引入代码执行风险	❌

2.2 配置结构：统一配置文件（单文件）

将所有外部仓库和用例配置在一个 YAML 文件中，按 repo 分组：

# ── 全局开关：是否启用外部测试（默认 false，需用户主动开启）──
# 也可通过环境变量 TA_EXTERNAL_TEST_ENABLED=true 覆盖
enabled: false

# ── 运行模式（仅在 enabled=true 时生效）───────────────────────
mode: inline                    # inline | standalone | off

# ── 全局默认值 ──────────────────────────────────────────────
test_procs: 8                   # pytest -n 并行数
max_retries: 5                  # 每仓最大 AI 修复次数
timeout: 7200                   # 单仓测试超时（秒）

# external_test_repos 列表中每个元素是一个外部算子仓
external_test_repos:
  - name: "Liger-Kernel"                          # 显示名称
    url: "https://github.com/linkedin/Liger-Kernel.git"
    branch: "main"                                 # 可选，默认 main
    # install_cmd: "pip install -e ."              # 可选，为空则跳过安装
    test_cases:                                    # 以测试文件为粒度
      - "test/transformers/test_attn_res.py"
      - "test/transformers/test_auto_model.py"
      - "test/transformers/test_cute_moe_autograd.py"

  - name: "flash-linear-attention"
    url: "https://github.com/fla-org/flash-linear-attention.git"
    branch: "main"
    test_cases:
      - "tests/context_parallel/test_cp_conv.py"
      - "tests/context_parallel/test_cp_dplr.py"
      - "tests/context_parallel/test_cp_gdn.py"

理由：单文件便于版本管理和整体审阅；按 repo 分组使每个算子仓的用例一目了然；顺序即执行顺序（先配置先跑）。无需按 repo 拆分配置文件，避免文件散落。

2.3 克隆位置

workspace/external_repos/
├── Liger-Kernel/          # 从 https://github.com/linkedin/Liger-Kernel.git 克隆
├── flash-linear-attention/ # 从 https://github.com/fla-org/flash-linear-attention.git 克隆
└── ...

workspace/ 是项目已有的工作区根目录（TA_MAIN2MAIN_WORKSPACE 环境变量决定）。

2.4 执行顺序：按算子仓串行，仓内并行

repo-1: clone → [install_deps (可选)] → pytest -n 8 → [失败→AI修复→重测] → 通过
repo-2: clone → [install_deps (可选)] → pytest -n 8 → [失败→AI修复→重测] → 通过
...

• 仓与仓之间：严格串行。一个算子仓全部跑通后再进行下一个。
• 仓内测试：使用 pytest -n 8（可配置）并行执行该仓配置的所有用例文件。
• 失败处理：复用现有的 AI fix + rebuild/retest 循环（最多 max_retries 次）。

2.5 工作流集成位置

在现有标准 test_and_fix_loop 之后、commit 之前插入外部测试阶段：

merge → resolve → build_and_fix → test_and_fix (pytest_ut) → ★ external_test → commit → ...

理由：

• 与现有 pytest_ut 测试平级，不阻塞核心测试
• 外部测试在 commit 之前运行，失败可以触发 AI 修复
• 如果 SKIP_E2E_TEST=true，外部测试同样跳过

提供两种运行模式（通过配置开关）：

模式	配置	行为
内联模式（默认）	TA_EXTERNAL_TEST_MODE=inline	集成到主流程，失败阻塞 commit
独立模式	TA_EXTERNAL_TEST_MODE=standalone	作为独立阶段运行，失败不阻塞主流程
禁用	TA_EXTERNAL_TEST_MODE=off	不运行外部测试

---

3. 详细设计

3.1 文件结构

src/TA_main2main_workflow/
├── external_test/                          # 新增目录
│   ├── __init__.py
│   ├── external_test_config.yaml           # 默认配置文件
│   ├── config_loader.py                    # 配置加载与校验
│   └── runner.py                           # 外部测试执行器
├── pipeline/
│   └── test.py                             # 修改：集成外部测试调用
├── utils/
│   └── config.py                           # 修改：新增外部测试配置字段
└── flow.py                                 # 修改：在 test 阶段后插入外部测试

3.2 配置加载器（config_loader.py）

ExternalTestRepoConfig (dataclass):
    name: str            # 算子仓显示名称
    url: str             # Git 克隆 URL
    branch: str          # 分支（默认 "main"）
    test_cases: list[str] # 测试文件相对路径列表
    install_cmd: str     # 可选：依赖安装命令（默认 ""，为空则跳过安装）

ExternalTestConfig (dataclass):
    enabled: bool         # 是否启用外部测试（默认 false）
    repos: list[ExternalTestRepoConfig]
    test_procs: int       # 并行进程数（默认 8）
    mode: str             # "inline" | "standalone" | "off"
    max_retries: int      # 每仓最大修复重试次数
    timeout: int          # 单仓测试超时（秒）

load_config(path: str) -> ExternalTestConfig   # 从 YAML 加载
validate_config(cfg) -> bool                    # 校验 URL、路径合法性

配置文件加载优先级：
    1. 显式传入 path 参数
    2. TA_EXTERNAL_TEST_CONFIG 环境变量
    3. 默认 external_test/external_test_config.yaml（内置配置）

3.3 执行器（runner.py）

核心执行流程：

def run_external_tests(ctx, config, external_cfg) -> WorkflowContext:
    """
    1. 检查 repos 列表是否为空，为空则打印警告并跳过
    2. 确保 workspace/external_repos/ 目录存在
    3. 遍历 external_cfg.repos（按配置顺序）：
       a. clone_or_update_repo(repo_cfg)       # 克隆或更新仓库
       b. install_dependencies(repo_cfg)        # 安装依赖（install_cmd 为空则跳过）
       c. run_repo_tests(repo_cfg, ...)         # pytest -n 8 执行用例
       d. 如果失败：external_test_fix_loop()    # AI 修复循环（直接调用 run_opencode_adapter）
       e. 记录结果，继续下一个
    4. 写入汇总 JSON → 返回更新的 WorkflowContext
    """

关键函数签名：

函数	职责
clone_or_update_repo(cfg, workspace)	git clone 或 git pull（已存在时）
install_dependencies(repo_path, cfg)	进入仓库执行自定义命令（install_cmd 为空则跳过）
run_repo_tests(repo_path, test_cases, procs)	pytest -n {procs} {test_case_1} {test_case_2} ...，输出写入 test-output-external-{name}.log
_external_test_ai_fix(...)	直接调用 run_opencode_adapter(mode="external_test_fix")，校验变更范围限定在外部仓库内

3.4 TAConfig 新增字段

# 在 TAConfig dataclass 中新增：
external_test_enabled: bool = False    # 是否启用外部测试（默认关闭）
external_test_config: str = ""         # 外部测试配置文件路径
external_test_mode: str = "inline"     # "inline" | "standalone" | "off"

# 对应环境变量：
#   TA_EXTERNAL_TEST_ENABLED  — 启用/禁用外部测试（true/false）
#   TA_EXTERNAL_TEST_CONFIG   — 配置文件路径（默认：external_test/external_test_config.yaml）
#   TA_EXTERNAL_TEST_MODE     — 运行模式

3.5 WorkflowContext 新增字段

external_test_passed: bool = False
external_test_results: list[dict] = field(default_factory=list)
# 每仓一条：{"repo": "Liger-Kernel", "passed": True, "failed_cases": [], "fix_count": 0}

3.6 Flow 集成点

在 flow.py 的 TA_Main2MainFlow.run() 中，标准 test_and_fix_loop 之后、commit_step 之前：

# ── 外部算子仓测试 ──────────────────────────────
if external_cfg and external_cfg.enabled and external_cfg.mode != "off":
    log.section(f"External Test — {external_cfg.mode} mode")
    with timed("external-test"):
        ctx = run_external_tests(ctx, self.config, external_cfg)
    if external_cfg.mode == "inline" and not ctx.external_test_passed:
        log.error("External tests failed (inline mode)")
        ctx = ctx.copy_with(final_status=UpgradeFailed)
        return UpgradeFailed
elif external_cfg and not external_cfg.enabled:
    log.info("External test is disabled — skipped")

3.7 AI 修复适配

外部测试失败时，AI 修复需要知道：

• 失败在哪个外部仓库（repo_path）
• 该仓库的测试日志位置
• 允许修改的文件范围（外部仓库全部文件，不限制 third_party/ascend/ 前缀）

因此在调用 ai_fix() 时需扩展参数，传递外部仓库路径和允许修改的路径范围：

# 扩展现有 ai_fix 调用参数
result = run_opencode_adapter({
    ...
    "external_repo_path": str(repo_path),       # 外部仓库路径
    "allowed_fix_paths": str(repo_path),         # AI 修复允许的根路径
    "mode": "external_test_fix",                 # 标识为外部测试修复
})

3.8 目录结构总览

workspace/
├── external_repos/                    # 新增：外部算子仓克隆目录
│   ├── Liger-Kernel/                  # 克隆的 Liger-Kernel 仓库
│   ├── flash-linear-attention/        # 克隆的 flash-linear-attention 仓库
│   └── ...
├── test-logs/                         # 现有：测试日志目录
│   ├── pytest-junit-primary.xml       # 现有：主测试结果
│   ├── pytest-junit-external-Liger-Kernel.xml       # 新增：JUnit XML
│   ├── test-output-external-Liger-Kernel.log        # 新增：pytest 完整输出
│   ├── test-result-external-Liger-Kernel.json       # 新增：结果摘要
│   ├── external-test-summary.json                   # 新增：汇总
│   └── ...
├── steps/                             # 现有：步骤目录
├── fixes/                             # 现有：修复日志 / 外部测试修复
│   ├── external-Liger-Kernel-fix-1/   # 新增：外部测试 AI 修复目录
│   └── ...
└── ...

---

4. 执行流程图

4.1 主流程集成位置

flowchart TD
    A[prepare] --> B[detect]
    B --> C[plan]
    C --> D[build_baseline_llvm]
    D --> E{for each step}
    E --> F[merge]
    F --> G[resolve]
    G --> H[build_and_fix]
    H --> I[test_and_fix]
    I --> J["★ External Test<br/>(enabled 时执行)"]
    J --> K[commit]
    K --> E
    E -- "all steps done" --> L[finalize]
    L --> M[push_pr]

    style J fill:#fff3cd,stroke:#f0ad4e,stroke-width:2px

Loading

4.2 外部测试详细流程

flowchart TD
    A["加载配置<br/>(TA_EXTERNAL_TEST_CONFIG or 默认)"] --> B{enabled 开关}
    B -- "false" --> Z1["跳过，记录日志"]
    B -- "true" --> C{repos 列表}
    C -- "为空" --> Z2["警告，跳过"]
    C -- "非空" --> D[for each repo]

    D --> E["git clone &lt;url&gt;<br/>→ workspace/external_repos/"]
    E --> F["[可选] install_cmd<br/>(空则跳过)"]
    F --> G["pytest -n 8 &lt;test_case_1&gt; ...<br/>输出 → test-output-external-{name}.log"]
    G --> H{测试结果}
    H -- "PASS" --> I["记录结果<br/>→ 下一个 repo"]
    H -- "FAIL" --> J["AI Fix<br/>(mode=external_test_fix)"]
    J --> K{"验证修改<br/>(限定 repo root)"}
    K -- "违规" --> L["revert 修改"]
    K -- "合规" --> M[重测]
    M --> G
    L --> N{"超过<br/>max_retries?"}
    N -- "no" --> G
    N -- "yes" --> O["记录 FAIL<br/>→ 下一个 repo"]

    I --> P{还有下一个 repo?}
    O --> P
    P -- "yes" --> D
    P -- "no" --> Q["汇总结果<br/>→ external-test-summary.json"]
    Q --> R[更新 WorkflowContext]

    style J fill:#ffe6cc,stroke:#f0ad4e
    style O fill:#f8d7da,stroke:#dc3545

Loading

---

5. 详细步骤

Step 1：创建目录和配置文件

创建 src/TA_main2main_workflow/external_test/ 目录，包含：

• __init__.py
• external_test_config.yaml（默认配置模板）
• config_loader.py（配置加载）
• runner.py（测试执行器）

Step 2：实现配置加载器

# config_loader.py
from dataclasses import dataclass, field
from pathlib import Path
import yaml

@dataclass
class ExternalTestRepoConfig:
    name: str
    url: str
    branch: str = "main"
    test_cases: list[str] = field(default_factory=list)
    install_cmd: str = ""          # 可选，为空则跳过安装

@dataclass
class ExternalTestConfig:
    enabled: bool = False
    repos: list[ExternalTestRepoConfig] = field(default_factory=list)
    test_procs: int = 8
    mode: str = "inline"
    max_retries: int = 5
    timeout: int = 3600

def load_external_test_config(path: str) -> ExternalTestConfig | None:
    """加载外部测试配置，文件不存在时返回 None"""
    ...

Step 3：实现执行器

实现 runner.py，核心逻辑为：

1. ensure_cloned(repo_cfg) — clone 或更新仓库
2. install_deps(repo_cfg) — 安装依赖
3. run_pytest(repo_cfg, procs) — 执行 pytest
4. external_fix_loop(...) — AI 修复循环（复用 fix.py 的 ai_fix）
5. run_external_tests(ctx, config, external_cfg) — 主入口

Step 4：修改 TAConfig

在 config.py 中新增：

external_test_enabled: bool = False     # TA_EXTERNAL_TEST_ENABLED env
external_test_config: str = ""          # TA_EXTERNAL_TEST_CONFIG env
external_test_mode: str = "inline"      # TA_EXTERNAL_TEST_MODE env

Step 5：修改 WorkflowContext

在 context.py 中新增：

external_test_passed: bool = False
external_test_results: list[dict] = field(default_factory=list)

Step 6：集成到 Flow

在 flow.py 的 run() 方法中，test_and_fix_loop 之后插入外部测试调用。

Step 7：扩展 AI Fix

在 fix.py 中支持 mode="external_test_fix"，扩大允许修改的路径范围为外部仓库根目录。

---

6. 测试验证

单元测试

测试项	验证内容
test_config_loader_valid	合法 YAML 正确加载
test_config_loader_missing_file	配置文件不存在返回 None
test_config_loader_invalid_url	非法 URL 校验失败
test_clone_repo	仓库克隆到正确路径
test_clone_repo_existing	仓库已存在时执行 pull
test_run_repo_tests_pass	测试全部通过返回 passed=True
test_run_repo_tests_fail	测试失败返回错误日志路径
test_sequential_execution	多个 repo 按顺序执行

集成测试

1. 创建测试用外部仓库（或使用 mock）
2. 验证完整的外部测试流程
3. 验证 enabled=false 时跳过所有外部测试
4. 验证 inline 模式失败时阻塞主流程
5. 验证 standalone 模式失败时不阻塞主流程
6. 验证 off 模式跳过外部测试
7. 验证 CLI --no-external-test 覆盖 YAML 配置中的 enabled: true

---

7. 环境变量参考

变量名	默认值	说明
TA_EXTERNAL_TEST_ENABLED	false	控制开关：是否启用外部测试（true / false，默认关闭）
TA_EXTERNAL_TEST_CONFIG	src/TA_main2main_workflow/external_test/external_test_config.yaml	外部测试配置文件路径（不设则用内置默认配置）
TA_EXTERNAL_TEST_MODE	inline	运行模式：inline / standalone / off（仅在 enabled=true 时生效）
TA_EXTERNAL_TEST_PROCS	8	每个外部算子仓测试的并行进程数
TA_EXTERNAL_TEST_MAX_RETRIES	5	每个外部算子仓最大 AI 修复次数
TA_EXTERNAL_TEST_TIMEOUT	7200	每个外部算子仓测试超时时间（秒）

---

8. 风险与缓解

风险	缓解措施
外部仓库依赖与主项目冲突	每个外部仓库在独立 conda env 或 venv 中运行（通过 install_cmd 自定义）
外部仓库网络不可达	克隆前检查连通性；失败跳过并记录
外部仓库版本变更导致用例失败	支持指定 branch 或 tag，锁定版本
磁盘空间不足	克隆前检查磁盘空间；standalone 模式下可跳过
AI 修复修改了外部仓库之外的代码	validate_fix() 限制路径为外部仓库根目录

---