code: def test_submit_event(session, create_task_instance):
      """
      Tests that events submitted to a trigger re-wake their dependent
      task instances.
      """
      # Make a trigger
      trigger = Trigger(
          classpath="airflow.triggers.testing.SuccessTrigger", kwargs={}
      )
      trigger.id = 1
      session.add(trigger)
      session.commit()
      # Make a TaskInstance that's deferred and waiting on it
      task_instance = create_task_instance(
          session=session,
          execution_date=timezone.utcnow(),
          state=State.DEFERRED,
      )
      task_instance.trigger_id = trigger.id
      task_instance.next_kwargs = {"cheesecake": True}
      session.commit()
      # Call submit_event
      Trigger.submit_event(
          trigger.id, TriggerEvent(42), session=session
      )
      # commit changes made by submit event and expire all cache to read from db.
      session.flush()
      session.expunge_all()
      # Check that the task instance is now scheduled
      updated_task_instance = session.query(TaskInstance).one()
      assert updated_task_instance.state == State.SCHEDULED
      assert updated_task_instance.next_kwargs == {
          "event": 42,
          "cheesecake": True,
      }
  
imports: from __future__ import annotations
  import datetime
  from typing import Any, AsyncIterator
  import pytest
  import pytz
  from cryptography.fernet import Fernet
  from airflow.jobs.job import Job
  from airflow.jobs.triggerer_job_runner import TriggererJobRunner, TriggerRunner
  from airflow.models import TaskInstance, Trigger
  from airflow.operators.empty import EmptyOperator
  from airflow.triggers.base import BaseTrigger, TriggerEvent
  from airflow.utils import timezone
  from airflow.utils.session import create_session
  from airflow.utils.state import State
  from tests.test_utils.config import conf_vars
label_desc: Use image node_id 4 for calling a global function with example usage: test_submit_event(session, create_task_instance) without return types