python类SerializeToString()的实例源码

def testNegativeInfinity(self, message_module):
    if message_module is unittest_pb2:
      golden_data = (b'\x5D\x00\x00\x80\xFF'
                     b'\x61\x00\x00\x00\x00\x00\x00\xF0\xFF'
                     b'\xCD\x02\x00\x00\x80\xFF'
                     b'\xD1\x02\x00\x00\x00\x00\x00\x00\xF0\xFF')
    else:
      golden_data = (b'\x5D\x00\x00\x80\xFF'
                     b'\x61\x00\x00\x00\x00\x00\x00\xF0\xFF'
                     b'\xCA\x02\x04\x00\x00\x80\xFF'
                     b'\xD2\x02\x08\x00\x00\x00\x00\x00\x00\xF0\xFF')

    golden_message = message_module.TestAllTypes()
    golden_message.ParseFromString(golden_data)
    self.assertTrue(IsNegInf(golden_message.optional_float))
    self.assertTrue(IsNegInf(golden_message.optional_double))
    self.assertTrue(IsNegInf(golden_message.repeated_float[0]))
    self.assertTrue(IsNegInf(golden_message.repeated_double[0]))
    self.assertEqual(golden_data, golden_message.SerializeToString())

def testNotANumber(self, message_module):
    golden_data = (b'\x5D\x00\x00\xC0\x7F'
                   b'\x61\x00\x00\x00\x00\x00\x00\xF8\x7F'
                   b'\xCD\x02\x00\x00\xC0\x7F'
                   b'\xD1\x02\x00\x00\x00\x00\x00\x00\xF8\x7F')
    golden_message = message_module.TestAllTypes()
    golden_message.ParseFromString(golden_data)
    self.assertTrue(isnan(golden_message.optional_float))
    self.assertTrue(isnan(golden_message.optional_double))
    self.assertTrue(isnan(golden_message.repeated_float[0]))
    self.assertTrue(isnan(golden_message.repeated_double[0]))

    # The protocol buffer may serialize to any one of multiple different
    # representations of a NaN.  Rather than verify a specific representation,
    # verify the serialized string can be converted into a correctly
    # behaving protocol buffer.
    serialized = golden_message.SerializeToString()
    message = message_module.TestAllTypes()
    message.ParseFromString(serialized)
    self.assertTrue(isnan(message.optional_float))
    self.assertTrue(isnan(message.optional_double))
    self.assertTrue(isnan(message.repeated_float[0]))
    self.assertTrue(isnan(message.repeated_double[0]))

def testMapAssignmentCausesPresence(self):
    msg = map_unittest_pb2.TestMapSubmessage()
    msg.test_map.map_int32_int32[123] = 456

    serialized = msg.SerializeToString()
    msg2 = map_unittest_pb2.TestMapSubmessage()
    msg2.ParseFromString(serialized)

    self.assertEqual(msg, msg2)

    # Now test that various mutations of the map properly invalidate the
    # cached size of the submessage.
    msg.test_map.map_int32_int32[888] = 999
    serialized = msg.SerializeToString()
    msg2.ParseFromString(serialized)
    self.assertEqual(msg, msg2)

    msg.test_map.map_int32_int32.clear()
    serialized = msg.SerializeToString()
    msg2.ParseFromString(serialized)
    self.assertEqual(msg, msg2)

def testPackedFields(self):
    message = packed_field_test_pb2.TestPackedTypes()
    self.setMessage(message)
    golden_data = (b'\x0A\x01\x01'
                   b'\x12\x01\x01'
                   b'\x1A\x01\x01'
                   b'\x22\x01\x01'
                   b'\x2A\x01\x02'
                   b'\x32\x01\x02'
                   b'\x3A\x04\x01\x00\x00\x00'
                   b'\x42\x08\x01\x00\x00\x00\x00\x00\x00\x00'
                   b'\x4A\x04\x01\x00\x00\x00'
                   b'\x52\x08\x01\x00\x00\x00\x00\x00\x00\x00'
                   b'\x5A\x04\x00\x00\x80\x3f'
                   b'\x62\x08\x00\x00\x00\x00\x00\x00\xf0\x3f'
                   b'\x6A\x01\x01'
                   b'\x72\x01\x01')
    self.assertEqual(golden_data, message.SerializeToString())

def testUnpackedFields(self):
    message = packed_field_test_pb2.TestUnpackedTypes()
    self.setMessage(message)
    golden_data = (b'\x08\x01'
                   b'\x10\x01'
                   b'\x18\x01'
                   b'\x20\x01'
                   b'\x28\x02'
                   b'\x30\x02'
                   b'\x3D\x01\x00\x00\x00'
                   b'\x41\x01\x00\x00\x00\x00\x00\x00\x00'
                   b'\x4D\x01\x00\x00\x00'
                   b'\x51\x01\x00\x00\x00\x00\x00\x00\x00'
                   b'\x5D\x00\x00\x80\x3f'
                   b'\x61\x00\x00\x00\x00\x00\x00\xf0\x3f'
                   b'\x68\x01'
                   b'\x70\x01')
    self.assertEqual(golden_data, message.SerializeToString())

def testGoldenPackedMessage(self, message_module):
    golden_data = test_util.GoldenFileData('golden_packed_fields_message')
    golden_message = message_module.TestPackedTypes()
    golden_message.ParseFromString(golden_data)
    all_set = message_module.TestPackedTypes()
    test_util.SetAllPackedFields(all_set)
    self.assertEqual(all_set, golden_message)
    self.assertEqual(golden_data, all_set.SerializeToString())
    golden_copy = copy.deepcopy(golden_message)
    self.assertEqual(golden_data, golden_copy.SerializeToString())

def testNegativeInfinityPacked(self, message_module):
    golden_data = (b'\xA2\x06\x04\x00\x00\x80\xFF'
                   b'\xAA\x06\x08\x00\x00\x00\x00\x00\x00\xF0\xFF')
    golden_message = message_module.TestPackedTypes()
    golden_message.ParseFromString(golden_data)
    self.assertTrue(IsNegInf(golden_message.packed_float[0]))
    self.assertTrue(IsNegInf(golden_message.packed_double[0]))
    self.assertEqual(golden_data, golden_message.SerializeToString())

def testNotANumberPacked(self, message_module):
    golden_data = (b'\xA2\x06\x04\x00\x00\xC0\x7F'
                   b'\xAA\x06\x08\x00\x00\x00\x00\x00\x00\xF8\x7F')
    golden_message = message_module.TestPackedTypes()
    golden_message.ParseFromString(golden_data)
    self.assertTrue(isnan(golden_message.packed_float[0]))
    self.assertTrue(isnan(golden_message.packed_double[0]))

    serialized = golden_message.SerializeToString()
    message = message_module.TestPackedTypes()
    message.ParseFromString(serialized)
    self.assertTrue(isnan(message.packed_float[0]))
    self.assertTrue(isnan(message.packed_double[0]))

def testUnknownFieldPrinting(self, message_module):
    populated = message_module.TestAllTypes()
    test_util.SetAllNonLazyFields(populated)
    empty = message_module.TestEmptyMessage()
    empty.ParseFromString(populated.SerializeToString())
    self.assertEqual(str(empty), '')

def testOneofDeserialize(self, message_module):
    m = message_module.TestAllTypes()
    m.oneof_uint32 = 11
    m2 = message_module.TestAllTypes()
    m2.ParseFromString(m.SerializeToString())
    self.assertEqual('oneof_uint32', m2.WhichOneof('oneof_field'))

def testGoldenExtensions(self):
    golden_data = test_util.GoldenFileData('golden_message')
    golden_message = unittest_pb2.TestAllExtensions()
    golden_message.ParseFromString(golden_data)
    all_set = unittest_pb2.TestAllExtensions()
    test_util.SetAllExtensions(all_set)
    self.assertEqual(all_set, golden_message)
    self.assertEqual(golden_data, golden_message.SerializeToString())
    golden_copy = copy.deepcopy(golden_message)
    self.assertEqual(golden_data, golden_copy.SerializeToString())

def testGoldenPackedExtensions(self):
    golden_data = test_util.GoldenFileData('golden_packed_fields_message')
    golden_message = unittest_pb2.TestPackedExtensions()
    golden_message.ParseFromString(golden_data)
    all_set = unittest_pb2.TestPackedExtensions()
    test_util.SetAllPackedExtensions(all_set)
    self.assertEqual(all_set, golden_message)
    self.assertEqual(golden_data, all_set.SerializeToString())
    golden_copy = copy.deepcopy(golden_message)
    self.assertEqual(golden_data, golden_copy.SerializeToString())

def testPickleIncompleteProto(self):
    golden_message = unittest_pb2.TestRequired(a=1)
    pickled_message = pickle.dumps(golden_message)

    unpickled_message = pickle.loads(pickled_message)
    self.assertEqual(unpickled_message, golden_message)
    self.assertEqual(unpickled_message.a, 1)
    # This is still an incomplete proto - so serializing should fail
    self.assertRaises(message.EncodeError, unpickled_message.SerializeToString)


  # TODO(haberman): this isn't really a proto2-specific test except that this
  # message has a required field in it.  Should probably be factored out so
  # that we can test the other parts with proto3.

def testMessageMap(self):
    msg = map_unittest_pb2.TestMap()

    self.assertEqual(0, len(msg.map_int32_foreign_message))
    self.assertFalse(5 in msg.map_int32_foreign_message)

    msg.map_int32_foreign_message[123]
    # get_or_create() is an alias for getitem.
    msg.map_int32_foreign_message.get_or_create(-456)

    self.assertEqual(2, len(msg.map_int32_foreign_message))
    self.assertIn(123, msg.map_int32_foreign_message)
    self.assertIn(-456, msg.map_int32_foreign_message)
    self.assertEqual(2, len(msg.map_int32_foreign_message))

    # Bad key.
    with self.assertRaises(TypeError):
      msg.map_int32_foreign_message['123']

    # Can't assign directly to submessage.
    with self.assertRaises(ValueError):
      msg.map_int32_foreign_message[999] = msg.map_int32_foreign_message[123]

    # Verify that trying to assign a bad key doesn't actually add a member to
    # the map.
    self.assertEqual(2, len(msg.map_int32_foreign_message))

    serialized = msg.SerializeToString()
    msg2 = map_unittest_pb2.TestMap()
    msg2.ParseFromString(serialized)

    self.assertEqual(2, len(msg2.map_int32_foreign_message))
    self.assertIn(123, msg2.map_int32_foreign_message)
    self.assertIn(-456, msg2.map_int32_foreign_message)
    self.assertEqual(2, len(msg2.map_int32_foreign_message))

def testIntegerMapWithLongs(self):
    msg = map_unittest_pb2.TestMap()
    msg.map_int32_int32[long(-123)] = long(-456)
    msg.map_int64_int64[long(-2**33)] = long(-2**34)
    msg.map_uint32_uint32[long(123)] = long(456)
    msg.map_uint64_uint64[long(2**33)] = long(2**34)

    serialized = msg.SerializeToString()
    msg2 = map_unittest_pb2.TestMap()
    msg2.ParseFromString(serialized)

    self.assertEqual(-456, msg2.map_int32_int32[-123])
    self.assertEqual(-2**34, msg2.map_int64_int64[-2**33])
    self.assertEqual(456, msg2.map_uint32_uint32[123])
    self.assertEqual(2**34, msg2.map_uint64_uint64[2**33])

def testMapAssignmentCausesPresenceForSubmessages(self):
    msg = map_unittest_pb2.TestMapSubmessage()
    msg.test_map.map_int32_foreign_message[123].c = 5

    serialized = msg.SerializeToString()
    msg2 = map_unittest_pb2.TestMapSubmessage()
    msg2.ParseFromString(serialized)

    self.assertEqual(msg, msg2)

    # Now test that various mutations of the map properly invalidate the
    # cached size of the submessage.
    msg.test_map.map_int32_foreign_message[888].c = 7
    serialized = msg.SerializeToString()
    msg2.ParseFromString(serialized)
    self.assertEqual(msg, msg2)

    msg.test_map.map_int32_foreign_message[888].MergeFrom(
        msg.test_map.map_int32_foreign_message[123])
    serialized = msg.SerializeToString()
    msg2.ParseFromString(serialized)
    self.assertEqual(msg, msg2)

    msg.test_map.map_int32_foreign_message.clear()
    serialized = msg.SerializeToString()
    msg2.ParseFromString(serialized)
    self.assertEqual(msg, msg2)

def testMapValidAfterFieldCleared(self):
    # Map needs to work even if field is cleared.
    # For the C++ implementation this tests the correctness of
    # ScalarMapContainer::Release()
    msg = map_unittest_pb2.TestMap()
    int32_map = msg.map_int32_int32

    int32_map[2] = 4
    int32_map[3] = 6
    int32_map[4] = 8

    msg.ClearField('map_int32_int32')
    self.assertEqual(b'', msg.SerializeToString())
    matching_dict = {2: 4, 3: 6, 4: 8}
    self.assertMapIterEquals(int32_map.items(), matching_dict)

def testMessageMapValidAfterFieldCleared(self):
    # Map needs to work even if field is cleared.
    # For the C++ implementation this tests the correctness of
    # ScalarMapContainer::Release()
    msg = map_unittest_pb2.TestMap()
    int32_foreign_message = msg.map_int32_foreign_message

    int32_foreign_message[2].c = 5

    msg.ClearField('map_int32_foreign_message')
    self.assertEqual(b'', msg.SerializeToString())
    self.assertTrue(2 in int32_foreign_message.keys())

def setUp(self):
    self.p = self.proto_cls()
    self.p.field.payload = 'c' * (1024 * 1024 * 64 + 1)
    self.p_serialized = self.p.SerializeToString()

def testGoldenPackedMessage(self, message_module):
    golden_data = test_util.GoldenFileData('golden_packed_fields_message')
    golden_message = message_module.TestPackedTypes()
    golden_message.ParseFromString(golden_data)
    all_set = message_module.TestPackedTypes()
    test_util.SetAllPackedFields(all_set)
    self.assertEqual(all_set, golden_message)
    self.assertEqual(golden_data, all_set.SerializeToString())
    golden_copy = copy.deepcopy(golden_message)
    self.assertEqual(golden_data, golden_copy.SerializeToString())