code: def test_parallel_logging_works_with_streamed_steps(
      pipeline_class: BaseQueuedPipeline,
      use_processes: bool,
      use_savers: bool,
  ):
      # TODO: parametrize using SequentialQueuedPipeline as well instead of ParallelQueuedFeatureUnion
      # Given
      minibatch_size = 50
      n_workers = 2
      names = [
          f"{i}-{pipeline_class.__name__}-{use_processes}-{use_savers}"
          for i in range(3)
      ]
      p = pipeline_class(
          [
              TransformOnlyCounterLoggingStep().set_name(names[0]),
              TransformOnlyCounterLoggingStep().set_name(names[1]),
              TransformOnlyCounterLoggingStep().set_name(names[2]),
          ],
          n_workers_per_step=n_workers,
          batch_size=minibatch_size,
          use_processes=use_processes,
          use_savers=use_savers,
      )
  
      # When
      p.transform(GIVEN_INPUTS)
  
      # Then
      log_history = list(CX().logger)
      n_calls = int(len(GIVEN_INPUTS) / minibatch_size) * 3
      # NOTE: logs are duplicated by pytest-xdist's handlers or something, and polluted by other parallel tests. See: https://github.com/pytest-dev/pytest/issues/10062
      shortened_log_history = [
          e[: e.index("#") + 1]
          for e in set(log_history)
          if "logging call" in e
      ]
      assert len(shortened_log_history) >= n_calls, str(
          shortened_log_history
      )
      for nm in names:
          log_line = f"{nm} - transform call - logging call #"
          assert log_line in shortened_log_history, str(
              shortened_log_history
          )
      pass
  
imports: import copy
  import random
  import time
  from typing import List, Type
  import numpy
  import pytest
  from neuraxle.base import BaseStep
  from neuraxle.base import ExecutionContext
  from neuraxle.base import NonFittableMixin
  from neuraxle.data_container import ARG_X_INPUTTED, ARG_Y_PREDICTD, DACT, PredsDACT, StripAbsentValues, TrainDACT
  from neuraxle.distributed.streaming import BaseQueuedPipeline, ParallelQueuedFeatureUnion, ParallelWorkersWrapper, QueuedMinibatchTask, SequentialQueuedPipeline, WorkersJoiner
  from neuraxle.metaopt.context import AutoMLContext
  from neuraxle.pipeline import Pipeline
  from neuraxle.steps.loop import ForEach
  from neuraxle.steps.misc import FitTransformCallbackStep, Sleep, TransformOnlyCounterLoggingStep
  from neuraxle.steps.numpy import MultiplyByN
label_desc: Use image node_id 22 for calling a global function with example usage: test_parallel_logging_works_with_streamed_steps(pipeline_class, use_processes, use_savers) without return types