pytest 不支持真正的 DAG,仅能通过插件实现线性依赖;复杂场景应交由 Airflow 等调度器,pytest 退为执行单元。

pytest 本身不支持 DAG,别硬塞依赖关系
pytest 的 pytest.mark.dependency 只能做线性前置约束(A 必须在 B 前跑),不是 DAG。强行用它拼复杂依赖,会遇到:测试节点重复执行、依赖环检测缺失、失败后无法跳过下游——这些不是配置问题,是模型不匹配。
真正需要 DAG 编排时,得换角色:让 pytest 专注“单个测试单元的断言与 fixture 管理”,把调度逻辑交给外部工具或轻量级 DAG 引擎。
- 如果只是几个测试函数之间有明确先后+条件跳过,用
pytest-dependency+depends参数足够,但别超过 5 个节点 - 如果涉及并行分支、重试策略、状态传递(比如 A 输出要给 C 当输入)、失败自动降级,就得引入
airflow或prefect这类调度器,pytest 退为 task 中的执行单元 - 避免在
conftest.py里写循环 import + 动态注册 test 函数来模拟 DAG——这会让pytest --collect-only失效,CI 里 collect 阶段就报错
用 pytest_generate_tests 实现参数化 DAG 节点
当你的“DAG”本质是同一逻辑在不同数据/环境上的多轮验证(比如:上传→校验→归档→通知),可以用 pytest_generate_tests 钩子动态生成带顺序标记的测试项,再靠 order 插件控制执行流。
关键点在于:生成的每个 test_* 函数名必须含唯一标识,且参数中显式携带 stage 名和 predecessor 列表,否则 order 插件无法识别依赖链。
立即学习“Python免费学习笔记(深入)”;
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac,无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启:辅助功能 + 屏幕录制权限
- 在
conftest.py中实现pytest_generate_tests,根据 YAML 定义的 DAG 配置生成metafunc.parametrize - 每个参数元组里必须包含
stage(如"validate")和depends_on(如["upload"]),供pytest-ordering插件读取 - 不要在
parametrize的ids里用 lambda 或闭包——pytest 会序列化失败,报PickleError: Can't pickle <function ...></function> - 示例片段:
def pytest_generate_tests(metafunc): if "stage" in metafunc.fixturenames: dag = load_yaml("pipeline.yaml") for node in dag["nodes"]: metafunc.parametrize("stage,depends_on", [(node["name"], node.get("depends_on", []))], ids=[node["name"]])
fixture 模拟 DAG 上下文传递,但别跨 stage 共享 mutable 对象
DAG 各阶段常需传递中间结果(如文件路径、响应 ID),用 fixture 是对的,但必须区分 scope 和生命周期。pytest 的 scope="session" 或 scope="package" fixture 在多 stage 场景下极易引发状态污染。
比如 stage A 的 fixture 返回一个 dict,stage B 修改了它,stage C 再用时已是脏数据——这不是 bug,是 scope 设计使然。
- 每个 stage 应定义独立 fixture,scope 设为
"function",通过参数依赖显式接收上游输出:def stage_b(stage_a_output): - 上游 fixture 返回值必须是不可变对象(
str,int,namedtuple),或深拷贝后的副本;避免返回dict/list并在下游修改 - 若需跨 stage 共享临时目录,用
tmp_path_factory创建 scoped 子目录,路径传参,不传Path对象本身 - 错误示范:
@pytest.fixture(scope="session") def shared_ctx(): return {"data": []}→ 多 stage 并发时数据混杂
CI 中 DAG 执行失败时,pytest 的 exit code 不反映真实失败粒度
默认情况下,只要有一个 test 函数失败,pytest 就返回 1;但 DAG 场景里你可能希望:核心路径失败才阻断流水线,非关键分支失败只告警。这时候不能只看 exit code。
根本问题是 pytest 把所有 test 当平级用例,没内置“任务组失败权重”概念。
- 用
--tb=no -q配合pytest-asyncio或自定义 hook,在pytest_runtest_makereport中按 stage 分类统计失败数,写入临时文件供 CI 解析 - 禁止在 CI 脚本里用
|| true掩盖失败——这会让 stage C 的崩溃被 stage A 的成功掩盖 - 如果用了
pytest-xdist并行,注意--dist=loadgroup无法保证 stage 顺序,必须加--max-worker-restart=0防止 worker 重启导致 fixture 状态丢失 - 真正复杂的 DAG 编排,最终还是得导出为
airflow DAG Python file,把每个 pytest 调用包装成BashOperator,由 Airflow 管理重试、超时、上下游通知

















