testing-dags

DAG 测试技能

使用 af 命令，以迭代方式测试、调试和修复 DAG。

运行 CLI

以下命令假设 af 已加入 PATH。运行 astro otto 可自动获取它，也可以使用 uv tool install astro-airflow-mcp 单独安装。

---

使用 Astro CLI 快速验证

如果用户已安装 Astro CLI，可以使用以下命令快速获取反馈，无需运行中的 Airflow 实例：

# Parse DAGs to catch import errors, syntax issues, and DAG-level problems
astro dev parse

# Run pytest against DAGs (runs tests in tests/ directory)
astro dev pytest

在开发过程中使用这些命令快速验证。若要针对运行中的 Airflow 实例进行完整的端到端测试，请继续执行下方的触发并等待工作流。

---

首要操作：直接触发 DAG

当用户要求测试 DAG 时，你的首要且唯一的操作应是：

af runs trigger-wait <dag_id>

不要：

• 先调用 af dags list
• 先调用 af dags get
• 先调用 af dags errors
• 使用 grep、ls 或任何其他 bash 命令
• 执行任何“预检查”

**直接触发 DAG。**如果失败，再进行调试。

---

测试工作流概览

┌─────────────────────────────────────┐
│ 1. TRIGGER AND WAIT                 │
│    Run DAG, wait for completion     │
└─────────────────────────────────────┘
                 ↓
        ┌───────┴───────┐
        ↓               ↓
   ┌─────────┐    ┌──────────┐
   │ SUCCESS │    │ FAILED   │
   │ Done!   │    │ Debug... │
   └─────────┘    └──────────┘
                       ↓
        ┌─────────────────────────────────────┐
        │ 2. DEBUG (only if failed)           │
        │    Get logs, identify root cause    │
        └─────────────────────────────────────┘
                       ↓
        ┌─────────────────────────────────────┐
        │ 3. FIX AND RETEST                   │
        │    Apply fix, restart from step 1   │
        └─────────────────────────────────────┘

**原则：先尝试，失败后再调试。**不要浪费时间做预检查——直接运行 DAG，出错后再诊断。

---

阶段 1：触发并等待

使用 af runs trigger-wait 测试 DAG：

主要方法：触发并等待

af runs trigger-wait <dag_id> --timeout 300

示例：

af runs trigger-wait my_dag --timeout 300

推荐使用此方法的原因：

• 一条命令即可完成触发和监控
• DAG 完成时立即返回（成功或失败）
• 如果运行失败，会包含失败任务的详细信息
• 无需手动轮询

解读响应

成功：

{
  "dag_run": {
    "dag_id": "my_dag",
    "dag_run_id": "manual__2025-01-14T...",
    "state": "success",
    "start_date": "...",
    "end_date": "..."
  },
  "timed_out": false,
  "elapsed_seconds": 45.2
}

失败：

{
  "dag_run": {
    "state": "failed"
  },
  "timed_out": false,
  "elapsed_seconds": 30.1,
  "failed_tasks": [
    {
      "task_id": "extract_data",
      "state": "failed",
      "try_number": 2
    }
  ]
}

超时：

{
  "dag_id": "my_dag",
  "dag_run_id": "manual__...",
  "state": "running",
  "timed_out": true,
  "elapsed_seconds": 300.0,
  "message": "Timed out after 300 seconds. DAG run is still running."
}

替代方法：分别触发和监控

仅在需要更多控制时使用：

# Step 1: Trigger
af runs trigger my_dag
# Returns: {"dag_run_id": "manual__...", "state": "queued"}

# Step 2: Check status
af runs get my_dag manual__2025-01-14T...
# Returns current state

---

处理结果

如果成功

DAG 已成功运行。向用户总结：

• 总耗时
• 已完成的任务数
• 值得注意的输出（如果日志中可见）

完成！

如果超时

DAG 仍在运行。可选操作：

1. 检查当前状态：af runs get <dag_id> <dag_run_id>
2. 询问用户是否要继续等待
3. 增加超时时间并重试

如果失败

进入阶段 2（调试）以确定根本原因。

---

阶段 2：调试失败（仅在需要时）

DAG 运行失败时，使用以下命令进行诊断：

获取全面诊断

af runs diagnose <dag_id> <dag_run_id>

一次调用即可返回：

• 运行元数据（状态、时间）
• 所有任务实例及其状态
• 失败任务摘要
• 状态计数（成功、失败、跳过等）

获取任务日志

af tasks logs <dag_id> <dag_run_id> <task_id>

示例：

af tasks logs my_dag manual__2025-01-14T... extract_data

针对特定重试尝试：

af tasks logs my_dag manual__2025-01-14T... extract_data --try 2

查找以下内容：

• 异常消息和堆栈跟踪
• 连接错误（数据库、API、S3）
• 权限错误
• 超时错误
• 缺失的依赖项

检查上游任务

如果任务显示 upstream_failed，根本原因在上游任务中。使用 af runs diagnose 找出实际失败的任务。

检查导入错误（DAG 未运行时）

如果触发失败是因为 DAG 不存在：

af dags errors

这会显示阻止 DAG 加载的语法错误或缺失的依赖项。

---

阶段 3：修复并重新测试

确定问题后：

常见修复方法

问题	修复方法
缺少导入	添加到 DAG 文件中
缺少软件包	添加到 requirements.txt
连接错误	检查 af config connections，验证凭据
缺少变量	检查 af config variables，并在需要时创建
超时	增加任务超时时间或优化查询
权限错误	检查连接中的凭据

修复后

1. 保存文件
2. 重新测试： af runs trigger-wait <dag_id>

重复“测试 → 调试 → 修复”循环，直到 DAG 成功。

---

CLI 快速参考

阶段	命令	用途
测试	af runs trigger-wait <dag_id>	主要测试方法——从这里开始
测试	af runs trigger <dag_id>	启动运行（备用方法）
测试	af runs get <dag_id> <run_id>	检查运行状态
调试	af runs diagnose <dag_id> <run_id>	全面诊断故障
调试	af tasks logs <dag_id> <run_id> <task_id>	获取任务输出／错误
调试	af dags errors	检查解析错误（如果 DAG 无法加载）
调试	af dags get <dag_id>	验证 DAG 配置
调试	af dags explore <dag_id>	全面检查 DAG
配置	af config connections	列出连接
配置	af config variables	列出变量

---

测试场景

场景 1：测试 DAG（顺利完成）

af runs trigger-wait my_dag
# Success! Done.

场景 2：测试 DAG（出现故障）

# 1. Run and wait
af runs trigger-wait my_dag
# Failed...

# 2. Find failed tasks
af runs diagnose my_dag manual__2025-01-14T...

# 3. Get error details
af tasks logs my_dag manual__2025-01-14T... extract_data

# 4. [Fix the issue in DAG code]

# 5. Retest
af runs trigger-wait my_dag

场景 3：DAG 不存在／无法加载

# 1. Trigger fails - DAG not found
af runs trigger-wait my_dag
# Error: DAG not found

# 2. Find parse error
af dags errors

# 3. [Fix the issue in DAG code]

# 4. Retest
af runs trigger-wait my_dag

场景 4：调试失败的定时运行

# 1. Get failure summary
af runs diagnose my_dag scheduled__2025-01-14T...

# 2. Get error from failed task
af tasks logs my_dag scheduled__2025-01-14T... failed_task_id

# 3. [Fix the issue]

# 4. Retest
af runs trigger-wait my_dag

场景 5：使用自定义配置进行测试

af runs trigger-wait my_dag --conf '{"env": "staging", "batch_size": 100}' --timeout 600

场景 6：长时间运行的 DAG

# Wait up to 1 hour
af runs trigger-wait my_dag --timeout 3600

# If timed out, check current state
af runs get my_dag manual__2025-01-14T...

---

调试提示

常见错误模式

连接被拒绝／超时：

• 使用 af config connections 检查主机／端口是否正确
• 验证与外部系统的网络连接
• 检查连接凭据是否正确

ModuleNotFoundError：

• requirements.txt 中缺少软件包
• 添加后可能需要重启环境

PermissionError：

• 检查 IAM 角色、数据库授权、API 密钥
• 确认连接具有正确的凭据

任务超时：

• 查询或操作耗时过长
• 考虑为任务添加超时参数
• 优化底层查询／操作

查看任务日志

任务日志通常会显示：

1. 任务开始时间戳
2. 任务代码中的打印／日志语句
3. 返回值（适用于使用 @task 装饰的函数）
4. 异常和完整堆栈跟踪（如果失败）
5. 任务结束时间戳和持续时间

重点查看失败任务日志底部的异常。

在 Astro 上

Astro 部署支持环境晋级，有助于构建测试工作流：

• 开发部署：使用 astro deploy --dags 自由测试 DAG，以便快速迭代
• 预发布部署：针对类似生产环境的数据运行集成测试
• 生产部署：仅在低级环境中完成验证后再部署
• 为每个环境使用单独的 Astro 部署，并通过这些环境逐步晋级代码

---

相关技能

• authoring-dags：用于创建新的 DAG（包含测试前的验证）
• debugging-dags：用于常规 Airflow 故障排查
• deploying-airflow：用于测试后将 DAG 部署到生产环境