| # -*- coding: utf-8 -*- |
| # Protocol Buffers - Google's data interchange format |
| # Copyright 2008 Google Inc. All rights reserved. |
| # |
| # Use of this source code is governed by a BSD-style |
| # license that can be found in the LICENSE file or at |
| # https://developers.google.com/open-source/licenses/bsd |
| |
| """Tests python protocol buffers against the golden message. |
| |
| Note that the golden messages exercise every known field type, thus this |
| test ends up exercising and verifying nearly all of the parsing and |
| serialization code in the whole library. |
| """ |
| |
| __author__ = 'gps@google.com (Gregory P. Smith)' |
| |
| import collections |
| import copy |
| import math |
| import operator |
| import pickle |
| import pydoc |
| import sys |
| import types |
| import unittest |
| from unittest import mock |
| import warnings |
| |
| cmp = lambda x, y: (x > y) - (x < y) |
| |
| from google.protobuf.internal import api_implementation # pylint: disable=g-import-not-at-top |
| from google.protobuf.internal import encoder |
| from google.protobuf.internal import enum_type_wrapper |
| from google.protobuf.internal import more_extensions_pb2 |
| from google.protobuf.internal import more_messages_pb2 |
| from google.protobuf.internal import packed_field_test_pb2 |
| from google.protobuf.internal import test_proto3_optional_pb2 |
| from google.protobuf.internal import test_util |
| from google.protobuf.internal import testing_refleaks |
| from google.protobuf import descriptor |
| from google.protobuf import message |
| from google.protobuf.internal import _parameterized |
| from google.protobuf import map_proto2_unittest_pb2 |
| from google.protobuf import map_unittest_pb2 |
| from google.protobuf import unittest_pb2 |
| from google.protobuf import unittest_proto3_arena_pb2 |
| |
| UCS2_MAXUNICODE = 65535 |
| |
| warnings.simplefilter('error', DeprecationWarning) |
| |
| @_parameterized.named_parameters(('_proto2', unittest_pb2), |
| ('_proto3', unittest_proto3_arena_pb2)) |
| @testing_refleaks.TestCase |
| class MessageTest(unittest.TestCase): |
| |
| def testBadUtf8String(self, message_module): |
| if api_implementation.Type() != 'python': |
| self.skipTest('Skipping testBadUtf8String, currently only the python ' |
| 'api implementation raises UnicodeDecodeError when a ' |
| 'string field contains bad utf-8.') |
| bad_utf8_data = test_util.GoldenFileData('bad_utf8_string') |
| with self.assertRaises(UnicodeDecodeError) as context: |
| message_module.TestAllTypes.FromString(bad_utf8_data) |
| self.assertIn('TestAllTypes.optional_string', str(context.exception)) |
| |
| def testGoldenMessage(self, message_module): |
| # Proto3 doesn't have the "default_foo" members or foreign enums, |
| # and doesn't preserve unknown fields, so for proto3 we use a golden |
| # message that doesn't have these fields set. |
| if message_module is unittest_pb2: |
| golden_data = test_util.GoldenFileData('golden_message_oneof_implemented') |
| else: |
| golden_data = test_util.GoldenFileData('golden_message_proto3') |
| |
| golden_message = message_module.TestAllTypes() |
| golden_message.ParseFromString(golden_data) |
| if message_module is unittest_pb2: |
| test_util.ExpectAllFieldsSet(self, golden_message) |
| self.assertEqual(golden_data, golden_message.SerializeToString()) |
| golden_copy = copy.deepcopy(golden_message) |
| self.assertEqual(golden_data, golden_copy.SerializeToString()) |
| |
| def testGoldenMessageBytearray(self, message_module): |
| # bytearray was broken, test that it works again |
| if message_module is unittest_pb2: |
| golden_data = test_util.GoldenFileData('golden_message_oneof_implemented') |
| else: |
| golden_data = test_util.GoldenFileData('golden_message_proto3') |
| |
| golden_message = message_module.TestAllTypes() |
| golden_message.ParseFromString(bytearray(golden_data)) |
| if message_module is unittest_pb2: |
| test_util.ExpectAllFieldsSet(self, golden_message) |
| self.assertEqual(golden_data, golden_message.SerializeToString()) |
| |
| def testGoldenPackedMessage(self, message_module): |
| golden_data = test_util.GoldenFileData('golden_packed_fields_message') |
| golden_message = message_module.TestPackedTypes() |
| parsed_bytes = golden_message.ParseFromString(golden_data) |
| all_set = message_module.TestPackedTypes() |
| test_util.SetAllPackedFields(all_set) |
| self.assertEqual(parsed_bytes, len(golden_data)) |
| 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 testParseErrors(self, message_module): |
| msg = message_module.TestAllTypes() |
| self.assertRaises(TypeError, msg.FromString, 0) |
| self.assertRaises(Exception, msg.FromString, '0') |
| # TODO: Fix cpp extension to raise error instead of warning. |
| # b/27494216 |
| end_tag = encoder.TagBytes(1, 4) |
| if (api_implementation.Type() == 'python' or |
| api_implementation.Type() == 'upb'): |
| with self.assertRaises(message.DecodeError) as context: |
| msg.FromString(end_tag) |
| if api_implementation.Type() == 'python': |
| # Only pure-Python has an error message this specific. |
| self.assertEqual('Unexpected end-group tag.', str(context.exception)) |
| |
| # Field number 0 is illegal. |
| self.assertRaises(message.DecodeError, msg.FromString, b'\3\4') |
| |
| def testDeterminismParameters(self, message_module): |
| # This message is always deterministically serialized, even if determinism |
| # is disabled, so we can use it to verify that all the determinism |
| # parameters work correctly. |
| golden_data = (b'\xe2\x02\nOne string' |
| b'\xe2\x02\nTwo string' |
| b'\xe2\x02\nRed string' |
| b'\xe2\x02\x0bBlue string') |
| golden_message = message_module.TestAllTypes() |
| golden_message.repeated_string.extend([ |
| 'One string', |
| 'Two string', |
| 'Red string', |
| 'Blue string', |
| ]) |
| self.assertEqual(golden_data, |
| golden_message.SerializeToString(deterministic=None)) |
| self.assertEqual(golden_data, |
| golden_message.SerializeToString(deterministic=False)) |
| self.assertEqual(golden_data, |
| golden_message.SerializeToString(deterministic=True)) |
| |
| class BadArgError(Exception): |
| pass |
| |
| class BadArg(object): |
| |
| def __nonzero__(self): |
| raise BadArgError() |
| |
| def __bool__(self): |
| raise BadArgError() |
| |
| with self.assertRaises(BadArgError): |
| golden_message.SerializeToString(deterministic=BadArg()) |
| |
| def testPickleSupport(self, message_module): |
| golden_data = test_util.GoldenFileData('golden_message') |
| golden_message = message_module.TestAllTypes() |
| golden_message.ParseFromString(golden_data) |
| pickled_message = pickle.dumps(golden_message) |
| |
| unpickled_message = pickle.loads(pickled_message) |
| self.assertEqual(unpickled_message, golden_message) |
| |
| def testPickleNestedMessage(self, message_module): |
| golden_message = message_module.TestPickleNestedMessage.NestedMessage(bb=1) |
| pickled_message = pickle.dumps(golden_message) |
| unpickled_message = pickle.loads(pickled_message) |
| self.assertEqual(unpickled_message, golden_message) |
| |
| def testPickleNestedNestedMessage(self, message_module): |
| cls = message_module.TestPickleNestedMessage.NestedMessage |
| golden_message = cls.NestedNestedMessage(cc=1) |
| pickled_message = pickle.dumps(golden_message) |
| unpickled_message = pickle.loads(pickled_message) |
| self.assertEqual(unpickled_message, golden_message) |
| |
| def testPositiveInfinity(self, message_module): |
| if message_module is unittest_pb2: |
| golden_data = (b'\x5D\x00\x00\x80\x7F' |
| b'\x61\x00\x00\x00\x00\x00\x00\xF0\x7F' |
| b'\xCD\x02\x00\x00\x80\x7F' |
| b'\xD1\x02\x00\x00\x00\x00\x00\x00\xF0\x7F') |
| else: |
| golden_data = (b'\x5D\x00\x00\x80\x7F' |
| b'\x61\x00\x00\x00\x00\x00\x00\xF0\x7F' |
| b'\xCA\x02\x04\x00\x00\x80\x7F' |
| b'\xD2\x02\x08\x00\x00\x00\x00\x00\x00\xF0\x7F') |
| |
| golden_message = message_module.TestAllTypes() |
| golden_message.ParseFromString(golden_data) |
| self.assertEqual(golden_message.optional_float, math.inf) |
| self.assertEqual(golden_message.optional_double, math.inf) |
| self.assertEqual(golden_message.repeated_float[0], math.inf) |
| self.assertEqual(golden_message.repeated_double[0], math.inf) |
| self.assertEqual(golden_data, golden_message.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.assertEqual(golden_message.optional_float, -math.inf) |
| self.assertEqual(golden_message.optional_double, -math.inf) |
| self.assertEqual(golden_message.repeated_float[0], -math.inf) |
| self.assertEqual(golden_message.repeated_double[0], -math.inf) |
| 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(math.isnan(golden_message.optional_float)) |
| self.assertTrue(math.isnan(golden_message.optional_double)) |
| self.assertTrue(math.isnan(golden_message.repeated_float[0])) |
| self.assertTrue(math.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(math.isnan(message.optional_float)) |
| self.assertTrue(math.isnan(message.optional_double)) |
| self.assertTrue(math.isnan(message.repeated_float[0])) |
| self.assertTrue(math.isnan(message.repeated_double[0])) |
| |
| def testPositiveInfinityPacked(self, message_module): |
| golden_data = (b'\xA2\x06\x04\x00\x00\x80\x7F' |
| b'\xAA\x06\x08\x00\x00\x00\x00\x00\x00\xF0\x7F') |
| golden_message = message_module.TestPackedTypes() |
| golden_message.ParseFromString(golden_data) |
| self.assertEqual(golden_message.packed_float[0], math.inf) |
| self.assertEqual(golden_message.packed_double[0], math.inf) |
| self.assertEqual(golden_data, golden_message.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.assertEqual(golden_message.packed_float[0], -math.inf) |
| self.assertEqual(golden_message.packed_double[0], -math.inf) |
| 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(math.isnan(golden_message.packed_float[0])) |
| self.assertTrue(math.isnan(golden_message.packed_double[0])) |
| |
| serialized = golden_message.SerializeToString() |
| message = message_module.TestPackedTypes() |
| message.ParseFromString(serialized) |
| self.assertTrue(math.isnan(message.packed_float[0])) |
| self.assertTrue(math.isnan(message.packed_double[0])) |
| |
| def testExtremeFloatValues(self, message_module): |
| message = message_module.TestAllTypes() |
| |
| # Most positive exponent, no significand bits set. |
| kMostPosExponentNoSigBits = math.pow(2, 127) |
| message.optional_float = kMostPosExponentNoSigBits |
| message.ParseFromString(message.SerializeToString()) |
| self.assertTrue(message.optional_float == kMostPosExponentNoSigBits) |
| |
| # Most positive exponent, one significand bit set. |
| kMostPosExponentOneSigBit = 1.5 * math.pow(2, 127) |
| message.optional_float = kMostPosExponentOneSigBit |
| message.ParseFromString(message.SerializeToString()) |
| self.assertTrue(message.optional_float == kMostPosExponentOneSigBit) |
| |
| # Repeat last two cases with values of same magnitude, but negative. |
| message.optional_float = -kMostPosExponentNoSigBits |
| message.ParseFromString(message.SerializeToString()) |
| self.assertTrue(message.optional_float == -kMostPosExponentNoSigBits) |
| |
| message.optional_float = -kMostPosExponentOneSigBit |
| message.ParseFromString(message.SerializeToString()) |
| self.assertTrue(message.optional_float == -kMostPosExponentOneSigBit) |
| |
| # Most negative exponent, no significand bits set. |
| kMostNegExponentNoSigBits = math.pow(2, -127) |
| message.optional_float = kMostNegExponentNoSigBits |
| message.ParseFromString(message.SerializeToString()) |
| self.assertTrue(message.optional_float == kMostNegExponentNoSigBits) |
| |
| # Most negative exponent, one significand bit set. |
| kMostNegExponentOneSigBit = 1.5 * math.pow(2, -127) |
| message.optional_float = kMostNegExponentOneSigBit |
| message.ParseFromString(message.SerializeToString()) |
| self.assertTrue(message.optional_float == kMostNegExponentOneSigBit) |
| |
| # Repeat last two cases with values of the same magnitude, but negative. |
| message.optional_float = -kMostNegExponentNoSigBits |
| message.ParseFromString(message.SerializeToString()) |
| self.assertTrue(message.optional_float == -kMostNegExponentNoSigBits) |
| |
| message.optional_float = -kMostNegExponentOneSigBit |
| message.ParseFromString(message.SerializeToString()) |
| self.assertTrue(message.optional_float == -kMostNegExponentOneSigBit) |
| |
| # Max 4 bytes float value |
| max_float = float.fromhex('0x1.fffffep+127') |
| message.optional_float = max_float |
| self.assertAlmostEqual(message.optional_float, max_float) |
| serialized_data = message.SerializeToString() |
| message.ParseFromString(serialized_data) |
| self.assertAlmostEqual(message.optional_float, max_float) |
| |
| # Test set double to float field. |
| message.optional_float = 3.4028235e+39 |
| self.assertEqual(message.optional_float, float('inf')) |
| serialized_data = message.SerializeToString() |
| message.ParseFromString(serialized_data) |
| self.assertEqual(message.optional_float, float('inf')) |
| |
| message.optional_float = -3.4028235e+39 |
| self.assertEqual(message.optional_float, float('-inf')) |
| |
| message.optional_float = 1.4028235e-39 |
| self.assertAlmostEqual(message.optional_float, 1.4028235e-39) |
| |
| def testExtremeDoubleValues(self, message_module): |
| message = message_module.TestAllTypes() |
| |
| # Most positive exponent, no significand bits set. |
| kMostPosExponentNoSigBits = math.pow(2, 1023) |
| message.optional_double = kMostPosExponentNoSigBits |
| message.ParseFromString(message.SerializeToString()) |
| self.assertTrue(message.optional_double == kMostPosExponentNoSigBits) |
| |
| # Most positive exponent, one significand bit set. |
| kMostPosExponentOneSigBit = 1.5 * math.pow(2, 1023) |
| message.optional_double = kMostPosExponentOneSigBit |
| message.ParseFromString(message.SerializeToString()) |
| self.assertTrue(message.optional_double == kMostPosExponentOneSigBit) |
| |
| # Repeat last two cases with values of same magnitude, but negative. |
| message.optional_double = -kMostPosExponentNoSigBits |
| message.ParseFromString(message.SerializeToString()) |
| self.assertTrue(message.optional_double == -kMostPosExponentNoSigBits) |
| |
| message.optional_double = -kMostPosExponentOneSigBit |
| message.ParseFromString(message.SerializeToString()) |
| self.assertTrue(message.optional_double == -kMostPosExponentOneSigBit) |
| |
| # Most negative exponent, no significand bits set. |
| kMostNegExponentNoSigBits = math.pow(2, -1023) |
| message.optional_double = kMostNegExponentNoSigBits |
| message.ParseFromString(message.SerializeToString()) |
| self.assertTrue(message.optional_double == kMostNegExponentNoSigBits) |
| |
| # Most negative exponent, one significand bit set. |
| kMostNegExponentOneSigBit = 1.5 * math.pow(2, -1023) |
| message.optional_double = kMostNegExponentOneSigBit |
| message.ParseFromString(message.SerializeToString()) |
| self.assertTrue(message.optional_double == kMostNegExponentOneSigBit) |
| |
| # Repeat last two cases with values of the same magnitude, but negative. |
| message.optional_double = -kMostNegExponentNoSigBits |
| message.ParseFromString(message.SerializeToString()) |
| self.assertTrue(message.optional_double == -kMostNegExponentNoSigBits) |
| |
| message.optional_double = -kMostNegExponentOneSigBit |
| message.ParseFromString(message.SerializeToString()) |
| self.assertTrue(message.optional_double == -kMostNegExponentOneSigBit) |
| |
| def testFloatPrinting(self, message_module): |
| message = message_module.TestAllTypes() |
| message.optional_float = 2.0 |
| self.assertEqual(str(message), 'optional_float: 2.0\n') |
| |
| def testFloatNanPrinting(self, message_module): |
| message = message_module.TestAllTypes() |
| message.optional_float = float('nan') |
| self.assertEqual(str(message), 'optional_float: nan\n') |
| |
| def testHighPrecisionFloatPrinting(self, message_module): |
| msg = message_module.TestAllTypes() |
| msg.optional_float = 0.12345678912345678 |
| old_float = msg.optional_float |
| msg.ParseFromString(msg.SerializeToString()) |
| self.assertEqual(old_float, msg.optional_float) |
| |
| def testDoubleNanPrinting(self, message_module): |
| message = message_module.TestAllTypes() |
| message.optional_double = float('nan') |
| self.assertEqual(str(message), 'optional_double: nan\n') |
| |
| def testHighPrecisionDoublePrinting(self, message_module): |
| msg = message_module.TestAllTypes() |
| msg.optional_double = 0.12345678912345678 |
| self.assertEqual(str(msg), 'optional_double: 0.12345678912345678\n') |
| |
| 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 testCopyFromEmpty(self, message_module): |
| msg = message_module.NestedTestAllTypes() |
| test_msg = message_module.NestedTestAllTypes() |
| test_util.SetAllFields(test_msg.payload) |
| self.assertTrue(test_msg.HasField('payload')) |
| # Copy from empty message |
| test_msg.CopyFrom(msg) |
| self.assertEqual(0, len(test_msg.ListFields())) |
| |
| test_util.SetAllFields(test_msg.payload) |
| self.assertTrue(test_msg.HasField('payload')) |
| # Copy from a non exist message |
| test_msg.CopyFrom(msg.child) |
| self.assertFalse(test_msg.HasField('payload')) |
| self.assertEqual(0, len(test_msg.ListFields())) |
| |
| def testAppendRepeatedCompositeField(self, message_module): |
| msg = message_module.TestAllTypes() |
| msg.repeated_nested_message.append( |
| message_module.TestAllTypes.NestedMessage(bb=1)) |
| nested = message_module.TestAllTypes.NestedMessage(bb=2) |
| msg.repeated_nested_message.append(nested) |
| try: |
| msg.repeated_nested_message.append(1) |
| except TypeError: |
| pass |
| self.assertEqual(2, len(msg.repeated_nested_message)) |
| self.assertEqual([1, 2], [m.bb for m in msg.repeated_nested_message]) |
| |
| def testInsertRepeatedCompositeField(self, message_module): |
| msg = message_module.TestAllTypes() |
| msg.repeated_nested_message.insert( |
| -1, message_module.TestAllTypes.NestedMessage(bb=1)) |
| sub_msg = msg.repeated_nested_message[0] |
| msg.repeated_nested_message.insert( |
| 0, message_module.TestAllTypes.NestedMessage(bb=2)) |
| msg.repeated_nested_message.insert( |
| 99, message_module.TestAllTypes.NestedMessage(bb=3)) |
| msg.repeated_nested_message.insert( |
| -2, message_module.TestAllTypes.NestedMessage(bb=-1)) |
| msg.repeated_nested_message.insert( |
| -1000, message_module.TestAllTypes.NestedMessage(bb=-1000)) |
| try: |
| msg.repeated_nested_message.insert(1, 999) |
| except TypeError: |
| pass |
| self.assertEqual(5, len(msg.repeated_nested_message)) |
| self.assertEqual([-1000, 2, -1, 1, 3], |
| [m.bb for m in msg.repeated_nested_message]) |
| self.assertEqual( |
| str(msg), 'repeated_nested_message {\n' |
| ' bb: -1000\n' |
| '}\n' |
| 'repeated_nested_message {\n' |
| ' bb: 2\n' |
| '}\n' |
| 'repeated_nested_message {\n' |
| ' bb: -1\n' |
| '}\n' |
| 'repeated_nested_message {\n' |
| ' bb: 1\n' |
| '}\n' |
| 'repeated_nested_message {\n' |
| ' bb: 3\n' |
| '}\n') |
| self.assertEqual(sub_msg.bb, 1) |
| |
| def testAssignRepeatedField(self, message_module): |
| msg = message_module.NestedTestAllTypes() |
| msg.payload.repeated_int32[:] = [1, 2, 3, 4] |
| self.assertEqual(4, len(msg.payload.repeated_int32)) |
| self.assertEqual([1, 2, 3, 4], msg.payload.repeated_int32) |
| |
| def testMergeFromRepeatedField(self, message_module): |
| msg = message_module.TestAllTypes() |
| msg.repeated_int32.append(1) |
| msg.repeated_int32.append(3) |
| msg.repeated_nested_message.add(bb=1) |
| msg.repeated_nested_message.add(bb=2) |
| other_msg = message_module.TestAllTypes() |
| other_msg.repeated_nested_message.add(bb=3) |
| other_msg.repeated_nested_message.add(bb=4) |
| other_msg.repeated_int32.append(5) |
| other_msg.repeated_int32.append(7) |
| |
| msg.repeated_int32.MergeFrom(other_msg.repeated_int32) |
| self.assertEqual(4, len(msg.repeated_int32)) |
| |
| msg.repeated_nested_message.MergeFrom(other_msg.repeated_nested_message) |
| self.assertEqual([1, 2, 3, 4], [m.bb for m in msg.repeated_nested_message]) |
| |
| def testInternalMergeWithMissingRequiredField(self, message_module): |
| req = more_messages_pb2.RequiredField() |
| more_messages_pb2.RequiredWrapper(request=req) |
| |
| def testMergeFromMissingRequiredField(self, message_module): |
| msg = more_messages_pb2.RequiredField() |
| message = more_messages_pb2.RequiredField() |
| message.MergeFrom(msg) |
| self.assertEqual(msg, message) |
| |
| def testAddWrongRepeatedNestedField(self, message_module): |
| msg = message_module.TestAllTypes() |
| try: |
| msg.repeated_nested_message.add('wrong') |
| except TypeError: |
| pass |
| try: |
| msg.repeated_nested_message.add(value_field='wrong') |
| except ValueError: |
| pass |
| self.assertEqual(len(msg.repeated_nested_message), 0) |
| |
| def testRepeatedContains(self, message_module): |
| msg = message_module.TestAllTypes() |
| msg.repeated_int32.extend([1, 2, 3]) |
| self.assertIn(2, msg.repeated_int32) |
| self.assertNotIn(0, msg.repeated_int32) |
| |
| msg.repeated_nested_message.add(bb=1) |
| sub_msg1 = msg.repeated_nested_message[0] |
| sub_msg2 = message_module.TestAllTypes.NestedMessage(bb=2) |
| sub_msg3 = message_module.TestAllTypes.NestedMessage(bb=3) |
| msg.repeated_nested_message.append(sub_msg2) |
| msg.repeated_nested_message.insert(0, sub_msg3) |
| self.assertIn(sub_msg1, msg.repeated_nested_message) |
| self.assertIn(sub_msg2, msg.repeated_nested_message) |
| self.assertIn(sub_msg3, msg.repeated_nested_message) |
| |
| def testRepeatedScalarIterable(self, message_module): |
| msg = message_module.TestAllTypes() |
| msg.repeated_int32.extend([1, 2, 3]) |
| add = 0 |
| for item in msg.repeated_int32: |
| add += item |
| self.assertEqual(add, 6) |
| |
| def testRepeatedNestedFieldIteration(self, message_module): |
| msg = message_module.TestAllTypes() |
| msg.repeated_nested_message.add(bb=1) |
| msg.repeated_nested_message.add(bb=2) |
| msg.repeated_nested_message.add(bb=3) |
| msg.repeated_nested_message.add(bb=4) |
| |
| self.assertEqual([1, 2, 3, 4], [m.bb for m in msg.repeated_nested_message]) |
| self.assertEqual([4, 3, 2, 1], |
| [m.bb for m in reversed(msg.repeated_nested_message)]) |
| self.assertEqual([4, 3, 2, 1], |
| [m.bb for m in msg.repeated_nested_message[::-1]]) |
| |
| def testSortEmptyRepeated(self, message_module): |
| message = message_module.NestedTestAllTypes() |
| self.assertFalse(message.HasField('child')) |
| self.assertFalse(message.HasField('payload')) |
| message.child.repeated_child.sort() |
| message.payload.repeated_int32.sort() |
| self.assertFalse(message.HasField('child')) |
| self.assertFalse(message.HasField('payload')) |
| |
| def testSortingRepeatedScalarFieldsDefaultComparator(self, message_module): |
| """Check some different types with the default comparator.""" |
| message = message_module.TestAllTypes() |
| |
| # TODO: would testing more scalar types strengthen test? |
| message.repeated_int32.append(1) |
| message.repeated_int32.append(3) |
| message.repeated_int32.append(2) |
| message.repeated_int32.sort() |
| self.assertEqual(message.repeated_int32[0], 1) |
| self.assertEqual(message.repeated_int32[1], 2) |
| self.assertEqual(message.repeated_int32[2], 3) |
| self.assertEqual(str(message.repeated_int32), str([1, 2, 3])) |
| |
| message.repeated_float.append(1.1) |
| message.repeated_float.append(1.3) |
| message.repeated_float.append(1.2) |
| message.repeated_float.sort() |
| self.assertAlmostEqual(message.repeated_float[0], 1.1) |
| self.assertAlmostEqual(message.repeated_float[1], 1.2) |
| self.assertAlmostEqual(message.repeated_float[2], 1.3) |
| |
| message.repeated_string.append('a') |
| message.repeated_string.append('c') |
| message.repeated_string.append('b') |
| message.repeated_string.sort() |
| self.assertEqual(message.repeated_string[0], 'a') |
| self.assertEqual(message.repeated_string[1], 'b') |
| self.assertEqual(message.repeated_string[2], 'c') |
| self.assertEqual(str(message.repeated_string), str([u'a', u'b', u'c'])) |
| |
| message.repeated_bytes.append(b'a') |
| message.repeated_bytes.append(b'c') |
| message.repeated_bytes.append(b'b') |
| message.repeated_bytes.sort() |
| self.assertEqual(message.repeated_bytes[0], b'a') |
| self.assertEqual(message.repeated_bytes[1], b'b') |
| self.assertEqual(message.repeated_bytes[2], b'c') |
| self.assertEqual(str(message.repeated_bytes), str([b'a', b'b', b'c'])) |
| |
| def testSortingRepeatedScalarFieldsCustomComparator(self, message_module): |
| """Check some different types with custom comparator.""" |
| message = message_module.TestAllTypes() |
| |
| message.repeated_int32.append(-3) |
| message.repeated_int32.append(-2) |
| message.repeated_int32.append(-1) |
| message.repeated_int32.sort(key=abs) |
| self.assertEqual(message.repeated_int32[0], -1) |
| self.assertEqual(message.repeated_int32[1], -2) |
| self.assertEqual(message.repeated_int32[2], -3) |
| |
| message.repeated_string.append('aaa') |
| message.repeated_string.append('bb') |
| message.repeated_string.append('c') |
| message.repeated_string.sort(key=len) |
| self.assertEqual(message.repeated_string[0], 'c') |
| self.assertEqual(message.repeated_string[1], 'bb') |
| self.assertEqual(message.repeated_string[2], 'aaa') |
| |
| def testSortingRepeatedCompositeFieldsCustomComparator(self, message_module): |
| """Check passing a custom comparator to sort a repeated composite field.""" |
| message = message_module.TestAllTypes() |
| |
| message.repeated_nested_message.add().bb = 1 |
| message.repeated_nested_message.add().bb = 3 |
| message.repeated_nested_message.add().bb = 2 |
| message.repeated_nested_message.add().bb = 6 |
| message.repeated_nested_message.add().bb = 5 |
| message.repeated_nested_message.add().bb = 4 |
| message.repeated_nested_message.sort(key=operator.attrgetter('bb')) |
| self.assertEqual(message.repeated_nested_message[0].bb, 1) |
| self.assertEqual(message.repeated_nested_message[1].bb, 2) |
| self.assertEqual(message.repeated_nested_message[2].bb, 3) |
| self.assertEqual(message.repeated_nested_message[3].bb, 4) |
| self.assertEqual(message.repeated_nested_message[4].bb, 5) |
| self.assertEqual(message.repeated_nested_message[5].bb, 6) |
| self.assertEqual( |
| str(message.repeated_nested_message), |
| '[bb: 1\n, bb: 2\n, bb: 3\n, bb: 4\n, bb: 5\n, bb: 6\n]') |
| |
| def testSortingRepeatedCompositeFieldsStable(self, message_module): |
| """Check passing a custom comparator to sort a repeated composite field.""" |
| message = message_module.TestAllTypes() |
| |
| message.repeated_nested_message.add().bb = 21 |
| message.repeated_nested_message.add().bb = 20 |
| message.repeated_nested_message.add().bb = 13 |
| message.repeated_nested_message.add().bb = 33 |
| message.repeated_nested_message.add().bb = 11 |
| message.repeated_nested_message.add().bb = 24 |
| message.repeated_nested_message.add().bb = 10 |
| message.repeated_nested_message.sort(key=lambda z: z.bb // 10) |
| self.assertEqual([13, 11, 10, 21, 20, 24, 33], |
| [n.bb for n in message.repeated_nested_message]) |
| |
| # Make sure that for the C++ implementation, the underlying fields |
| # are actually reordered. |
| pb = message.SerializeToString() |
| message.Clear() |
| message.MergeFromString(pb) |
| self.assertEqual([13, 11, 10, 21, 20, 24, 33], |
| [n.bb for n in message.repeated_nested_message]) |
| |
| def testRepeatedCompositeFieldSortArguments(self, message_module): |
| """Check sorting a repeated composite field using list.sort() arguments.""" |
| message = message_module.TestAllTypes() |
| |
| get_bb = operator.attrgetter('bb') |
| message.repeated_nested_message.add().bb = 1 |
| message.repeated_nested_message.add().bb = 3 |
| message.repeated_nested_message.add().bb = 2 |
| message.repeated_nested_message.add().bb = 6 |
| message.repeated_nested_message.add().bb = 5 |
| message.repeated_nested_message.add().bb = 4 |
| message.repeated_nested_message.sort(key=get_bb) |
| self.assertEqual([k.bb for k in message.repeated_nested_message], |
| [1, 2, 3, 4, 5, 6]) |
| message.repeated_nested_message.sort(key=get_bb, reverse=True) |
| self.assertEqual([k.bb for k in message.repeated_nested_message], |
| [6, 5, 4, 3, 2, 1]) |
| |
| def testRepeatedScalarFieldSortArguments(self, message_module): |
| """Check sorting a scalar field using list.sort() arguments.""" |
| message = message_module.TestAllTypes() |
| |
| message.repeated_int32.append(-3) |
| message.repeated_int32.append(-2) |
| message.repeated_int32.append(-1) |
| message.repeated_int32.sort(key=abs) |
| self.assertEqual(list(message.repeated_int32), [-1, -2, -3]) |
| message.repeated_int32.sort(key=abs, reverse=True) |
| self.assertEqual(list(message.repeated_int32), [-3, -2, -1]) |
| |
| message.repeated_string.append('aaa') |
| message.repeated_string.append('bb') |
| message.repeated_string.append('c') |
| message.repeated_string.sort(key=len) |
| self.assertEqual(list(message.repeated_string), ['c', 'bb', 'aaa']) |
| message.repeated_string.sort(key=len, reverse=True) |
| self.assertEqual(list(message.repeated_string), ['aaa', 'bb', 'c']) |
| |
| def testRepeatedFieldsComparable(self, message_module): |
| m1 = message_module.TestAllTypes() |
| m2 = message_module.TestAllTypes() |
| m1.repeated_int32.append(0) |
| m1.repeated_int32.append(1) |
| m1.repeated_int32.append(2) |
| m2.repeated_int32.append(0) |
| m2.repeated_int32.append(1) |
| m2.repeated_int32.append(2) |
| m1.repeated_nested_message.add().bb = 1 |
| m1.repeated_nested_message.add().bb = 2 |
| m1.repeated_nested_message.add().bb = 3 |
| m2.repeated_nested_message.add().bb = 1 |
| m2.repeated_nested_message.add().bb = 2 |
| m2.repeated_nested_message.add().bb = 3 |
| |
| def testRepeatedFieldsAreSequences(self, message_module): |
| m = message_module.TestAllTypes() |
| self.assertIsInstance(m.repeated_int32, collections.abc.MutableSequence) |
| self.assertIsInstance(m.repeated_nested_message, |
| collections.abc.MutableSequence) |
| |
| def testRepeatedFieldsNotHashable(self, message_module): |
| m = message_module.TestAllTypes() |
| with self.assertRaises(TypeError): |
| hash(m.repeated_int32) |
| with self.assertRaises(TypeError): |
| hash(m.repeated_nested_message) |
| |
| def testRepeatedFieldInsideNestedMessage(self, message_module): |
| m = message_module.NestedTestAllTypes() |
| m.payload.repeated_int32.extend([]) |
| self.assertTrue(m.HasField('payload')) |
| |
| def testMergeFrom(self, message_module): |
| m1 = message_module.TestAllTypes() |
| m2 = message_module.TestAllTypes() |
| # Cpp extension will lazily create a sub message which is immutable. |
| nested = m1.optional_nested_message |
| self.assertEqual(0, nested.bb) |
| m2.optional_nested_message.bb = 1 |
| # Make sure cmessage pointing to a mutable message after merge instead of |
| # the lazily created message. |
| m1.MergeFrom(m2) |
| self.assertEqual(1, nested.bb) |
| |
| # Test more nested sub message. |
| msg1 = message_module.NestedTestAllTypes() |
| msg2 = message_module.NestedTestAllTypes() |
| nested = msg1.child.payload.optional_nested_message |
| self.assertEqual(0, nested.bb) |
| msg2.child.payload.optional_nested_message.bb = 1 |
| msg1.MergeFrom(msg2) |
| self.assertEqual(1, nested.bb) |
| |
| # Test repeated field. |
| self.assertEqual(msg1.payload.repeated_nested_message, |
| msg1.payload.repeated_nested_message) |
| nested = msg2.payload.repeated_nested_message.add() |
| nested.bb = 1 |
| msg1.MergeFrom(msg2) |
| self.assertEqual(1, len(msg1.payload.repeated_nested_message)) |
| self.assertEqual(1, nested.bb) |
| |
| def testMergeFromString(self, message_module): |
| m1 = message_module.TestAllTypes() |
| m2 = message_module.TestAllTypes() |
| # Cpp extension will lazily create a sub message which is immutable. |
| self.assertEqual(0, m1.optional_nested_message.bb) |
| m2.optional_nested_message.bb = 1 |
| # Make sure cmessage pointing to a mutable message after merge instead of |
| # the lazily created message. |
| m1.MergeFromString(m2.SerializeToString()) |
| self.assertEqual(1, m1.optional_nested_message.bb) |
| |
| def testMergeFromStringUsingMemoryView(self, message_module): |
| m2 = message_module.TestAllTypes() |
| m2.optional_string = 'scalar string' |
| m2.repeated_string.append('repeated string') |
| m2.optional_bytes = b'scalar bytes' |
| m2.repeated_bytes.append(b'repeated bytes') |
| |
| serialized = m2.SerializeToString() |
| memview = memoryview(serialized) |
| m1 = message_module.TestAllTypes.FromString(memview) |
| |
| self.assertEqual(m1.optional_bytes, b'scalar bytes') |
| self.assertEqual(m1.repeated_bytes, [b'repeated bytes']) |
| self.assertEqual(m1.optional_string, 'scalar string') |
| self.assertEqual(m1.repeated_string, ['repeated string']) |
| # Make sure that the memoryview was correctly converted to bytes, and |
| # that a sub-sliced memoryview is not being used. |
| self.assertIsInstance(m1.optional_bytes, bytes) |
| self.assertIsInstance(m1.repeated_bytes[0], bytes) |
| self.assertIsInstance(m1.optional_string, str) |
| self.assertIsInstance(m1.repeated_string[0], str) |
| |
| def testMergeFromEmpty(self, message_module): |
| m1 = message_module.TestAllTypes() |
| # Cpp extension will lazily create a sub message which is immutable. |
| self.assertEqual(0, m1.optional_nested_message.bb) |
| self.assertFalse(m1.HasField('optional_nested_message')) |
| # Make sure the sub message is still immutable after merge from empty. |
| m1.MergeFromString(b'') # field state should not change |
| self.assertFalse(m1.HasField('optional_nested_message')) |
| |
| def ensureNestedMessageExists(self, msg, attribute): |
| """Make sure that a nested message object exists. |
| |
| As soon as a nested message attribute is accessed, it will be present in the |
| _fields dict, without being marked as actually being set. |
| """ |
| getattr(msg, attribute) |
| self.assertFalse(msg.HasField(attribute)) |
| |
| def testOneofGetCaseNonexistingField(self, message_module): |
| m = message_module.TestAllTypes() |
| self.assertRaises(ValueError, m.WhichOneof, 'no_such_oneof_field') |
| self.assertRaises(Exception, m.WhichOneof, 0) |
| |
| def testOneofDefaultValues(self, message_module): |
| m = message_module.TestAllTypes() |
| self.assertIs(None, m.WhichOneof('oneof_field')) |
| self.assertFalse(m.HasField('oneof_field')) |
| self.assertFalse(m.HasField('oneof_uint32')) |
| |
| # Oneof is set even when setting it to a default value. |
| m.oneof_uint32 = 0 |
| self.assertEqual('oneof_uint32', m.WhichOneof('oneof_field')) |
| self.assertTrue(m.HasField('oneof_field')) |
| self.assertTrue(m.HasField('oneof_uint32')) |
| self.assertFalse(m.HasField('oneof_string')) |
| |
| m.oneof_string = '' |
| self.assertEqual('oneof_string', m.WhichOneof('oneof_field')) |
| self.assertTrue(m.HasField('oneof_string')) |
| self.assertFalse(m.HasField('oneof_uint32')) |
| |
| def testOneofSemantics(self, message_module): |
| m = message_module.TestAllTypes() |
| self.assertIs(None, m.WhichOneof('oneof_field')) |
| |
| m.oneof_uint32 = 11 |
| self.assertEqual('oneof_uint32', m.WhichOneof('oneof_field')) |
| self.assertTrue(m.HasField('oneof_uint32')) |
| |
| m.oneof_string = u'foo' |
| self.assertEqual('oneof_string', m.WhichOneof('oneof_field')) |
| self.assertFalse(m.HasField('oneof_uint32')) |
| self.assertTrue(m.HasField('oneof_string')) |
| |
| # Read nested message accessor without accessing submessage. |
| m.oneof_nested_message |
| self.assertEqual('oneof_string', m.WhichOneof('oneof_field')) |
| self.assertTrue(m.HasField('oneof_string')) |
| self.assertFalse(m.HasField('oneof_nested_message')) |
| |
| # Read accessor of nested message without accessing submessage. |
| m.oneof_nested_message.bb |
| self.assertEqual('oneof_string', m.WhichOneof('oneof_field')) |
| self.assertTrue(m.HasField('oneof_string')) |
| self.assertFalse(m.HasField('oneof_nested_message')) |
| |
| m.oneof_nested_message.bb = 11 |
| self.assertEqual('oneof_nested_message', m.WhichOneof('oneof_field')) |
| self.assertFalse(m.HasField('oneof_string')) |
| self.assertTrue(m.HasField('oneof_nested_message')) |
| |
| m.oneof_bytes = b'bb' |
| self.assertEqual('oneof_bytes', m.WhichOneof('oneof_field')) |
| self.assertFalse(m.HasField('oneof_nested_message')) |
| self.assertTrue(m.HasField('oneof_bytes')) |
| |
| def testOneofCompositeFieldReadAccess(self, message_module): |
| m = message_module.TestAllTypes() |
| m.oneof_uint32 = 11 |
| |
| self.ensureNestedMessageExists(m, 'oneof_nested_message') |
| self.assertEqual('oneof_uint32', m.WhichOneof('oneof_field')) |
| self.assertEqual(11, m.oneof_uint32) |
| |
| def testOneofWhichOneof(self, message_module): |
| m = message_module.TestAllTypes() |
| self.assertIs(None, m.WhichOneof('oneof_field')) |
| if message_module is unittest_pb2: |
| self.assertFalse(m.HasField('oneof_field')) |
| |
| m.oneof_uint32 = 11 |
| self.assertEqual('oneof_uint32', m.WhichOneof('oneof_field')) |
| if message_module is unittest_pb2: |
| self.assertTrue(m.HasField('oneof_field')) |
| |
| m.oneof_bytes = b'bb' |
| self.assertEqual('oneof_bytes', m.WhichOneof('oneof_field')) |
| |
| m.ClearField('oneof_bytes') |
| self.assertIs(None, m.WhichOneof('oneof_field')) |
| if message_module is unittest_pb2: |
| self.assertFalse(m.HasField('oneof_field')) |
| |
| def testOneofClearField(self, message_module): |
| m = message_module.TestAllTypes() |
| m.ClearField('oneof_field') |
| m.oneof_uint32 = 11 |
| m.ClearField('oneof_field') |
| if message_module is unittest_pb2: |
| self.assertFalse(m.HasField('oneof_field')) |
| self.assertFalse(m.HasField('oneof_uint32')) |
| self.assertIs(None, m.WhichOneof('oneof_field')) |
| |
| def testOneofClearSetField(self, message_module): |
| m = message_module.TestAllTypes() |
| m.oneof_uint32 = 11 |
| m.ClearField('oneof_uint32') |
| if message_module is unittest_pb2: |
| self.assertFalse(m.HasField('oneof_field')) |
| self.assertFalse(m.HasField('oneof_uint32')) |
| self.assertIs(None, m.WhichOneof('oneof_field')) |
| |
| def testOneofClearUnsetField(self, message_module): |
| m = message_module.TestAllTypes() |
| m.oneof_uint32 = 11 |
| self.ensureNestedMessageExists(m, 'oneof_nested_message') |
| m.ClearField('oneof_nested_message') |
| self.assertEqual(11, m.oneof_uint32) |
| if message_module is unittest_pb2: |
| self.assertTrue(m.HasField('oneof_field')) |
| self.assertTrue(m.HasField('oneof_uint32')) |
| self.assertEqual('oneof_uint32', m.WhichOneof('oneof_field')) |
| |
| 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 testOneofCopyFrom(self, message_module): |
| m = message_module.TestAllTypes() |
| m.oneof_uint32 = 11 |
| m2 = message_module.TestAllTypes() |
| m2.CopyFrom(m) |
| self.assertEqual('oneof_uint32', m2.WhichOneof('oneof_field')) |
| |
| def testOneofNestedMergeFrom(self, message_module): |
| m = message_module.NestedTestAllTypes() |
| m.payload.oneof_uint32 = 11 |
| m2 = message_module.NestedTestAllTypes() |
| m2.payload.oneof_bytes = b'bb' |
| m2.child.payload.oneof_bytes = b'bb' |
| m2.MergeFrom(m) |
| self.assertEqual('oneof_uint32', m2.payload.WhichOneof('oneof_field')) |
| self.assertEqual('oneof_bytes', m2.child.payload.WhichOneof('oneof_field')) |
| |
| def testOneofMessageMergeFrom(self, message_module): |
| m = message_module.NestedTestAllTypes() |
| m.payload.oneof_nested_message.bb = 11 |
| m.child.payload.oneof_nested_message.bb = 12 |
| m2 = message_module.NestedTestAllTypes() |
| m2.payload.oneof_uint32 = 13 |
| m2.MergeFrom(m) |
| self.assertEqual('oneof_nested_message', |
| m2.payload.WhichOneof('oneof_field')) |
| self.assertEqual('oneof_nested_message', |
| m2.child.payload.WhichOneof('oneof_field')) |
| |
| def testOneofNestedMessageInit(self, message_module): |
| m = message_module.TestAllTypes( |
| oneof_nested_message=message_module.TestAllTypes.NestedMessage()) |
| self.assertEqual('oneof_nested_message', m.WhichOneof('oneof_field')) |
| |
| def testOneofClear(self, message_module): |
| m = message_module.TestAllTypes() |
| m.oneof_uint32 = 11 |
| m.Clear() |
| self.assertIsNone(m.WhichOneof('oneof_field')) |
| m.oneof_bytes = b'bb' |
| self.assertEqual('oneof_bytes', m.WhichOneof('oneof_field')) |
| |
| def testAssignByteStringToUnicodeField(self, message_module): |
| """Assigning a byte string to a string field should result |
| |
| in the value being converted to a Unicode string. |
| """ |
| m = message_module.TestAllTypes() |
| m.optional_string = str('') |
| self.assertIsInstance(m.optional_string, str) |
| |
| def testLongValuedSlice(self, message_module): |
| """It should be possible to use int-valued indices in slices. |
| |
| This didn't used to work in the v2 C++ implementation. |
| """ |
| m = message_module.TestAllTypes() |
| |
| # Repeated scalar |
| m.repeated_int32.append(1) |
| sl = m.repeated_int32[int(0):int(len(m.repeated_int32))] |
| self.assertEqual(len(m.repeated_int32), len(sl)) |
| |
| # Repeated composite |
| m.repeated_nested_message.add().bb = 3 |
| sl = m.repeated_nested_message[int(0):int(len(m.repeated_nested_message))] |
| self.assertEqual(len(m.repeated_nested_message), len(sl)) |
| |
| def testExtendShouldNotSwallowExceptions(self, message_module): |
| """This didn't use to work in the v2 C++ implementation.""" |
| m = message_module.TestAllTypes() |
| with self.assertRaises(NameError) as _: |
| m.repeated_int32.extend(a for i in range(10)) # pylint: disable=undefined-variable |
| with self.assertRaises(NameError) as _: |
| m.repeated_nested_enum.extend(a for i in range(10)) # pylint: disable=undefined-variable |
| |
| FALSY_VALUES = [None, False, 0, 0.0] |
| EMPTY_VALUES = [b'', u'', bytearray(), [], {}, set()] |
| |
| def testExtendInt32WithNothing(self, message_module): |
| """Test no-ops extending repeated int32 fields.""" |
| m = message_module.TestAllTypes() |
| self.assertSequenceEqual([], m.repeated_int32) |
| |
| for falsy_value in MessageTest.FALSY_VALUES: |
| with self.assertRaises(TypeError) as context: |
| m.repeated_int32.extend(falsy_value) |
| self.assertIn('iterable', str(context.exception)) |
| self.assertSequenceEqual([], m.repeated_int32) |
| |
| for empty_value in MessageTest.EMPTY_VALUES: |
| m.repeated_int32.extend(empty_value) |
| self.assertSequenceEqual([], m.repeated_int32) |
| |
| def testExtendFloatWithNothing(self, message_module): |
| """Test no-ops extending repeated float fields.""" |
| m = message_module.TestAllTypes() |
| self.assertSequenceEqual([], m.repeated_float) |
| |
| for falsy_value in MessageTest.FALSY_VALUES: |
| with self.assertRaises(TypeError) as context: |
| m.repeated_float.extend(falsy_value) |
| self.assertIn('iterable', str(context.exception)) |
| self.assertSequenceEqual([], m.repeated_float) |
| |
| for empty_value in MessageTest.EMPTY_VALUES: |
| m.repeated_float.extend(empty_value) |
| self.assertSequenceEqual([], m.repeated_float) |
| |
| def testExtendStringWithNothing(self, message_module): |
| """Test no-ops extending repeated string fields.""" |
| m = message_module.TestAllTypes() |
| self.assertSequenceEqual([], m.repeated_string) |
| |
| for falsy_value in MessageTest.FALSY_VALUES: |
| with self.assertRaises(TypeError) as context: |
| m.repeated_string.extend(falsy_value) |
| self.assertIn('iterable', str(context.exception)) |
| self.assertSequenceEqual([], m.repeated_string) |
| |
| for empty_value in MessageTest.EMPTY_VALUES: |
| m.repeated_string.extend(empty_value) |
| self.assertSequenceEqual([], m.repeated_string) |
| |
| def testExtendInt32WithPythonList(self, message_module): |
| """Test extending repeated int32 fields with python lists.""" |
| m = message_module.TestAllTypes() |
| self.assertSequenceEqual([], m.repeated_int32) |
| m.repeated_int32.extend([0]) |
| self.assertSequenceEqual([0], m.repeated_int32) |
| m.repeated_int32.extend([1, 2]) |
| self.assertSequenceEqual([0, 1, 2], m.repeated_int32) |
| m.repeated_int32.extend([3, 4]) |
| self.assertSequenceEqual([0, 1, 2, 3, 4], m.repeated_int32) |
| |
| def testExtendFloatWithPythonList(self, message_module): |
| """Test extending repeated float fields with python lists.""" |
| m = message_module.TestAllTypes() |
| self.assertSequenceEqual([], m.repeated_float) |
| m.repeated_float.extend([0.0]) |
| self.assertSequenceEqual([0.0], m.repeated_float) |
| m.repeated_float.extend([1.0, 2.0]) |
| self.assertSequenceEqual([0.0, 1.0, 2.0], m.repeated_float) |
| m.repeated_float.extend([3.0, 4.0]) |
| self.assertSequenceEqual([0.0, 1.0, 2.0, 3.0, 4.0], m.repeated_float) |
| |
| def testExtendStringWithPythonList(self, message_module): |
| """Test extending repeated string fields with python lists.""" |
| m = message_module.TestAllTypes() |
| self.assertSequenceEqual([], m.repeated_string) |
| m.repeated_string.extend(['']) |
| self.assertSequenceEqual([''], m.repeated_string) |
| m.repeated_string.extend(['11', '22']) |
| self.assertSequenceEqual(['', '11', '22'], m.repeated_string) |
| m.repeated_string.extend(['33', '44']) |
| self.assertSequenceEqual(['', '11', '22', '33', '44'], m.repeated_string) |
| |
| def testExtendStringWithString(self, message_module): |
| """Test extending repeated string fields with characters from a string.""" |
| m = message_module.TestAllTypes() |
| self.assertSequenceEqual([], m.repeated_string) |
| m.repeated_string.extend('abc') |
| self.assertSequenceEqual(['a', 'b', 'c'], m.repeated_string) |
| |
| class TestIterable(object): |
| """This iterable object mimics the behavior of numpy.array. |
| |
| __nonzero__ fails for length > 1, and returns bool(item[0]) for length == 1. |
| |
| """ |
| |
| def __init__(self, values=None): |
| self._list = values or [] |
| |
| def __nonzero__(self): |
| size = len(self._list) |
| if size == 0: |
| return False |
| if size == 1: |
| return bool(self._list[0]) |
| raise ValueError('Truth value is ambiguous.') |
| |
| def __len__(self): |
| return len(self._list) |
| |
| def __iter__(self): |
| return self._list.__iter__() |
| |
| def testExtendInt32WithIterable(self, message_module): |
| """Test extending repeated int32 fields with iterable.""" |
| m = message_module.TestAllTypes() |
| self.assertSequenceEqual([], m.repeated_int32) |
| m.repeated_int32.extend(MessageTest.TestIterable([])) |
| self.assertSequenceEqual([], m.repeated_int32) |
| m.repeated_int32.extend(MessageTest.TestIterable([0])) |
| self.assertSequenceEqual([0], m.repeated_int32) |
| m.repeated_int32.extend(MessageTest.TestIterable([1, 2])) |
| self.assertSequenceEqual([0, 1, 2], m.repeated_int32) |
| m.repeated_int32.extend(MessageTest.TestIterable([3, 4])) |
| self.assertSequenceEqual([0, 1, 2, 3, 4], m.repeated_int32) |
| |
| def testExtendFloatWithIterable(self, message_module): |
| """Test extending repeated float fields with iterable.""" |
| m = message_module.TestAllTypes() |
| self.assertSequenceEqual([], m.repeated_float) |
| m.repeated_float.extend(MessageTest.TestIterable([])) |
| self.assertSequenceEqual([], m.repeated_float) |
| m.repeated_float.extend(MessageTest.TestIterable([0.0])) |
| self.assertSequenceEqual([0.0], m.repeated_float) |
| m.repeated_float.extend(MessageTest.TestIterable([1.0, 2.0])) |
| self.assertSequenceEqual([0.0, 1.0, 2.0], m.repeated_float) |
| m.repeated_float.extend(MessageTest.TestIterable([3.0, 4.0])) |
| self.assertSequenceEqual([0.0, 1.0, 2.0, 3.0, 4.0], m.repeated_float) |
| |
| def testExtendStringWithIterable(self, message_module): |
| """Test extending repeated string fields with iterable.""" |
| m = message_module.TestAllTypes() |
| self.assertSequenceEqual([], m.repeated_string) |
| m.repeated_string.extend(MessageTest.TestIterable([])) |
| self.assertSequenceEqual([], m.repeated_string) |
| m.repeated_string.extend(MessageTest.TestIterable([''])) |
| self.assertSequenceEqual([''], m.repeated_string) |
| m.repeated_string.extend(MessageTest.TestIterable(['1', '2'])) |
| self.assertSequenceEqual(['', '1', '2'], m.repeated_string) |
| m.repeated_string.extend(MessageTest.TestIterable(['3', '4'])) |
| self.assertSequenceEqual(['', '1', '2', '3', '4'], m.repeated_string) |
| |
| class TestIndex(object): |
| """This index object mimics the behavior of numpy.int64 and other types.""" |
| |
| def __init__(self, value=None): |
| self.value = value |
| |
| def __index__(self): |
| return self.value |
| |
| def testRepeatedIndexingWithIntIndex(self, message_module): |
| msg = message_module.TestAllTypes() |
| msg.repeated_int32.extend([1, 2, 3]) |
| self.assertEqual(1, msg.repeated_int32[MessageTest.TestIndex(0)]) |
| |
| def testRepeatedIndexingWithNegative1IntIndex(self, message_module): |
| msg = message_module.TestAllTypes() |
| msg.repeated_int32.extend([1, 2, 3]) |
| self.assertEqual(3, msg.repeated_int32[MessageTest.TestIndex(-1)]) |
| |
| def testRepeatedIndexingWithNegative1Int(self, message_module): |
| msg = message_module.TestAllTypes() |
| msg.repeated_int32.extend([1, 2, 3]) |
| self.assertEqual(3, msg.repeated_int32[-1]) |
| |
| def testPickleRepeatedScalarContainer(self, message_module): |
| # Pickle repeated scalar container is not supported. |
| m = message_module.TestAllTypes() |
| with self.assertRaises(pickle.PickleError) as _: |
| pickle.dumps(m.repeated_int32, pickle.HIGHEST_PROTOCOL) |
| |
| def testSortEmptyRepeatedCompositeContainer(self, message_module): |
| """Exercise a scenario that has led to segfaults in the past.""" |
| m = message_module.TestAllTypes() |
| m.repeated_nested_message.sort() |
| |
| def testHasFieldOnRepeatedField(self, message_module): |
| """Using HasField on a repeated field should raise an exception.""" |
| m = message_module.TestAllTypes() |
| with self.assertRaises(ValueError) as _: |
| m.HasField('repeated_int32') |
| |
| def testRepeatedScalarFieldPop(self, message_module): |
| m = message_module.TestAllTypes() |
| with self.assertRaises(IndexError) as _: |
| m.repeated_int32.pop() |
| m.repeated_int32.extend(range(5)) |
| self.assertEqual(4, m.repeated_int32.pop()) |
| self.assertEqual(0, m.repeated_int32.pop(0)) |
| self.assertEqual(2, m.repeated_int32.pop(1)) |
| self.assertEqual([1, 3], m.repeated_int32) |
| |
| def testRepeatedCompositeFieldPop(self, message_module): |
| m = message_module.TestAllTypes() |
| with self.assertRaises(IndexError) as _: |
| m.repeated_nested_message.pop() |
| with self.assertRaises(TypeError) as _: |
| m.repeated_nested_message.pop('0') |
| for i in range(5): |
| n = m.repeated_nested_message.add() |
| n.bb = i |
| self.assertEqual(4, m.repeated_nested_message.pop().bb) |
| self.assertEqual(0, m.repeated_nested_message.pop(0).bb) |
| self.assertEqual(2, m.repeated_nested_message.pop(1).bb) |
| self.assertEqual([1, 3], [n.bb for n in m.repeated_nested_message]) |
| |
| def testRepeatedCompareWithSelf(self, message_module): |
| m = message_module.TestAllTypes() |
| for i in range(5): |
| m.repeated_int32.insert(i, i) |
| n = m.repeated_nested_message.add() |
| n.bb = i |
| self.assertSequenceEqual(m.repeated_int32, m.repeated_int32) |
| self.assertEqual(m.repeated_nested_message, m.repeated_nested_message) |
| |
| def testReleasedNestedMessages(self, message_module): |
| """A case that lead to a segfault when a message detached from its parent |
| |
| container has itself a child container. |
| """ |
| m = message_module.NestedTestAllTypes() |
| m = m.repeated_child.add() |
| m = m.child |
| m = m.repeated_child.add() |
| self.assertEqual(m.payload.optional_int32, 0) |
| |
| def testSetRepeatedComposite(self, message_module): |
| m = message_module.TestAllTypes() |
| with self.assertRaises(AttributeError): |
| m.repeated_int32 = [] |
| m.repeated_int32.append(1) |
| with self.assertRaises(AttributeError): |
| m.repeated_int32 = [] |
| |
| def testReturningType(self, message_module): |
| m = message_module.TestAllTypes() |
| self.assertEqual(float, type(m.optional_float)) |
| self.assertEqual(float, type(m.optional_double)) |
| self.assertEqual(bool, type(m.optional_bool)) |
| m.optional_float = 1 |
| m.optional_double = 1 |
| m.optional_bool = 1 |
| m.repeated_float.append(1) |
| m.repeated_double.append(1) |
| m.repeated_bool.append(1) |
| m.ParseFromString(m.SerializeToString()) |
| self.assertEqual(float, type(m.optional_float)) |
| self.assertEqual(float, type(m.optional_double)) |
| self.assertEqual('1.0', str(m.optional_double)) |
| self.assertEqual(bool, type(m.optional_bool)) |
| self.assertEqual(float, type(m.repeated_float[0])) |
| self.assertEqual(float, type(m.repeated_double[0])) |
| self.assertEqual(bool, type(m.repeated_bool[0])) |
| self.assertEqual(True, m.repeated_bool[0]) |
| |
| def testEquality(self, message_module): |
| m = message_module.TestAllTypes() |
| m2 = message_module.TestAllTypes() |
| self.assertEqual(m, m) |
| self.assertEqual(m, m2) |
| self.assertEqual(m2, m) |
| |
| different_m = message_module.TestAllTypes() |
| different_m.repeated_float.append(1) |
| self.assertNotEqual(m, different_m) |
| self.assertNotEqual(different_m, m) |
| |
| self.assertIsNotNone(m) |
| self.assertIsNotNone(m) |
| self.assertNotEqual(42, m) |
| self.assertNotEqual(m, 42) |
| self.assertNotEqual('foo', m) |
| self.assertNotEqual(m, 'foo') |
| |
| self.assertEqual(mock.ANY, m) |
| self.assertEqual(m, mock.ANY) |
| |
| class ComparesWithFoo(object): |
| |
| def __eq__(self, other): |
| if getattr(other, 'optional_string', 'not_foo') == 'foo': |
| return True |
| return NotImplemented |
| |
| m.optional_string = 'foo' |
| self.assertEqual(m, ComparesWithFoo()) |
| self.assertEqual(ComparesWithFoo(), m) |
| m.optional_string = 'bar' |
| self.assertNotEqual(m, ComparesWithFoo()) |
| self.assertNotEqual(ComparesWithFoo(), m) |
| |
| def testTypeUnion(self, message_module): |
| # Below python 3.10 you cannot create union types with the | operator, so we |
| # skip testing for unions with old versions. |
| if sys.version_info < (3, 10): |
| return |
| enum_type = enum_type_wrapper.EnumTypeWrapper( |
| message_module.TestAllTypes.NestedEnum.DESCRIPTOR |
| ) |
| union_type = enum_type | int |
| self.assertIsInstance(union_type, types.UnionType) |
| |
| def get_union() -> union_type: |
| return enum_type |
| |
| union = get_union() |
| self.assertIsInstance(union, enum_type_wrapper.EnumTypeWrapper) |
| self.assertEqual( |
| union.DESCRIPTOR, message_module.TestAllTypes.NestedEnum.DESCRIPTOR |
| ) |
| |
| def testIn(self, message_module): |
| m = message_module.TestAllTypes() |
| self.assertNotIn('optional_nested_message', m) |
| self.assertNotIn('oneof_bytes', m) |
| self.assertNotIn('oneof_string', m) |
| with self.assertRaises(ValueError) as e: |
| 'repeated_int32' in m |
| with self.assertRaises(ValueError) as e: |
| 'repeated_nested_message' in m |
| with self.assertRaises(ValueError) as e: |
| 1 in m |
| with self.assertRaises(ValueError) as e: |
| 'not_a_field' in m |
| test_util.SetAllFields(m) |
| self.assertIn('optional_nested_message', m) |
| self.assertIn('oneof_bytes', m) |
| self.assertNotIn('oneof_string', m) |
| |
| |
| # Class to test proto2-only features (required, extensions, etc.) |
| @testing_refleaks.TestCase |
| class Proto2Test(unittest.TestCase): |
| |
| def testFieldPresence(self): |
| message = unittest_pb2.TestAllTypes() |
| |
| self.assertFalse(message.HasField('optional_int32')) |
| self.assertFalse(message.HasField('optional_bool')) |
| self.assertFalse(message.HasField('optional_nested_message')) |
| |
| with self.assertRaises(ValueError): |
| message.HasField('field_doesnt_exist') |
| |
| with self.assertRaises(ValueError): |
| message.HasField('repeated_int32') |
| with self.assertRaises(ValueError): |
| message.HasField('repeated_nested_message') |
| |
| self.assertEqual(0, message.optional_int32) |
| self.assertEqual(False, message.optional_bool) |
| self.assertEqual(0, message.optional_nested_message.bb) |
| |
| # Fields are set even when setting the values to default values. |
| message.optional_int32 = 0 |
| message.optional_bool = False |
| message.optional_nested_message.bb = 0 |
| self.assertTrue(message.HasField('optional_int32')) |
| self.assertTrue(message.HasField('optional_bool')) |
| self.assertTrue(message.HasField('optional_nested_message')) |
| self.assertIn('optional_int32', message) |
| self.assertIn('optional_bool', message) |
| self.assertIn('optional_nested_message', message) |
| |
| # Set the fields to non-default values. |
| message.optional_int32 = 5 |
| message.optional_bool = True |
| message.optional_nested_message.bb = 15 |
| |
| self.assertTrue(message.HasField(u'optional_int32')) |
| self.assertTrue(message.HasField('optional_bool')) |
| self.assertTrue(message.HasField('optional_nested_message')) |
| |
| # Clearing the fields unsets them and resets their value to default. |
| message.ClearField('optional_int32') |
| message.ClearField(u'optional_bool') |
| message.ClearField('optional_nested_message') |
| |
| self.assertFalse(message.HasField('optional_int32')) |
| self.assertFalse(message.HasField('optional_bool')) |
| self.assertFalse(message.HasField('optional_nested_message')) |
| self.assertNotIn('optional_int32', message) |
| self.assertNotIn('optional_bool', message) |
| self.assertNotIn('optional_nested_message', message) |
| self.assertEqual(0, message.optional_int32) |
| self.assertEqual(False, message.optional_bool) |
| self.assertEqual(0, message.optional_nested_message.bb) |
| |
| def testDel(self): |
| msg = unittest_pb2.TestAllTypes() |
| |
| # Fields cannot be deleted. |
| with self.assertRaises(AttributeError): |
| del msg.optional_int32 |
| with self.assertRaises(AttributeError): |
| del msg.optional_bool |
| with self.assertRaises(AttributeError): |
| del msg.repeated_nested_message |
| |
| def testAssignInvalidEnum(self): |
| """Assigning an invalid enum number is not allowed in proto2.""" |
| m = unittest_pb2.TestAllTypes() |
| |
| # Proto2 can not assign unknown enum. |
| with self.assertRaises(ValueError) as _: |
| m.optional_nested_enum = 1234567 |
| self.assertRaises(ValueError, m.repeated_nested_enum.append, 1234567) |
| # Assignment is a different code path than append for the C++ impl. |
| m.repeated_nested_enum.append(2) |
| m.repeated_nested_enum[0] = 2 |
| with self.assertRaises(ValueError): |
| m.repeated_nested_enum[0] = 123456 |
| |
| # Unknown enum value can be parsed but is ignored. |
| m2 = unittest_proto3_arena_pb2.TestAllTypes() |
| m2.optional_nested_enum = 1234567 |
| m2.repeated_nested_enum.append(7654321) |
| serialized = m2.SerializeToString() |
| |
| m3 = unittest_pb2.TestAllTypes() |
| m3.ParseFromString(serialized) |
| self.assertFalse(m3.HasField('optional_nested_enum')) |
| # 1 is the default value for optional_nested_enum. |
| self.assertEqual(1, m3.optional_nested_enum) |
| self.assertEqual(0, len(m3.repeated_nested_enum)) |
| m2.Clear() |
| m2.ParseFromString(m3.SerializeToString()) |
| self.assertEqual(1234567, m2.optional_nested_enum) |
| self.assertEqual(7654321, m2.repeated_nested_enum[0]) |
| |
| def testUnknownEnumMap(self): |
| m = map_proto2_unittest_pb2.TestEnumMap() |
| m.known_map_field[123] = 0 |
| with self.assertRaises(ValueError): |
| m.unknown_map_field[1] = 123 |
| |
| def testDeepCopyClosedEnum(self): |
| m = map_proto2_unittest_pb2.TestEnumMap() |
| m.known_map_field[123] = 0 |
| m2 = copy.deepcopy(m) |
| self.assertEqual(m, m2) |
| |
| def testExtensionsErrors(self): |
| msg = unittest_pb2.TestAllTypes() |
| self.assertRaises(AttributeError, getattr, msg, 'Extensions') |
| |
| def testMergeFromExtensions(self): |
| msg1 = more_extensions_pb2.TopLevelMessage() |
| msg2 = more_extensions_pb2.TopLevelMessage() |
| # Cpp extension will lazily create a sub message which is immutable. |
| self.assertEqual( |
| 0, |
| msg1.submessage.Extensions[more_extensions_pb2.optional_int_extension]) |
| self.assertFalse(msg1.HasField('submessage')) |
| msg2.submessage.Extensions[more_extensions_pb2.optional_int_extension] = 123 |
| # Make sure cmessage and extensions pointing to a mutable message |
| # after merge instead of the lazily created message. |
| msg1.MergeFrom(msg2) |
| self.assertEqual( |
| 123, |
| msg1.submessage.Extensions[more_extensions_pb2.optional_int_extension]) |
| |
| def testCopyFromAll(self): |
| message = unittest_pb2.TestAllTypes() |
| test_util.SetAllFields(message) |
| copy = unittest_pb2.TestAllTypes() |
| copy.CopyFrom(message) |
| self.assertEqual(message, copy) |
| message.repeated_nested_message.add().bb = 123 |
| self.assertNotEqual(message, copy) |
| |
| def testCopyFromAllExtensions(self): |
| all_set = unittest_pb2.TestAllExtensions() |
| test_util.SetAllExtensions(all_set) |
| copy = unittest_pb2.TestAllExtensions() |
| copy.CopyFrom(all_set) |
| self.assertEqual(all_set, copy) |
| all_set.Extensions[unittest_pb2.repeatedgroup_extension].add().a = 321 |
| self.assertNotEqual(all_set, copy) |
| |
| def testCopyFromAllPackedExtensions(self): |
| all_set = unittest_pb2.TestPackedExtensions() |
| test_util.SetAllPackedExtensions(all_set) |
| copy = unittest_pb2.TestPackedExtensions() |
| copy.CopyFrom(all_set) |
| self.assertEqual(all_set, copy) |
| all_set.Extensions[unittest_pb2.packed_float_extension].extend([61.0, 71.0]) |
| self.assertNotEqual(all_set, copy) |
| |
| 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_message, golden_copy) |
| # Depend on a specific serialization order for extensions is not |
| # reasonable to guarantee. |
| if api_implementation.Type() != 'upb': |
| 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_message, golden_copy) |
| # Depend on a specific serialization order for extensions is not |
| # reasonable to guarantee. |
| if api_implementation.Type() != 'upb': |
| 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: 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 testParsingMerge(self): |
| """Check the merge behavior when a required or optional field appears |
| |
| multiple times in the input. |
| """ |
| messages = [ |
| unittest_pb2.TestAllTypes(), |
| unittest_pb2.TestAllTypes(), |
| unittest_pb2.TestAllTypes() |
| ] |
| messages[0].optional_int32 = 1 |
| messages[1].optional_int64 = 2 |
| messages[2].optional_int32 = 3 |
| messages[2].optional_string = 'hello' |
| |
| merged_message = unittest_pb2.TestAllTypes() |
| merged_message.optional_int32 = 3 |
| merged_message.optional_int64 = 2 |
| merged_message.optional_string = 'hello' |
| |
| generator = unittest_pb2.TestParsingMerge.RepeatedFieldsGenerator() |
| generator.field1.extend(messages) |
| generator.field2.extend(messages) |
| generator.field3.extend(messages) |
| generator.ext1.extend(messages) |
| generator.ext2.extend(messages) |
| generator.group1.add().field1.MergeFrom(messages[0]) |
| generator.group1.add().field1.MergeFrom(messages[1]) |
| generator.group1.add().field1.MergeFrom(messages[2]) |
| generator.group2.add().field1.MergeFrom(messages[0]) |
| generator.group2.add().field1.MergeFrom(messages[1]) |
| generator.group2.add().field1.MergeFrom(messages[2]) |
| |
| data = generator.SerializeToString() |
| parsing_merge = unittest_pb2.TestParsingMerge() |
| parsing_merge.ParseFromString(data) |
| |
| # Required and optional fields should be merged. |
| self.assertEqual(parsing_merge.required_all_types, merged_message) |
| self.assertEqual(parsing_merge.optional_all_types, merged_message) |
| self.assertEqual(parsing_merge.optionalgroup.optional_group_all_types, |
| merged_message) |
| self.assertEqual( |
| parsing_merge.Extensions[unittest_pb2.TestParsingMerge.optional_ext], |
| merged_message) |
| |
| # Repeated fields should not be merged. |
| self.assertEqual(len(parsing_merge.repeated_all_types), 3) |
| self.assertEqual(len(parsing_merge.repeatedgroup), 3) |
| self.assertEqual( |
| len(parsing_merge.Extensions[ |
| unittest_pb2.TestParsingMerge.repeated_ext]), 3) |
| |
| def testPythonicInit(self): |
| message = unittest_pb2.TestAllTypes( |
| optional_int32=100, |
| optional_fixed32=200, |
| optional_float=300.5, |
| optional_bytes=b'x', |
| optionalgroup={'a': 400}, |
| optional_nested_message={'bb': 500}, |
| optional_foreign_message={}, |
| optional_nested_enum='BAZ', |
| repeatedgroup=[{ |
| 'a': 600 |
| }, { |
| 'a': 700 |
| }], |
| repeated_nested_enum=['FOO', unittest_pb2.TestAllTypes.BAR], |
| default_int32=800, |
| oneof_string='y') |
| self.assertIsInstance(message, unittest_pb2.TestAllTypes) |
| self.assertEqual(100, message.optional_int32) |
| self.assertEqual(200, message.optional_fixed32) |
| self.assertEqual(300.5, message.optional_float) |
| self.assertEqual(b'x', message.optional_bytes) |
| self.assertEqual(400, message.optionalgroup.a) |
| self.assertIsInstance(message.optional_nested_message, |
| unittest_pb2.TestAllTypes.NestedMessage) |
| self.assertEqual(500, message.optional_nested_message.bb) |
| self.assertTrue(message.HasField('optional_foreign_message')) |
| self.assertEqual(message.optional_foreign_message, |
| unittest_pb2.ForeignMessage()) |
| self.assertEqual(unittest_pb2.TestAllTypes.BAZ, |
| message.optional_nested_enum) |
| self.assertEqual(2, len(message.repeatedgroup)) |
| self.assertEqual(600, message.repeatedgroup[0].a) |
| self.assertEqual(700, message.repeatedgroup[1].a) |
| self.assertEqual(2, len(message.repeated_nested_enum)) |
| self.assertEqual(unittest_pb2.TestAllTypes.FOO, |
| message.repeated_nested_enum[0]) |
| self.assertEqual(unittest_pb2.TestAllTypes.BAR, |
| message.repeated_nested_enum[1]) |
| self.assertEqual(800, message.default_int32) |
| self.assertEqual('y', message.oneof_string) |
| self.assertFalse(message.HasField('optional_int64')) |
| self.assertEqual(0, len(message.repeated_float)) |
| self.assertEqual(42, message.default_int64) |
| |
| message = unittest_pb2.TestAllTypes(optional_nested_enum=u'BAZ') |
| self.assertEqual(unittest_pb2.TestAllTypes.BAZ, |
| message.optional_nested_enum) |
| |
| with self.assertRaises(ValueError): |
| unittest_pb2.TestAllTypes( |
| optional_nested_message={'INVALID_NESTED_FIELD': 17}) |
| |
| with self.assertRaises(TypeError): |
| unittest_pb2.TestAllTypes( |
| optional_nested_message={'bb': 'INVALID_VALUE_TYPE'}) |
| |
| with self.assertRaises(ValueError): |
| unittest_pb2.TestAllTypes(optional_nested_enum='INVALID_LABEL') |
| |
| with self.assertRaises(ValueError): |
| unittest_pb2.TestAllTypes(repeated_nested_enum='FOO') |
| |
| def testPythonicInitWithDict(self): |
| # Both string/unicode field name keys should work. |
| kwargs = { |
| 'optional_int32': 100, |
| u'optional_fixed32': 200, |
| } |
| msg = unittest_pb2.TestAllTypes(**kwargs) |
| self.assertEqual(100, msg.optional_int32) |
| self.assertEqual(200, msg.optional_fixed32) |
| |
| def test_documentation(self): |
| # Also used by the interactive help() function. |
| doc = pydoc.html.document(unittest_pb2.TestAllTypes, 'message') |
| self.assertIn('class TestAllTypes', doc) |
| self.assertIn('SerializePartialToString', doc) |
| self.assertIn('repeated_float', doc) |
| base = unittest_pb2.TestAllTypes.__bases__[0] |
| self.assertRaises(AttributeError, getattr, base, '_extensions_by_name') |
| |
| |
| # Class to test proto3-only features/behavior (updated field presence & enums) |
| @testing_refleaks.TestCase |
| class Proto3Test(unittest.TestCase): |
| |
| # Utility method for comparing equality with a map. |
| def assertMapIterEquals(self, map_iter, dict_value): |
| # Avoid mutating caller's copy. |
| dict_value = dict(dict_value) |
| |
| for k, v in map_iter: |
| self.assertEqual(v, dict_value[k]) |
| del dict_value[k] |
| |
| self.assertEqual({}, dict_value) |
| |
| def testFieldPresence(self): |
| message = unittest_proto3_arena_pb2.TestAllTypes() |
| |
| # We can't test presence of non-repeated, non-submessage fields. |
| with self.assertRaises(ValueError): |
| message.HasField('optional_int32') |
| with self.assertRaises(ValueError): |
| message.HasField('optional_float') |
| with self.assertRaises(ValueError): |
| message.HasField('optional_string') |
| with self.assertRaises(ValueError): |
| message.HasField('optional_bool') |
| |
| # But we can still test presence of submessage fields. |
| self.assertFalse(message.HasField('optional_nested_message')) |
| |
| # As with proto2, we can't test presence of fields that don't exist, or |
| # repeated fields. |
| with self.assertRaises(ValueError): |
| message.HasField('field_doesnt_exist') |
| |
| with self.assertRaises(ValueError): |
| message.HasField('repeated_int32') |
| with self.assertRaises(ValueError): |
| message.HasField('repeated_nested_message') |
| |
| # Can not test "in" operator. |
| with self.assertRaises(ValueError): |
| 'repeated_int32' in message |
| with self.assertRaises(ValueError): |
| 'repeated_nested_message' in message |
| |
| # Fields should default to their type-specific default. |
| self.assertEqual(0, message.optional_int32) |
| self.assertEqual(0, message.optional_float) |
| self.assertEqual('', message.optional_string) |
| self.assertEqual(False, message.optional_bool) |
| self.assertEqual(0, message.optional_nested_message.bb) |
| |
| # Setting a submessage should still return proper presence information. |
| message.optional_nested_message.bb = 0 |
| self.assertTrue(message.HasField('optional_nested_message')) |
| self.assertIn('optional_nested_message', message) |
| |
| # Set the fields to non-default values. |
| message.optional_int32 = 5 |
| message.optional_float = 1.1 |
| message.optional_string = 'abc' |
| message.optional_bool = True |
| message.optional_nested_message.bb = 15 |
| |
| # Clearing the fields unsets them and resets their value to default. |
| message.ClearField('optional_int32') |
| message.ClearField('optional_float') |
| message.ClearField('optional_string') |
| message.ClearField('optional_bool') |
| message.ClearField('optional_nested_message') |
| |
| self.assertEqual(0, message.optional_int32) |
| self.assertEqual(0, message.optional_float) |
| self.assertEqual('', message.optional_string) |
| self.assertEqual(False, message.optional_bool) |
| self.assertEqual(0, message.optional_nested_message.bb) |
| |
| def testProto3ParserDropDefaultScalar(self): |
| message_proto2 = unittest_pb2.TestAllTypes() |
| message_proto2.optional_int32 = 0 |
| message_proto2.optional_string = '' |
| message_proto2.optional_bytes = b'' |
| self.assertEqual(len(message_proto2.ListFields()), 3) |
| |
| message_proto3 = unittest_proto3_arena_pb2.TestAllTypes() |
| message_proto3.ParseFromString(message_proto2.SerializeToString()) |
| self.assertEqual(len(message_proto3.ListFields()), 0) |
| |
| def testProto3Optional(self): |
| msg = test_proto3_optional_pb2.TestProto3Optional() |
| self.assertFalse(msg.HasField('optional_int32')) |
| self.assertFalse(msg.HasField('optional_float')) |
| self.assertFalse(msg.HasField('optional_string')) |
| self.assertFalse(msg.HasField('optional_nested_message')) |
| self.assertFalse(msg.optional_nested_message.HasField('bb')) |
| |
| # Set fields. |
| msg.optional_int32 = 1 |
| msg.optional_float = 1.0 |
| msg.optional_string = '123' |
| msg.optional_nested_message.bb = 1 |
| self.assertTrue(msg.HasField('optional_int32')) |
| self.assertTrue(msg.HasField('optional_float')) |
| self.assertTrue(msg.HasField('optional_string')) |
| self.assertTrue(msg.HasField('optional_nested_message')) |
| self.assertTrue(msg.optional_nested_message.HasField('bb')) |
| # Set to default value does not clear the fields |
| msg.optional_int32 = 0 |
| msg.optional_float = 0.0 |
| msg.optional_string = '' |
| msg.optional_nested_message.bb = 0 |
| self.assertTrue(msg.HasField('optional_int32')) |
| self.assertTrue(msg.HasField('optional_float')) |
| self.assertTrue(msg.HasField('optional_string')) |
| self.assertTrue(msg.HasField('optional_nested_message')) |
| self.assertTrue(msg.optional_nested_message.HasField('bb')) |
| |
| # Test serialize |
| msg2 = test_proto3_optional_pb2.TestProto3Optional() |
| msg2.ParseFromString(msg.SerializeToString()) |
| self.assertTrue(msg2.HasField('optional_int32')) |
| self.assertTrue(msg2.HasField('optional_float')) |
| self.assertTrue(msg2.HasField('optional_string')) |
| self.assertTrue(msg2.HasField('optional_nested_message')) |
| self.assertTrue(msg2.optional_nested_message.HasField('bb')) |
| |
| self.assertEqual(msg.WhichOneof('_optional_int32'), 'optional_int32') |
| |
| # Clear these fields. |
| msg.ClearField('optional_int32') |
| msg.ClearField('optional_float') |
| msg.ClearField('optional_string') |
| msg.ClearField('optional_nested_message') |
| self.assertFalse(msg.HasField('optional_int32')) |
| self.assertFalse(msg.HasField('optional_float')) |
| self.assertFalse(msg.HasField('optional_string')) |
| self.assertFalse(msg.HasField('optional_nested_message')) |
| self.assertFalse(msg.optional_nested_message.HasField('bb')) |
| |
| self.assertEqual(msg.WhichOneof('_optional_int32'), None) |
| |
| # Test has presence: |
| for field in test_proto3_optional_pb2.TestProto3Optional.DESCRIPTOR.fields: |
| if field.name.startswith('optional_'): |
| self.assertTrue(field.has_presence) |
| for field in unittest_pb2.TestAllTypes.DESCRIPTOR.fields: |
| if field.label == descriptor.FieldDescriptor.LABEL_REPEATED: |
| self.assertFalse(field.has_presence) |
| else: |
| self.assertTrue(field.has_presence) |
| proto3_descriptor = unittest_proto3_arena_pb2.TestAllTypes.DESCRIPTOR |
| repeated_field = proto3_descriptor.fields_by_name['repeated_int32'] |
| self.assertFalse(repeated_field.has_presence) |
| singular_field = proto3_descriptor.fields_by_name['optional_int32'] |
| self.assertFalse(singular_field.has_presence) |
| optional_field = proto3_descriptor.fields_by_name['proto3_optional_int32'] |
| self.assertTrue(optional_field.has_presence) |
| message_field = proto3_descriptor.fields_by_name['optional_nested_message'] |
| self.assertTrue(message_field.has_presence) |
| oneof_field = proto3_descriptor.fields_by_name['oneof_uint32'] |
| self.assertTrue(oneof_field.has_presence) |
| |
| def testAssignUnknownEnum(self): |
| """Assigning an unknown enum value is allowed and preserves the value.""" |
| m = unittest_proto3_arena_pb2.TestAllTypes() |
| |
| # Proto3 can assign unknown enums. |
| m.optional_nested_enum = 1234567 |
| self.assertEqual(1234567, m.optional_nested_enum) |
| m.repeated_nested_enum.append(22334455) |
| self.assertEqual(22334455, m.repeated_nested_enum[0]) |
| # Assignment is a different code path than append for the C++ impl. |
| m.repeated_nested_enum[0] = 7654321 |
| self.assertEqual(7654321, m.repeated_nested_enum[0]) |
| serialized = m.SerializeToString() |
| |
| m2 = unittest_proto3_arena_pb2.TestAllTypes() |
| m2.ParseFromString(serialized) |
| self.assertEqual(1234567, m2.optional_nested_enum) |
| self.assertEqual(7654321, m2.repeated_nested_enum[0]) |
| |
| # Map isn't really a proto3-only feature. But there is no proto2 equivalent |
| # of google/protobuf/map_unittest.proto right now, so it's not easy to |
| # test both with the same test like we do for the other proto2/proto3 tests. |
| # (google/protobuf/map_proto2_unittest.proto is very different in the set |
| # of messages and fields it contains). |
| def testScalarMapDefaults(self): |
| msg = map_unittest_pb2.TestMap() |
| |
| # Scalars start out unset. |
| self.assertFalse(-123 in msg.map_int32_int32) |
| self.assertFalse(-2**33 in msg.map_int64_int64) |
| self.assertFalse(123 in msg.map_uint32_uint32) |
| self.assertFalse(2**33 in msg.map_uint64_uint64) |
| self.assertFalse(123 in msg.map_int32_double) |
| self.assertFalse(False in msg.map_bool_bool) |
| self.assertFalse('abc' in msg.map_string_string) |
| self.assertFalse(111 in msg.map_int32_bytes) |
| self.assertFalse(888 in msg.map_int32_enum) |
| |
| # Accessing an unset key returns the default. |
| self.assertEqual(0, msg.map_int32_int32[-123]) |
| self.assertEqual(0, msg.map_int64_int64[-2**33]) |
| self.assertEqual(0, msg.map_uint32_uint32[123]) |
| self.assertEqual(0, msg.map_uint64_uint64[2**33]) |
| self.assertEqual(0.0, msg.map_int32_double[123]) |
| self.assertTrue(isinstance(msg.map_int32_double[123], float)) |
| self.assertEqual(False, msg.map_bool_bool[False]) |
| self.assertTrue(isinstance(msg.map_bool_bool[False], bool)) |
| self.assertEqual('', msg.map_string_string['abc']) |
| self.assertEqual(b'', msg.map_int32_bytes[111]) |
| self.assertEqual(0, msg.map_int32_enum[888]) |
| |
| # It also sets the value in the map |
| self.assertTrue(-123 in msg.map_int32_int32) |
| self.assertTrue(-2**33 in msg.map_int64_int64) |
| self.assertTrue(123 in msg.map_uint32_uint32) |
| self.assertTrue(2**33 in msg.map_uint64_uint64) |
| self.assertTrue(123 in msg.map_int32_double) |
| self.assertTrue(False in msg.map_bool_bool) |
| self.assertTrue('abc' in msg.map_string_string) |
| self.assertTrue(111 in msg.map_int32_bytes) |
| self.assertTrue(888 in msg.map_int32_enum) |
| |
| self.assertIsInstance(msg.map_string_string['abc'], str) |
| |
| # Accessing an unset key still throws TypeError if the type of the key |
| # is incorrect. |
| with self.assertRaises(TypeError): |
| msg.map_string_string[123] |
| |
| with self.assertRaises(TypeError): |
| 123 in msg.map_string_string |
| |
| with self.assertRaises(TypeError): |
| msg.map_string_string.__contains__(123) |
| |
| def testScalarMapComparison(self): |
| msg1 = map_unittest_pb2.TestMap() |
| msg2 = map_unittest_pb2.TestMap() |
| |
| self.assertEqual(msg1.map_int32_int32, msg2.map_int32_int32) |
| |
| def testMessageMapComparison(self): |
| msg1 = map_unittest_pb2.TestMap() |
| msg2 = map_unittest_pb2.TestMap() |
| |
| self.assertEqual(msg1.map_int32_foreign_message, |
| msg2.map_int32_foreign_message) |
| |
| def testMapGet(self): |
| # Need to test that get() properly returns the default, even though the dict |
| # has defaultdict-like semantics. |
| msg = map_unittest_pb2.TestMap() |
| |
| self.assertIsNone(msg.map_int32_int32.get(5)) |
| self.assertEqual(10, msg.map_int32_int32.get(5, 10)) |
| self.assertEqual(10, msg.map_int32_int32.get(key=5, default=10)) |
| self.assertIsNone(msg.map_int32_int32.get(5)) |
| |
| msg.map_int32_int32[5] = 15 |
| self.assertEqual(15, msg.map_int32_int32.get(5)) |
| self.assertEqual(15, msg.map_int32_int32.get(5)) |
| with self.assertRaises(TypeError): |
| msg.map_int32_int32.get('') |
| |
| self.assertIsNone(msg.map_int32_foreign_message.get(5)) |
| self.assertEqual(10, msg.map_int32_foreign_message.get(5, 10)) |
| self.assertEqual(10, msg.map_int32_foreign_message.get(key=5, default=10)) |
| |
| submsg = msg.map_int32_foreign_message[5] |
| self.assertIs(submsg, msg.map_int32_foreign_message.get(5)) |
| with self.assertRaises(TypeError): |
| msg.map_int32_foreign_message.get('') |
| |
| def testScalarMap(self): |
| msg = map_unittest_pb2.TestMap() |
| |
| self.assertEqual(0, len(msg.map_int32_int32)) |
| self.assertFalse(5 in msg.map_int32_int32) |
| |
| msg.map_int32_int32[-123] = -456 |
| msg.map_int64_int64[-2**33] = -2**34 |
| msg.map_uint32_uint32[123] = 456 |
| msg.map_uint64_uint64[2**33] = 2**34 |
| msg.map_int32_float[2] = 1.2 |
| msg.map_int32_double[1] = 3.3 |
| msg.map_string_string['abc'] = '123' |
| msg.map_bool_bool[True] = True |
| msg.map_int32_enum[888] = 2 |
| # Unknown numeric enum is supported in proto3. |
| msg.map_int32_enum[123] = 456 |
| |
| self.assertEqual([], msg.FindInitializationErrors()) |
| |
| self.assertEqual(1, len(msg.map_string_string)) |
| |
| # Bad key. |
| with self.assertRaises(TypeError): |
| msg.map_string_string[123] = '123' |
| |
| # Verify that trying to assign a bad key doesn't actually add a member to |
| # the map. |
| self.assertEqual(1, len(msg.map_string_string)) |
| |
| # Bad value. |
| with self.assertRaises(TypeError): |
| msg.map_string_string['123'] = 123 |
| |
| serialized = msg.SerializeToString() |
| msg2 = map_unittest_pb2.TestMap() |
| msg2.ParseFromString(serialized) |
| |
| # Bad key. |
| with self.assertRaises(TypeError): |
| msg2.map_string_string[123] = '123' |
| |
| # Bad value. |
| with self.assertRaises(TypeError): |
| msg2.map_string_string['123'] = 123 |
| |
| 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]) |
| self.assertAlmostEqual(1.2, msg.map_int32_float[2]) |
| self.assertEqual(3.3, msg.map_int32_double[1]) |
| self.assertEqual('123', msg2.map_string_string['abc']) |
| self.assertEqual(True, msg2.map_bool_bool[True]) |
| self.assertEqual(2, msg2.map_int32_enum[888]) |
| self.assertEqual(456, msg2.map_int32_enum[123]) |
| self.assertEqual('{-123: -456}', str(msg2.map_int32_int32)) |
| |
| def testMapEntryAlwaysSerialized(self): |
| msg = map_unittest_pb2.TestMap() |
| msg.map_int32_int32[0] = 0 |
| msg.map_string_string[''] = '' |
| self.assertEqual(msg.ByteSize(), 12) |
| self.assertEqual(b'\n\x04\x08\x00\x10\x00r\x04\n\x00\x12\x00', |
| msg.SerializeToString()) |
| |
| def testStringUnicodeConversionInMap(self): |
| msg = map_unittest_pb2.TestMap() |
| |
| unicode_obj = u'\u1234' |
| bytes_obj = unicode_obj.encode('utf8') |
| |
| msg.map_string_string[bytes_obj] = bytes_obj |
| |
| (key, value) = list(msg.map_string_string.items())[0] |
| |
| self.assertEqual(key, unicode_obj) |
| self.assertEqual(value, unicode_obj) |
| |
| self.assertIsInstance(key, str) |
| self.assertIsInstance(value, str) |
| |
| 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'] |
| |
| with self.assertRaises(TypeError): |
| '123' in msg.map_int32_foreign_message |
| |
| with self.assertRaises(TypeError): |
| msg.map_int32_foreign_message.__contains__('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)) |
| msg2.map_int32_foreign_message[123].c = 1 |
| # TODO: Fix text format for message map. |
| self.assertIn( |
| str(msg2.map_int32_foreign_message), |
| ('{-456: , 123: c: 1\n}', '{123: c: 1\n, -456: }')) |
| |
| def testNestedMessageMapItemDelete(self): |
| msg = map_unittest_pb2.TestMap() |
| msg.map_int32_all_types[1].optional_nested_message.bb = 1 |
| del msg.map_int32_all_types[1] |
| msg.map_int32_all_types[2].optional_nested_message.bb = 2 |
| self.assertEqual(1, len(msg.map_int32_all_types)) |
| msg.map_int32_all_types[1].optional_nested_message.bb = 1 |
| self.assertEqual(2, len(msg.map_int32_all_types)) |
| |
| serialized = msg.SerializeToString() |
| msg2 = map_unittest_pb2.TestMap() |
| msg2.ParseFromString(serialized) |
| keys = [1, 2] |
| # The loop triggers PyErr_Occurred() in c extension. |
| for key in keys: |
| del msg2.map_int32_all_types[key] |
| |
| def testMapByteSize(self): |
| msg = map_unittest_pb2.TestMap() |
| msg.map_int32_int32[1] = 1 |
| size = msg.ByteSize() |
| msg.map_int32_int32[1] = 128 |
| self.assertEqual(msg.ByteSize(), size + 1) |
| |
| msg.map_int32_foreign_message[19].c = 1 |
| size = msg.ByteSize() |
| msg.map_int32_foreign_message[19].c = 128 |
| self.assertEqual(msg.ByteSize(), size + 1) |
| |
| def testMergeFrom(self): |
| msg = map_unittest_pb2.TestMap() |
| msg.map_int32_int32[12] = 34 |
| msg.map_int32_int32[56] = 78 |
| msg.map_int64_int64[22] = 33 |
| msg.map_int32_foreign_message[111].c = 5 |
| msg.map_int32_foreign_message[222].c = 10 |
| |
| msg2 = map_unittest_pb2.TestMap() |
| msg2.map_int32_int32[12] = 55 |
| msg2.map_int64_int64[88] = 99 |
| msg2.map_int32_foreign_message[222].c = 15 |
| msg2.map_int32_foreign_message[222].d = 20 |
| old_map_value = msg2.map_int32_foreign_message[222] |
| |
| msg2.MergeFrom(msg) |
| # Compare with expected message instead of call |
| # msg2.map_int32_foreign_message[222] to make sure MergeFrom does not |
| # sync with repeated field and there is no duplicated keys. |
| expected_msg = map_unittest_pb2.TestMap() |
| expected_msg.CopyFrom(msg) |
| expected_msg.map_int64_int64[88] = 99 |
| self.assertEqual(msg2, expected_msg) |
| |
| self.assertEqual(34, msg2.map_int32_int32[12]) |
| self.assertEqual(78, msg2.map_int32_int32[56]) |
| self.assertEqual(33, msg2.map_int64_int64[22]) |
| self.assertEqual(99, msg2.map_int64_int64[88]) |
| self.assertEqual(5, msg2.map_int32_foreign_message[111].c) |
| self.assertEqual(10, msg2.map_int32_foreign_message[222].c) |
| self.assertFalse(msg2.map_int32_foreign_message[222].HasField('d')) |
| if api_implementation.Type() != 'cpp': |
| # During the call to MergeFrom(), the C++ implementation will have |
| # deallocated the underlying message, but this is very difficult to detect |
| # properly. The line below is likely to cause a segmentation fault. |
| # With the Python implementation, old_map_value is just 'detached' from |
| # the main message. Using it will not crash of course, but since it still |
| # have a reference to the parent message I'm sure we can find interesting |
| # ways to cause inconsistencies. |
| self.assertEqual(15, old_map_value.c) |
| |
| # Verify that there is only one entry per key, even though the MergeFrom |
| # may have internally created multiple entries for a single key in the |
| # list representation. |
| as_dict = {} |
| for key in msg2.map_int32_foreign_message: |
| self.assertFalse(key in as_dict) |
| as_dict[key] = msg2.map_int32_foreign_message[key].c |
| |
| self.assertEqual({111: 5, 222: 10}, as_dict) |
| |
| # Special case: test that delete of item really removes the item, even if |
| # there might have physically been duplicate keys due to the previous merge. |
| # This is only a special case for the C++ implementation which stores the |
| # map as an array. |
| del msg2.map_int32_int32[12] |
| self.assertFalse(12 in msg2.map_int32_int32) |
| |
| del msg2.map_int32_foreign_message[222] |
| self.assertFalse(222 in msg2.map_int32_foreign_message) |
| with self.assertRaises(TypeError): |
| del msg2.map_int32_foreign_message[''] |
| |
| def testMapMergeFrom(self): |
| msg = map_unittest_pb2.TestMap() |
| msg.map_int32_int32[12] = 34 |
| msg.map_int32_int32[56] = 78 |
| msg.map_int64_int64[22] = 33 |
| msg.map_int32_foreign_message[111].c = 5 |
| msg.map_int32_foreign_message[222].c = 10 |
| |
| msg2 = map_unittest_pb2.TestMap() |
| msg2.map_int32_int32[12] = 55 |
| msg2.map_int64_int64[88] = 99 |
| msg2.map_int32_foreign_message[222].c = 15 |
| msg2.map_int32_foreign_message[222].d = 20 |
| |
| msg2.map_int32_int32.MergeFrom(msg.map_int32_int32) |
| self.assertEqual(34, msg2.map_int32_int32[12]) |
| self.assertEqual(78, msg2.map_int32_int32[56]) |
| |
| msg2.map_int64_int64.MergeFrom(msg.map_int64_int64) |
| self.assertEqual(33, msg2.map_int64_int64[22]) |
| self.assertEqual(99, msg2.map_int64_int64[88]) |
| |
| msg2.map_int32_foreign_message.MergeFrom(msg.map_int32_foreign_message) |
| # Compare with expected message instead of call |
| # msg.map_int32_foreign_message[222] to make sure MergeFrom does not |
| # sync with repeated field and no duplicated keys. |
| expected_msg = map_unittest_pb2.TestMap() |
| expected_msg.CopyFrom(msg) |
| expected_msg.map_int64_int64[88] = 99 |
| self.assertEqual(msg2, expected_msg) |
| |
| # Test when cpp extension cache a map. |
| m1 = map_unittest_pb2.TestMap() |
| m2 = map_unittest_pb2.TestMap() |
| self.assertEqual(m1.map_int32_foreign_message, m1.map_int32_foreign_message) |
| m2.map_int32_foreign_message[123].c = 10 |
| m1.MergeFrom(m2) |
| self.assertEqual(10, m2.map_int32_foreign_message[123].c) |
| |
| # Test merge maps within different message types. |
| m1 = map_unittest_pb2.TestMap() |
| m2 = map_unittest_pb2.TestMessageMap() |
| m2.map_int32_message[123].optional_int32 = 10 |
| m1.map_int32_all_types.MergeFrom(m2.map_int32_message) |
| self.assertEqual(10, m1.map_int32_all_types[123].optional_int32) |
| |
| # Test overwrite message value map |
| msg = map_unittest_pb2.TestMap() |
| msg.map_int32_foreign_message[222].c = 123 |
| msg2 = map_unittest_pb2.TestMap() |
| msg2.map_int32_foreign_message[222].d = 20 |
| msg.MergeFromString(msg2.SerializeToString()) |
| self.assertEqual(msg.map_int32_foreign_message[222].d, 20) |
| self.assertNotEqual(msg.map_int32_foreign_message[222].c, 123) |
| |
| # Merge a dict to map field is not accepted |
| with self.assertRaises(AttributeError): |
| m1.map_int32_all_types.MergeFrom( |
| {1: unittest_proto3_arena_pb2.TestAllTypes()}) |
| |
| def testMergeFromBadType(self): |
| msg = map_unittest_pb2.TestMap() |
| with self.assertRaisesRegex( |
| TypeError, |
| r'Parameter to MergeFrom\(\) must be instance of same class: expected ' |
| r'.+TestMap got int\.'): |
| msg.MergeFrom(1) |
| |
| def testCopyFromBadType(self): |
| msg = map_unittest_pb2.TestMap() |
| with self.assertRaisesRegex( |
| TypeError, |
| r'Parameter to [A-Za-z]*From\(\) must be instance of same class: ' |
| r'expected .+TestMap got int\.'): |
| msg.CopyFrom(1) |
| |
| def testIntegerMapWithLongs(self): |
| msg = map_unittest_pb2.TestMap() |
| msg.map_int32_int32[int(-123)] = int(-456) |
| msg.map_int64_int64[int(-2**33)] = int(-2**34) |
| msg.map_uint32_uint32[int(123)] = int(456) |
| msg.map_uint64_uint64[int(2**33)] = int(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 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 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 testModifyMapWhileIterating(self): |
| msg = map_unittest_pb2.TestMap() |
| |
| string_string_iter = iter(msg.map_string_string) |
| int32_foreign_iter = iter(msg.map_int32_foreign_message) |
| |
| msg.map_string_string['abc'] = '123' |
| msg.map_int32_foreign_message[5].c = 5 |
| |
| with self.assertRaises(RuntimeError): |
| for key in string_string_iter: |
| pass |
| |
| with self.assertRaises(RuntimeError): |
| for key in int32_foreign_iter: |
| pass |
| |
| def testModifyMapEntryWhileIterating(self): |
| msg = map_unittest_pb2.TestMap() |
| |
| msg.map_string_string['abc'] = '123' |
| msg.map_string_string['def'] = '456' |
| msg.map_string_string['ghi'] = '789' |
| |
| msg.map_int32_foreign_message[5].c = 5 |
| msg.map_int32_foreign_message[6].c = 6 |
| msg.map_int32_foreign_message[7].c = 7 |
| |
| string_string_keys = list(msg.map_string_string.keys()) |
| int32_foreign_keys = list(msg.map_int32_foreign_message.keys()) |
| |
| keys = [] |
| for key in msg.map_string_string: |
| keys.append(key) |
| msg.map_string_string[key] = '000' |
| self.assertEqual(keys, string_string_keys) |
| self.assertEqual(keys, list(msg.map_string_string.keys())) |
| |
| keys = [] |
| for key in msg.map_int32_foreign_message: |
| keys.append(key) |
| msg.map_int32_foreign_message[key].c = 0 |
| self.assertEqual(keys, int32_foreign_keys) |
| self.assertEqual(keys, list(msg.map_int32_foreign_message.keys())) |
| |
| def testSubmessageMap(self): |
| msg = map_unittest_pb2.TestMap() |
| |
| submsg = msg.map_int32_foreign_message[111] |
| self.assertIs(submsg, msg.map_int32_foreign_message[111]) |
| self.assertIsInstance(submsg, unittest_pb2.ForeignMessage) |
| |
| submsg.c = 5 |
| |
| serialized = msg.SerializeToString() |
| msg2 = map_unittest_pb2.TestMap() |
| msg2.ParseFromString(serialized) |
| |
| self.assertEqual(5, msg2.map_int32_foreign_message[111].c) |
| |
| # Doesn't allow direct submessage assignment. |
| with self.assertRaises(ValueError): |
| msg.map_int32_foreign_message[88] = unittest_pb2.ForeignMessage() |
| |
| def testMapIteration(self): |
| msg = map_unittest_pb2.TestMap() |
| |
| for k, v in msg.map_int32_int32.items(): |
| # Should not be reached. |
| self.assertTrue(False) |
| |
| msg.map_int32_int32[2] = 4 |
| msg.map_int32_int32[3] = 6 |
| msg.map_int32_int32[4] = 8 |
| self.assertEqual(3, len(msg.map_int32_int32)) |
| |
| matching_dict = {2: 4, 3: 6, 4: 8} |
| self.assertMapIterEquals(msg.map_int32_int32.items(), matching_dict) |
| |
| def testMapItems(self): |
| # Map items used to have strange behaviors when use c extension. Because |
| # [] may reorder the map and invalidate any existing iterators. |
| # TODO: Check if [] reordering the map is a bug or intended |
| # behavior. |
| msg = map_unittest_pb2.TestMap() |
| msg.map_string_string['local_init_op'] = '' |
| msg.map_string_string['trainable_variables'] = '' |
| msg.map_string_string['variables'] = '' |
| msg.map_string_string['init_op'] = '' |
| msg.map_string_string['summaries'] = '' |
| items1 = msg.map_string_string.items() |
| items2 = msg.map_string_string.items() |
| self.assertEqual(items1, items2) |
| |
| def testMapDeterministicSerialization(self): |
| golden_data = (b'r\x0c\n\x07init_op\x12\x01d' |
| b'r\n\n\x05item1\x12\x01e' |
| b'r\n\n\x05item2\x12\x01f' |
| b'r\n\n\x05item3\x12\x01g' |
| b'r\x0b\n\x05item4\x12\x02QQ' |
| b'r\x12\n\rlocal_init_op\x12\x01a' |
| b'r\x0e\n\tsummaries\x12\x01e' |
| b'r\x18\n\x13trainable_variables\x12\x01b' |
| b'r\x0e\n\tvariables\x12\x01c') |
| msg = map_unittest_pb2.TestMap() |
| msg.map_string_string['local_init_op'] = 'a' |
| msg.map_string_string['trainable_variables'] = 'b' |
| msg.map_string_string['variables'] = 'c' |
| msg.map_string_string['init_op'] = 'd' |
| msg.map_string_string['summaries'] = 'e' |
| msg.map_string_string['item1'] = 'e' |
| msg.map_string_string['item2'] = 'f' |
| msg.map_string_string['item3'] = 'g' |
| msg.map_string_string['item4'] = 'QQ' |
| |
| # If deterministic serialization is not working correctly, this will be |
| # "flaky" depending on the exact python dict hash seed. |
| # |
| # Fortunately, there are enough items in this map that it is extremely |
| # unlikely to ever hit the "right" in-order combination, so the test |
| # itself should fail reliably. |
| self.assertEqual(golden_data, msg.SerializeToString(deterministic=True)) |
| |
| def testMapIterationClearMessage(self): |
| # Iterator needs to work even if message and map are deleted. |
| msg = map_unittest_pb2.TestMap() |
| |
| msg.map_int32_int32[2] = 4 |
| msg.map_int32_int32[3] = 6 |
| msg.map_int32_int32[4] = 8 |
| |
| it = msg.map_int32_int32.items() |
| del msg |
| |
| matching_dict = {2: 4, 3: 6, 4: 8} |
| self.assertMapIterEquals(it, matching_dict) |
| |
| def testMapConstruction(self): |
| msg = map_unittest_pb2.TestMap(map_int32_int32={1: 2, 3: 4}) |
| self.assertEqual(2, msg.map_int32_int32[1]) |
| self.assertEqual(4, msg.map_int32_int32[3]) |
| |
| msg = map_unittest_pb2.TestMap( |
| map_int32_foreign_message={3: unittest_pb2.ForeignMessage(c=5)}) |
| self.assertEqual(5, msg.map_int32_foreign_message[3].c) |
| |
| def testMapScalarFieldConstruction(self): |
| msg1 = map_unittest_pb2.TestMap() |
| msg1.map_int32_int32[1] = 42 |
| msg2 = map_unittest_pb2.TestMap(map_int32_int32=msg1.map_int32_int32) |
| self.assertEqual(42, msg2.map_int32_int32[1]) |
| |
| def testMapMessageFieldConstruction(self): |
| msg1 = map_unittest_pb2.TestMap() |
| msg1.map_string_foreign_message['test'].c = 42 |
| msg2 = map_unittest_pb2.TestMap( |
| map_string_foreign_message=msg1.map_string_foreign_message) |
| self.assertEqual(42, msg2.map_string_foreign_message['test'].c) |
| |
| def testMapFieldRaisesCorrectError(self): |
| # Should raise a TypeError when given a non-iterable. |
| with self.assertRaises(TypeError): |
| map_unittest_pb2.TestMap(map_string_foreign_message=1) |
| |
| def testMapValidAfterFieldCleared(self): |
| # Map needs to work even if field is cleared. |
| # For the C++ implementation this tests the correctness of |
| # MapContainer::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 |
| # MapContainer::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 testMessageMapItemValidAfterTopMessageCleared(self): |
| # Message map item needs to work even if it is cleared. |
| # For the C++ implementation this tests the correctness of |
| # MapContainer::Release() |
| msg = map_unittest_pb2.TestMap() |
| msg.map_int32_all_types[2].optional_string = 'bar' |
| |
| if api_implementation.Type() == 'cpp': |
| # Need to keep the map reference because of b/27942626. |
| # TODO: Remove it. |
| unused_map = msg.map_int32_all_types # pylint: disable=unused-variable |
| msg_value = msg.map_int32_all_types[2] |
| msg.Clear() |
| |
| # Reset to trigger sync between repeated field and map in c++. |
| msg.map_int32_all_types[3].optional_string = 'foo' |
| self.assertEqual(msg_value.optional_string, 'bar') |
| |
| def testMapIterInvalidatedByClearField(self): |
| # Map iterator is invalidated when field is cleared. |
| # But this case does need to not crash the interpreter. |
| # For the C++ implementation this tests the correctness of |
| # ScalarMapContainer::Release() |
| msg = map_unittest_pb2.TestMap() |
| |
| it = iter(msg.map_int32_int32) |
| |
| msg.ClearField('map_int32_int32') |
| with self.assertRaises(RuntimeError): |
| for _ in it: |
| pass |
| |
| it = iter(msg.map_int32_foreign_message) |
| msg.ClearField('map_int32_foreign_message') |
| with self.assertRaises(RuntimeError): |
| for _ in it: |
| pass |
| |
| def testMapDelete(self): |
| msg = map_unittest_pb2.TestMap() |
| |
| self.assertEqual(0, len(msg.map_int32_int32)) |
| |
| msg.map_int32_int32[4] = 6 |
| self.assertEqual(1, len(msg.map_int32_int32)) |
| |
| with self.assertRaises(KeyError): |
| del msg.map_int32_int32[88] |
| |
| del msg.map_int32_int32[4] |
| self.assertEqual(0, len(msg.map_int32_int32)) |
| |
| with self.assertRaises(KeyError): |
| del msg.map_int32_all_types[32] |
| |
| def testMapsAreMapping(self): |
| msg = map_unittest_pb2.TestMap() |
| self.assertIsInstance(msg.map_int32_int32, collections.abc.Mapping) |
| self.assertIsInstance(msg.map_int32_int32, collections.abc.MutableMapping) |
| self.assertIsInstance(msg.map_int32_foreign_message, |
| collections.abc.Mapping) |
| self.assertIsInstance(msg.map_int32_foreign_message, |
| collections.abc.MutableMapping) |
| |
| def testMapsCompare(self): |
| msg = map_unittest_pb2.TestMap() |
| msg.map_int32_int32[-123] = -456 |
| self.assertEqual(msg.map_int32_int32, msg.map_int32_int32) |
| self.assertEqual(msg.map_int32_foreign_message, |
| msg.map_int32_foreign_message) |
| self.assertNotEqual(msg.map_int32_int32, 0) |
| |
| def testMapFindInitializationErrorsSmokeTest(self): |
| msg = map_unittest_pb2.TestMap() |
| msg.map_string_string['abc'] = '123' |
| msg.map_int32_int32[35] = 64 |
| msg.map_string_foreign_message['foo'].c = 5 |
| self.assertEqual(0, len(msg.FindInitializationErrors())) |
| |
| @unittest.skipIf(sys.maxunicode == UCS2_MAXUNICODE, 'Skip for ucs2') |
| def testStrictUtf8Check(self): |
| # Test u'\ud801' is rejected at parser in both python2 and python3. |
| serialized = (b'r\x03\xed\xa0\x81') |
| msg = unittest_proto3_arena_pb2.TestAllTypes() |
| with self.assertRaises(Exception) as context: |
| msg.MergeFromString(serialized) |
|
