code: def test_get_unprocessed_inputs(self):
      with self._mlmd_connection as m:
          contexts = m.store.get_contexts()
          with self.subTest(name="NoInput"):
              # There is no input.
              resolved_info = task_gen_utils.ResolvedInfo(
                  contexts=contexts, input_and_params=[]
              )
              unprocessed_inputs = (
                  task_gen_utils.get_unprocessed_inputs(
                      m, resolved_info, self._transform
                  )
              )
              self.assertEmpty(unprocessed_inputs)
  
          # Fake 2 artifacts for _example_gen.
          otu.fake_upstream_node_run(
              m,
              self._example_gen,
              fake_result="Tflex rocks.",
              tmp_path=self.create_tempfile().full_path,
          )
          otu.fake_upstream_node_run(
              m,
              self._example_gen,
              fake_result="Tflex rocks.",
              tmp_path=self.create_tempfile().full_path,
          )
          artifact_types = m.store.get_artifact_types()
          artifacts = artifact_utils.deserialize_artifacts(
              artifact_types[0], m.store.get_artifacts()
          )
          artifacts.sort(
              key=lambda a: a.mlmd_artifact.create_time_since_epoch
          )
          input_and_param = task_gen_utils.InputAndParam(
              input_artifacts={"examples": artifacts}
          )
          resolved_info_for_transform = task_gen_utils.ResolvedInfo(
              contexts=contexts,
              input_and_params=[input_and_param],
          )
  
          with self.subTest(name="OneUnprocessedInput"):
              mock.patch.object(
                  m.store,
                  "get_executions",
                  wraps=m.store.get_executions,
              ).start()
              unprocessed_inputs = (
                  task_gen_utils.get_unprocessed_inputs(
                      m, resolved_info_for_transform, self._transform
                  )
              )
              m.store.get_executions.assert_called_once()
              self.assertLen(unprocessed_inputs, 1)
              self.assertEqual(unprocessed_inputs[0], input_and_param)
  
          with self.subTest(
              name="ResolvedArtifactsMatchProcessedArtifacts"
          ):
              mock.patch.object(
                  m.store,
                  "get_executions",
                  wraps=m.store.get_executions,
              ).start()
              # Simulate that the output for _example_gen is processed, so no
              # unprocessed input for _transform.
              execution = otu.fake_start_node_with_handle(
                  m,
                  self._transform,
                  input_artifacts={"examples": artifacts},
              )
              otu.fake_finish_node_with_handle(
                  m, self._transform, execution.id
              )
              unprocessed_inputs = (
                  task_gen_utils.get_unprocessed_inputs(
                      m, resolved_info_for_transform, self._transform
                  )
              )
              m.store.get_executions.assert_called_once()
              self.assertEqual(
                  m.store.get_executions.call_args[1][
                      "list_options"
                  ].filter_query,
                  "(contexts_0.type = 'node') AND (contexts_0.name ="
                  " 'my_pipeline.my_transform') AND (create_time_since_epoch >="
                  f" {artifacts[-1].mlmd_artifact.create_time_since_epoch}) AND"
                  " ((last_known_state = COMPLETE)"
                  " OR (last_known_state = CACHED) OR (last_known_state = FAILED))",
              )
              self.assertEmpty(unprocessed_inputs)
  
imports: import os
  import time
  from unittest import mock
  import uuid
  from absl.testing import parameterized
  import tensorflow
  from tfx import types
  from tfx import version
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import node_proto_view
  from tfx.orchestration.experimental.core import constants
  from tfx.orchestration.experimental.core import pipeline_state
  from tfx.orchestration.experimental.core import task
  from tfx.orchestration.experimental.core import task_gen_utils
  from tfx.orchestration.experimental.core import test_utils
  from tfx.orchestration.experimental.core.testing import test_async_pipeline
  from tfx.orchestration.experimental.core.testing import test_dynamic_exec_properties_pipeline
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable import outputs_utils
  from tfx.orchestration.portable.mlmd import execution_lib
  from tfx.proto.orchestration import execution_result_pb2
  from tfx.proto.orchestration import placeholder_pb2
  from tfx.types import artifact_utils
  from tfx.types import standard_artifacts
  from tfx.utils import status
  from tfx.utils import test_case_utils
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 16 for calling the TaskGenUtilsTest obj's underlying member method code with example usage: obj.test_get_unprocessed_inputs() without return types