code: def _get_test_json_data_source(self):
      import json
      import os
      from dataclasses import dataclass
  
      class TestJsonReader(DataSourceReader):
          def __init__(self, options):
              self.options = options
  
          def read(self, partition):
              path = self.options.get("path")
              if path is None:
                  raise Exception("path is not specified")
              with open(path, "r") as file:
                  for line in file.readlines():
                      if line.strip():
                          data = json.loads(line)
                          yield data.get("name"), data.get("age")
  
      @dataclass
      class TestCommitMessage(WriterCommitMessage):
          count: int
  
      class TestJsonWriter(DataSourceWriter):
          def __init__(self, options):
              self.options = options
              self.path = self.options.get("path")
  
          def write(self, iterator):
              from pyspark import TaskContext
  
              context = TaskContext.get()
              output_path = os.path.join(
                  self.path, f"{context.partitionId}.json"
              )
              count = 0
              with open(output_path, "w") as file:
                  for row in iterator:
                      count += 1
                      if "id" in row and row.id > 5:
                          raise Exception("id > 5")
                      file.write(json.dumps(row.asDict()) + "\n")
              return TestCommitMessage(count=count)
  
          def commit(self, messages):
              total_count = sum(message.count for message in messages)
              with open(
                  os.path.join(self.path, "_success.txt"), "a"
              ) as file:
                  file.write(f"count: {total_count}\n")
  
          def abort(self, messages):
              with open(
                  os.path.join(self.path, "_failed.txt"), "a"
              ) as file:
                  file.write("failed")
  
      class TestJsonDataSource(DataSource):
          @classmethod
          def name(cls):
              return "my-json"
  
          def schema(self):
              return "name STRING, age INT"
  
          def reader(self, schema) -> "DataSourceReader":
              return TestJsonReader(self.options)
  
          def writer(self, schema, overwrite):
              return TestJsonWriter(self.options)
  
      return TestJsonDataSource
  
imports: import os
  import tempfile
  import unittest
  from typing import Callable, Union
  from pyspark.errors import PythonException
  from pyspark.sql.datasource import DataSource, DataSourceReader, InputPartition, DataSourceWriter, WriterCommitMessage, CaseInsensitiveDict
  from pyspark.sql.types import Row, StructType
  from pyspark.testing import assertDataFrameEqual
  from pyspark.testing.sqlutils import ReusedSQLTestCase
  from pyspark.testing.utils import SPARK_HOME
label_desc: Use image node_id 17 for calling the BasePythonDataSourceTestsMixin obj's underlying member method code with example usage: obj._get_test_json_data_source() and returns: TestJsonDataSource, TestCommitMessage, str, str, TestJsonReader, TestJsonWriter