code: def testDriverWithSpan(self):
      # Test align of span number.
      span1_split1 = os.path.join(
          _TEST_INPUT_DIR, "span1", "split1", "data"
      )
      io_utils.write_string_file(span1_split1, "testing11")
      span1_split2 = os.path.join(
          _TEST_INPUT_DIR, "span1", "split2", "data"
      )
      io_utils.write_string_file(span1_split2, "testing12")
      span2_split1 = os.path.join(
          _TEST_INPUT_DIR, "span2", "split1", "data"
      )
      io_utils.write_string_file(span2_split1, "testing21")
  
      serialized_args = [
          "--json_serialized_invocation_args",
          json_format.MessageToJson(message=self._executor_invocation),
      ]
      with self.assertRaisesRegex(
          ValueError, "Latest span should be the same for each split"
      ):
          driver.main(driver._parse_flags(serialized_args))
  
      # Test if latest span is selected when span aligns for each split.
      span2_split2 = os.path.join(
          _TEST_INPUT_DIR, "span2", "split2", "data"
      )
      io_utils.write_string_file(span2_split2, "testing22")
  
      driver.main(driver._parse_flags(serialized_args))
  
      # Check the output metadata file for the expected outputs
      with fileio.open(
          _TEST_OUTPUT_METADATA_JSON, "rb"
      ) as output_meta_json:
          output_metadata = pipeline_pb2.ExecutorOutput()
          json_format.Parse(
              output_meta_json.read(),
              output_metadata,
              ignore_unknown_fields=True,
          )
          self.assertEqual(
              output_metadata.parameters["span"].int_value, 2
          )
          self.assertEqual(
              output_metadata.parameters["input_config"].string_value,
              json_format.MessageToJson(
                  example_gen_pb2.Input(
                      splits=[
                          example_gen_pb2.Input.Split(
                              name="s1", pattern="span2/split1/*"
                          ),
                          example_gen_pb2.Input.Split(
                              name="s2", pattern="span2/split2/*"
                          ),
                      ]
                  )
              ),
          )
  
file_path: tfx/tfx/orchestration/kubeflow/v2/file_based_example_gen/driver_test.py
imports: import json
  import os
  from absl import logging
  from kfp.pipeline_spec import pipeline_spec_pb2
  import tensorflow
  from tfx.dsl.io import fileio
  from tfx.orchestration.kubeflow.v2 import compiler_utils
  from tfx.orchestration.kubeflow.v2.file_based_example_gen import driver
  from tfx.proto import example_gen_pb2
  from tfx.types import standard_artifacts
  from tfx.utils import io_utils
  from tfx.utils import test_case_utils
  from google.protobuf import json_format
label_desc: Use image node_id 3 for calling the RunDriverTest obj's underlying member method code with example usage: obj.testDriverWithSpan() without return types