code: def test_external_source_gpu():
      class ExternalSourcePipeline(Pipeline):
          def __init__(
              self, batch_size, num_threads, device_id, use_list
          ):
              super().__init__(batch_size, num_threads, device_id)
              self.input = ops.ExternalSource(device="gpu")
              self.crop = ops.Crop(
                  device="gpu",
                  crop_h=32,
                  crop_w=32,
                  crop_pos_x=0.2,
                  crop_pos_y=0.2,
              )
              self.use_list = use_list
  
          def define_graph(self):
              self.batch = self.input()
              output = self.crop(self.batch)
              return output
  
          def iter_setup(self):
              if use_list:
                  batch_data = [
                      cast_to(
                          random_array([100, 100, 3]) * 256,
                          datapy.uint8,
                      )
                      for _ in range(self.batch_size)
                  ]
              else:
                  batch_data = cast_to(
                      random_array([self.batch_size, 100, 100, 3])
                      * 256,
                      datapy.uint8,
                  )
              self.feed_input(self.batch, batch_data, layout="HWC")
  
      for batch_size in [1, 10]:
          for use_list in (True, False):
              pipe = ExternalSourcePipeline(batch_size, 3, 0, use_list)
              pipe.build()
              pipe.run()
  
imports: import functools
  import numpy
  import nvidia.dali.fn
  import nvidia.dali.ops
  from nvidia.dali import Pipeline
  from nose_utils import assert_raises
  from test_utils import check_output
label_desc: Use image node_id 31 for calling a global function with example usage: test_external_source_gpu() and returns: output