code: def test_get_filtered_upstream_artifacts_by_artifact_ids(self):
      # Test: get upstream artifacts by examples, with max_num_hops = 0, filter
      # by artifact name.
      result_from_exps = (
          self.resolver.get_upstream_artifacts_by_artifact_ids(
              [self.e1.id, self.e2.id, self.e3.id],
              max_num_hops=0,
              filter_query=f'name = "{self.e1.name}" ',
          )
      )
      self.assertLen(result_from_exps, 1)
      self.assertIn(self.e1.id, result_from_exps)
      self.assertCountEqual(
          [result_from_exps[self.e1.id][0].name], [self.e1.name]
      )
  
      # Test: get upstream artifacts by examples, with max_num_hops = 1, filter
      # by artifact name.
      result_from_exps = (
          self.resolver.get_upstream_artifacts_by_artifact_ids(
              [self.e1.id, self.e2.id, self.e3.id],
              max_num_hops=1,
              filter_query=f'name = "{self.e1.name}" ',
          )
      )
      self.assertLen(result_from_exps, 1)
      self.assertIn(self.e1.id, result_from_exps)
      self.assertCountEqual(
          [result_from_exps[self.e1.id][0].name], [self.e1.name]
      )
  
      # Test: get upstream artifacts by examples, with max_num_hops = 0, filter
      # by artifact type = Example.
      artifact_names_filter_query = '","'.join(
          [self.e1.name, self.e2.name, self.e3.name]
      )
      result_from_exps = (
          self.resolver.get_upstream_artifacts_by_artifact_ids(
              [self.e1.id, self.e2.id, self.e3.id],
              max_num_hops=0,
              filter_query=f'name IN ("{artifact_names_filter_query}")',
          )
      )
      self.assertLen(result_from_exps, 3)
      self.assertIn(self.e1.id, result_from_exps)
      self.assertIn(self.e2.id, result_from_exps)
      self.assertIn(self.e3.id, result_from_exps)
      self.assertCountEqual(
          [artifact.name for artifact in result_from_exps[self.e1.id]],
          [self.e1.name],
      )
      self.assertCountEqual(
          [artifact.name for artifact in result_from_exps[self.e2.id]],
          [self.e2.name],
      )
      self.assertCountEqual(
          [artifact.name for artifact in result_from_exps[self.e3.id]],
          [self.e3.name],
      )
  
      # Test: get upstream artifacts by examples, with max_num_hops = 0, filter
      # by artifact type = Evaluation.
      result_from_exps = (
          self.resolver.get_upstream_artifacts_by_artifact_ids(
              [self.e1.id, self.e2.id, self.e3.id],
              max_num_hops=0,
              filter_query=f'name = "{self.evaluation_type.name}"',
          )
      )
      self.assertEmpty(result_from_exps)
  
      # Test: get upstream artifacts by evaluation, with max_num_hops = 20, filter
      # by artifact type.
      result_from_eva = (
          self.resolver.get_upstream_artifacts_by_artifact_ids(
              [self.ev1.id],
              max_num_hops=20,
              filter_query=f'type = "{self.model_type.name}"',
          )
      )
      self.assertLen(result_from_eva, 1)
      self.assertIn(self.ev1.id, result_from_eva)
      self.assertCountEqual(
          [artifact.name for artifact in result_from_eva[self.ev1.id]],
          [self.m1.name],
      )
  
      # Test: get upstream artifacts by examples, models and evaluation, with
      # max_num_hops = 20, filter by artifact type = Model or Evaluation.
      result_from_exps_model_eva = (
          self.resolver.get_upstream_artifacts_by_artifact_ids(
              [self.e1.id, self.m2.id, self.ev1.id],
              max_num_hops=20,
              filter_query=(
                  f'type = "{self.model_type.name}" OR type ='
                  f' "{self.evaluation_type.name}"'
              ),
          )
      )
      self.assertLen(result_from_exps_model_eva, 2)
      self.assertIn(self.m2.id, result_from_exps_model_eva)
      self.assertIn(self.ev1.id, result_from_exps_model_eva)
      self.assertCountEqual(
          [
              artifact.name
              for artifact in result_from_exps_model_eva[self.m2.id]
          ],
          [self.m2.name],
      )
      self.assertCountEqual(
          [
              artifact.name
              for artifact in result_from_exps_model_eva[self.ev1.id]
          ],
          [self.ev1.name, self.m1.name],
      )
  
      # Test: get upstream artifacts by evaluation, with max_num_hops and
      # filter_query unspecified.
      result_from_ev1 = (
          self.resolver.get_upstream_artifacts_by_artifact_ids(
              [self.ev1.id]
          )
      )
      self.assertLen(result_from_ev1, 1)
      self.assertIn(self.ev1.id, result_from_ev1)
      self.assertCountEqual(
          [artifact.name for artifact in result_from_ev1[self.ev1.id]],
          [
              self.e1.name,
              self.e2.name,
              self.e3.name,
              self.m1.name,
              self.ev1.name,
          ],
      )
  
      def _is_input_event_or_valid_output_event(
          event: metadata_store_pb2.Event,
      ) -> bool:
          return (
              event.type != metadata_store_pb2.Event.Type.PENDING_OUTPUT
          )
  
      # Test: get upstream artifacts filtered by events from models. Only
      # artifacts connected to model_1 and model_2 itself will be included.
      result_from_m12 = (
          self.resolver.get_upstream_artifacts_by_artifact_ids(
              [self.m1.id, self.m2.id],
              max_num_hops=20,
              event_filter=_is_input_event_or_valid_output_event,
          )
      )
      self.assertLen(result_from_m12, 2)
      self.assertIn(self.m1.id, result_from_m12)
      self.assertIn(self.m2.id, result_from_m12)
      self.assertCountEqual(
          [artifact.name for artifact in result_from_m12[self.m1.id]],
          [self.e1.name, self.e2.name, self.m1.name],
      )
      self.assertCountEqual(
          [artifact.name for artifact in result_from_m12[self.m2.id]],
          [self.m2.name],
      )
  
      # Test: get upstream artifacts filtered by events from models, with filter
      # query for filtering upstream artifacts with type = Model. Only model_1
      # and model_2 will included.
      result_from_m12 = (
          self.resolver.get_upstream_artifacts_by_artifact_ids(
              [self.m1.id, self.m2.id],
              max_num_hops=20,
              filter_query=f'type = "{self.model_type.name}"',
              event_filter=_is_input_event_or_valid_output_event,
          )
      )
      self.assertLen(result_from_m12, 2)
      self.assertIn(self.m1.id, result_from_m12)
      self.assertIn(self.m2.id, result_from_m12)
      self.assertCountEqual(
          [artifact.name for artifact in result_from_m12[self.m1.id]],
          [self.m1.name],
      )
      self.assertCountEqual(
          [artifact.name for artifact in result_from_m12[self.m2.id]],
          [self.m2.name],
      )
  
imports: from typing import Dict, List
  from absl.testing import absltest
  from ml_metadata import metadata_store
  from ml_metadata.proto import metadata_store_pb2
  from ml_metadata.tools.mlmd_resolver import metadata_resolver
  from ml_metadata.tools.mlmd_resolver import metadata_resolver_utils
label_desc: Use image node_id 7 for calling the MetadataResolverTest obj's underlying member method code with example usage: obj.test_get_filtered_upstream_artifacts_by_artifact_ids() and returns: unknown