code: def setUp(self):
      super().setUp()
      # Set up MLMD connection.
      pipeline_root = os.path.join(
          os.environ.get(
              "TEST_UNDECLARED_OUTPUTS_DIR", self.get_temp_dir()
          ),
          self.id(),
      )
      metadata_path = os.path.join(
          pipeline_root, "metadata", "metadata.db"
      )
      connection_config = metadata.sqlite_metadata_connection_config(
          metadata_path
      )
      connection_config.sqlite.SetInParent()
      self._mlmd_connection = metadata.Metadata(
          connection_config=connection_config
      )
      with self._mlmd_connection as m:
          self._execution = execution_publish_utils.register_execution(
              metadata_handle=m,
              execution_type=metadata_store_pb2.ExecutionType(
                  name="test_execution_type"
              ),
              contexts=[],
              input_artifacts=[],
          )
      # Set up gRPC stub.
      port = portpicker.pick_unused_port()
      self.sidecar = execution_watcher.ExecutionWatcher(
          port,
          mlmd_connection=self._mlmd_connection,
          execution=self._execution,
          creds=grpc.local_server_credentials(),
      )
      self.sidecar.start()
      self.stub = execution_watcher.generate_service_stub(
          self.sidecar.address, grpc.local_channel_credentials()
      )
  
file_path: tfx/tfx/orchestration/portable/execution_watcher_test.py
imports: import os
  import grpc
  import portpicker
  import tensorflow
  from tfx.orchestration import metadata
  from tfx.orchestration.portable import execution_publish_utils
  from tfx.orchestration.portable import execution_watcher
  from tfx.proto.orchestration import execution_watcher_pb2
  from tfx.utils import test_case_utils
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 1 for calling the ExecutionWatcherTest obj's underlying member method code with example usage: obj.setUp() without return types