Function test_run

tests/helpers/airflow_tests/test_airflow_wrapper.py:256–295

()

def test_run() -> None:
    task: BaseOperator = None

    pipeline_standalone = dlt.pipeline(
        pipeline_name="pipeline_standalone",
        dataset_name="mock_data_" + uniq_id(),
        destination=dlt.destinations.duckdb(credentials=":pipeline:"),
    )
    pipeline_standalone.run(mock_data_source())
    pipeline_standalone_counts = load_table_counts(pipeline_standalone)

    @dag(schedule=None, start_date=DEFAULT_DATE, catchup=False, default_args=default_args)
    def dag_regular():
        nonlocal task
        tasks = PipelineTasksGroup(
            "pipeline_dag_regular", local_data_folder=get_test_storage_root(), wipe_local_data=False
        )

        # set duckdb to be outside of pipeline folder which is dropped on each task
        pipeline_dag_regular = dlt.pipeline(
            pipeline_name="pipeline_dag_regular",
            dataset_name="mock_data_" + uniq_id(),
            destination=dlt.destinations.duckdb(destination_name="dag_regular_b"),
        )
        task = tasks.run(pipeline_dag_regular, mock_data_source())

    dag_def: DAG = dag_regular()
    assert task.task_id == "mock_data_source__r_init-_t_init_post-_t1-_t2-2-more"

    dag_def.test()

    pipeline_dag_regular = dlt.attach(
        pipeline_name="pipeline_dag_regular",
    )
    assert pipeline_dag_regular.first_run is False

    pipeline_dag_regular_counts = load_table_counts(pipeline_dag_regular)
    assert pipeline_dag_regular_counts == pipeline_standalone_counts

    assert isinstance(task, PythonOperator)


@pytest.mark.parametrize("serialize_first_task", [True, False])