code: def testSideInputNotCopied(self):
      with DeepCopyTest._MakeBeamPipeline() as p:
          side = (
              p
              | "CreateSide" >> beam.Create(["s1", "s2", "s3"])
              | beam.Map(
                  lambda x: DeepCopyTest._CountingIdentityFn(
                      "SideInput", x
                  )
              )
          )
          main = (
              p
              | "CreateMain" >> beam.Create([1, 2, 3])
              | beam.Map(
                  lambda x: DeepCopyTest._CountingIdentityFn("Main", x)
              )
              | beam.Map(lambda e, s: (e, list(s)), pvalue.AsList(side))
          )
          copied = deep_copy.deep_copy(main)
  
          # Check that deep copy was performed.
          self.assertIsNot(main, copied)
          self.assertIsNot(main.producer, copied.producer)
  
          # Check that deep copy stops at the side input materialization boundary.
          self.assertIs(
              main.producer.side_inputs[0],
              copied.producer.side_inputs[0],
          )
          self.assertIs(main.producer.side_inputs[0].pvalue, side)
  
      # Check counts of processed items.
      self.assertEqual(DeepCopyTest._counts["SideInput"], 3)
      self.assertEqual(DeepCopyTest._counts["Main"], 6)
  
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 11 for calling the DeepCopyTest obj's underlying member method code with example usage: obj.testSideInputNotCopied() without return types