docstr: Test that ensures that when a task is killed with sigkill or sigterm
  on_failure_callback does not get executed by LocalTaskJob.
  
  Callbacks should not be executed by LocalTaskJob.  If the task killed via sigkill,
  it will be reaped as zombie, then the callback is executed
docstr_tok: ["Test","that","ensures","that","when","a","task","is","killed","with","sigkill","or","sigterm","on_failure_callback","does","not","get","executed","by","LocalTaskJob",".","Callbacks","should","not","be","executed","by","LocalTaskJob",".","If","the","task","killed","via","sigkill",",","it","will","be","reaped","as","zombie",",","then","the","callback","is","executed"]
code: def test_process_os_signal_calls_on_failure_callback(
      self, monkeypatch, tmp_path, get_test_dag, signal_type
  ):
      """
      Test that ensures that when a task is killed with sigkill or sigterm
      on_failure_callback does not get executed by LocalTaskJob.
  
      Callbacks should not be executed by LocalTaskJob.  If the task killed via sigkill,
      it will be reaped as zombie, then the callback is executed
      """
      callback_file = tmp_path.joinpath("callback.txt")
      # callback_file will be created by the task: bash_sleep
      monkeypatch.setenv("AIRFLOW_CALLBACK_FILE", str(callback_file))
      dag = get_test_dag("test_on_failure_callback")
      with create_session() as session:
          dag.create_dagrun(
              state=State.RUNNING,
              execution_date=DEFAULT_DATE,
              run_type=DagRunType.SCHEDULED,
              session=session,
          )
      task = dag.get_task(task_id="bash_sleep")
      ti = TaskInstance(task=task, execution_date=DEFAULT_DATE)
      ti.refresh_from_db()
  
      signal_sent_status = {"sent": False}
  
      def get_ti_current_pid(ti) -> str:
          with create_session() as session:
              pid = (
                  session.query(TaskInstance.pid)
                  .filter(
                      TaskInstance.dag_id == ti.dag_id,
                      TaskInstance.task_id == ti.task_id,
                      TaskInstance.run_id == ti.run_id,
                  )
                  .one_or_none()
              )
              return pid[0]
  
      def send_signal(ti, signal_sent, sig):
          while True:
              task_pid = get_ti_current_pid(
                  ti
              )  # get pid from the db, which is the pid of airflow run --raw
              if (
                  task_pid
                  and ti.current_state() == State.RUNNING
                  and os.path.isfile(callback_file)
              ):  # ensure task is running before sending sig
                  signal_sent["sent"] = True
                  os.kill(task_pid, sig)
                  break
              time.sleep(1)
  
      thread = threading.Thread(
          name="signaler",
          target=send_signal,
          args=(ti, signal_sent_status, signal_type),
      )
      thread.daemon = True
      thread.start()
  
      job1 = Job(dag_id=ti.dag_id, executor=SequentialExecutor())
      job_runner = LocalTaskJobRunner(
          job=job1, task_instance=ti, ignore_ti_state=True
      )
      run_job(job=job1, execute_callable=job_runner._execute)
  
      ti.refresh_from_db()
  
      assert signal_sent_status["sent"]
  
      if signal_type == signal.SIGTERM:
          assert ti.state == State.FAILED
          with open(callback_file) as f:
              lines = f.readlines()
  
          assert len(lines) == 1
          assert lines[0].startswith(ti.key.primary)
  
          m = re.match(r"^.+pid: (\d+)$", lines[0])
          assert m, "pid expected in output."
          pid = int(m.group(1))
          assert (
              os.getpid() != pid
          )  # ensures callback is NOT run by LocalTaskJob
          assert (
              ti.pid == pid
          )  # ensures callback is run by airflow run --raw (TaskInstance#_run_raw_task)
      elif signal_type == signal.SIGKILL:
          assert (
              ti.state == State.RUNNING
          )  # task exits with running state, will be reaped as zombie by scheduler
          with open(callback_file) as f:
              lines = f.readlines()
          assert len(lines) == 0
  
imports: from __future__ import annotations
  import datetime
  import logging
  import multiprocessing
  import os
  import re
  import signal
  import threading
  import time
  import uuid
  import warnings
  from unittest import mock
  from unittest.mock import patch
  import psutil
  import pytest
  from airflow import settings
  from airflow.exceptions import AirflowException
  from airflow.executors.sequential_executor import SequentialExecutor
  from airflow.jobs.job import Job, run_job
  from airflow.jobs.local_task_job_runner import SIGSEGV_MESSAGE, LocalTaskJobRunner
  from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
  from airflow.models.dag import DAG
  from airflow.models.dagbag import DagBag
  from airflow.models.serialized_dag import SerializedDagModel
  from airflow.models.taskinstance import TaskInstance
  from airflow.operators.empty import EmptyOperator
  from airflow.operators.python import PythonOperator
  from airflow.task.task_runner.standard_task_runner import StandardTaskRunner
  from airflow.utils import timezone
  from airflow.utils.net import get_hostname
  from airflow.utils.session import create_session
  from airflow.utils.state import State
  from airflow.utils.timeout import timeout
  from airflow.utils.types import DagRunType
  from tests.test_utils import db
  from tests.test_utils.asserts import assert_queries_count
  from tests.test_utils.config import conf_vars
  from tests.test_utils.mock_executor import MockExecutor
label_desc: Use image node_id 16 for calling the TestLocalTaskJob obj's underlying member method code with example usage: obj.test_process_os_signal_calls_on_failure_callback(monkeypatch, tmp_path, get_test_dag, signal_type) and returns: pid