code: def test_listener_events(self):
      def verify(test_listener, table_postfix):
          try:
              self.spark.streams.addListener(test_listener)
  
              # This ensures the read socket on the server won't crash (i.e. because of timeout)
              # when there hasn't been a new event for a long time
              time.sleep(30)
  
              df = (
                  self.spark.readStream.format("rate")
                  .option("rowsPerSecond", 10)
                  .load()
              )
              df_observe = df.observe(
                  "my_event", count(lit(1)).alias("rc")
              )
              df_stateful = (
                  df_observe.groupBy().count()
              )  # make query stateful
              q = (
                  df_stateful.writeStream.format("noop")
                  .queryName("test")
                  .outputMode("complete")
                  .start()
              )
  
              self.assertTrue(q.isActive)
              # ensure at least one batch is ran
              while (
                  q.lastProgress is None
                  or q.lastProgress["batchId"] == 0
              ):
                  time.sleep(5)
              q.stop()
              self.assertFalse(q.isActive)
  
              # Sleep to make sure listener_terminated_events is written successfully
              time.sleep(60)
  
              start_table_name = "listener_start_events" + table_postfix
              progress_tbl_name = (
                  "listener_progress_events" + table_postfix
              )
              terminated_tbl_name = (
                  "listener_terminated_events" + table_postfix
              )
  
              start_event = pyspark.cloudpickle.loads(
                  self.spark.read.table(start_table_name).collect()[0][
                      0
                  ]
              )
  
              progress_event = pyspark.cloudpickle.loads(
                  self.spark.read.table(progress_tbl_name).collect()[0][
                      0
                  ]
              )
  
              terminated_event = pyspark.cloudpickle.loads(
                  self.spark.read.table(terminated_tbl_name).collect()[
                      0
                  ][0]
              )
  
              self.check_start_event(start_event)
              self.check_progress_event(progress_event)
              self.check_terminated_event(terminated_event)
  
          finally:
              self.spark.streams.removeListener(test_listener)
  
              # Remove again to verify this won't throw any error
              self.spark.streams.removeListener(test_listener)
  
      verify(TestListenerV1(), "_v1")
      verify(TestListenerV2(), "_v2")
  
imports: import unittest
  import time
  import pyspark.cloudpickle
  from pyspark.errors import PySparkPicklingError
  from pyspark.sql.tests.streaming.test_streaming_listener import StreamingListenerTestsMixin
  from pyspark.sql.streaming.listener import StreamingQueryListener
  from pyspark.sql.functions import count, lit
  from pyspark.testing.connectutils import ReusedConnectTestCase
label_desc: Use image node_id 1 for calling the StreamingListenerParityTests obj's underlying member method code with example usage: obj.test_listener_events() without return types