■ はじめに

Airflow の単体試験をすることになりそうなので予習。

【１】Airflow の単体試験
　１）Airflow's Best Practices
　２）テスティングフレームワークの選定
【２】Airflowのテストモード
　１）設定方法１：環境変数 UNIT_TEST_MODE に指定
　２）設定方法２：airflow.cfg に unit_test_mode を指定
【３】サンプル
　１）テストコード

【１】Airflow の単体試験

１）Airflow's Best Practices

https://airflow.apache.org/docs/apache-airflow/stable/best-practices.html

Best Practices
Creating a new DAG is a three-step process:

1. writing Python code to create a DAG object,
2. testing if the code meets your expectations, <<★
3. configuring environment dependencies to run your DAG

This tutorial will introduce you to the best practices for these three steps

https://airflow.apache.org/docs/apache-airflow/stable/best-practices.html#testing-a-dag

２）テスティングフレームワークの選定

* 以下のどちらかが主流っぽく、業務条件で選定すれば良さそう
 
[1] pytest ... 上の公式ドキュメントでも使用
[2] unittest ... 標準なのでインストール不要

【２】Airflowのテストモード

* 公式ドキュメントに記載

https://airflow.apache.org/docs/apache-airflow/2.1.0/howto/use-test-config.html

* 以下も分かりやすい

https://dev.classmethod.jp/articles/howto-unittestmode-on-airflow/

１）設定方法１：環境変数 UNIT_TEST_MODE に指定

export AIRFLOW__CORE__UNIT_TEST_MODE=True

２）設定方法２：airflow.cfg に unit_test_mode を指定

[tests]
unit_test_mode = True

【３】サンプル

* pytest を選定した場合の例
* 環境設定や実行方法は、以下の関連記事を参照のこと

Apache Airflow ～単体試験 / 環境設定編～
https://dk521123.hatenablog.com/entry/2024/06/16/002630

hello_world.py

import airflow
from airflow.decorators import dag, task
from airflow.utils.dates import days_ago

default_args = {
  "owner": "airflow"
}
@dag(
  default_args=default_args,
  dag_id="hello_world",
  schedule_interval="@daily",
  start_date=days_ago(2),
  tags=['example']
)
def main_dag():
  @task
  def get_hello_world():
     return 'hello world'

  @task
  def say_hello(result):
    print(result)

  result = get_hello_world()
  say_hello(result)

main_dag = main_dag()

test_hello_world.py

import pytest

from airflow.models import DagBag


@pytest.fixture()
def dagbag():
    return DagBag()


def test_dag_loaded(dagbag):
    dag = dagbag.get_dag(dag_id="hello_world")
    assert dagbag.import_errors == {}
    assert dag is not None
    assert len(dag.tasks) == 2

参考文献

https://zenn.dev/jcc/articles/c339865bd9a4b5
https://goodbyegangster.hatenablog.com/entry/2022/08/05/024037
https://takemikami.com/2021/1205-airflowdag.html
https://zenn.dev/masahito/scraps/8f5256223c013b

プログラムの超個人的なメモ

Memo for Programming.

【Airflow】Apache Airflow ～単体試験 / 入門編～

■ はじめに

目次

【１】Airflow の単体試験

１）Airflow's Best Practices

２）テスティングフレームワークの選定

【２】Airflowのテストモード

１）設定方法１：環境変数 UNIT_TEST_MODE に指定

２）設定方法２：airflow.cfg に unit_test_mode を指定

【３】サンプル

参考文献

関連記事