code: def runtest_refleak(
      test_name, test_func, hunt_refleak: HuntRefleak, quiet: bool
  ):
      """Run a test multiple times, looking for reference leaks.
  
      Returns:
          False if the test didn't leak references; True if we detected refleaks.
      """
      # This code is hackish and inelegant, but it seems to do the job.
      import copyreg
      import collections.abc
  
      if not hasattr(sys, "gettotalrefcount"):
          raise Exception(
              "Tracking reference leaks requires a debug build "
              "of Python"
          )
  
      # Avoid false positives due to various caches
      # filling slowly with random data:
      warm_caches()
  
      # Save current values for dash_R_cleanup() to restore.
      fs = warnings.filters[:]
      ps = copyreg.dispatch_table.copy()
      pic = sys.path_importer_cache.copy()
      zdc: dict[str, Any] | None
      try:
          import zipimport
      except ImportError:
          zdc = None  # Run unmodified on platforms without zipimport support
      else:
          # private attribute that mypy doesn't know about:
          zdc = zipimport._zip_directory_cache.copy()  # type: ignore[attr-defined]
      abcs = {}
      for abc in [
          getattr(collections.abc, a) for a in collections.abc.__all__
      ]:
          if not isabstract(abc):
              continue
          for obj in abc.__subclasses__() + [abc]:
              abcs[obj] = _get_dump(obj)[0]
  
      # bpo-31217: Integer pool to get a single integer object for the same
      # value. The pool is used to prevent false alarm when checking for memory
      # block leaks. Fill the pool with values in -1000..1000 which are the most
      # common (reference, memory block, file descriptor) differences.
      int_pool = {value: value for value in range(-1000, 1000)}
  
      def get_pooled_int(value):
          return int_pool.setdefault(value, value)
  
      warmups = hunt_refleak.warmups
      runs = hunt_refleak.runs
      filename = hunt_refleak.filename
      repcount = warmups + runs
  
      # Pre-allocate to ensure that the loop doesn't allocate anything new
      rep_range = list(range(repcount))
      rc_deltas = [0] * repcount
      alloc_deltas = [0] * repcount
      fd_deltas = [0] * repcount
      getallocatedblocks = sys.getallocatedblocks
      gettotalrefcount = sys.gettotalrefcount
      getunicodeinternedsize = sys.getunicodeinternedsize
      fd_count = os_helper.fd_count
      # initialize variables to make pyflakes quiet
      rc_before = alloc_before = fd_before = interned_before = 0
  
      if not quiet:
          print("beginning", repcount, "repetitions", file=sys.stderr)
          print(
              ("1234567890" * (repcount // 10 + 1))[:repcount],
              file=sys.stderr,
              flush=True,
          )
  
      results = None
      dash_R_cleanup(fs, ps, pic, zdc, abcs)
      support.gc_collect()
  
      for i in rep_range:
          results = test_func()
  
          dash_R_cleanup(fs, ps, pic, zdc, abcs)
          support.gc_collect()
  
          # Read memory statistics immediately after the garbage collection.
          # Also, readjust the reference counts and alloc blocks by ignoring
          # any strings that might have been interned during test_func. These
          # strings will be deallocated at runtime shutdown
          interned_after = getunicodeinternedsize()
          alloc_after = getallocatedblocks() - interned_after
          rc_after = gettotalrefcount() - interned_after * 2
          fd_after = fd_count()
  
          if not quiet:
              print(".", end="", file=sys.stderr, flush=True)
  
          rc_deltas[i] = get_pooled_int(rc_after - rc_before)
          alloc_deltas[i] = get_pooled_int(alloc_after - alloc_before)
          fd_deltas[i] = get_pooled_int(fd_after - fd_before)
  
          alloc_before = alloc_after
          rc_before = rc_after
          fd_before = fd_after
          interned_before = interned_after
  
      if not quiet:
          print(file=sys.stderr)
  
      # These checkers return False on success, True on failure
      def check_rc_deltas(deltas):
          # Checker for reference counters and memory blocks.
          #
          # bpo-30776: Try to ignore false positives:
          #
          #   [3, 0, 0]
          #   [0, 1, 0]
          #   [8, -8, 1]
          #
          # Expected leaks:
          #
          #   [5, 5, 6]
          #   [10, 1, 1]
          return all(delta >= 1 for delta in deltas)
  
      def check_fd_deltas(deltas):
          return any(deltas)
  
      failed = False
      for deltas, item_name, checker in [
          (rc_deltas, "references", check_rc_deltas),
          (alloc_deltas, "memory blocks", check_rc_deltas),
          (fd_deltas, "file descriptors", check_fd_deltas),
      ]:
          # ignore warmup runs
          deltas = deltas[warmups:]
          if checker(deltas):
              msg = "%s leaked %s %s, sum=%s" % (
                  test_name,
                  deltas,
                  item_name,
                  sum(deltas),
              )
              print(msg, file=sys.stderr, flush=True)
              with open(filename, "a", encoding="utf-8") as refrep:
                  print(msg, file=refrep)
                  refrep.flush()
              failed = True
      return (failed, results)
  
file_path: cpython/Lib/test/libregrtest/refleak.py
imports: import sys
  import warnings
  from inspect import isabstract
  from typing import Any
  from test import support
  from test.support import os_helper
  from .runtests import HuntRefleak
  from .utils import clear_caches
label_desc: Use image node_id 1 for calling a global function with example usage: runtest_refleak(test_name, test_func, hunt_refleak, quiet) and returns: failed,  results, int_pool, all, any