code: def test_trigger_from_expired_triggerer(
      session, create_task_instance
  ):
      """
      Checks that the triggerer will correctly claim a Trigger that is assigned to a
      triggerer that has an expired heartbeat.
      """
      # Use a trigger assigned to the expired triggerer
      trigger = TimeDeltaTrigger(datetime.timedelta(days=7))
      trigger_orm = Trigger.from_object(trigger)
      trigger_orm.id = 1
      trigger_orm.triggerer_id = 42
      session.add(trigger_orm)
      ti_orm = create_task_instance(
          task_id="ti_orm",
          execution_date=datetime.datetime.utcnow(),
          run_id="orm_run_id",
      )
      ti_orm.trigger_id = trigger_orm.id
      session.add(trigger_orm)
      # Use a TriggererJobRunner with an expired heartbeat
      triggerer_job_orm = Job(TriggererJobRunner.job_type)
      triggerer_job_orm.id = 42
      triggerer_job_orm.start_date = (
          timezone.utcnow() - datetime.timedelta(hours=1)
      )
      triggerer_job_orm.end_date = None
      triggerer_job_orm.latest_heartbeat = (
          timezone.utcnow() - datetime.timedelta(hours=1)
      )
      session.add(triggerer_job_orm)
      session.commit()
      # Make a TriggererJobRunner and have it retrieve DB tasks
      job = Job(TriggererJobRunner.job_type)
      job_runner = TriggererJobRunner(job)
      job_runner.load_triggers()
      # Make sure it turned up in TriggerRunner's queue
      assert [x for x, y in job_runner.trigger_runner.to_create] == [1]
  
imports: from __future__ import annotations
  import asyncio
  import datetime
  import importlib
  import time
  from threading import Thread
  from unittest.mock import MagicMock, patch
  import pendulum
  import pytest
  from airflow.config_templates import airflow_local_settings
  from airflow.jobs.job import Job
  from airflow.jobs.triggerer_job_runner import TriggererJobRunner, TriggerRunner, setup_queue_listener
  from airflow.logging_config import configure_logging
  from airflow.models import DagModel, DagRun, TaskInstance, Trigger
  from airflow.models.baseoperator import BaseOperator
  from airflow.models.dag import DAG
  from airflow.operators.empty import EmptyOperator
  from airflow.operators.python import PythonOperator
  from airflow.triggers.base import TriggerEvent
  from airflow.triggers.temporal import DateTimeTrigger, TimeDeltaTrigger
  from airflow.triggers.testing import FailureTrigger, SuccessTrigger
  from airflow.utils import timezone
  from airflow.utils.log.logging_mixin import RedirectStdHandler
  from airflow.utils.log.trigger_handler import LocalQueueHandler
  from airflow.utils.session import create_session
  from airflow.utils.state import State, TaskInstanceState
  from airflow.utils.types import DagRunType
  from tests.core.test_logging_config import reset_logging
  from tests.test_utils.db import clear_db_dags, clear_db_runs
label_desc: Use image node_id 11 for calling a global function with example usage: test_trigger_from_expired_triggerer(session, create_task_instance) without return types