气流Python单元测试？

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

来源

2017-07-31 Chengzhi

我已经在隔离工作都是我自定义逻辑放入'./ plugins'目录并为那里的逻辑创建一个测试套件。在此期间，不是端到端，但可能是一个很好的方法：https：//airflow.incubator.apache.org/plugins.html。如果你将大量未经测试的逻辑填充到你的'。/ dags'目录中，你可能需要考虑将其分解为插件。 – Mike

目前我无法找到任何优于单纯使用BashOperator：

with DAG('platform-test', start_date=datetime(2017, 8, 29)) as dag: 
    test_command = "python3 -m unittest --verbose {}".format(platform_test_fname) 
    op = BashOperator(
     task_id="platform-test", 
     bash_command=test_command, 
    )

来源

2017-08-29 07:10:12

测试你的运营商是这样的：

class TestMyOperator(TestCase): 

    def test_execute(self): 
     dag = DAG(dag_id='foo', start_date=datetime.now()) 
     task = MyOperator(dag=dag, task_id='foo') 
     ti = TaskInstance(task=task, execution_date=datetime.now()) 
     result = task.execute(ti.get_template_context()) 
     self.assertEqual(result, 'foo')

Source

来源

2017-12-23 06:49:24

气流Python单元测试？

回答

相关问题