class_docstr: Abstract class for integration testing concrete Metadata implementations.
  
  Subclasses should override metadata() to return an instance of Metadata that
  has not been entered.
class_docstr_tok: ["Abstract","class","for","integration","testing","concrete","Metadata","implementations",".","Subclasses","should","override","metadata","(",")","to","return","an","instance","of","Metadata","that","has","not","been","entered","."]
code: def testExecution(self):
      with self.metadata() as m:
          contexts = m.register_pipeline_contexts_if_not_exists(
              self._pipeline_info
          )
          # Test prepare_execution.
          exec_properties = {"arg_one": 1}
          input_artifact = standard_artifacts.Examples()
          output_artifact = standard_artifacts.Examples()
          input_artifacts = {"input": [input_artifact]}
          output_artifacts = {"output": [output_artifact]}
          m.register_execution(
              input_artifacts=input_artifacts,
              exec_properties=exec_properties,
              pipeline_info=self._pipeline_info,
              component_info=self._component_info,
              contexts=contexts,
          )
          [execution] = m.store.get_executions_by_context(
              contexts[0].id
          )
          # Skip verifying time sensitive fields.
          execution.ClearField("type_id")
          execution.ClearField("type")
          execution.ClearField("create_time_since_epoch")
          execution.ClearField("last_update_time_since_epoch")
          self.assertProtoEquals(
              """
          id: 1
          last_known_state: RUNNING
          properties {
            key: "state"
            value {
              string_value: "new"
            }
          }
          properties {
            key: "pipeline_name"
            value {
              string_value: "my_pipeline"
            }
          }
          properties {
            key: "pipeline_root"
            value {
              string_value: "/tmp"
            }
          }
          properties {
            key: "run_id"
            value {
              string_value: "my_run_id"
            }
          }
          properties {
            key: "component_id"
            value {
              string_value: "my_component"
            }
          }
          properties {
            key: "arg_one"
            value {
              string_value: "1"
            }
          }""",
              execution,
          )
  
          # Test publish_execution.
          m.publish_execution(
              component_info=self._component_info,
              output_artifacts=output_artifacts,
          )
          # Make sure artifacts in output_dict are published.
          self.assertEqual(
              ArtifactState.PUBLISHED, output_artifact.state
          )
          # Make sure execution state are changed.
          [execution] = m.store.get_executions_by_id([execution.id])
          self.assertEqual(
              metadata.EXECUTION_STATE_COMPLETE,
              execution.properties["state"].string_value,
          )
          # Make sure events are published.
          events = m.store.get_events_by_execution_ids([execution.id])
          self.assertEqual(2, len(events))
          self.assertEqual(input_artifact.id, events[0].artifact_id)
          self.assertEqual(
              metadata_store_pb2.Event.INPUT, events[0].type
          )
          self.assertProtoEquals(
              """
            steps {
              key: "input"
            }
            steps {
              index: 0
            }""",
              events[0].path,
          )
          self.assertEqual(output_artifact.id, events[1].artifact_id)
          self.assertEqual(
              metadata_store_pb2.Event.OUTPUT, events[1].type
          )
          self.assertProtoEquals(
              """
            steps {
              key: "output"
            }
            steps {
              index: 0
            }""",
              events[1].path,
          )
  
imports: from typing import Mapping
  import tensorflow
  from tfx.orchestration import data_types
  from tfx.orchestration import metadata
  from tfx.types import artifact_utils
  from tfx.types import standard_artifacts
  from tfx.types.artifact import ArtifactState
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 7 for calling the MetadataTest obj's underlying member method code with example usage: obj.testExecution() without return types