docstr: Test if DagRun is successful, and if Success callbacks is defined, it is sent to DagFileProcessor.
docstr_tok: ["Test","if","DagRun","is","successful",",","and","if","Success","callbacks","is","defined",",","it","is","sent","to","DagFileProcessor","."]
code: def test_base_job_respects_plugin_lifecycle(self, dag_maker):
      """
      Test if DagRun is successful, and if Success callbacks is defined, it is sent to DagFileProcessor.
      """
      get_listener_manager().add_listener(lifecycle_listener)
  
      job = Job()
      job_runner = MockJobRunner(job=job, func=lambda: sys.exit(0))
      run_job(job=job, execute_callable=job_runner._execute)
  
      assert lifecycle_listener.started_component is job
      assert lifecycle_listener.stopped_component is job
  
imports: from __future__ import annotations
  import datetime
  import sys
  from unittest.mock import ANY, Mock, patch
  import pytest
  from sqlalchemy.exc import OperationalError
  from airflow.executors.sequential_executor import SequentialExecutor
  from airflow.jobs.job import Job, most_recent_job, perform_heartbeat, run_job
  from airflow.listeners.listener import get_listener_manager
  from airflow.utils import timezone
  from airflow.utils.session import create_session
  from airflow.utils.state import State
  from tests.listeners import lifecycle_listener
  from tests.test_utils.config import conf_vars
  from tests.utils.test_helpers import MockJobRunner, SchedulerJobRunner, TriggererJobRunner
label_desc: Use image node_id 4 for calling the TestJob obj's underlying member method code with example usage: obj.test_base_job_respects_plugin_lifecycle(dag_maker) without return types