code: def test_batch_info_reports(self):
      batch_collector = self.BatchInfoCollector()
      self.ssc.addStreamingListener(batch_collector)
      input = [[1], [2], [3], [4]]
  
      def func(dstream):
          return dstream.map(int)
  
      expected = [[1], [2], [3], [4]]
      self._test_func(input, func, expected)
  
      batchInfosSubmitted = batch_collector.batchInfosSubmitted
      batchInfosStarted = batch_collector.batchInfosStarted
      batchInfosCompleted = batch_collector.batchInfosCompleted
      streamingStartedTime = batch_collector.streamingStartedTime
  
      self.wait_for(batchInfosCompleted, 4)
  
      self.assertEqual(len(streamingStartedTime), 1)
  
      self.assertGreaterEqual(len(batchInfosSubmitted), 4)
      for info in batchInfosSubmitted:
          self.assertGreaterEqual(info.batchTime().milliseconds(), 0)
          self.assertGreaterEqual(info.submissionTime(), 0)
  
          for streamId in info.streamIdToInputInfo():
              streamInputInfo = info.streamIdToInputInfo()[streamId]
              self.assertGreaterEqual(
                  streamInputInfo.inputStreamId(), 0
              )
              self.assertGreaterEqual(streamInputInfo.numRecords, 0)
              for key in streamInputInfo.metadata():
                  self.assertIsNotNone(streamInputInfo.metadata()[key])
              self.assertIsNotNone(
                  streamInputInfo.metadataDescription()
              )
  
          for outputOpId in info.outputOperationInfos():
              outputInfo = info.outputOperationInfos()[outputOpId]
              self.assertGreaterEqual(
                  outputInfo.batchTime().milliseconds(), 0
              )
              self.assertGreaterEqual(outputInfo.id(), 0)
              self.assertIsNotNone(outputInfo.name())
              self.assertIsNotNone(outputInfo.description())
              self.assertGreaterEqual(outputInfo.startTime(), -1)
              self.assertGreaterEqual(outputInfo.endTime(), -1)
              self.assertIsNone(outputInfo.failureReason())
  
          self.assertEqual(info.schedulingDelay(), -1)
          self.assertEqual(info.processingDelay(), -1)
          self.assertEqual(info.totalDelay(), -1)
          self.assertEqual(info.numRecords(), 0)
  
      self.assertGreaterEqual(len(batchInfosStarted), 4)
      for info in batchInfosStarted:
          self.assertGreaterEqual(info.batchTime().milliseconds(), 0)
          self.assertGreaterEqual(info.submissionTime(), 0)
  
          for streamId in info.streamIdToInputInfo():
              streamInputInfo = info.streamIdToInputInfo()[streamId]
              self.assertGreaterEqual(
                  streamInputInfo.inputStreamId(), 0
              )
              self.assertGreaterEqual(streamInputInfo.numRecords, 0)
              for key in streamInputInfo.metadata():
                  self.assertIsNotNone(streamInputInfo.metadata()[key])
              self.assertIsNotNone(
                  streamInputInfo.metadataDescription()
              )
  
          for outputOpId in info.outputOperationInfos():
              outputInfo = info.outputOperationInfos()[outputOpId]
              self.assertGreaterEqual(
                  outputInfo.batchTime().milliseconds(), 0
              )
              self.assertGreaterEqual(outputInfo.id(), 0)
              self.assertIsNotNone(outputInfo.name())
              self.assertIsNotNone(outputInfo.description())
              self.assertGreaterEqual(outputInfo.startTime(), -1)
              self.assertGreaterEqual(outputInfo.endTime(), -1)
              self.assertIsNone(outputInfo.failureReason())
  
          self.assertGreaterEqual(info.schedulingDelay(), 0)
          self.assertEqual(info.processingDelay(), -1)
          self.assertEqual(info.totalDelay(), -1)
          self.assertEqual(info.numRecords(), 0)
  
      self.assertGreaterEqual(len(batchInfosCompleted), 4)
      for info in batchInfosCompleted:
          self.assertGreaterEqual(info.batchTime().milliseconds(), 0)
          self.assertGreaterEqual(info.submissionTime(), 0)
  
          for streamId in info.streamIdToInputInfo():
              streamInputInfo = info.streamIdToInputInfo()[streamId]
              self.assertGreaterEqual(
                  streamInputInfo.inputStreamId(), 0
              )
              self.assertGreaterEqual(streamInputInfo.numRecords, 0)
              for key in streamInputInfo.metadata():
                  self.assertIsNotNone(streamInputInfo.metadata()[key])
              self.assertIsNotNone(
                  streamInputInfo.metadataDescription()
              )
  
          for outputOpId in info.outputOperationInfos():
              outputInfo = info.outputOperationInfos()[outputOpId]
              self.assertGreaterEqual(
                  outputInfo.batchTime().milliseconds(), 0
              )
              self.assertGreaterEqual(outputInfo.id(), 0)
              self.assertIsNotNone(outputInfo.name())
              self.assertIsNotNone(outputInfo.description())
              self.assertGreaterEqual(outputInfo.startTime(), 0)
              self.assertGreaterEqual(outputInfo.endTime(), 0)
              self.assertIsNone(outputInfo.failureReason())
  
          self.assertGreaterEqual(info.schedulingDelay(), 0)
          self.assertGreaterEqual(info.processingDelay(), 0)
          self.assertGreaterEqual(info.totalDelay(), 0)
          self.assertEqual(info.numRecords(), 0)
  
imports: from pyspark.streaming import StreamingListener
  from pyspark.testing.streamingutils import PySparkStreamingTestCase
label_desc: Use image node_id 1 for calling the StreamingListenerTests obj's underlying member method code with example usage: obj.test_batch_info_reports() and returns: dstream