Airflow 코드 테스트

Airflow로 데이터 파이프라인 구축하기

Volker Janz

Senior Developer Advocate at Astronomer

테스트의 세 단계

Dag 테스트 피라미드

무결성: Dag가 오류 없이 로드되나요?
단위: 비즈니스 로직이 올바른 결과를 반환하나요?
통합: Dag 전체가 처음부터 끝까지 실행되나요?

DagBag로 무결성 테스트하기

from airflow.models import DagBag


dag_bag = DagBag(include_examples=False)


def test_no_import_errors():
    assert len(dag_bag.import_errors) == 0


def test_dag_loaded():
    assert "daily_etl" in dag_bag.dags

Dag 임포트 오류의 원인

ModuleNotFoundError

ModuleNotFoundError: 프로바이더 패키지 누락 또는 잘못된 임포트 경로
NameError: 변수나 함수 이름 변경 또는 오타
ImportError: Dag 파일 간 순환 임포트
무결성 테스트로 이러한 문제를 예방할 수 있습니다

태스크 함수 단위 테스트

Dag 파일:

def clean_record(record):
    return {
        "name": record["name"].strip(),
        "email": record["email"].lower(),
    }

@task
def transform(records):
    return [clean_record(r)
            for r in records]

비즈니스 로직을 별도로 분리하세요

테스트 파일:

from dags.data_cleaning import (
    clean_record,
)

def test_strips_whitespace():
    result = clean_record(
      {"name": "  Alice  ",
       "email": "[email protected]"}
    )
    assert result["name"] == "Alice"

단위 테스트는 비즈니스 로직에 집중하세요

dag.test()로 통합 테스트하기

import pytest
from airflow.models import DagBag
from pendulum import datetime

dag_bag = DagBag(include_examples=False)


def test_etl_pipeline():
    dag = dag_bag.get_dag("etl_output")
    assert dag is not None

    dag.test(logical_date=datetime(2026, 1, 15))

    output = Path("/tmp/etl_results.json")
    assert output.exists()
    results = json.loads(output.read_text())
    assert len(results) == 2

제어된 입력으로 실제 Dag를 실행하고 출력을 검증합니다

CI에서의 테스트

CI 파이프라인

무결성 + 단위: 매 커밋 시 실행 (빠름)
통합: 풀 리퀘스트 또는 야간 실행 (느림)
세 가지 테스트를 모두 통과해야 프로덕션에 배포됩니다

연습해 봅시다!

Airflow로 데이터 파이프라인 구축하기