code: def test_processes_terminate(self):
      def acquire_lock(lock):
          lock.acquire()
  
      mp_context = self.get_context()
      if mp_context.get_start_method(allow_none=False) == "fork":
          # fork pre-spawns, not on demand.
          expected_num_processes = self.worker_count
      else:
          expected_num_processes = 3
  
      sem = mp_context.Semaphore(0)
      for _ in range(3):
          self.executor.submit(acquire_lock, sem)
      self.assertEqual(
          len(self.executor._processes), expected_num_processes
      )
      for _ in range(3):
          sem.release()
      processes = self.executor._processes
      self.executor.shutdown()
  
      for p in processes.values():
          p.join()
  
imports: import signal
  import sys
  import threading
  import time
  import unittest
  from concurrent import futures
  from test import support
  from test.support.script_helper import assert_python_ok
  from .util import BaseTestCase, ThreadPoolMixin, ProcessPoolForkMixin, ProcessPoolForkserverMixin, ProcessPoolSpawnMixin, create_executor_tests, setup_module
label_desc: Use image node_id 1 for calling the ProcessPoolShutdownTest obj's underlying member method code with example usage: obj.test_processes_terminate() without return types