DAG Generator Testing
Relevant source files
The following files were used as context for generating this wiki page:
Introduction
DAG Generator Testing ensures that the Meltano Airflow extension correctly transforms Meltano schedules into functional Apache Airflow Directed Acyclic Graphs (DAGs). The core logic resides in a bundled generator file, meltano.py, which is dynamically loaded by Airflow to orchestrate data pipelines defined in a Meltano project.
The testing framework validates that different versions of Meltano schedule exports (v1 and v2) result in valid Airflow objects without import errors. It specifically checks for correct task mapping, dependency ordering, and the handling of edge cases like one-off schedules.
Sources: tests/test_dag_generator.py:1-5, airflow_ext/files/orchestrate/meltano.py:1-15
Test Architecture and Environment
Testing the DAG generator requires simulating a Meltano project environment and an Airflow runtime. This is achieved through pytest fixtures and stubs that mimic the Meltano CLI behavior.
Meltano Project Stubbing
A specialized fixture, meltano_project, is used to create a temporary project structure. It writes a fake meltano executable to the system PATH. This stub is configured to intercept calls to meltano schedule list --format=json and return predefined JSON payloads representing various pipeline configurations.
Sources: tests/conftest.py:38-65
Airflow Integration
The tests utilize the airflow.models.DagBag class to process the generated DAGs. Instead of scanning a real directory, the _load_dag_bag helper function points Airflow directly to the meltano.py template located within the extension's package resources.
Sources: tests/test_dag_generator.py:53-63
This diagram illustrates how the test environment stubs Meltano outputs to feed the DAG generator for validation.
Testing Scenarios
The test suite covers multiple generations of Meltano schedule formats and specific scheduling logic.
Legacy (v1) Schedule Validation
The suite validates that legacy schedules—consisting of a flat list of ELT tasks—generate a single DAG with an extract_load task.
| Feature | Expectation |
|---|---|
| DAG ID Format | meltano_<schedule_name> |
| Task ID | extract_load |
| Import Errors | Must be empty |
Sources: tests/test_dag_generator.py:11-20, tests/test_dag_generator.py:66-74
Modern (v2) Schedule Validation
V2 schedules include both standard elt tasks and complex job tasks. Testing ensures that jobs with multiple tasks are correctly mapped to an Airflow task sequence with appropriate dependencies.
Sources: tests/test_dag_generator.py:23-50, tests/test_dag_generator.py:77-91
The generator transforms sequential Meltano job tasks into an Airflow dependency chain.
One-Off and Empty Schedules
The generator is tested to ensure it ignores schedules where cron_interval is null (representing @once schedules) and gracefully handles empty schedule lists.
Sources: tests/test_dag_generator.py:94-110, airflow_ext/files/orchestrate/meltano.py:56-61
Cross-Version Compatibility Testing
The project uses tox to run the DAG generator tests across multiple Python and Airflow versions. This ensures the generator logic remains compatible with both Airflow 2.x and Airflow 3.x syntax requirements, specifically regarding the change from schedule_interval to schedule.
| Factor | Versions Tested |
|---|---|
| Python | 3.10, 3.11, 3.12, 3.13, 3.14 |
| Airflow | >=2.9, <3.0 (Airflow 2), >=3.0, <4.0 (Airflow 3) |
Sources: tox.toml:2-25, pyproject.toml:7-14
Implementation Details in Generator
The meltano.py generator, which is the subject of these tests, uses subprocess.run to call the Meltano binary. The tests verify this interaction by checking if the generator correctly parses the JSON output and populates the globals() dictionary with DAG objects.
# From airflow_ext/files/orchestrate/meltano.py:155-163
def create_dags() -> None:
"""Create DAGs for Meltano schedules."""
list_result = subprocess.run(
[MELTANO_BIN, "schedule", "list", "--format=json"],
cwd=PROJECT_ROOT,
stdout=subprocess.PIPE,
text=True,
check=True,
)
schedule_export = json.loads(list_result.stdout)
Sources: airflow_ext/files/orchestrate/meltano.py:155-163
Conclusion
DAG Generator Testing provides critical assurance that the mapping between Meltano's orchestration metadata and Airflow's execution model is accurate. By stubbing the Meltano CLI and utilizing the native Airflow DagBag for validation, the suite confirms that generated DAGs are syntactically correct and logically sound across different Airflow versions.
Sources: tests/test_dag_generator.py, tests/conftest.py