本文介绍了气流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")来源
更多推荐
气流Python单元测试?
发布评论