docstr: When SubDagOperator executes, it creates a DagRun if there is no existing one
  and wait until the DagRun succeeds.
docstr_tok: ["When","SubDagOperator","executes",",","it","creates","a","DagRun","if","there","is","no","existing","one","and","wait","until","the","DagRun","succeeds","."]
code: def test_execute_create_dagrun_with_conf(self):
      """
      When SubDagOperator executes, it creates a DagRun if there is no existing one
      and wait until the DagRun succeeds.
      """
      conf = {"key": "value"}
      dag = DAG("parent", default_args=default_args)
      subdag = DAG("parent.test", default_args=default_args)
      subdag_task = SubDagOperator(
          task_id="test",
          subdag=subdag,
          dag=dag,
          poke_interval=1,
          conf=conf,
      )
  
      subdag.create_dagrun = Mock()
      subdag.create_dagrun.return_value = self.dag_run_running
  
      subdag_task._get_dagrun = Mock()
      subdag_task._get_dagrun.side_effect = [
          None,
          self.dag_run_success,
          self.dag_run_success,
      ]
  
      context = {
          "data_interval_start": None,
          "data_interval_end": None,
          "execution_date": DEFAULT_DATE,
      }
  
      subdag_task.pre_execute(context=context)
      subdag_task.execute(context=context)
      subdag_task.post_execute(context=context)
  
      subdag.create_dagrun.assert_called_once_with(
          run_type=DagRunType.SCHEDULED,
          execution_date=DEFAULT_DATE,
          data_interval=None,
          conf=conf,
          state=State.RUNNING,
          external_trigger=True,
      )
  
      assert 3 == subdag_task._get_dagrun.call_count
  
imports: from __future__ import annotations
  from unittest import mock
  from unittest.mock import Mock
  import pytest
  import airflow
  from airflow.exceptions import AirflowException
  from airflow.models.dag import DAG
  from airflow.models.dagrun import DagRun
  from airflow.models.taskinstance import TaskInstance
  from airflow.operators.empty import EmptyOperator
  from airflow.operators.subdag import SkippedStatePropagationOptions, SubDagOperator
  from airflow.utils.session import create_session
  from airflow.utils.state import State
  from airflow.utils.timezone import datetime
  from airflow.utils.types import DagRunType
  from tests.test_utils.db import clear_db_runs
label_desc: Use image node_id 8 for calling the TestSubDagOperator obj's underlying member method code with example usage: obj.test_execute_create_dagrun_with_conf() without return types