code: def testDeepCopyTags(self):
      if not resources.ResourceHint.is_registered("tags"):
          self.skipTest("Resource hint tags are not available.")
  
      with DeepCopyTest._MakeBeamPipeline() as p:
          grouped = (
              p
              | beam.Create([(1, "a"), (2, "b"), (3, "c")])
              | beam.Map(
                  lambda x: DeepCopyTest._CountingIdentityFn(
                      "PreGroup", x
                  )
              )
          )
  
          modified = (
              grouped
              | "Add1"
              >> beam.Map(
                  DeepCopyTest._MakeAdd1CountingIdentityFn("Add1")
              )
              | "Add2"
              >> beam.Map(
                  DeepCopyTest._MakeAdd1CountingIdentityFn("Add2")
              )
          )
  
          num_copies = 6
  
          for i in range(num_copies):
              copied = deep_copy.deep_copy(modified)
              # Check labels.
              self.assertEqual(
                  copied.producer.full_label, "Add2.Copy%d" % i
              )
              self.assertEqual(
                  copied.producer.inputs[0].producer.full_label,
                  "Add1.Copy%d" % i,
              )
  
              # Check resource hints.
              self.assertEqual(
                  modified.producer.resource_hints,
                  {
                      "beam:resources:close_to_resources:v1": b"/fake/DeepCopy.Original[0]"
                  },
              )
              self.assertEqual(
                  modified.producer.inputs[0].producer.resource_hints,
                  {
                      "beam:resources:close_to_resources:v1": b"/fake/DeepCopy.Original[0]"
                  },
              )
              self.assertEqual(
                  copied.producer.resource_hints,
                  {
                      "beam:resources:close_to_resources:v1": b"/fake/DeepCopy.Copy%d[0]"
                      % i
                  },
              )
              self.assertEqual(
                  copied.producer.inputs[0].producer.resource_hints,
                  {
                      "beam:resources:close_to_resources:v1": b"/fake/DeepCopy.Copy%d[0]"
                      % i
                  },
              )
  
          # pylint: disable=expression-not-assigned
          modified | "Add3" >> beam.Map(
              DeepCopyTest._MakeAdd1CountingIdentityFn("Add3")
          )
          # pylint: enable=expression-not-assigned
  
      # Check counts of processed items. Without the materialization boundary,
      # e.g. GroupByKey, PreGroup is also copied.
      self.assertEqual(
          DeepCopyTest._counts["PreGroup"], 3 * (num_copies + 1)
      )
      self.assertEqual(
          DeepCopyTest._counts["Add1"], 3 * (num_copies + 1)
      )
      self.assertEqual(
          DeepCopyTest._counts["Add2"], 3 * (num_copies + 1)
      )
      self.assertEqual(DeepCopyTest._counts["Add3"], 3)
  
imports: import collections
  import apache_beam
  from apache_beam import pvalue
  from apache_beam.transforms import resources
  from tensorflow_transform.beam import deep_copy
  from tensorflow_transform.beam import test_helpers
  import unittest
label_desc: Use image node_id 12 for calling the DeepCopyTest obj's underlying member method code with example usage: obj.testDeepCopyTags() without return types