code: def test_localtaskjob_heartbeat_with_run_as_user(
      self, psutil_mock, _, dag_maker
  ):
      session = settings.Session()
      with dag_maker("test_localtaskjob_heartbeat"):
          op1 = EmptyOperator(task_id="op1", run_as_user="myuser")
      dr = dag_maker.create_dagrun()
      ti = dr.get_task_instance(task_id=op1.task_id, session=session)
      ti.state = State.RUNNING
      ti.pid = 2
      ti.hostname = get_hostname()
      session.commit()
  
      job1 = Job(dag_id=ti.dag_id, executor=SequentialExecutor())
      job_runner = LocalTaskJobRunner(
          job=job1, task_instance=ti, ignore_ti_state=True
      )
      ti.task = op1
      ti.refresh_from_task(op1)
      job1.task_runner = StandardTaskRunner(job_runner)
      job1.task_runner.process = mock.Mock()
      job1.task_runner.process.pid = 2
      job_runner.task_runner = job1.task_runner
      # Here, ti.pid is 2, the parent process of ti.pid is a mock(different).
      # And task_runner process is 2. Should fail
      with pytest.raises(
          AirflowException, match="PID of job runner does not match"
      ):
          job_runner.heartbeat_callback()
  
      job1.task_runner.process.pid = 1
      # We make the parent process of ti.pid to equal the task_runner process id
      psutil_mock.Process.return_value.ppid.return_value = 1
      ti.state = State.RUNNING
      ti.pid = 2
      # The task_runner process id is 1, same as the parent process of ti.pid
      # as seen above
      assert ti.run_as_user
      session.merge(ti)
      session.commit()
      job_runner.heartbeat_callback(session=None)
  
      # Here the task_runner process id is changed to 2
      # while parent process of ti.pid is kept at 1, which is different
      job1.task_runner.process.pid = 2
      with pytest.raises(
          AirflowException, match="PID of job runner does not match"
      ):
          job_runner.heartbeat_callback()
  
      # Here we set the ti.pid to None and test that no error is
      # raised
      ti.pid = None
      session.merge(ti)
      session.commit()
      assert ti.run_as_user
      assert job1.task_runner.run_as_user == ti.run_as_user
      assert ti.pid != job1.task_runner.process.pid
      job_runner.heartbeat_callback()
  
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 5 for calling the TestLocalTaskJob obj's underlying member method code with example usage: obj.test_localtaskjob_heartbeat_with_run_as_user(psutil_mock, _, dag_maker) without return types