气流一体化试验

Question

针对BigQuery运行气流集成测试的好方法/推荐方法是什么？

气流似乎有相当多的操作人员为测试设计，例如BigQueryCheckOperator。我很难找到任何最佳实践或示例来展示如何将其用于集成测试。

建议以何种方式进行集成测试：

• 具体任务
• 整个DAGs

最好使用用Python编写的测试框架，因为气流也是用Python编写的，比如pytest或类似的。

Answer 1

您可以像这样在整个dag上运行测试：

import unittest
from airflow.models import DagBag

class TestDagIntegrity(unittest.TestCase):

LOAD_SECOND_THRESHOLD = 2

def setUp(self):
    self.dagbag = DagBag()

def test_import_dags(self):
    """ Test if dags works, no fail in import
    """
    self.assertFalse(
        len(self.dagbag.import_errors),
        'DAG import failures. Errors: {}'.format(
            self.dagbag.import_errors
        )
    )

suite = unittest.TestLoader().loadTestsFromTestCase(TestDagIntegrity)
unittest.TextTestRunner(verbosity=2).run(suite)

它将测试您是否工作，如果导入没有失败或缺少变量。