code: def test_streaming_query_progress_fromJson(self):
      progress_json = """
              {
                "id" : "00000000-0000-0001-0000-000000000001",
                "runId" : "00000000-0000-0001-0000-000000000002",
                "name" : "test",
                "timestamp" : "2016-12-05T20:54:20.827Z",
                "batchId" : 2,
                "numInputRows" : 678,
                "inputRowsPerSecond" : 10.0,
                "processedRowsPerSecond" : 5.4,
                "batchDuration": 5,
                "durationMs" : {
                  "getBatch" : 0
                },
                "eventTime" : {
                  "min" : "2016-12-05T20:54:20.827Z",
                  "avg" : "2016-12-05T20:54:20.827Z",
                  "watermark" : "2016-12-05T20:54:20.827Z",
                  "max" : "2016-12-05T20:54:20.827Z"
                },
                "stateOperators" : [ {
                  "operatorName" : "op1",
                  "numRowsTotal" : 0,
                  "numRowsUpdated" : 1,
                  "allUpdatesTimeMs" : 1,
                  "numRowsRemoved" : 2,
                  "allRemovalsTimeMs" : 34,
                  "commitTimeMs" : 23,
                  "memoryUsedBytes" : 3,
                  "numRowsDroppedByWatermark" : 0,
                  "numShufflePartitions" : 2,
                  "numStateStoreInstances" : 2,
                  "customMetrics" : {
                    "loadedMapCacheHitCount" : 1,
                    "loadedMapCacheMissCount" : 0,
                    "stateOnCurrentVersionSizeBytes" : 2
                  }
                } ],
                "sources" : [ {
                  "description" : "source",
                  "startOffset" : 123,
                  "endOffset" : 456,
                  "latestOffset" : 789,
                  "numInputRows" : 678,
                  "inputRowsPerSecond" : 10.0,
                  "processedRowsPerSecond" : 5.4,
                  "metrics": {}
                } ],
                "sink" : {
                  "description" : "sink",
                  "numOutputRows" : -1,
                  "metrics": {}
                },
                "observedMetrics" : {
                  "event1" : {
                    "c1" : 1,
                    "c2" : 3.0
                  },
                  "event2" : {
                    "rc" : 1,
                    "min_q" : "hello",
                    "max_q" : "world"
                  }
                }
              }
          """
      progress = StreamingQueryProgress.fromJson(
          json.loads(progress_json)
      )
  
      self.check_streaming_query_progress(progress)
  
      # checks for progress
      self.assertEqual(
          progress.id, uuid.UUID("00000000-0000-0001-0000-000000000001")
      )
      self.assertEqual(
          progress.runId,
          uuid.UUID("00000000-0000-0001-0000-000000000002"),
      )
      self.assertEqual(progress.name, "test")
      self.assertEqual(progress.timestamp, "2016-12-05T20:54:20.827Z")
      self.assertEqual(progress.batchId, 2)
      self.assertEqual(progress.numInputRows, 678)
      self.assertEqual(progress.inputRowsPerSecond, 10.0)
      self.assertEqual(progress.batchDuration, 5)
      self.assertEqual(progress.durationMs, {"getBatch": 0})
      self.assertEqual(
          progress.eventTime,
          {
              "min": "2016-12-05T20:54:20.827Z",
              "avg": "2016-12-05T20:54:20.827Z",
              "watermark": "2016-12-05T20:54:20.827Z",
              "max": "2016-12-05T20:54:20.827Z",
          },
      )
      self.assertEqual(
          progress.observedMetrics,
          {
              "event1": Row("c1", "c2")(1, 3.0),
              "event2": Row("rc", "min_q", "max_q")(
                  1, "hello", "world"
              ),
          },
      )
  
      # Check stateOperators list
      self.assertEqual(len(progress.stateOperators), 1)
      state_operator = progress.stateOperators[0]
      self.assertTrue(isinstance(state_operator, StateOperatorProgress))
      self.assertEqual(state_operator.operatorName, "op1")
      self.assertEqual(state_operator.numRowsTotal, 0)
      self.assertEqual(state_operator.numRowsUpdated, 1)
      self.assertEqual(state_operator.allUpdatesTimeMs, 1)
      self.assertEqual(state_operator.numRowsRemoved, 2)
      self.assertEqual(state_operator.allRemovalsTimeMs, 34)
      self.assertEqual(state_operator.commitTimeMs, 23)
      self.assertEqual(state_operator.memoryUsedBytes, 3)
      self.assertEqual(state_operator.numRowsDroppedByWatermark, 0)
      self.assertEqual(state_operator.numShufflePartitions, 2)
      self.assertEqual(state_operator.numStateStoreInstances, 2)
      self.assertEqual(
          state_operator.customMetrics,
          {
              "loadedMapCacheHitCount": 1,
              "loadedMapCacheMissCount": 0,
              "stateOnCurrentVersionSizeBytes": 2,
          },
      )
  
      # Check sources list
      self.assertEqual(len(progress.sources), 1)
      source = progress.sources[0]
      self.assertTrue(isinstance(source, SourceProgress))
      self.assertEqual(source.description, "source")
      self.assertEqual(source.startOffset, "123")
      self.assertEqual(source.endOffset, "456")
      self.assertEqual(source.latestOffset, "789")
      self.assertEqual(source.numInputRows, 678)
      self.assertEqual(source.inputRowsPerSecond, 10.0)
      self.assertEqual(source.processedRowsPerSecond, 5.4)
      self.assertEqual(source.metrics, {})
  
      # Check sink
      sink = progress.sink
      self.assertTrue(isinstance(sink, SinkProgress))
      self.assertEqual(sink.description, "sink")
      self.assertEqual(sink.numOutputRows, -1)
      self.assertEqual(sink.metrics, {})
  
imports: import json
  import time
  import uuid
  from datetime import datetime
  from pyspark import Row
  from pyspark.sql.streaming import StreamingQueryListener
  from pyspark.sql.streaming.listener import QueryStartedEvent, QueryProgressEvent, QueryTerminatedEvent, SinkProgress, SourceProgress, StateOperatorProgress, StreamingQueryProgress
  from pyspark.testing.sqlutils import ReusedSQLTestCase
label_desc: Use image node_id 6 for calling the StreamingListenerTests obj's underlying member method code with example usage: obj.test_streaming_query_progress_fromJson() without return types