dag_id = "test_subdag"
        self._clean_up(dag_id)
        task_id = "t1"
        dag = DAG(dag_id, start_date=DEFAULT_DATE, max_active_runs=1)
        t_1 = EmptyOperator(task_id=task_id, dag=dag)
        subdag = DAG(dag_id + ".test", start_date=DEFAULT_DATE, max_active_runs=1)
        SubDagOperator(task_id="test", subdag=subdag, dag=dag)
        t_2 = EmptyOperator(task_id="task", dag=subdag)
        subdag.parent_dag = dag
        dag.sync_to_db()
        session = settings.Session()
        dag.create_dagrun(
            run_type=DagRunType.MANUAL,
            state=State.FAILED,
            start_date=DEFAULT_DATE,
            execution_date=DEFAULT_DATE,
            session=session,
        )
        subdag.create_dagrun(
            run_type=DagRunType.MANUAL,
            state=State.FAILED,
            start_date=DEFAULT_DATE,
            execution_date=DEFAULT_DATE,
            session=session,