code: def test_get_upstream_artifacts_by_artifact_uri(self):
      # Test: get upstream artifacts by model_1, with max_num_hops = 0
      result_from_m1 = (
          self.resolver.get_upstream_artifacts_by_artifact_uri(
              self.m1.uri, max_num_hops=0
          )
      )
      self.assertLen(result_from_m1, 1)
      self.assertIn(self.m1.id, result_from_m1)
      self.assertEqual(
          [result_from_m1[self.m1.id][0].name], [self.m1.name]
      )
  
      # Test: get upstream artifacts by model_1, with max_num_hops = 2
      result_from_m1 = (
          self.resolver.get_upstream_artifacts_by_artifact_uri(
              self.m1.uri, max_num_hops=2
          )
      )
      self.assertLen(result_from_m1, 1)
      self.assertIn(self.m1.id, result_from_m1)
      self.assertCountEqual(
          [artifact.name for artifact in result_from_m1[self.m1.id]],
          [self.e1.name, self.m1.name, self.e2.name],
      )
  
      # Test: get upstream artifacts by evaluation_1, with max_num_hops = 2
      result_from_ev1 = (
          self.resolver.get_upstream_artifacts_by_artifact_uri(
              self.ev1.uri, max_num_hops=2
          )
      )
      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.ev1.name, self.e3.name, self.m1.name],
      )
  
      # Test: get upstream artifacts by evaluation_1, with max_num_hops = 20
      result_from_ev1 = (
          self.resolver.get_upstream_artifacts_by_artifact_uri(
              self.ev1.uri, max_num_hops=20
          )
      )
      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.ev1.name,
              self.e3.name,
              self.m1.name,
              self.e1.name,
              self.e2.name,
          ],
      )
  
      # Test: get upstream artifacts by evaluation_1, with max_num_hops
      # unspecified.
      result_from_ev1 = (
          self.resolver.get_upstream_artifacts_by_artifact_uri(
              self.ev1.uri
          )
      )
      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.ev1.name,
              self.e3.name,
              self.m1.name,
              self.e1.name,
              self.e2.name,
          ],
      )
      # Test: raise ValueError if `artifact_uri` is empty.
      with self.assertRaisesRegex(
          ValueError, "`artifact_uri` is empty."
      ):
          self.resolver.get_upstream_artifacts_by_artifact_uri("")
  
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 6 for calling the MetadataResolverTest obj's underlying member method code with example usage: obj.test_get_upstream_artifacts_by_artifact_uri() without return types