氣流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單元測試？

回答

相關問題