code: def _test_partially_utilized_external_source_warning(
      usage_mask, source_type
  ):
      np_rng = np.random.default_rng(12345)
      max_batch_size = 8
      num_outputs = len(usage_mask)
  
      def rand_batch():
          batch_size = np.int32(np_rng.uniform(1, max_batch_size + 1))
          return np.float32(
              np_rng.uniform(-1, 1, shape=(batch_size, 10, 100, 3))
          )
  
      def rand_tuple(rand_elem):
          return tuple(rand_elem() for _ in range(num_outputs))
  
      def sample_cb_source(sample_info):
          def rand_sample():
              return np.float32(
                  np_rng.uniform(-1, 1, shape=(10, 100, 3))
              )
  
          return rand_tuple(rand_sample)
  
      def batch_cb_source():
          return rand_tuple(rand_batch)
  
      def gen_fun_source():
          while True:
              yield rand_tuple(rand_batch)
  
      class IteratorSource:
          def __iter__(self):
              return self
  
          def __next__(self):
              return rand_tuple(rand_batch)
  
      sources = {
          "sample_cb_source": sample_cb_source,
          "batch_cb_source": batch_cb_source,
          "gen_fun_source": gen_fun_source,
          "generator": gen_fun_source(),
          "IteratorSource": IteratorSource(),
      }
  
      @pipeline_def
      def pipeline():
          outputs = fn.external_source(
              source=sources[source_type],
              num_outputs=num_outputs,
              batch=source_type != "sample_cb_source",
          )
          assert len(outputs) == num_outputs
          utilized_outputs = (
              out
              for out, is_used in zip(outputs, usage_mask)
              if is_used
          )
          return tuple(
              fn.gaussian_blur(out, window_size=3)
              for out in utilized_outputs
          )
  
      pipe = pipeline(
          batch_size=max_batch_size, num_threads=4, device_id=0
      )
      unused_output_idxs = [
          i for i, is_used in enumerate(usage_mask) if not is_used
      ]
      assert len(unused_output_idxs) > 0
      pruned_idx_str = ", ".join(str(idx) for idx in unused_output_idxs)
      if len(unused_output_idxs) == 1:
          pruned_str = f"output at the index {pruned_idx_str} is"
      else:
          pruned_str = f"outputs at the indices {pruned_idx_str} are"
      expected_error_msg = (
          f"The external source node '*{source_type}*' produces {num_outputs} outputs, "
          f"but the {pruned_str} not used."
      )
      with assert_warns(Warning, glob=expected_error_msg):
          pipe.build()
  
imports: import random
  import nvidia.dali.fn
  import nvidia.dali.ops
  import nvidia.dali.types
  import numpy
  from nvidia.dali import Pipeline, pipeline_def
  from test_utils import check_batch
  from nose_utils import raises, assert_warns, assert_raises
  from nvidia.dali.types import DALIDataType
  from numpy.random import default_rng
label_desc: Use image node_id 25 for calling a global function with example usage: _test_partially_utilized_external_source_warning(usage_mask, source_type) and returns: np, tuple, rand_tuple, rand_tuple, tuple, np, self, rand_tuple