Skip to content

Commit

Permalink
Add integration test to cover the expected behaviour when rendering D…
Browse files Browse the repository at this point in the history
…AG with multiple tests
  • Loading branch information
tatiana committed Dec 27, 2024
1 parent 1ebe514 commit 0f0c0da
Showing 1 changed file with 41 additions and 0 deletions.
41 changes: 41 additions & 0 deletions tests/test_converter.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
SAMPLE_PROFILE_YML = Path(__file__).parent / "sample/profiles.yml"
SAMPLE_DBT_PROJECT = Path(__file__).parent / "sample/"
SAMPLE_DBT_MANIFEST = Path(__file__).parent / "sample/manifest.json"
MULTIPLE_PARENTS_TEST_DBT_PROJECT = Path(__file__).parent.parent / "dev/dags/dbt/multiple_parents_test/"


@pytest.mark.parametrize("argument_key", ["tags", "paths"])
Expand Down Expand Up @@ -164,6 +165,46 @@ def test_converter_creates_dag_with_seed(mock_load_dbt_graph, execution_mode, op
assert converter


@pytest.mark.integration
def test_converter_creates_dag_with_test_with_multiple_parents():
"""
Validate topology of a project that uses the MULTIPLE_PARENTS_TEST_DBT_PROJECT project
"""
project_config = ProjectConfig(dbt_project_path=MULTIPLE_PARENTS_TEST_DBT_PROJECT)
execution_config = ExecutionConfig(execution_mode=ExecutionMode.LOCAL)
profile_config = ProfileConfig(
profile_name="default",
target_name="dev",
profile_mapping=PostgresUserPasswordProfileMapping(
conn_id="example_conn",
profile_args={"schema": "public"},
disable_event_tracking=True,
),
)
with DAG("sample_dag", start_date=datetime(2024, 4, 16)) as dag:
converter = DbtToAirflowConverter(
dag=dag, project_config=project_config, profile_config=profile_config, execution_config=execution_config
)
tasks = converter.tasks_map

assert len(converter.tasks_map) == 4

# We exclude the test that depends on combined_model and model_a from their commands
args = tasks["model.my_dbt_project.combined_model"].children["combined_model.test"].build_cmd({})[0]
assert args[1:] == ["test", "--exclude", "custom_test_combined_model_combined_model_", "--models", "combined_model"]

args = tasks["model.my_dbt_project.model_a"].children["model_a.test"].build_cmd({})[0]
assert args[1:] == ["test", "--exclude", "custom_test_combined_model_combined_model_", "--models", "model_a"]

# The test for model_b should not be changed, since it is not a parent of this test
args = tasks["model.my_dbt_project.model_b"].children["model_b.test"].build_cmd({})[0]
assert args[1:] == ["test", "--models", "model_b"]

# We should have a task dedicated to run the test with multiple parents
args = tasks["test.my_dbt_project.custom_test_combined_model_combined_model_.c6e4587380"].build_cmd({})[0]
assert args[1:] == ["test", "--select", "custom_test_combined_model_combined_model_.c6e4587380"]


@pytest.mark.parametrize(
"execution_mode,operator_args",
[
Expand Down

0 comments on commit 0f0c0da

Please sign in to comment.