code: def test_mark_mapped_task_instance_state(self, session):
      # set mapped task instance to success
      mapped = self.dag4.get_task("consumer")
      tasks = [(mapped, 0), (mapped, 1)]
      dr = DagRun.find(
          dag_id=self.dag4.dag_id,
          execution_date=self.execution_dates[0],
          session=session,
      )[0]
      expand_mapped_task(
          mapped, dr.run_id, "make_arg_lists", length=3, session=session
      )
      snapshot = TestMarkTasks.snapshot_state(
          self.dag4, self.execution_dates
      )
      altered = set_state(
          tasks=tasks,
          run_id=dr.run_id,
          upstream=True,
          downstream=False,
          future=False,
          past=False,
          state=State.SUCCESS,
          commit=True,
          session=session,
      )
      assert len(altered) == 3
      self.verify_state(
          self.dag4,
          ["consumer", "make_arg_lists"],
          [self.execution_dates[0]],
          State.SUCCESS,
          snapshot,
          map_task_pairs=[
              (task.task_id, map_index) for (task, map_index) in tasks
          ]
          + [("make_arg_lists", -1)],
          session=session,
      )
  
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 12 for calling the TestMarkTasks obj's underlying member method code with example usage: obj.test_mark_mapped_task_instance_state(session) without return types