code: def test_set_state_with_multiple_dagruns(self, session=None):
      self.dag2.create_dagrun(
          run_type=DagRunType.MANUAL,
          state=State.FAILED,
          execution_date=self.execution_dates[0],
          data_interval=(
              self.execution_dates[0],
              self.execution_dates[0],
          ),
          session=session,
      )
      dr2 = self.dag2.create_dagrun(
          run_type=DagRunType.MANUAL,
          state=State.FAILED,
          execution_date=self.execution_dates[1],
          data_interval=(
              self.execution_dates[1],
              self.execution_dates[1],
          ),
          session=session,
      )
      self.dag2.create_dagrun(
          run_type=DagRunType.MANUAL,
          state=State.RUNNING,
          execution_date=self.execution_dates[2],
          data_interval=(
              self.execution_dates[2],
              self.execution_dates[2],
          ),
          session=session,
      )
  
      altered = set_dag_run_state_to_success(
          dag=self.dag2, run_id=dr2.run_id, commit=True
      )
  
      # Recursively count number of tasks in the dag
      def count_dag_tasks(dag):
          count = len(dag.tasks)
          subdag_counts = [
              count_dag_tasks(subdag) for subdag in dag.subdags
          ]
          count += sum(subdag_counts)
          return count
  
      assert len(altered) == count_dag_tasks(self.dag2)
      self._verify_dag_run_state(
          self.dag2, self.execution_dates[1], State.SUCCESS
      )
  
      # Make sure other dag status are not changed
      models.DagRun.find(
          dag_id=self.dag2.dag_id,
          execution_date=self.execution_dates[0],
      )
      self._verify_dag_run_state(
          self.dag2, self.execution_dates[0], State.FAILED
      )
      models.DagRun.find(
          dag_id=self.dag2.dag_id,
          execution_date=self.execution_dates[2],
      )
      self._verify_dag_run_state(
          self.dag2, self.execution_dates[2], State.RUNNING
      )
  
imports: from __future__ import annotations
  import datetime
  from typing import Callable
  import pytest
  from sqlalchemy.orm import eagerload
  from airflow import models
  from airflow.api.common.mark_tasks import _create_dagruns, _DagRunInfo, set_dag_run_state_to_failed, set_dag_run_state_to_queued, set_dag_run_state_to_running, set_dag_run_state_to_success, set_state
  from airflow.models import DagRun
  from airflow.utils import timezone
  from airflow.utils.session import create_session, provide_session
  from airflow.utils.state import State
  from airflow.utils.types import DagRunType
  from tests.test_utils.db import clear_db_runs
  from tests.test_utils.mapping import expand_mapped_task
label_desc: Use image node_id 20 for calling the TestMarkDAGRun obj's underlying member method code with example usage: obj.test_set_state_with_multiple_dagruns(session) and returns: count