code: def test_utils_suppress_similar():
      tu = TestUtils()
  
      # Check suppression boundary for straight chain of similar
      # messages.
  
      # yield test results immediately to make test run fast
      sleep_dur = 0.0
  
      def n_foo(number):
          for i in range(number):
              yield dict(
                  action="foo", status="ok", path="path{}".format(i)
              )
              sleep(sleep_dur)
  
      with _swallow_outputs() as cmo:
          cmo.isatty = lambda: True
          list(tu(9, result_fn=n_foo, result_renderer="default"))
          assert_in("path8", cmo.out)
          assert_not_in("suppressed", cmo.out)
  
      with _swallow_outputs() as cmo:
          list(tu(10, result_fn=n_foo, result_renderer="default"))
          assert_in("path9", cmo.out)
          assert_not_in("suppressed", cmo.out)
  
      with _swallow_outputs() as cmo:
          list(tu(11, result_fn=n_foo, result_renderer="default"))
          assert_not_in("path10", cmo.out)
          assert_re_in(r"[^-0-9]1 .* suppressed", cmo.out, match=False)
  
      with _swallow_outputs() as cmo:
          # for this one test yield results slightly slower than 2Hz
          # such that we can see each individual suppression message
          # and no get caught by the rate limiter
          sleep_dur = 0.51
          list(tu(13, result_fn=n_foo, result_renderer="default"))
          assert_not_in("path10", cmo.out)
          # We see an update for each result.
          assert_re_in(r"1 .* suppressed", cmo.out, match=False)
          assert_re_in(r"2 .* suppressed", cmo.out, match=False)
          assert_re_in(r"3 .* suppressed", cmo.out, match=False)
  
      # make tests run fast again
      sleep_dur = 0.0
  
      with _swallow_outputs(isatty=False) as cmo:
          list(tu(11, result_fn=n_foo, result_renderer="default"))
          assert_in("path10", cmo.out)
  
      # Check a chain of similar messages, split in half by a distinct one.
  
      def n_foo_split_by_a_bar(number):
          half = number // 2 - 1
          for i in range(number):
              yield dict(
                  action="foo", status="ok", path="path{}".format(i)
              )
              if i == half:
                  yield dict(action="bar", status="ok", path="path")
  
      with _swallow_outputs() as cmo:
          list(
              tu(
                  20,
                  result_fn=n_foo_split_by_a_bar,
                  result_renderer="default",
              )
          )
          assert_in("path10", cmo.out)
          assert_in("path19", cmo.out)
          assert_not_in("suppressed", cmo.out)
  
      with _swallow_outputs() as cmo:
          list(
              tu(
                  21,
                  result_fn=n_foo_split_by_a_bar,
                  result_renderer="default",
              )
          )
          assert_in("path10", cmo.out)
          assert_not_in("path20", cmo.out)
          assert_re_in("[^-0-9]1 .* suppressed", cmo.out, match=False)
  
imports: import logging
  import re
  from contextlib import contextmanager
  from os.path import exists
  from os.path import join
  from time import sleep
  from datalad.distribution.dataset import Dataset, EnsureDataset, datasetmethod
  from datalad.interface.base import build_doc, eval_results
  from datalad.support.constraints import EnsureKeyChoice, EnsureNone, EnsureStr
  from datalad.support.param import Parameter
  from datalad.tests.utils_pytest import assert_dict_equal, assert_equal, assert_in, assert_not_equal, assert_not_in, assert_raises, assert_re_in, assert_repo_status, assert_result_count, assert_true, ok_, slow, with_tempfile, with_tree
  from datalad.utils import swallow_logs, swallow_outputs
  from ..base import Interface
  from ..results import get_status_dict
  from ..utils import discover_dataset_trace_to_targets, handle_dirty_dataset
label_desc: Use image node_id 10 for calling a global function with example usage: test_utils_suppress_similar() without return types