docstr: Tests that skipped state of leaf tasks propagates to the parent dag.
  Note that the skipped state propagation only takes affect when the dagrun's state is SUCCESS.
docstr_tok: ["Tests","that","skipped","state","of","leaf","tasks","propagates","to","the","parent","dag",".","Note","that","the","skipped","state","propagation","only","takes","affect","when","the","dagrun","'s","state","is","SUCCESS","."]
code: def test_subdag_with_propagate_skipped_state(
      self,
      mock_get_task_instance,
      mock_skip,
      dag_maker,
      propagate_option,
      states,
      skip_parent,
  ):
      """
      Tests that skipped state of leaf tasks propagates to the parent dag.
      Note that the skipped state propagation only takes affect when the dagrun's state is SUCCESS.
      """
      with dag_maker(
          "parent.test", default_args=default_args
      ) as subdag:
          dummy_subdag_tasks = [
              EmptyOperator(task_id=f"dummy_subdag_{i}")
              for i in range(len(states))
          ]
      dag_maker.create_dagrun(execution_date=DEFAULT_DATE)
  
      with dag_maker("parent", default_args=default_args):
          subdag_task = SubDagOperator(
              task_id="test",
              subdag=subdag,
              poke_interval=1,
              propagate_skipped_state=propagate_option,
          )
          dummy_dag_task = EmptyOperator(task_id="dummy_dag")
          subdag_task >> dummy_dag_task
      dag_run = dag_maker.create_dagrun(execution_date=DEFAULT_DATE)
  
      subdag_task._get_dagrun = Mock(return_value=self.dag_run_success)
  
      mock_get_task_instance.side_effect = [
          TaskInstance(task=task, run_id=dag_run.run_id, state=state)
          for task, state in zip(dummy_subdag_tasks, states)
      ]
  
      context = {
          "execution_date": DEFAULT_DATE,
          "dag_run": dag_run,
          "task": subdag_task,
          "ti": mock.MagicMock(map_index=-1),
      }
      subdag_task.post_execute(context)
  
      if skip_parent:
          mock_skip.assert_called_once_with(
              context["dag_run"],
              context["execution_date"],
              [dummy_dag_task],
              map_index=-1,
          )
      else:
          mock_skip.assert_not_called()
  
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 12 for calling the TestSubDagOperator obj's underlying member method code with example usage: obj.test_subdag_with_propagate_skipped_state(mock_get_task_instance, mock_skip, dag_maker, propagate_option, states, skip_parent) without return types