text: #! /usr/bin/env python
  #
  # Protocol Buffers - Google's data interchange format
  # Copyright 2008 Google Inc.  All rights reserved.
  # https://developers.google.com/protocol-buffers/
  #
  # Redistribution and use in source and binary forms, with or without
  # modification, are permitted provided that the following conditions are
  # met:
  #
  #     * Redistributions of source code must retain the above copyright
  # notice, this list of conditions and the following disclaimer.
  #     * Redistributions in binary form must reproduce the above
  # copyright notice, this list of conditions and the following disclaimer
  # in the documentation and/or other materials provided with the
  # distribution.
  #     * Neither the name of Google Inc. nor the names of its
  # contributors may be used to endorse or promote products derived from
  # this software without specific prior written permission.
  #
  # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  
  """Test for google.protobuf.internal.well_known_types."""
  
  __author__ = 'jieluo@google.com (Jie Luo)'
  
  from datetime import datetime
  
  try:
    import unittest2 as unittest  #PY26
  except ImportError:
    import unittest
  
  from google.protobuf import any_pb2
  from google.protobuf import duration_pb2
  from google.protobuf import field_mask_pb2
  from google.protobuf import struct_pb2
  from google.protobuf import timestamp_pb2
  from google.protobuf import unittest_pb2
  from google.protobuf.internal import any_test_pb2
  from google.protobuf.internal import test_util
  from google.protobuf.internal import well_known_types
  from google.protobuf import descriptor
  from google.protobuf import text_format
  
  
  class TimeUtilTestBase(unittest.TestCase):
  
    def CheckTimestampConversion(self, message, text):
      self.assertEqual(text, message.ToJsonString())
      parsed_message = timestamp_pb2.Timestamp()
      parsed_message.FromJsonString(text)
      self.assertEqual(message, parsed_message)
  
    def CheckDurationConversion(self, message, text):
      self.assertEqual(text, message.ToJsonString())
      parsed_message = duration_pb2.Duration()
      parsed_message.FromJsonString(text)
      self.assertEqual(message, parsed_message)
  
  
  class TimeUtilTest(TimeUtilTestBase):
  
    def testTimestampSerializeAndParse(self):
      message = timestamp_pb2.Timestamp()
      # Generated output should contain 3, 6, or 9 fractional digits.
      message.seconds = 0
      message.nanos = 0
      self.CheckTimestampConversion(message, '1970-01-01T00:00:00Z')
      message.nanos = 10000000
      self.CheckTimestampConversion(message, '1970-01-01T00:00:00.010Z')
      message.nanos = 10000
      self.CheckTimestampConversion(message, '1970-01-01T00:00:00.000010Z')
      message.nanos = 10
      self.CheckTimestampConversion(message, '1970-01-01T00:00:00.000000010Z')
      # Test min timestamps.
      message.seconds = -62135596800
      message.nanos = 0
      self.CheckTimestampConversion(message, '0001-01-01T00:00:00Z')
      # Test max timestamps.
      message.seconds = 253402300799
      message.nanos = 999999999
      self.CheckTimestampConversion(message, '9999-12-31T23:59:59.999999999Z')
      # Test negative timestamps.
      message.seconds = -1
      self.CheckTimestampConversion(message, '1969-12-31T23:59:59.999999999Z')
  
      # Parsing accepts an fractional digits as long as they fit into nano
      # precision.
      message.FromJsonString('1970-01-01T00:00:00.1Z')
      self.assertEqual(0, message.seconds)
      self.assertEqual(100000000, message.nanos)
      # Parsing accpets offsets.
      message.FromJsonString('1970-01-01T00:00:00-08:00')
      self.assertEqual(8 * 3600, message.seconds)
      self.assertEqual(0, message.nanos)
  
    def testDurationSerializeAndParse(self):
      message = duration_pb2.Duration()
      # Generated output should contain 3, 6, or 9 fractional digits.
      message.seconds = 0
      message.nanos = 0
      self.CheckDurationConversion(message, '0s')
      message.nanos = 10000000
      self.CheckDurationConversion(message, '0.010s')
      message.nanos = 10000
      self.CheckDurationConversion(message, '0.000010s')
      message.nanos = 10
      self.CheckDurationConversion(message, '0.000000010s')
  
      # Test min and max
      message.seconds = 315576000000
      message.nanos = 999999999
      self.CheckDurationConversion(message, '315576000000.999999999s')
      message.seconds = -315576000000
      message.nanos = -999999999
      self.CheckDurationConversion(message, '-315576000000.999999999s')
  
      # Parsing accepts an fractional digits as long as they fit into nano
      # precision.
      message.FromJsonString('0.1s')
      self.assertEqual(100000000, message.nanos)
      message.FromJsonString('0.0000001s')
      self.assertEqual(100, message.nanos)
  
    def testTimestampIntegerConversion(self):
      message = timestamp_pb2.Timestamp()
      message.FromNanoseconds(1)
      self.assertEqual('1970-01-01T00:00:00.000000001Z',
                       message.ToJsonString())
      self.assertEqual(1, message.ToNanoseconds())
  
      message.FromNanoseconds(-1)
      self.assertEqual('1969-12-31T23:59:59.999999999Z',
                       message.ToJsonString())
      self.assertEqual(-1, message.ToNanoseconds())
  
      message.FromMicroseconds(1)
      self.assertEqual('1970-01-01T00:00:00.000001Z',
                       message.ToJsonString())
      self.assertEqual(1, message.ToMicroseconds())
  
      message.FromMicroseconds(-1)
      self.assertEqual('1969-12-31T23:59:59.999999Z',
                       message.ToJsonString())
      self.assertEqual(-1, message.ToMicroseconds())
  
      message.FromMilliseconds(1)
      self.assertEqual('1970-01-01T00:00:00.001Z',
                       message.ToJsonString())
      self.assertEqual(1, message.ToMilliseconds())
  
      message.FromMilliseconds(-1)
      self.assertEqual('1969-12-31T23:59:59.999Z',
                       message.ToJsonString())
      self.assertEqual(-1, message.ToMilliseconds())
  
      message.FromSeconds(1)
      self.assertEqual('1970-01-01T00:00:01Z',
                       message.ToJsonString())
      self.assertEqual(1, message.ToSeconds())
  
      message.FromSeconds(-1)
      self.assertEqual('1969-12-31T23:59:59Z',
                       message.ToJsonString())
      self.assertEqual(-1, message.ToSeconds())
  
      message.FromNanoseconds(1999)
      self.assertEqual(1, message.ToMicroseconds())
      # For negative values, Timestamp will be rounded down.
      # For example, "1969-12-31T23:59:59.5Z" (i.e., -0.5s) rounded to seconds
      # will be "1969-12-31T23:59:59Z" (i.e., -1s) rather than
      # "1970-01-01T00:00:00Z" (i.e., 0s).
      message.FromNanoseconds(-1999)
      self.assertEqual(-2, message.ToMicroseconds())
  
    def testDurationIntegerConversion(self):
      message = duration_pb2.Duration()
      message.FromNanoseconds(1)
      self.assertEqual('0.000000001s',
                       message.ToJsonString())
      self.assertEqual(1, message.ToNanoseconds())
  
      message.FromNanoseconds(-1)
      self.assertEqual('-0.000000001s',
                       message.ToJsonString())
      self.assertEqual(-1, message.ToNanoseconds())
  
      message.FromMicroseconds(1)
      self.assertEqual('0.000001s',
                       message.ToJsonString())
      self.assertEqual(1, message.ToMicroseconds())
  
      message.FromMicroseconds(-1)
      self.assertEqual('-0.000001s',
                       message.ToJsonString())
      self.assertEqual(-1, message.ToMicroseconds())
  
      message.FromMilliseconds(1)
      self.assertEqual('0.001s',
                       message.ToJsonString())
      self.assertEqual(1, message.ToMilliseconds())
  
      message.FromMilliseconds(-1)
      self.assertEqual('-0.001s',
                       message.ToJsonString())
      self.assertEqual(-1, message.ToMilliseconds())
  
      message.FromSeconds(1)
      self.assertEqual('1s', message.ToJsonString())
      self.assertEqual(1, message.ToSeconds())
  
      message.FromSeconds(-1)
      self.assertEqual('-1s',
                       message.ToJsonString())
      self.assertEqual(-1, message.ToSeconds())
  
      # Test truncation behavior.
      message.FromNanoseconds(1999)
      self.assertEqual(1, message.ToMicroseconds())
  
      # For negative values, Duration will be rounded towards 0.
      message.FromNanoseconds(-1999)
      self.assertEqual(-1, message.ToMicroseconds())
  
    def testDatetimeConverison(self):
      message = timestamp_pb2.Timestamp()
      dt = datetime(1970, 1, 1)
      message.FromDatetime(dt)
      self.assertEqual(dt, message.ToDatetime())
  
      message.FromMilliseconds(1999)
      self.assertEqual(datetime(1970, 1, 1, 0, 0, 1, 999000),
                       message.ToDatetime())
  
    def testTimedeltaConversion(self):
      message = duration_pb2.Duration()
      message.FromNanoseconds(1999999999)
      td = message.ToTimedelta()
      self.assertEqual(1, td.seconds)
      self.assertEqual(999999, td.microseconds)
  
      message.FromNanoseconds(-1999999999)
      td = message.ToTimedelta()
      self.assertEqual(-1, td.days)
      self.assertEqual(86398, td.seconds)
      self.assertEqual(1, td.microseconds)
  
      message.FromMicroseconds(-1)
      td = message.ToTimedelta()
      self.assertEqual(-1, td.days)
      self.assertEqual(86399, td.seconds)
      self.assertEqual(999999, td.microseconds)
      converted_message = duration_pb2.Duration()
      converted_message.FromTimedelta(td)
      self.assertEqual(message, converted_message)
  
    def testInvalidTimestamp(self):
      message = timestamp_pb2.Timestamp()
      self.assertRaisesRegexp(
          ValueError,
          'time data \'10000-01-01T00:00:00\' does not match'
          ' format \'%Y-%m-%dT%H:%M:%S\'',
          message.FromJsonString, '10000-01-01T00:00:00.00Z')
      self.assertRaisesRegexp(
          well_known_types.ParseError,
          'nanos 0123456789012 more than 9 fractional digits.',
          message.FromJsonString,
          '1970-01-01T00:00:00.0123456789012Z')
      self.assertRaisesRegexp(
          well_known_types.ParseError,
          (r'Invalid timezone offset value: \+08.'),
          message.FromJsonString,
          '1972-01-01T01:00:00.01+08',)
      self.assertRaisesRegexp(
          ValueError,
          'year is out of range',
          message.FromJsonString,
          '0000-01-01T00:00:00Z')
      message.seconds = 253402300800
      self.assertRaisesRegexp(
          OverflowError,
          'date value out of range',
          message.ToJsonString)
  
    def testInvalidDuration(self):
      message = duration_pb2.Duration()
      self.assertRaisesRegexp(
          well_known_types.ParseError,
          'Duration must end with letter "s": 1.',
          message.FromJsonString, '1')
      self.assertRaisesRegexp(
          well_known_types.ParseError,
          'Couldn\'t parse duration: 1...2s.',
          message.FromJsonString, '1...2s')
  
  
  class FieldMaskTest(unittest.TestCase):
  
    def testStringFormat(self):
      mask = field_mask_pb2.FieldMask()
      self.assertEqual('', mask.ToJsonString())
      mask.paths.append('foo')
      self.assertEqual('foo', mask.ToJsonString())
      mask.paths.append('bar')
      self.assertEqual('foo,bar', mask.ToJsonString())
  
      mask.FromJsonString('')
      self.assertEqual('', mask.ToJsonString())
      mask.FromJsonString('foo')
      self.assertEqual(['foo'], mask.paths)
      mask.FromJsonString('foo,bar')
      self.assertEqual(['foo', 'bar'], mask.paths)
  
    def testDescriptorToFieldMask(self):
      mask = field_mask_pb2.FieldMask()
      msg_descriptor = unittest_pb2.TestAllTypes.DESCRIPTOR
      mask.AllFieldsFromDescriptor(msg_descriptor)
      self.assertEqual(75, len(mask.paths))
      self.assertTrue(mask.IsValidForDescriptor(msg_descriptor))
      for field in msg_descriptor.fields:
        self.assertTrue(field.name in mask.paths)
      mask.paths.append('optional_nested_message.bb')
      self.assertTrue(mask.IsValidForDescriptor(msg_descriptor))
      mask.paths.append('repeated_nested_message.bb')
      self.assertFalse(mask.IsValidForDescriptor(msg_descriptor))
  
    def testCanonicalFrom(self):
      mask = field_mask_pb2.FieldMask()
      out_mask = field_mask_pb2.FieldMask()
      # Paths will be sorted.
      mask.FromJsonString('baz.quz,bar,foo')
      out_mask.CanonicalFormFromMask(mask)
      self.assertEqual('bar,baz.quz,foo', out_mask.ToJsonString())
      # Duplicated paths will be removed.
      mask.FromJsonString('foo,bar,foo')
      out_mask.CanonicalFormFromMask(mask)
      self.assertEqual('bar,foo', out_mask.ToJsonString())
      # Sub-paths of other paths will be removed.
      mask.FromJsonString('foo.b1,bar.b1,foo.b2,bar')
      out_mask.CanonicalFormFromMask(mask)
      self.assertEqual('bar,foo.b1,foo.b2', out_mask.ToJsonString())
  
      # Test more deeply nested cases.
      mask.FromJsonString(
          'foo.bar.baz1,foo.bar.baz2.quz,foo.bar.baz2')
      out_mask.CanonicalFormFromMask(mask)
      self.assertEqual('foo.bar.baz1,foo.bar.baz2',
                       out_mask.ToJsonString())
      mask.FromJsonString(
          'foo.bar.baz1,foo.bar.baz2,foo.bar.baz2.quz')
      out_mask.CanonicalFormFromMask(mask)
      self.assertEqual('foo.bar.baz1,foo.bar.baz2',
                       out_mask.ToJsonString())
      mask.FromJsonString(
          'foo.bar.baz1,foo.bar.baz2,foo.bar.baz2.quz,foo.bar')
      out_mask.CanonicalFormFromMask(mask)
      self.assertEqual('foo.bar', out_mask.ToJsonString())
      mask.FromJsonString(
          'foo.bar.baz1,foo.bar.baz2,foo.bar.baz2.quz,foo')
      out_mask.CanonicalFormFromMask(mask)
      self.assertEqual('foo', out_mask.ToJsonString())
  
    def testUnion(self):
      mask1 = field_mask_pb2.FieldMask()
      mask2 = field_mask_pb2.FieldMask()
      out_mask = field_mask_pb2.FieldMask()
      mask1.FromJsonString('foo,baz')
      mask2.FromJsonString('bar,quz')
      out_mask.Union(mask1, mask2)
      self.assertEqual('bar,baz,foo,quz', out_mask.ToJsonString())
      # Overlap with duplicated paths.
      mask1.FromJsonString('foo,baz.bb')
      mask2.FromJsonString('baz.bb,quz')
      out_mask.Union(mask1, mask2)
      self.assertEqual('baz.bb,foo,quz', out_mask.ToJsonString())
      # Overlap with paths covering some other paths.
      mask1.FromJsonString('foo.bar.baz,quz')
      mask2.FromJsonString('foo.bar,bar')
      out_mask.Union(mask1, mask2)
      self.assertEqual('bar,foo.bar,quz', out_mask.ToJsonString())
  
    def testIntersect(self):
      mask1 = field_mask_pb2.FieldMask()
      mask2 = field_mask_pb2.FieldMask()
      out_mask = field_mask_pb2.FieldMask()
      # Test cases without overlapping.
      mask1.FromJsonString('foo,baz')
      mask2.FromJsonString('bar,quz')
      out_mask.Intersect(mask1, mask2)
      self.assertEqual('', out_mask.ToJsonString())
      # Overlap with duplicated paths.
      mask1.FromJsonString('foo,baz.bb')
      mask2.FromJsonString('baz.bb,quz')
      out_mask.Intersect(mask1, mask2)
      self.assertEqual('baz.bb', out_mask.ToJsonString())
      # Overlap with paths covering some other paths.
      mask1.FromJsonString('foo.bar.baz,quz')
      mask2.FromJsonString('foo.bar,bar')
      out_mask.Intersect(mask1, mask2)
      self.assertEqual('foo.bar.baz', out_mask.ToJsonString())
      mask1.FromJsonString('foo.bar,bar')
      mask2.FromJsonString('foo.bar.baz,quz')
      out_mask.Intersect(mask1, mask2)
      self.assertEqual('foo.bar.baz', out_mask.ToJsonString())
  
    def testMergeMessage(self):
      # Test merge one field.
      src = unittest_pb2.TestAllTypes()
      test_util.SetAllFields(src)
      for field in src.DESCRIPTOR.fields:
        if field.containing_oneof:
          continue
        field_name = field.name
        dst = unittest_pb2.TestAllTypes()
        # Only set one path to mask.
        mask = field_mask_pb2.FieldMask()
        mask.paths.append(field_name)
        mask.MergeMessage(src, dst)
        # The expected result message.
        msg = unittest_pb2.TestAllTypes()
        if field.label == descriptor.FieldDescriptor.LABEL_REPEATED:
          repeated_src = getattr(src, field_name)
          repeated_msg = getattr(msg, field_name)
          if field.cpp_type == descriptor.FieldDescriptor.CPPTYPE_MESSAGE:
            for item in repeated_src:
              repeated_msg.add().CopyFrom(item)
          else:
            repeated_msg.extend(repeated_src)
        elif field.cpp_type == descriptor.FieldDescriptor.CPPTYPE_MESSAGE:
          getattr(msg, field_name).CopyFrom(getattr(src, field_name))
        else:
          setattr(msg, field_name, getattr(src, field_name))
        # Only field specified in mask is merged.
        self.assertEqual(msg, dst)
  
      # Test merge nested fields.
      nested_src = unittest_pb2.NestedTestAllTypes()
      nested_dst = unittest_pb2.NestedTestAllTypes()
      nested_src.child.payload.optional_int32 = 1234
      nested_src.child.child.payload.optional_int32 = 5678
      mask = field_mask_pb2.FieldMask()
      mask.FromJsonString('child.payload')
      mask.MergeMessage(nested_src, nested_dst)
      self.assertEqual(1234, nested_dst.child.payload.optional_int32)
      self.assertEqual(0, nested_dst.child.child.payload.optional_int32)
  
      mask.FromJsonString('child.child.payload')
      mask.MergeMessage(nested_src, nested_dst)
      self.assertEqual(1234, nested_dst.child.payload.optional_int32)
      self.assertEqual(5678, nested_dst.child.child.payload.optional_int32)
  
      nested_dst.Clear()
      mask.FromJsonString('child.child.payload')
      mask.MergeMessage(nested_src, nested_dst)
      self.assertEqual(0, nested_dst.child.payload.optional_int32)
      self.assertEqual(5678, nested_dst.child.child.payload.optional_int32)
  
      nested_dst.Clear()
      mask.FromJsonString('child')
      mask.MergeMessage(nested_src, nested_dst)
      self.assertEqual(1234, nested_dst.child.payload.optional_int32)
      self.assertEqual(5678, nested_dst.child.child.payload.optional_int32)
  
      # Test MergeOptions.
      nested_dst.Clear()
      nested_dst.child.payload.optional_int64 = 4321
      # Message fields will be merged by default.
      mask.FromJsonString('child.payload')
      mask.MergeMessage(nested_src, nested_dst)
      self.assertEqual(1234, nested_dst.child.payload.optional_int32)
      self.assertEqual(4321, nested_dst.child.payload.optional_int64)
      # Change the behavior to replace message fields.
      mask.FromJsonString('child.payload')
      mask.MergeMessage(nested_src, nested_dst, True, False)
      self.assertEqual(1234, nested_dst.child.payload.optional_int32)
      self.assertEqual(0, nested_dst.child.payload.optional_int64)
  
      # By default, fields missing in source are not cleared in destination.
      nested_dst.payload.optional_int32 = 1234
      self.assertTrue(nested_dst.HasField('payload'))
      mask.FromJsonString('payload')
      mask.MergeMessage(nested_src, nested_dst)
      self.assertTrue(nested_dst.HasField('payload'))
      # But they are cleared when replacing message fields.
      nested_dst.Clear()
      nested_dst.payload.optional_int32 = 1234
      mask.FromJsonString('payload')
      mask.MergeMessage(nested_src, nested_dst, True, False)
      self.assertFalse(nested_dst.HasField('payload'))
  
      nested_src.payload.repeated_int32.append(1234)
      nested_dst.payload.repeated_int32.append(5678)
      # Repeated fields will be appended by default.
      mask.FromJsonString('payload.repeated_int32')
      mask.MergeMessage(nested_src, nested_dst)
      self.assertEqual(2, len(nested_dst.payload.repeated_int32))
      self.assertEqual(5678, nested_dst.payload.repeated_int32[0])
      self.assertEqual(1234, nested_dst.payload.repeated_int32[1])
      # Change the behavior to replace repeated fields.
      mask.FromJsonString('payload.repeated_int32')
      mask.MergeMessage(nested_src, nested_dst, False, True)
      self.assertEqual(1, len(nested_dst.payload.repeated_int32))
      self.assertEqual(1234, nested_dst.payload.repeated_int32[0])
  
  
  class StructTest(unittest.TestCase):
  
    def testStruct(self):
      struct = struct_pb2.Struct()
      struct_class = struct.__class__
  
      struct['key1'] = 5
      struct['key2'] = 'abc'
      struct['key3'] = True
      struct.get_or_create_struct('key4')['subkey'] = 11.0
      struct_list = struct.get_or_create_list('key5')
      struct_list.extend([6, 'seven', True, False, None])
      struct_list.add_struct()['subkey2'] = 9
  
      self.assertTrue(isinstance(struct, well_known_types.Struct))
      self.assertEquals(5, struct['key1'])
      self.assertEquals('abc', struct['key2'])
      self.assertIs(True, struct['key3'])
      self.assertEquals(11, struct['key4']['subkey'])
      inner_struct = struct_class()
      inner_struct['subkey2'] = 9
      self.assertEquals([6, 'seven', True, False, None, inner_struct],
                        list(struct['key5'].items()))
  
      serialized = struct.SerializeToString()
  
      struct2 = struct_pb2.Struct()
      struct2.ParseFromString(serialized)
  
      self.assertEquals(struct, struct2)
  
      self.assertTrue(isinstance(struct2, well_known_types.Struct))
      self.assertEquals(5, struct2['key1'])
      self.assertEquals('abc', struct2['key2'])
      self.assertIs(True, struct2['key3'])
      self.assertEquals(11, struct2['key4']['subkey'])
      self.assertEquals([6, 'seven', True, False, None, inner_struct],
                        list(struct2['key5'].items()))
  
      struct_list = struct2['key5']
      self.assertEquals(6, struct_list[0])
      self.assertEquals('seven', struct_list[1])
      self.assertEquals(True, struct_list[2])
      self.assertEquals(False, struct_list[3])
      self.assertEquals(None, struct_list[4])
      self.assertEquals(inner_struct, struct_list[5])
  
      struct_list[1] = 7
      self.assertEquals(7, struct_list[1])
  
      struct_list.add_list().extend([1, 'two', True, False, None])
      self.assertEquals([1, 'two', True, False, None],
                        list(struct_list[6].items()))
  
      text_serialized = str(struct)
      struct3 = struct_pb2.Struct()
      text_format.Merge(text_serialized, struct3)
      self.assertEquals(struct, struct3)
  
      struct.get_or_create_struct('key3')['replace'] = 12
      self.assertEquals(12, struct['key3']['replace'])
  
  
  class AnyTest(unittest.TestCase):
  
    def testAnyMessage(self):
      # Creates and sets message.
      msg = any_test_pb2.TestAny()
      msg_descriptor = msg.DESCRIPTOR
      all_types = unittest_pb2.TestAllTypes()
      all_descriptor = all_types.DESCRIPTOR
      all_types.repeated_string.append('\u00fc\ua71f')
      # Packs to Any.
      msg.value.Pack(all_types)
      self.assertEqual(msg.value.type_url,
                       'type.googleapis.com/%s' % all_descriptor.full_name)
      self.assertEqual(msg.value.value,
                       all_types.SerializeToString())
      # Tests Is() method.
      self.assertTrue(msg.value.Is(all_descriptor))
      self.assertFalse(msg.value.Is(msg_descriptor))
      # Unpacks Any.
      unpacked_message = unittest_pb2.TestAllTypes()
      self.assertTrue(msg.value.Unpack(unpacked_message))
      self.assertEqual(all_types, unpacked_message)
      # Unpacks to different type.
      self.assertFalse(msg.value.Unpack(msg))
      # Only Any messages have Pack method.
      try:
        msg.Pack(all_types)
      except AttributeError:
        pass
      else:
        raise AttributeError('%s should not have Pack method.' %
                             msg_descriptor.full_name)
  
    def testMessageName(self):
      # Creates and sets message.
      submessage = any_test_pb2.TestAny()
      submessage.int_value = 12345
      msg = any_pb2.Any()
      msg.Pack(submessage)
      self.assertEqual(msg.TypeName(), 'google.protobuf.internal.TestAny')
  
    def testPackWithCustomTypeUrl(self):
      submessage = any_test_pb2.TestAny()
      submessage.int_value = 12345
      msg = any_pb2.Any()
      # Pack with a custom type URL prefix.
      msg.Pack(submessage, 'type.myservice.com')
      self.assertEqual(msg.type_url,
                       'type.myservice.com/%s' % submessage.DESCRIPTOR.full_name)
      # Pack with a custom type URL prefix ending with '/'.
      msg.Pack(submessage, 'type.myservice.com/')
      self.assertEqual(msg.type_url,
                       'type.myservice.com/%s' % submessage.DESCRIPTOR.full_name)
      # Pack with an empty type URL prefix.
      msg.Pack(submessage, '')
      self.assertEqual(msg.type_url,
                       '/%s' % submessage.DESCRIPTOR.full_name)
      # Test unpacking the type.
      unpacked_message = any_test_pb2.TestAny()
      self.assertTrue(msg.Unpack(unpacked_message))
      self.assertEqual(submessage, unpacked_message)
  
  
  if __name__ == '__main__':
    unittest.main()
  