Testing a DAG Without a Cluster
A team wants CI to catch broken DAGs before deploy. Their DAGs import a custom internal library, call Variable.get() at module import time, and use PostgresHook inside task functions. CI has no Airflow metadata database, no Postgres, and no network access. A previous attempt ran airflow dags test and failed because the scheduler tried to reach the database. The team has 40 DAG files and a 10-minute CI budget. They want fast feedback on import errors, missing dependencies, and obvious logic bugs without spinning up a full environment.
1. Explain why `Variable.get()` at import time and `PostgresHook` in task functions cause problems for CI, and what categories of failure CI can still catch.
Read the worked answer
Variable.get() at module import time queries the metadata database during DAG parsing. In CI without a database, parsing fails, and in production it also causes a database query per parse cycle, which is a known performance and hygiene problem. PostgresHook inside a task function is only instantiated at execution time, so it does not break parsing, but it means unit tests that call the function need a database or a mock.
CI without a full environment can still catch: Python syntax and import errors, DAG import errors from missing internal library versions, duplicate DAG IDs, cycles in the task graph, missing required arguments, and template rendering errors that can be checked statically. It can also run the DAG's task functions with mocked hooks to test branching logic and XCom handling.
The team should move Variable.get() into task functions or use Variable.get with a default, and inject configuration through params or environment variables so parsing is side-effect free.
2. Design a CI test strategy that fits the 10-minute budget and covers the stated goals.
Read the worked answer
Use a two-layer strategy. Layer one: a fast static check that imports every DAG file with a minimal Airflow environment (SQLite or AIRFLOW__CORE__UNIT_TEST_MODE style settings) and asserts no import errors, no duplicate DAG IDs, and no cycles. This should take under a minute for 40 files. Use DagBag with a temporary folder and assert bag.import_errors is empty.
Layer two: unit tests for task functions with mocked hooks and variables, using unittest.mock.patch on PostgresHook and Variable.get. Test branching and skip logic with @task.short_circuit. Keep these targeted to the highest-risk DAGs to stay within budget.
Optionally add a contract test that renders templates with sample params. Do not run airflow dags test in CI without a database; instead use a containerized Airflow with SQLite only for a nightly job. Cache dependencies and parallelize test files to fit 10 minutes.