code: def testSignalHandling(self):
      infra_validator = executor.Executor(self._context)
      ready_to_kill_event = threading.Event()
  
      def send_sigterm(pid):
          ready_to_kill_event.wait()
          os.kill(pid, signal.SIGTERM)
  
      def validate_side_effect(*args, **kwargs):
          del args, kwargs  # Unused.
          ready_to_kill_event.set()
          while True:  # Wait until killed.
              pass
  
      with mock.patch.object(
          infra_validator, "_ValidateOnce"
      ) as mock_validate:
          mock_validate.side_effect = validate_side_effect
          killer = threading.Thread(
              target=send_sigterm, args=(os.getpid(),)
          )
          killer.start()
  
          # SIGTERM should raise GracefulShutdown error.
          with self.assertRaises(error_types.GracefulShutdown):
              infra_validator.Do(
                  self._input_dict,
                  self._output_dict,
                  self._exec_properties,
              )
  
imports: import os
  import signal
  import threading
  from typing import Any, Dict
  from unittest import mock
  import tensorflow
  from tfx.components.infra_validator import error_types
  from tfx.components.infra_validator import executor
  from tfx.components.infra_validator import request_builder
  from tfx.components.infra_validator import serving_bins
  from tfx.dsl.io import fileio
  from tfx.proto import infra_validator_pb2
  from tfx.types import artifact_utils
  from tfx.types import standard_artifacts
  from tfx.types import standard_component_specs
  from tfx.utils import path_utils
  from tfx.utils import proto_utils
  from google.protobuf import json_format
label_desc: Use image node_id 13 for calling the ExecutorTest obj's underlying member method code with example usage: obj.testSignalHandling() without return types