气流Python单元测试？

本文介绍了气流Python单元测试？的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

我想为我们的DAG添加一些单元测试，但找不到任何单元测试。是否有DAG单元测试的框架？存在一个端到端测试框架，但我想它已经死了： https：/ /issues.apache/jira/browse/AIRFLOW-79 。请建议，谢谢！

I'd like to add some unit tests for our DAGs, but could not find any. Is there a framework for unit test for DAGs? There is an End-to-End testing framework that exists but I guess it's dead: issues.apache/jira/browse/AIRFLOW-79. Please suggest, Thanks!

推荐答案

像这样测试您的运营商：

Test your operators like this:

class TestMyOperator(TestCase): def test_execute(self): with DAG(dag_id="foo", start_date=datetime.now()): task = MyOperator(task_id="foo") ti = TaskInstance(task=task, execution_date=datetime.now()) result = task.execute(ti.get_template_context()) self.assertEqual(result, "foo")

来源