[Airflow - 7단계: DAG CI 구축: test_dag_integrity.py + GitHub Actions]

import pytest
from airflow.models import DagBag

DAG_FOLDER = "dags"

@pytest.fixture(scope="module")
def dagbag():
    return DagBag(dag_folder=DAG_FOLDER, include_examples=False)

def test_no_import_errors(dagbag):
    """DAG 파일 import 오류가 없어야 한다."""
    assert dagbag.import_errors == {}, (
        f"DAG import errors: {dagbag.import_errors}"
    )

EXPECTED_DAG_IDS = [
    "e2e_full",
    "dp_feature_pipeline",
    "feast_materialize_dev",
    "feast_full_refresh_dev_manual",
    "rollback_manual",
]

def test_expected_dag_ids_exist(dagbag):
    """예상된 모든 DAG이 정상 로드되어야 한다."""
    for dag_id in EXPECTED_DAG_IDS:
        assert dag_id in dagbag.dags, (
            f"Expected DAG '{dag_id}' not found in DagBag"
        )

def test_dag_has_tasks(dagbag):
    """각 DAG에 최소 1개 이상의 Task가 있어야 한다."""
    for dag_id, dag in dagbag.dags.items():
        assert len(dag.tasks) > 0, f"DAG '{dag_id}' has no tasks"

E2E_FULL_KEY_TASK_IDS = [
    "train_and_evaluate",
    "check_result",
    "observe_post_deploy_metrics",
    "rollback_minimal",
]

def test_e2e_full_key_task_ids(dagbag):
    dag = dagbag.dags["e2e_full"]
    task_ids = {t.task_id for t in dag.tasks}
    for tid in E2E_FULL_KEY_TASK_IDS:
        assert tid in task_ids, f"e2e_full: required task '{tid}' not found"

def test_e2e_full_max_active_runs(dagbag):
    dag = dagbag.dags["e2e_full"]
    assert dag.max_active_runs == 1

def test_e2e_full_catchup_disabled(dagbag):
    dag = dagbag.dags["e2e_full"]
    assert dag.catchup is False

from datetime import timedelta

def test_e2e_full_sla_in_default_args(dagbag):
    dag = dagbag.get_dag("e2e_full")
    sla = dag.default_args.get("sla")
    assert sla is not None
    assert isinstance(sla, timedelta)
    assert sla > timedelta(minutes=30)

def test_e2e_full_sla_miss_callback(dagbag):
    dag = dagbag.get_dag("e2e_full")
    assert dag.sla_miss_callback is not None

# .github/workflows/ci.yml
name: DAG Integrity

on:
  push:
    branches: ["**"]
  pull_request:
    branches: ["**"]

jobs:
  test:
    name: DagBag Integrity Test
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install dependencies
        run: |
          pip install --upgrade pip
          if [ -f requirements.txt ]; then
            pip install -r requirements.txt
          else
            pip install "apache-airflow>=2.10.0,<3.0" pytest
          fi

      - name: Initialize Airflow DB
        run: |
          export AIRFLOW_HOME="${GITHUB_WORKSPACE}/.airflow"
          airflow db init
        env:
          AIRFLOW__CORE__DAGS_FOLDER: ${{ github.workspace }}/dags
          AIRFLOW__CORE__LOAD_EXAMPLES: "False"

      - name: Run DAG integrity tests
        run: pytest tests/test_dag_integrity.py -v --tb=short
        env:
          AIRFLOW_HOME: ${{ github.workspace }}/.airflow
          AIRFLOW__CORE__DAGS_FOLDER: ${{ github.workspace }}/dags
          AIRFLOW__CORE__LOAD_EXAMPLES: "False"
          MLFLOW_TRACKING_URI: "sqlite:///mlflow.db"

# 의존성 설치
pip install "apache-airflow>=2.10.0,<3.0" pytest

# DB 초기화
export AIRFLOW_HOME="${PWD}/.airflow"
export AIRFLOW__CORE__DAGS_FOLDER="${PWD}/dags"
export AIRFLOW__CORE__LOAD_EXAMPLES=False
airflow db init

# 테스트 실행
pytest tests/test_dag_integrity.py -v --tb=short