气流Python单元测试？ [英] Airflow Python Unit Test?

问题描述

我想为我们的DAG添加一些单元测试，但找不到任何单元测试。是否有DAG单元测试的框架？存在一个端到端测试框架，但我想它已经死了： https：/ /issues.apache.org/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: https://issues.apache.org/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")

来源

这篇关于气流Python单元测试？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！