mmo-ref/dep-protobuf
1
0
Fork 0
mirror of https://github.com/protocolbuffers/protobuf synced 2026-08-26 02:23:14 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions
dep-protobuf/python/google/protobuf/internal/text_format_test.py
Protobuf Team Bot 6cd6ca4116 Auto-format Py Proto python files. 
PiperOrigin-RevId: 907755617
2026-04-29 13:40:58 -07:00

3465 lines
117 KiB
Python
Raw Permalink Blame History

# -*- 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
"""Test for google.protobuf.text_format."""
import io
import math
import re
import string
import textwrap
import unittest
from google.protobuf import any_pb2
from google.protobuf import struct_pb2
from google.protobuf import descriptor_pb2
from google.protobuf.internal import well_known_types_test_pb2 as test_extend_any
from google.protobuf.internal import api_implementation
from google.protobuf.internal import message_set_extensions_pb2
from google.protobuf.internal import test_proto3_optional_pb2
from google.protobuf.internal import test_util
from google.protobuf import descriptor_pool
from google.protobuf import text_format
from absl.testing import parameterized
from google.protobuf import any_test_pb2
from google.protobuf import map_unittest_pb2
from google.protobuf import unittest_mset_pb2
from google.protobuf import unittest_custom_options_pb2
from google.protobuf import unittest_delimited_pb2
from google.protobuf import unittest_delimited_import_pb2
from google.protobuf import unittest_pb2
from google.protobuf import unittest_proto3_arena_pb2
# pylint: enable=g-import-not-at-top
# Low-level nuts-n-bolts tests.
class SimpleTextFormatTests(unittest.TestCase):
# The members of _QUOTES are formatted into a regexp template that
# expects single characters. Therefore it's an error (in addition to being
# non-sensical in the first place) to try to specify a "quote mark" that is
# more than one character.
def testQuoteMarksAreSingleChars(self):
for quote in text_format._QUOTES:
self.assertEqual(1, len(quote))
# Base class with some common functionality.
class TextFormatBase(unittest.TestCase):
def ReadGolden(self, golden_filename):
with test_util.GoldenFile(golden_filename) as f:
return [golden_line.decode('utf-8') for golden_line in f]
def CompareToGoldenFile(self, text, golden_filename):
golden_lines = self.ReadGolden(golden_filename)
self.assertMultiLineEqual(text, ''.join(golden_lines))
def CompareToGoldenText(self, text, golden_text):
self.assertEqual(text, golden_text)
def RemoveRedundantZeros(self, text):
# Some platforms print 1e+5 as 1e+005. This is fine, but we need to remove
# these zeros in order to match the golden file.
text = (
text.replace('e+0', 'e+')
.replace('e+0', 'e+')
.replace('e-0', 'e-')
.replace('e-0', 'e-')
)
# Floating point fields are printed with .0 suffix even if they are
# actually integer numbers.
text = re.compile(r'\.0$', re.MULTILINE).sub('', text)
return text
@parameterized.parameters(unittest_pb2, unittest_proto3_arena_pb2)
class TextFormatMessageToStringTests(TextFormatBase):
def testPrintExotic(self, message_module):
message = message_module.TestAllTypes()
message.repeated_int64.append(-9223372036854775808)
message.repeated_uint64.append(18446744073709551615)
message.repeated_double.append(123.456)
message.repeated_double.append(1.23e22)
message.repeated_double.append(1.23e-18)
message.repeated_string.append('\000\001\a\b\f\n\r\t\v\\\'"')
message.repeated_string.append('\u00fc\ua71f')
self.CompareToGoldenText(
self.RemoveRedundantZeros(
text_format.MessageToString(message, as_utf8=True)
),
'repeated_int64: -9223372036854775808\n'
'repeated_uint64: 18446744073709551615\n'
'repeated_double: 123.456\n'
'repeated_double: 1.23e+22\n'
'repeated_double: 1.23e-18\n'
'repeated_string:'
' "\\000\\001\\007\\010\\014\\n\\r\\t\\013\\\\\\\'\\""\n'
'repeated_string: "üꜟ"\n',
)
def testPrintFloatPrecision(self, message_module):
message = message_module.TestAllTypes()
message.repeated_float.append(0.0)
message.repeated_float.append(0.8)
message.repeated_float.append(1.0)
message.repeated_float.append(1.2)
message.repeated_float.append(1.23)
message.repeated_float.append(1.234)
message.repeated_float.append(1.2345)
message.repeated_float.append(1.23456)
message.repeated_float.append(1.2e10)
message.repeated_float.append(1.23e10)
message.repeated_float.append(1.234e10)
message.repeated_float.append(1.2345e10)
message.repeated_float.append(1.23456e10)
message.repeated_float.append(float('NaN'))
message.repeated_float.append(float('inf'))
message.repeated_double.append(0.0)
message.repeated_double.append(0.8)
message.repeated_double.append(1.0)
message.repeated_double.append(1.2)
message.repeated_double.append(1.23)
message.repeated_double.append(1.234)
message.repeated_double.append(1.2345)
message.repeated_double.append(1.23456)
message.repeated_double.append(1.234567)
message.repeated_double.append(1.2345678)
message.repeated_double.append(1.23456789)
message.repeated_double.append(1.234567898)
message.repeated_double.append(1.2345678987)
message.repeated_double.append(1.23456789876)
message.repeated_double.append(1.234567898765)
message.repeated_double.append(1.2345678987654)
message.repeated_double.append(1.23456789876543)
message.repeated_double.append(1.2e100)
message.repeated_double.append(1.23e100)
message.repeated_double.append(1.234e100)
message.repeated_double.append(1.2345e100)
message.repeated_double.append(1.23456e100)
message.repeated_double.append(1.234567e100)
message.repeated_double.append(1.2345678e100)
message.repeated_double.append(1.23456789e100)
message.repeated_double.append(1.234567898e100)
message.repeated_double.append(1.2345678987e100)
message.repeated_double.append(1.23456789876e100)
message.repeated_double.append(1.234567898765e100)
message.repeated_double.append(1.2345678987654e100)
message.repeated_double.append(1.23456789876543e100)
# pylint: disable=g-long-ternary
self.CompareToGoldenText(
self.RemoveRedundantZeros(text_format.MessageToString(message)),
'repeated_float: 0\n'
'repeated_float: 0.8\n'
'repeated_float: 1\n'
'repeated_float: 1.2\n'
'repeated_float: 1.23\n'
'repeated_float: 1.234\n'
'repeated_float: 1.2345\n'
'repeated_float: 1.23456\n'
# Note that these don't use scientific notation.
'repeated_float: 12000000000\n'
'repeated_float: 12300000000\n'
'repeated_float: 12340000000\n'
'repeated_float: 12345000000\n'
'repeated_float: 12345600000\n'
'repeated_float: nan\n'
'repeated_float: inf\n'
'repeated_double: 0\n'
'repeated_double: 0.8\n'
'repeated_double: 1\n'
'repeated_double: 1.2\n'
'repeated_double: 1.23\n'
'repeated_double: 1.234\n'
'repeated_double: 1.2345\n'
'repeated_double: 1.23456\n'
'repeated_double: 1.234567\n'
'repeated_double: 1.2345678\n'
'repeated_double: 1.23456789\n'
'repeated_double: 1.234567898\n'
'repeated_double: 1.2345678987\n'
'repeated_double: 1.23456789876\n'
'repeated_double: 1.234567898765\n'
'repeated_double: 1.2345678987654\n'
'repeated_double: 1.23456789876543\n'
'repeated_double: 1.2e+100\n'
'repeated_double: 1.23e+100\n'
'repeated_double: 1.234e+100\n'
'repeated_double: 1.2345e+100\n'
'repeated_double: 1.23456e+100\n'
'repeated_double: 1.234567e+100\n'
'repeated_double: 1.2345678e+100\n'
'repeated_double: 1.23456789e+100\n'
'repeated_double: 1.234567898e+100\n'
'repeated_double: 1.2345678987e+100\n'
'repeated_double: 1.23456789876e+100\n'
'repeated_double: 1.234567898765e+100\n'
'repeated_double: 1.2345678987654e+100\n'
'repeated_double: 1.23456789876543e+100\n',
)
def testPrintExoticUnicodeSubclass(self, message_module):
class UnicodeSub(str):
pass
message = message_module.TestAllTypes()
message.repeated_string.append(UnicodeSub('\u00fc\ua71f'))
self.CompareToGoldenText(
text_format.MessageToString(message, as_utf8=True),
'repeated_string: "üꜟ"\n',
)
def testPrintNestedMessageAsOneLine(self, message_module):
message = message_module.TestAllTypes()
msg = message.repeated_nested_message.add()
msg.bb = 42
self.CompareToGoldenText(
text_format.MessageToString(message, as_one_line=True),
'repeated_nested_message { bb: 42 }',
)
def testPrintRepeatedFieldsAsOneLine(self, message_module):
message = message_module.TestAllTypes()
message.repeated_int32.append(1)
message.repeated_int32.append(1)
message.repeated_int32.append(3)
message.repeated_string.append('Google')
message.repeated_string.append('Zurich')
self.CompareToGoldenText(
text_format.MessageToString(message, as_one_line=True),
'repeated_int32: 1 repeated_int32: 1 repeated_int32: 3 '
'repeated_string: "Google" repeated_string: "Zurich"',
)
def VerifyPrintShortFormatRepeatedFields(self, message_module, as_one_line):
message = message_module.TestAllTypes()
message.repeated_int32.append(1)
message.repeated_string.append('Google')
message.repeated_string.append('Hello,World')
message.repeated_foreign_enum.append(unittest_pb2.FOREIGN_FOO)
message.repeated_foreign_enum.append(unittest_pb2.FOREIGN_BAR)
message.repeated_foreign_enum.append(unittest_pb2.FOREIGN_BAZ)
message.optional_nested_message.bb = 3
for i in (21, 32):
msg = message.repeated_nested_message.add()
msg.bb = i
expected_ascii = (
'optional_nested_message {\n bb: 3\n}\n'
'repeated_int32: [1]\n'
'repeated_string: "Google"\n'
'repeated_string: "Hello,World"\n'
'repeated_nested_message {\n bb: 21\n}\n'
'repeated_nested_message {\n bb: 32\n}\n'
'repeated_foreign_enum: [FOREIGN_FOO, FOREIGN_BAR, FOREIGN_BAZ]\n'
)
if as_one_line:
expected_ascii = expected_ascii.replace('\n', ' ')
expected_ascii = re.sub(r'\s+', ' ', expected_ascii)
expected_ascii = re.sub(r'\s$', '', expected_ascii)
actual_ascii = text_format.MessageToString(
message, use_short_repeated_primitives=True, as_one_line=as_one_line
)
self.CompareToGoldenText(actual_ascii, expected_ascii)
parsed_message = message_module.TestAllTypes()
text_format.Parse(actual_ascii, parsed_message)
self.assertEqual(parsed_message, message)
def testPrintShortFormatRepeatedFields(self, message_module):
self.VerifyPrintShortFormatRepeatedFields(message_module, False)
self.VerifyPrintShortFormatRepeatedFields(message_module, True)
def testPrintNestedNewLineInStringAsOneLine(self, message_module):
message = message_module.TestAllTypes()
message.optional_string = 'a\nnew\nline'
self.CompareToGoldenText(
text_format.MessageToString(message, as_one_line=True),
'optional_string: "a\\nnew\\nline"',
)
def testPrintExoticAsOneLine(self, message_module):
message = message_module.TestAllTypes()
message.repeated_int64.append(-9223372036854775808)
message.repeated_uint64.append(18446744073709551615)
message.repeated_double.append(123.456)
message.repeated_double.append(1.23e22)
message.repeated_double.append(1.23e-18)
message.repeated_string.append('\000\001\a\b\f\n\r\t\v\\\'"')
message.repeated_string.append('\u00fc\ua71f')
self.CompareToGoldenText(
self.RemoveRedundantZeros(
text_format.MessageToString(message, as_one_line=True, as_utf8=True)
),
'repeated_int64: -9223372036854775808'
' repeated_uint64: 18446744073709551615'
' repeated_double: 123.456'
' repeated_double: 1.23e+22'
' repeated_double: 1.23e-18'
' repeated_string: '
'"\\000\\001\\007\\010\\014\\n\\r\\t\\013\\\\\\\'\\""'
' repeated_string: "üꜟ"',
)
def testRoundTripExoticAsOneLine(self, message_module):
message = message_module.TestAllTypes()
message.repeated_int64.append(-9223372036854775808)
message.repeated_uint64.append(18446744073709551615)
message.repeated_double.append(123.456)
message.repeated_double.append(1.23e22)
message.repeated_double.append(1.23e-18)
message.repeated_string.append('\000\001\a\b\f\n\r\t\v\\\'"')
message.repeated_string.append('\u00fc\ua71f')
# Test as_utf8 = False.
wire_text = text_format.MessageToString(
message, as_one_line=True, as_utf8=False
)
parsed_message = message_module.TestAllTypes()
r = text_format.Parse(wire_text, parsed_message)
self.assertIs(r, parsed_message)
self.assertEqual(message, parsed_message)
# Test as_utf8 = True.
wire_text = text_format.MessageToString(
message, as_one_line=True, as_utf8=True
)
parsed_message = message_module.TestAllTypes()
r = text_format.Parse(wire_text, parsed_message)
self.assertIs(r, parsed_message)
self.assertEqual(
message, parsed_message, '\n%s != %s' % (message, parsed_message)
)
def testPrintRawUtf8String(self, message_module):
message = message_module.TestAllTypes()
message.repeated_string.append('\u00fc\t\ua71f')
text = text_format.MessageToString(message, as_utf8=True)
golden_unicode = 'repeated_string: "\u00fc\\t\ua71f"\n'
golden_text = golden_unicode
# MessageToString always returns a native str.
self.CompareToGoldenText(text, golden_text)
parsed_message = message_module.TestAllTypes()
text_format.Parse(text, parsed_message)
self.assertEqual(
message,
parsed_message,
'\n%s != %s (%s != %s)'
% (
message,
parsed_message,
message.repeated_string[0],
parsed_message.repeated_string[0],
),
)
def testPrintFloatFormat(self, message_module):
message = message_module.NestedTestAllTypes()
message.payload.optional_float = 1.25
message.payload.optional_double = -0.000003456789012345678
# Check no decimal point.
message.payload.repeated_float.append(-5642)
# Check no trailing zeros.
message.payload.repeated_double.append(0.000078900)
formatted_fields = [
'optional_float: 1.25',
'optional_double: -3.456789012345678e-6',
'repeated_float: -5642',
'repeated_double: 7.89e-5',
]
text_message = text_format.MessageToString(message)
self.CompareToGoldenText(
self.RemoveRedundantZeros(text_message),
'payload {{\n {0}\n {1}\n {2}\n {3}\n}}\n'.format(
*formatted_fields
),
)
# 32-bit 1.2 is noisy when extended to 64-bit:
# >>> struct.unpack('f', struct.pack('f', 1.2))[0]
# 1.2000000476837158
message.payload.optional_float = 1.2
formatted_fields = [
'optional_float: 1.2',
'optional_double: -3.456789012345678e-6',
'repeated_float: -5642',
'repeated_double: 7.89e-5',
]
text_message = text_format.MessageToString(message)
self.CompareToGoldenText(
self.RemoveRedundantZeros(text_message),
'payload {{\n {0}\n {1}\n {2}\n {3}\n}}\n'.format(
*formatted_fields
),
)
# Test print shortest float.
message.payload.optional_float = 1.2345678912
message.payload.optional_double = 1.2345678912
formatted_fields = [
'optional_float: 1.2345679',
'optional_double: 1.2345678912',
'repeated_float: -5642',
'repeated_double: 7.89e-5',
]
text_message = text_format.MessageToString(message)
self.CompareToGoldenText(
self.RemoveRedundantZeros(text_message),
'payload {{\n {0}\n {1}\n {2}\n {3}\n}}\n'.format(
*formatted_fields
),
)
message.Clear()
message.payload.optional_float = 1.1000000000011
self.assertEqual(
text_format.MessageToString(message),
'payload {\n optional_float: 1.1\n}\n',
)
message.payload.optional_float = 1.00000075e-36
self.assertEqual(
text_format.MessageToString(message),
'payload {\n optional_float: 1.00000075e-36\n}\n',
)
message.payload.optional_float = 12345678912345e11
self.assertEqual(
text_format.MessageToString(message),
'payload {\n optional_float: 1.234568e+24\n}\n',
)
def testMessageToString(self, message_module):
message = message_module.ForeignMessage()
message.c = 123
self.assertEqual('c: 123\n', str(message))
def testMessageToStringUnicode(self, message_module):
golden_unicode = 'Á short desçription and a 🍌.'
golden_bytes = golden_unicode.encode('utf-8')
message = message_module.TestAllTypes()
message.optional_string = golden_unicode
message.optional_bytes = golden_bytes
text = text_format.MessageToString(message, as_utf8=True)
golden_message = textwrap.dedent(
'optional_string: "Á short desçription and a 🍌."\n'
'optional_bytes: '
r'"\303\201 short des\303\247ription and a \360\237\215\214."'
'\n'
)
self.CompareToGoldenText(text, golden_message)
def testMessageToStringASCII(self, message_module):
golden_unicode = 'Á short desçription and a 🍌.'
golden_bytes = golden_unicode.encode('utf-8')
message = message_module.TestAllTypes()
message.optional_string = golden_unicode
message.optional_bytes = golden_bytes
text = text_format.MessageToString(message, as_utf8=False) # ASCII
golden_message = (
'optional_string: '
r'"\303\201 short des\303\247ription and a \360\237\215\214."'
'\n'
'optional_bytes: '
r'"\303\201 short des\303\247ription and a \360\237\215\214."'
'\n'
)
self.CompareToGoldenText(text, golden_message)
def testPrintField(self, message_module):
message = message_module.TestAllTypes()
field = message.DESCRIPTOR.fields_by_name['optional_float']
value = message.optional_float
out = text_format.TextWriter(False)
text_format.PrintField(field, value, out)
self.assertEqual('optional_float: 0.0\n', out.getvalue())
out.close()
# Test Printer
out = text_format.TextWriter(False)
printer = text_format._Printer(out)
printer.PrintField(field, value)
self.assertEqual('optional_float: 0.0\n', out.getvalue())
out.close()
def testPrintFieldValue(self, message_module):
message = message_module.TestAllTypes()
field = message.DESCRIPTOR.fields_by_name['optional_float']
value = message.optional_float
out = text_format.TextWriter(False)
text_format.PrintFieldValue(field, value, out)
self.assertEqual('0.0', out.getvalue())
out.close()
# Test Printer
out = text_format.TextWriter(False)
printer = text_format._Printer(out)
printer.PrintFieldValue(field, value)
self.assertEqual('0.0', out.getvalue())
out.close()
def testCustomOptions(self, message_module):
message_descriptor = (
unittest_custom_options_pb2.TestMessageWithCustomOptions.DESCRIPTOR
)
message_proto = descriptor_pb2.DescriptorProto()
message_descriptor.CopyToProto(message_proto)
expected_text = (
'name: "TestMessageWithCustomOptions"\n'
'field {\n'
' name: "field1"\n'
' number: 1\n'
' label: LABEL_OPTIONAL\n'
' type: TYPE_STRING\n'
' options {\n'
' ctype: CORD\n'
' [proto2_unittest.field_opt1]: 8765432109\n'
' }\n'
'}\n'
'field {\n'
' name: "oneof_field"\n'
' number: 2\n'
' label: LABEL_OPTIONAL\n'
' type: TYPE_INT32\n'
' oneof_index: 0\n'
'}\n'
'field {\n'
' name: "map_field"\n'
' number: 3\n'
' label: LABEL_REPEATED\n'
' type: TYPE_MESSAGE\n'
' type_name: ".proto2_unittest.TestMessageWithCustomOptions.'
'MapFieldEntry"\n'
' options {\n'
' [proto2_unittest.field_opt1]: 12345\n'
' }\n'
'}\n'
'nested_type {\n'
' name: "MapFieldEntry"\n'
' field {\n'
' name: "key"\n'
' number: 1\n'
' label: LABEL_OPTIONAL\n'
' type: TYPE_STRING\n'
' }\n'
' field {\n'
' name: "value"\n'
' number: 2\n'
' label: LABEL_OPTIONAL\n'
' type: TYPE_STRING\n'
' }\n'
' options {\n'
' map_entry: true\n'
' }\n'
'}\n'
'enum_type {\n'
' name: "AnEnum"\n'
' value {\n'
' name: "ANENUM_VAL1"\n'
' number: 1\n'
' }\n'
' value {\n'
' name: "ANENUM_VAL2"\n'
' number: 2\n'
' options {\n'
' [proto2_unittest.enum_value_opt1]: 123\n'
' }\n'
' }\n'
' options {\n'
' [proto2_unittest.enum_opt1]: -789\n'
' }\n'
'}\n'
'options {\n'
' message_set_wire_format: false\n'
' [proto2_unittest.message_opt1]: -56\n'
'}\n'
'oneof_decl {\n'
' name: "AnOneof"\n'
' options {\n'
' [proto2_unittest.oneof_opt1]: -99\n'
' }\n'
'}\n'
)
self.assertEqual(expected_text, text_format.MessageToString(message_proto))
parsed_proto = descriptor_pb2.DescriptorProto()
text_format.Parse(expected_text, parsed_proto)
self.assertEqual(message_proto, parsed_proto)
@unittest.skipIf(
api_implementation.Type() == 'upb',
"upb API doesn't support old UnknownField API. The TextFormat library "
'needs to convert to the new API.',
)
def testPrintUnknownFieldsEmbeddedMessageInBytes(self, message_module):
inner_msg = message_module.TestAllTypes()
inner_msg.optional_int32 = 101
inner_msg.optional_double = 102.0
inner_msg.optional_string = 'hello'
inner_msg.optional_bytes = b'103'
inner_msg.optional_nested_message.bb = 105
inner_data = inner_msg.SerializeToString()
outer_message = message_module.TestAllTypes()
outer_message.optional_int32 = 101
outer_message.optional_bytes = inner_data
all_data = outer_message.SerializeToString()
empty_message = message_module.TestEmptyMessage()
empty_message.ParseFromString(all_data)
self.assertEqual(
' 1: 101\n'
' 15 {\n'
' 1: 101\n'
' 12: 4636878028842991616\n'
' 14: "hello"\n'
' 15: "103"\n'
' 18 {\n'
' 1: 105\n'
' }\n'
' }\n',
text_format.MessageToString(
empty_message, indent=2, print_unknown_fields=True
),
)
self.assertEqual(
'1: 101 '
'15 { '
'1: 101 '
'12: 4636878028842991616 '
'14: "hello" '
'15: "103" '
'18 { 1: 105 } '
'}',
text_format.MessageToString(
empty_message, print_unknown_fields=True, as_one_line=True
),
)
def testBytestDoubleQuotes(self, message_module):
msg = message_module.TestAllTypes(optional_bytes=b'"')
self.assertEqual(str(msg), 'optional_bytes: "\\""\n')
def testBytesSingleQuote(self, message_module):
msg = message_module.TestAllTypes(optional_bytes=b"'")
self.assertEqual(str(msg), 'optional_bytes: "\\\'"\n')
@parameterized.parameters(unittest_pb2, unittest_proto3_arena_pb2)
class TextFormatMessageToTextBytesTests(TextFormatBase):
def testMessageToBytes(self, message_module):
message = message_module.ForeignMessage()
message.c = 123
self.assertEqual(b'c: 123\n', text_format.MessageToBytes(message))
def testRawUtf8RoundTrip(self, message_module):
message = message_module.TestAllTypes()
message.repeated_string.append('\u00fc\t\ua71f')
utf8_text = text_format.MessageToBytes(message, as_utf8=False)
golden_bytes = b'repeated_string: "\\303\\274\\t\\352\\234\\237"\n'
self.CompareToGoldenText(utf8_text, golden_bytes)
parsed_message = message_module.TestAllTypes()
text_format.Parse(utf8_text, parsed_message)
self.assertEqual(
message,
parsed_message,
'\n%s != %s (%s != %s)'
% (
message,
parsed_message,
message.repeated_string[0],
parsed_message.repeated_string[0],
),
)
def testRawUtf8RoundTripAsUtf8(self, message_module):
message = message_module.TestAllTypes()
message.repeated_string.append('\u00fc\t\ua71f')
utf8_text = text_format.MessageToString(message, as_utf8=True)
parsed_message = message_module.TestAllTypes()
text_format.Parse(utf8_text, parsed_message)
self.assertEqual(
message,
parsed_message,
'\n%s != %s (%s != %s)'
% (
message,
parsed_message,
message.repeated_string[0],
parsed_message.repeated_string[0],
),
)
# We can only test this case under proto2, because proto3 will reject invalid
# UTF-8 in the parser, so there should be no way of creating a string field
# that contains invalid UTF-8.
#
# We also can't test it in pure-Python, which validates all string fields for
# UTF-8 even when the spec says it shouldn't.
@unittest.skipIf(
api_implementation.Type() == 'python',
"Python can't create invalid UTF-8 strings",
)
def testInvalidUtf8RoundTrip(self, message_module):
if message_module is not unittest_pb2:
return
one_bytes = unittest_pb2.OneBytes()
one_bytes.data = b'ABC\xff123'
one_string = unittest_pb2.OneString()
one_string.ParseFromString(one_bytes.SerializeToString())
self.assertIn(
'data: "ABC\\377123"',
text_format.MessageToString(one_string, as_utf8=True),
)
def testEscapedUtf8ASCIIRoundTrip(self, message_module):
message = message_module.TestAllTypes()
message.repeated_string.append('\u00fc\t\ua71f')
ascii_text = text_format.MessageToBytes(message, as_utf8=False)
golden_bytes = b'repeated_string: "\\303\\274\\t\\352\\234\\237"\n'
self.CompareToGoldenText(ascii_text, golden_bytes)
parsed_message = message_module.TestAllTypes()
text_format.Parse(ascii_text, parsed_message)
self.assertEqual(
message,
parsed_message,
'\n%s != %s (%s != %s)'
% (
message,
parsed_message,
message.repeated_string[0],
parsed_message.repeated_string[0],
),
)
@parameterized.parameters(unittest_pb2, unittest_proto3_arena_pb2)
class TextFormatParserTests(TextFormatBase):
def testParseRecursionDepthLimit(self, message_module):
too_deep_text = textwrap.dedent("""\
child {
payload {}
child {
payload {}
child {}
}
}
""")
self.assertRaisesRegex(
text_format.ParseError,
'Message too deep. Max recursion depth is 3',
text_format.Parse,
too_deep_text,
message_module.NestedTestAllTypes(),
max_recursion_depth=3,
)
shallow_text = textwrap.dedent("""\
payload {}
child {
payload {}
}
""")
text_format.Parse(
shallow_text, message_module.NestedTestAllTypes(), max_recursion_depth=3
)
def testParseAnyRecursionDepthLimit(self, message_module):
del message_module
message = any_pb2.Any()
text = (
'[type.googleapis.com/google.protobuf.Any] {\n'
' [type.googleapis.com/google.protobuf.Any] {\n'
' [type.googleapis.com/google.protobuf.Any] {}\n'
' }\n'
'}\n'
)
with self.assertRaisesRegex(
text_format.ParseError,
'Message too deep. Max recursion depth is 2',
):
text_format.Parse(
text,
message,
descriptor_pool=descriptor_pool.Default(),
max_recursion_depth=2,
)
text_format.Parse(
text,
any_pb2.Any(),
descriptor_pool=descriptor_pool.Default(),
max_recursion_depth=4,
)
def testParseDefaultBehaviorRemainsUnbounded(self, message_module):
del message_module
message = descriptor_pb2.DescriptorProto()
text = 'nested_type {' * 110 + '}' * 110
try:
text_format.Parse(text, message)
except Exception as exc: # noqa: BLE001
self.fail(
'expected default parsing to remain unbounded, '
f'got {type(exc).__name__}: {exc}'
)
def testParseAllFields(self, message_module):
message = message_module.TestAllTypes()
test_util.SetAllFields(message)
ascii_text = text_format.MessageToString(message)
parsed_message = message_module.TestAllTypes()
text_format.Parse(ascii_text, parsed_message)
self.assertEqual(message, parsed_message)
if message_module is unittest_pb2:
test_util.ExpectAllFieldsSet(self, message)
def testParseAndMergeUtf8(self, message_module):
message = message_module.TestAllTypes()
test_util.SetAllFields(message)
ascii_text = text_format.MessageToString(message)
ascii_text = ascii_text.encode('utf-8')
parsed_message = message_module.TestAllTypes()
text_format.Parse(ascii_text, parsed_message)
self.assertEqual(message, parsed_message)
if message_module is unittest_pb2:
test_util.ExpectAllFieldsSet(self, message)
parsed_message.Clear()
text_format.Merge(ascii_text, parsed_message)
self.assertEqual(message, parsed_message)
if message_module is unittest_pb2:
test_util.ExpectAllFieldsSet(self, message)
msg2 = message_module.TestAllTypes()
text = 'optional_string: "café"'
text_format.Merge(text, msg2)
self.assertEqual(msg2.optional_string, 'café')
msg2.Clear()
self.assertEqual(msg2.optional_string, '')
text_format.Parse(text, msg2)
self.assertEqual(msg2.optional_string, 'café')
def testParseDoubleToFloat(self, message_module):
message = message_module.TestAllTypes()
text = 'repeated_float: 3.4028235e+39\nrepeated_float: 1.4028235e-39\n'
text_format.Parse(text, message)
self.assertEqual(message.repeated_float[0], float('inf'))
self.assertAlmostEqual(message.repeated_float[1], 1.4028235e-39)
def testParseExotic(self, message_module):
message = message_module.TestAllTypes()
text = (
'repeated_int64: -9223372036854775808\n'
'repeated_uint64: 18446744073709551615\n'
'repeated_double: 123.456\n'
'repeated_double: 1.23e+22\n'
'repeated_double: 1.23e-18\n'
'repeated_string: \n'
'"\\000\\001\\007\\010\\014\\n\\r\\t\\013\\\\\\\'\\""\n'
'repeated_string: "foo" \'corge\' "grault"\n'
'repeated_string: "\\303\\274\\352\\234\\237"\n'
'repeated_string: "\\xc3\\xbc"\n'
'repeated_string: "\xc3\xbc"\n'
)
text_format.Parse(text, message)
self.assertEqual(-9223372036854775808, message.repeated_int64[0])
self.assertEqual(18446744073709551615, message.repeated_uint64[0])
self.assertEqual(123.456, message.repeated_double[0])
self.assertEqual(1.23e22, message.repeated_double[1])
self.assertEqual(1.23e-18, message.repeated_double[2])
self.assertEqual('\000\001\a\b\f\n\r\t\v\\\'"', message.repeated_string[0])
self.assertEqual('foocorgegrault', message.repeated_string[1])
self.assertEqual('\u00fc\ua71f', message.repeated_string[2])
self.assertEqual('\u00fc', message.repeated_string[3])
def testParseTrailingCommas(self, message_module):
message = message_module.TestAllTypes()
text = (
'repeated_int64: 100;\n'
'repeated_int64: 200;\n'
'repeated_int64: 300,\n'
'repeated_string: "one",\n'
'repeated_string: "two";\n'
)
text_format.Parse(text, message)
self.assertEqual(100, message.repeated_int64[0])
self.assertEqual(200, message.repeated_int64[1])
self.assertEqual(300, message.repeated_int64[2])
self.assertEqual('one', message.repeated_string[0])
self.assertEqual('two', message.repeated_string[1])
def testParseRepeatedScalarShortFormat(self, message_module):
message = message_module.TestAllTypes()
text = (
'repeated_int64: [100, 200];\n'
'repeated_int64: []\n'
'repeated_int64: 300,\n'
'repeated_string: ["one", "two"];\n'
)
text_format.Parse(text, message)
self.assertEqual(100, message.repeated_int64[0])
self.assertEqual(200, message.repeated_int64[1])
self.assertEqual(300, message.repeated_int64[2])
self.assertEqual('one', message.repeated_string[0])
self.assertEqual('two', message.repeated_string[1])
def testParseRepeatedMessageShortFormat(self, message_module):
message = message_module.TestAllTypes()
text = (
'repeated_nested_message: [{bb: 100}, {bb: 200}],\n'
'repeated_nested_message: {bb: 300}\n'
'repeated_nested_message [{bb: 400}];\n'
)
text_format.Parse(text, message)
self.assertEqual(100, message.repeated_nested_message[0].bb)
self.assertEqual(200, message.repeated_nested_message[1].bb)
self.assertEqual(300, message.repeated_nested_message[2].bb)
self.assertEqual(400, message.repeated_nested_message[3].bb)
def testParseEmptyText(self, message_module):
message = message_module.TestAllTypes()
text = ''
text_format.Parse(text, message)
self.assertEqual(message_module.TestAllTypes(), message)
def testParseInvalidUtf8(self, message_module):
message = message_module.TestAllTypes()
text = b'invalid<\xc3\xc3>'
with self.assertRaises(text_format.ParseError):
text_format.Parse(text, message)
def testParseInvalidUtf8Value(self, message_module):
message = message_module.TestAllTypes()
text = 'repeated_string: "\\xc3\\xc3"'
with self.assertRaises(text_format.ParseError) as e:
text_format.Parse(text, message)
self.assertEqual(e.exception.GetLine(), 1)
self.assertEqual(e.exception.GetColumn(), 28)
def testParseSingleWord(self, message_module):
message = message_module.TestAllTypes()
text = 'foo'
self.assertRaisesRegex(
text_format.ParseError,
(
r'1:1 : Message type "\w+.TestAllTypes" has no field named '
r'"foo".'
),
text_format.Parse,
text,
message,
)
def testParseUnknownField(self, message_module):
message = message_module.TestAllTypes()
text = 'unknown_field: 8\n'
self.assertRaisesRegex(
text_format.ParseError,
(
r'1:1 : Message type "\w+.TestAllTypes" has no field named '
r'"unknown_field".'
),
text_format.Parse,
text,
message,
)
text = (
'optional_int32: 123\n'
'unknown_field: 8\n'
'optional_nested_message { bb: 45 }'
)
text_format.Parse(text, message, allow_unknown_field=True)
self.assertEqual(message.optional_nested_message.bb, 45)
self.assertEqual(message.optional_int32, 123)
def testParseUnknownRepeatedMessage(self, message_module):
message = message_module.TestAllTypes()
text = (
'unknown_repeated: [{}]\n'
'unknown_repeated2: [<>, {}]\n'
'optional_nested_message { bb: 45 }'
)
text_format.Parse(text, message, allow_unknown_field=True)
self.assertEqual(message.optional_nested_message.bb, 45)
def testParseBadEnumValue(self, message_module):
message = message_module.TestAllTypes()
text = 'optional_nested_enum: BARR'
self.assertRaisesRegex(
text_format.ParseError,
(
r'1:23 : \'optional_nested_enum: BARR\': '
r'Enum type "\w+.TestAllTypes.NestedEnum" '
r'has no value named BARR.'
),
text_format.Parse,
text,
message,
)
def testParseBadIntValue(self, message_module):
message = message_module.TestAllTypes()
text = 'optional_int32: bork'
self.assertRaisesRegex(
text_format.ParseError,
"1:17 : 'optional_int32: bork': Couldn't parse integer: bork",
text_format.Parse,
text,
message,
)
def testParseStringFieldUnescape(self, message_module):
message = message_module.TestAllTypes()
text = r'''repeated_string: "\xf\x62"
repeated_string: "\\xf\\x62"
repeated_string: "\\\xf\\\x62"
repeated_string: "\\\\xf\\\\x62"
repeated_string: "\\\\\xf\\\\\x62"
repeated_string: "\x5cx20"'''
text_format.Parse(text, message)
SLASH = '\\'
self.assertEqual('\x0fb', message.repeated_string[0])
self.assertEqual(SLASH + 'xf' + SLASH + 'x62', message.repeated_string[1])
self.assertEqual(SLASH + '\x0f' + SLASH + 'b', message.repeated_string[2])
self.assertEqual(
SLASH + SLASH + 'xf' + SLASH + SLASH + 'x62', message.repeated_string[3]
)
self.assertEqual(
SLASH + SLASH + '\x0f' + SLASH + SLASH + 'b', message.repeated_string[4]
)
self.assertEqual(SLASH + 'x20', message.repeated_string[5])
def testParseOneof(self, message_module):
m = message_module.TestAllTypes()
m.oneof_uint32 = 11
m2 = message_module.TestAllTypes()
text_format.Parse(text_format.MessageToString(m), m2)
self.assertEqual('oneof_uint32', m2.WhichOneof('oneof_field'))
def testParseMultipleOneof(self, message_module):
m_string = '\n'.join(['oneof_uint32: 11', 'oneof_string: "foo"'])
m2 = message_module.TestAllTypes()
with self.assertRaisesRegex(
text_format.ParseError, ' is specified along with field '
):
text_format.Parse(m_string, m2)
# This example contains non-ASCII codepoint unicode data as literals
# which should come through as utf-8 for bytes, and as the unicode
# itself for string fields. It also demonstrates escaped binary data.
# The ur"" string prefix is unfortunately missing from Python 3
# so we resort to double escaping our \s so that they come through.
_UNICODE_SAMPLE = """
optional_bytes: 'Á short desçription'
optional_string: 'Á short desçription'
repeated_bytes: '\\303\\201 short des\\303\\247ription'
repeated_bytes: '\\u00c1 short des\\u00e7ription'
repeated_bytes: '\\x12\\x34\\x56\\x78\\x90\\xab\\xcd\\xef'
repeated_string: '\\xd0\\x9f\\xd1\\x80\\xd0\\xb8\\xd0\\xb2\\xd0\\xb5\\xd1\\x82'
"""
_BYTES_SAMPLE = _UNICODE_SAMPLE.encode('utf-8')
_GOLDEN_UNICODE = 'Á short desçription'
_GOLDEN_BYTES = _GOLDEN_UNICODE.encode('utf-8')
_GOLDEN_BYTES_1 = b'\x12\x34\x56\x78\x90\xab\xcd\xef'
_GOLDEN_STR_0 = 'Привет'
def testParseUnicode(self, message_module):
m = message_module.TestAllTypes()
text_format.Parse(self._UNICODE_SAMPLE, m)
self.assertEqual(m.optional_bytes, self._GOLDEN_BYTES)
self.assertEqual(m.optional_string, self._GOLDEN_UNICODE)
self.assertEqual(m.repeated_bytes[0], self._GOLDEN_BYTES)
self.assertEqual(m.repeated_bytes[1], self._GOLDEN_BYTES)
# repeated_bytes[2] contained simple \ escaped non-UTF-8 raw binary data.
self.assertEqual(m.repeated_bytes[2], self._GOLDEN_BYTES_1)
# repeated_string[0] contained \ escaped data representing the UTF-8
# representation of _GOLDEN_STR_0 - it needs to decode as such.
self.assertEqual(m.repeated_string[0], self._GOLDEN_STR_0)
def testParseBytes(self, message_module):
m = message_module.TestAllTypes()
text_format.Parse(self._BYTES_SAMPLE, m)
self.assertEqual(m.optional_bytes, self._GOLDEN_BYTES)
self.assertEqual(m.optional_string, self._GOLDEN_UNICODE)
self.assertEqual(m.repeated_bytes[0], self._GOLDEN_BYTES)
self.assertEqual(m.repeated_bytes[1], self._GOLDEN_BYTES)
# repeated_bytes[1] contained simple \ escaped non-UTF-8 raw binary data.
self.assertEqual(m.repeated_bytes[2], self._GOLDEN_BYTES_1)
# repeated_string[0] contained \ escaped data representing the UTF-8
# representation of _GOLDEN_STR_0 - it needs to decode as such.
self.assertEqual(m.repeated_string[0], self._GOLDEN_STR_0)
def testFromBytesFile(self, message_module):
m = message_module.TestAllTypes()
f = io.BytesIO(self._BYTES_SAMPLE)
text_format.ParseLines(f, m)
self.assertEqual(m.optional_bytes, self._GOLDEN_BYTES)
self.assertEqual(m.optional_string, self._GOLDEN_UNICODE)
self.assertEqual(m.repeated_bytes[0], self._GOLDEN_BYTES)
def testFromUnicodeFile(self, message_module):
m = message_module.TestAllTypes()
f = io.StringIO(self._UNICODE_SAMPLE)
text_format.ParseLines(f, m)
self.assertEqual(m.optional_bytes, self._GOLDEN_BYTES)
self.assertEqual(m.optional_string, self._GOLDEN_UNICODE)
self.assertEqual(m.repeated_bytes[0], self._GOLDEN_BYTES)
def testFromBytesLines(self, message_module):
m = message_module.TestAllTypes()
text_format.ParseLines(self._BYTES_SAMPLE.split(b'\n'), m)
self.assertEqual(m.optional_bytes, self._GOLDEN_BYTES)
self.assertEqual(m.optional_string, self._GOLDEN_UNICODE)
self.assertEqual(m.repeated_bytes[0], self._GOLDEN_BYTES)
def testFromUnicodeLines(self, message_module):
m = message_module.TestAllTypes()
text_format.ParseLines(self._UNICODE_SAMPLE.split('\n'), m)
self.assertEqual(m.optional_bytes, self._GOLDEN_BYTES)
self.assertEqual(m.optional_string, self._GOLDEN_UNICODE)
self.assertEqual(m.repeated_bytes[0], self._GOLDEN_BYTES)
def testParseDuplicateMessages(self, message_module):
message = message_module.TestAllTypes()
text = 'optional_nested_message { bb: 1 } optional_nested_message { bb: 2 }'
self.assertRaisesRegex(
text_format.ParseError,
(
r'1:59 : Message type "\w+.TestAllTypes" '
r'should not have multiple "optional_nested_message" fields.'
),
text_format.Parse,
text,
message,
)
def testParseDuplicateScalars(self, message_module):
message = message_module.TestAllTypes()
text = 'optional_int32: 42 optional_int32: 67'
self.assertRaisesRegex(
text_format.ParseError,
(
r'1:36 : Message type "\w+.TestAllTypes" should not '
r'have multiple "optional_int32" fields.'
),
text_format.Parse,
text,
message,
)
def testParseDuplicateNegativeZero(self, message_module):
message = message_module.TestAllTypes()
text = 'optional_double: -0.0 optional_double: 3'
self.assertRaisesRegex(
text_format.ParseError,
(
r'1:40 : Message type "\w+.TestAllTypes" should not '
r'have multiple "optional_double" fields.'
),
text_format.Parse,
text,
message,
)
def testParseExistingScalarInMessage(self, message_module):
message = message_module.TestAllTypes(optional_int32=42)
text = 'optional_int32: 67'
self.assertRaisesRegex(
text_format.ParseError,
(
r'Message type "\w+.TestAllTypes" should not '
r'have multiple "optional_int32" fields.'
),
text_format.Parse,
text,
message,
)
@parameterized.parameters(unittest_pb2, unittest_proto3_arena_pb2)
class TextFormatMergeTests(TextFormatBase):
def testMergeDuplicateScalarsInText(self, message_module):
message = message_module.TestAllTypes()
text = 'optional_int32: 42 optional_int32: 67'
r = text_format.Merge(text, message)
self.assertIs(r, message)
self.assertEqual(67, message.optional_int32)
def testMergeDuplicateNestedMessageScalars(self, message_module):
message = message_module.TestAllTypes()
text = 'optional_nested_message { bb: 1 } optional_nested_message { bb: 2 }'
r = text_format.Merge(text, message)
self.assertTrue(r is message)
self.assertEqual(2, message.optional_nested_message.bb)
def testReplaceScalarInMessage(self, message_module):
message = message_module.TestAllTypes(optional_int32=42)
text = 'optional_int32: 67'
r = text_format.Merge(text, message)
self.assertIs(r, message)
self.assertEqual(67, message.optional_int32)
def testReplaceMessageInMessage(self, message_module):
message = message_module.TestAllTypes(
optional_int32=42, optional_nested_message=dict()
)
self.assertTrue(message.HasField('optional_nested_message'))
text = 'optional_nested_message{ bb: 3 }'
r = text_format.Merge(text, message)
self.assertIs(r, message)
self.assertEqual(3, message.optional_nested_message.bb)
def testMergeMultipleOneof(self, message_module):
m_string = '\n'.join(['oneof_uint32: 11', 'oneof_string: "foo"'])
m2 = message_module.TestAllTypes()
text_format.Merge(m_string, m2)
self.assertEqual('oneof_string', m2.WhichOneof('oneof_field'))
# These are tests that aren't fundamentally specific to proto2, but are at
# the moment because of differences between the proto2 and proto3 test schemas.
# Ideally the schemas would be made more similar so these tests could pass.
class OnlyWorksWithProto2RightNowTests(TextFormatBase):
def testPrintAllFieldsPointy(self):
message = unittest_pb2.TestAllTypes()
test_util.SetAllFields(message)
self.CompareToGoldenFile(
self.RemoveRedundantZeros(
text_format.MessageToString(message, pointy_brackets=True)
),
'text_format_unittest_data_pointy_oneof.txt',
)
def testParseGolden(self):
golden_text = '\n'.join(
self.ReadGolden('text_format_unittest_data_oneof_implemented.txt')
)
parsed_message = unittest_pb2.TestAllTypes()
r = text_format.Parse(golden_text, parsed_message)
self.assertIs(r, parsed_message)
message = unittest_pb2.TestAllTypes()
test_util.SetAllFields(message)
self.assertEqual(message, parsed_message)
def testPrintAllFields(self):
message = unittest_pb2.TestAllTypes()
test_util.SetAllFields(message)
self.CompareToGoldenFile(
self.RemoveRedundantZeros(text_format.MessageToString(message)),
'text_format_unittest_data_oneof_implemented.txt',
)
def testPrintUnknownFields(self):
message = unittest_pb2.TestAllTypes()
message.optional_int32 = 101
message.optional_double = 102.0
message.optional_string = 'hello'
message.optional_bytes = b'103'
message.optionalgroup.a = 104
message.optional_nested_message.bb = 105
all_data = message.SerializeToString()
empty_message = unittest_pb2.TestEmptyMessage()
empty_message.ParseFromString(all_data)
self.assertEqual(
' 1: 101\n'
' 12: 4636878028842991616\n'
' 14: "hello"\n'
' 15: "103"\n'
' 16 {\n'
' 17: 104\n'
' }\n'
' 18 {\n'
' 1: 105\n'
' }\n',
text_format.MessageToString(
empty_message, indent=2, print_unknown_fields=True
),
)
self.assertEqual(
'1: 101 '
'12: 4636878028842991616 '
'14: "hello" '
'15: "103" '
'16 { 17: 104 } '
'18 { 1: 105 }',
text_format.MessageToString(
empty_message, print_unknown_fields=True, as_one_line=True
),
)
def testPrintInIndexOrder(self):
message = unittest_pb2.TestFieldOrderings()
# Fields are listed in index order instead of field number.
message.my_string = 'str'
message.my_int = 101
message.my_float = 111
message.optional_nested_message.oo = 0
message.optional_nested_message.bb = 1
message.Extensions[unittest_pb2.my_extension_string] = 'ext_str0'
# Extensions are listed based on the order of extension number.
# Extension number 12.
message.Extensions[
unittest_pb2.TestExtensionOrderings2.test_ext_orderings2
].my_string = 'ext_str2'
# Extension number 13.
message.Extensions[
unittest_pb2.TestExtensionOrderings1.test_ext_orderings1
].my_string = 'ext_str1'
# Extension number 14.
message.Extensions[
unittest_pb2.TestExtensionOrderings2.TestExtensionOrderings3.test_ext_orderings3
].my_string = 'ext_str3'
# Print in index order.
self.CompareToGoldenText(
self.RemoveRedundantZeros(
text_format.MessageToString(message, use_index_order=True)
),
'my_string: "str"\n'
'my_int: 101\n'
'my_float: 111\n'
'optional_nested_message {\n'
' oo: 0\n'
' bb: 1\n'
'}\n'
'[proto2_unittest.TestExtensionOrderings2.test_ext_orderings2] {\n'
' my_string: "ext_str2"\n'
'}\n'
'[proto2_unittest.TestExtensionOrderings1.test_ext_orderings1] {\n'
' my_string: "ext_str1"\n'
'}\n'
'[proto2_unittest.TestExtensionOrderings2.TestExtensionOrderings3'
'.test_ext_orderings3] {\n'
' my_string: "ext_str3"\n'
'}\n'
'[proto2_unittest.my_extension_string]: "ext_str0"\n',
)
# By default, print in field number order.
self.CompareToGoldenText(
self.RemoveRedundantZeros(text_format.MessageToString(message)),
'my_int: 101\n'
'my_string: "str"\n'
'[proto2_unittest.TestExtensionOrderings2.test_ext_orderings2] {\n'
' my_string: "ext_str2"\n'
'}\n'
'[proto2_unittest.TestExtensionOrderings1.test_ext_orderings1] {\n'
' my_string: "ext_str1"\n'
'}\n'
'[proto2_unittest.TestExtensionOrderings2.TestExtensionOrderings3'
'.test_ext_orderings3] {\n'
' my_string: "ext_str3"\n'
'}\n'
'[proto2_unittest.my_extension_string]: "ext_str0"\n'
'my_float: 111\n'
'optional_nested_message {\n'
' bb: 1\n'
' oo: 0\n'
'}\n',
)
def testMergeLinesGolden(self):
opened = self.ReadGolden('text_format_unittest_data_oneof_implemented.txt')
parsed_message = unittest_pb2.TestAllTypes()
r = text_format.MergeLines(opened, parsed_message)
self.assertIs(r, parsed_message)
message = unittest_pb2.TestAllTypes()
test_util.SetAllFields(message)
self.assertEqual(message, parsed_message)
def testParseLinesGolden(self):
opened = self.ReadGolden('text_format_unittest_data_oneof_implemented.txt')
parsed_message = unittest_pb2.TestAllTypes()
r = text_format.ParseLines(opened, parsed_message)
self.assertIs(r, parsed_message)
message = unittest_pb2.TestAllTypes()
test_util.SetAllFields(message)
self.assertEqual(message, parsed_message)
def testPrintMapEmptyKeys(self):
message = map_unittest_pb2.TestMap()
message.map_int32_int32[0] = 123
message.map_int64_int64[0] = 2**33
message.map_uint32_uint32[0] = 123
message.map_uint64_uint64[0] = 2**33
message.map_string_string[''] = 'world'
message.map_int32_foreign_message[0].c = 111
self.CompareToGoldenText(
text_format.MessageToString(message),
'map_int32_int32 {\n'
' value: 123\n'
'}\n'
'map_int64_int64 {\n'
' value: 8589934592\n'
'}\n'
'map_uint32_uint32 {\n'
' value: 123\n'
'}\n'
'map_uint64_uint64 {\n'
' value: 8589934592\n'
'}\n'
'map_string_string {\n'
' value: "world"\n'
'}\n'
'map_int32_foreign_message {\n'
' value {\n'
' c: 111\n'
' }\n'
'}\n',
)
def testPrintMapEmptyValues(self):
message = map_unittest_pb2.TestMap()
message.map_int32_int32[-123] = 0
message.map_int64_int64[-(2**33)] = 0
message.map_uint32_uint32[123] = 0
message.map_uint64_uint64[2**33] = 0
message.map_string_string['hello'] = ''
message.map_int32_foreign_message[111].c = 0
self.CompareToGoldenText(
text_format.MessageToString(message),
'map_int32_int32 {\n'
' key: -123\n'
'}\n'
'map_int64_int64 {\n'
' key: -8589934592\n'
'}\n'
'map_uint32_uint32 {\n'
' key: 123\n'
'}\n'
'map_uint64_uint64 {\n'
' key: 8589934592\n'
'}\n'
'map_string_string {\n'
' key: "hello"\n'
'}\n'
'map_int32_foreign_message {\n'
' key: 111\n'
' value {\n'
' c: 0\n'
' }\n'
'}\n',
)
def testPrintMap(self):
message = map_unittest_pb2.TestMap()
message.map_int32_int32[-123] = -456
message.map_int64_int64[-(2**33)] = -(2**34)
message.map_uint32_uint32[123] = 456
message.map_uint64_uint64[2**33] = 2**34
message.map_string_string['abc'] = '123'
message.map_int32_foreign_message[111].c = 5
# Maps are serialized to text format using their underlying repeated
# representation.
self.CompareToGoldenText(
text_format.MessageToString(message),
'map_int32_int32 {\n'
' key: -123\n'
' value: -456\n'
'}\n'
'map_int64_int64 {\n'
' key: -8589934592\n'
' value: -17179869184\n'
'}\n'
'map_uint32_uint32 {\n'
' key: 123\n'
' value: 456\n'
'}\n'
'map_uint64_uint64 {\n'
' key: 8589934592\n'
' value: 17179869184\n'
'}\n'
'map_string_string {\n'
' key: "abc"\n'
' value: "123"\n'
'}\n'
'map_int32_foreign_message {\n'
' key: 111\n'
' value {\n'
' c: 5\n'
' }\n'
'}\n',
)
def testDuplicateMapKey(self):
message = map_unittest_pb2.TestMap()
text = (
'map_uint64_uint64 {\n'
' key: 123\n'
' value: 17179869184\n'
'}\n'
'map_string_string {\n'
' key: "abc"\n'
' value: "first"\n'
'}\n'
'map_int32_foreign_message {\n'
' key: 111\n'
' value {\n'
' c: 5\n'
' }\n'
'}\n'
'map_uint64_uint64 {\n'
' key: 123\n'
' value: 321\n'
'}\n'
'map_string_string {\n'
' key: "abc"\n'
' value: "second"\n'
'}\n'
'map_int32_foreign_message {\n'
' key: 111\n'
' value {\n'
' d: 5\n'
' }\n'
'}\n'
)
text_format.Parse(text, message)
self.CompareToGoldenText(
text_format.MessageToString(message),
'map_uint64_uint64 {\n'
' key: 123\n'
' value: 321\n'
'}\n'
'map_string_string {\n'
' key: "abc"\n'
' value: "second"\n'
'}\n'
'map_int32_foreign_message {\n'
' key: 111\n'
' value {\n'
' d: 5\n'
' }\n'
'}\n',
)
# In cpp implementation, __str__ calls the cpp implementation of text format.
def testPrintMapUsingCppImplementation(self):
message = map_unittest_pb2.TestMap()
inner_msg = message.map_int32_foreign_message[111]
inner_msg.c = 1
self.assertEqual(
str(message),
'map_int32_foreign_message {\n'
' key: 111\n'
' value {\n'
' c: 1\n'
' }\n'
'}\n',
)
inner_msg.c = 2
self.assertEqual(
str(message),
'map_int32_foreign_message {\n'
' key: 111\n'
' value {\n'
' c: 2\n'
' }\n'
'}\n',
)
def testMapOrderEnforcement(self):
message = map_unittest_pb2.TestMap()
for letter in string.ascii_uppercase[13:26]:
message.map_string_string[letter] = 'dummy'
for letter in reversed(string.ascii_uppercase[0:13]):
message.map_string_string[letter] = 'dummy'
golden = ''.join((
'map_string_string {\n key: "%c"\n value: "dummy"\n}\n' % (letter,)
for letter in string.ascii_uppercase
))
self.CompareToGoldenText(text_format.MessageToString(message), golden)
# TODO: In c/137553523, not serializing default value for map entry
# message has been fixed. This test needs to be disabled in order to submit
# that cl. Add this back when c/137553523 has been submitted.
# def testMapOrderSemantics(self):
# golden_lines = self.ReadGolden('map_test_data.txt')
# message = map_unittest_pb2.TestMap()
# text_format.ParseLines(golden_lines, message)
# candidate = text_format.MessageToString(message)
# # The Python implementation emits "1.0" for the double value that the C++
# # implementation emits as "1".
# candidate = candidate.replace('1.0', '1', 2)
# candidate = candidate.replace('0.0', '0', 2)
# self.assertMultiLineEqual(candidate, ''.join(golden_lines))
# Tests of proto2-only features (MessageSet, extensions, etc.).
class Proto2Tests(TextFormatBase):
def testPrintMessageSet(self):
message = unittest_mset_pb2.TestMessageSetContainer()
ext1 = unittest_mset_pb2.TestMessageSetExtension1.message_set_extension
ext2 = unittest_mset_pb2.TestMessageSetExtension2.message_set_extension
message.message_set.Extensions[ext1].i = 23
message.message_set.Extensions[ext2].str = 'foo'
self.CompareToGoldenText(
text_format.MessageToString(message),
'message_set {\n'
' [proto2_unittest.TestMessageSetExtension1] {\n'
' i: 23\n'
' }\n'
' [proto2_unittest.TestMessageSetExtension2] {\n'
' str: "foo"\n'
' }\n'
'}\n',
)
message = message_set_extensions_pb2.TestMessageSet()
ext = message_set_extensions_pb2.message_set_extension3
message.Extensions[ext].text = 'bar'
self.CompareToGoldenText(
text_format.MessageToString(message),
'[google.protobuf.internal.TestMessageSetExtension3] {\n'
' text: "bar"\n'
'}\n',
)
def testMessageSetExtensionNotFirst(self):
desc = message_set_extensions_pb2.TestMessageSetExtension1.DESCRIPTOR
self.assertEqual('first_extension', desc.extensions[0].name)
self.assertEqual('message_set_extension', desc.extensions[1].name)
message = message_set_extensions_pb2.TestMessageSet()
ext = (
message_set_extensions_pb2.TestMessageSetExtension1.message_set_extension
)
message.Extensions[ext].i = 123
expected_str = (
'[google.protobuf.internal.TestMessageSetExtension1] {\n i: 123\n}\n'
)
self.CompareToGoldenText(text_format.MessageToString(message), expected_str)
parsed = message_set_extensions_pb2.TestMessageSet()
text_format.Parse(expected_str, parsed)
self.CompareToGoldenText(text_format.MessageToString(parsed), expected_str)
def testPrintMessageSetByFieldNumber(self):
out = text_format.TextWriter(False)
message = unittest_mset_pb2.TestMessageSetContainer()
ext1 = unittest_mset_pb2.TestMessageSetExtension1.message_set_extension
ext2 = unittest_mset_pb2.TestMessageSetExtension2.message_set_extension
message.message_set.Extensions[ext1].i = 23
message.message_set.Extensions[ext2].str = 'foo'
text_format.PrintMessage(message, out, use_field_number=True)
self.CompareToGoldenText(
out.getvalue(),
'1 {\n'
' 1545008 {\n'
' 15: 23\n'
' }\n'
' 1547769 {\n'
' 25: "foo"\n'
' }\n'
'}\n',
)
out.close()
def testPrintMessageSetAsOneLine(self):
message = unittest_mset_pb2.TestMessageSetContainer()
ext1 = unittest_mset_pb2.TestMessageSetExtension1.message_set_extension
ext2 = unittest_mset_pb2.TestMessageSetExtension2.message_set_extension
message.message_set.Extensions[ext1].i = 23
message.message_set.Extensions[ext2].str = 'foo'
self.CompareToGoldenText(
text_format.MessageToString(message, as_one_line=True),
'message_set {'
' [proto2_unittest.TestMessageSetExtension1] {'
' i: 23'
' }'
' [proto2_unittest.TestMessageSetExtension2] {'
' str: "foo"'
' }'
' }',
)
def testParseMessageSet(self):
message = unittest_pb2.TestAllTypes()
text = 'repeated_uint64: 1\nrepeated_uint64: 2\n'
text_format.Parse(text, message)
self.assertEqual(1, message.repeated_uint64[0])
self.assertEqual(2, message.repeated_uint64[1])
message = unittest_mset_pb2.TestMessageSetContainer()
text = (
'message_set {\n'
' [proto2_unittest.TestMessageSetExtension1] {\n'
' i: 23\n'
' }\n'
' [proto2_unittest.TestMessageSetExtension2] {\n'
' str: "foo"\n'
' }\n'
'}\n'
)
text_format.Parse(text, message)
ext1 = unittest_mset_pb2.TestMessageSetExtension1.message_set_extension
ext2 = unittest_mset_pb2.TestMessageSetExtension2.message_set_extension
self.assertEqual(23, message.message_set.Extensions[ext1].i)
self.assertEqual('foo', message.message_set.Extensions[ext2].str)
def testExtensionInsideAnyMessage(self):
message = test_extend_any.TestAny()
text = (
'value {\n'
' [type.googleapis.com/google.protobuf.internal.TestAny] {\n'
' [google.protobuf.internal.TestAnyExtension1.extension1] {\n'
' i: 10\n'
' }\n'
' }\n'
'}\n'
)
text_format.Merge(text, message, descriptor_pool=descriptor_pool.Default())
self.CompareToGoldenText(
text_format.MessageToString(
message, descriptor_pool=descriptor_pool.Default()
),
text,
)
def testParseMessageByFieldNumber(self):
message = unittest_pb2.TestAllTypes()
text = '34: 1\nrepeated_uint64: 2\n'
text_format.Parse(text, message, allow_field_number=True)
self.assertEqual(1, message.repeated_uint64[0])
self.assertEqual(2, message.repeated_uint64[1])
message = unittest_mset_pb2.TestMessageSetContainer()
text = (
'1 {\n'
' 1545008 {\n'
' 15: 23\n'
' }\n'
' 1547769 {\n'
' 25: "foo"\n'
' }\n'
'}\n'
)
text_format.Parse(text, message, allow_field_number=True)
ext1 = unittest_mset_pb2.TestMessageSetExtension1.message_set_extension
ext2 = unittest_mset_pb2.TestMessageSetExtension2.message_set_extension
self.assertEqual(23, message.message_set.Extensions[ext1].i)
self.assertEqual('foo', message.message_set.Extensions[ext2].str)
# Can't parse field number without set allow_field_number=True.
message = unittest_pb2.TestAllTypes()
text = '34:1\n'
self.assertRaisesRegex(
text_format.ParseError,
(r'1:1 : Message type "\w+.TestAllTypes" has no field named ' r'"34".'),
text_format.Parse,
text,
message,
)
# Can't parse if field number is not found.
text = '1234:1\n'
self.assertRaisesRegex(
text_format.ParseError,
(
r'1:1 : Message type "\w+.TestAllTypes" has no field named '
r'"1234".'
),
text_format.Parse,
text,
message,
allow_field_number=True,
)
def testPrintAllExtensions(self):
message = unittest_pb2.TestAllExtensions()
test_util.SetAllExtensions(message)
self.CompareToGoldenFile(
self.RemoveRedundantZeros(text_format.MessageToString(message)),
'text_format_unittest_extensions_data.txt',
)
def testPrintAllExtensionsPointy(self):
message = unittest_pb2.TestAllExtensions()
test_util.SetAllExtensions(message)
self.CompareToGoldenFile(
self.RemoveRedundantZeros(
text_format.MessageToString(message, pointy_brackets=True)
),
'text_format_unittest_extensions_data_pointy.txt',
)
def testParseGoldenExtensions(self):
golden_text = '\n'.join(
self.ReadGolden('text_format_unittest_extensions_data.txt')
)
parsed_message = unittest_pb2.TestAllExtensions()
text_format.Parse(golden_text, parsed_message)
message = unittest_pb2.TestAllExtensions()
test_util.SetAllExtensions(message)
self.assertEqual(message, parsed_message)
def testParseAllExtensions(self):
message = unittest_pb2.TestAllExtensions()
test_util.SetAllExtensions(message)
ascii_text = text_format.MessageToString(message)
parsed_message = unittest_pb2.TestAllExtensions()
text_format.Parse(ascii_text, parsed_message)
self.assertEqual(message, parsed_message)
def testParseAllowedUnknownExtension(self):
# Skip over unknown extension correctly.
message = unittest_mset_pb2.TestMessageSetContainer()
text = (
'message_set {\n'
' [unknown_extension] {\n'
' i: 23\n'
' repeated_i: []\n'
' bin: "\xe0"\n'
' [nested_unknown_ext]: {\n'
' i: 23\n'
' repeated_i: [1, 2]\n'
' x: x\n'
' test: "test_string"\n'
' floaty_float: -0.315\n'
' num: -inf\n'
' multiline_str: "abc"\n'
' "def"\n'
' "xyz."\n'
' [nested_unknown_ext.ext]: <\n'
' i: 23\n'
' i: 24\n'
' pointfloat: .3\n'
' test: "test_string"\n'
' repeated_test: ["test_string1", "test_string2"]\n'
' floaty_float: -0.315\n'
' num: -inf\n'
' long_string: "test" "test2" \n'
' >\n'
' }\n'
' }\n'
' [unknown_extension]: 5\n'
' [unknown_extension_with_number_field] {\n'
' 1: "some_field"\n'
' 2: -0.451\n'
' }\n'
'}\n'
)
text_format.Parse(text, message, allow_unknown_extension=True)
golden = 'message_set {\n}\n'
self.CompareToGoldenText(text_format.MessageToString(message), golden)
# Catch parse errors in unknown extension.
message = unittest_mset_pb2.TestMessageSetContainer()
malformed = (
'message_set {\n'
' [unknown_extension] {\n'
' i:\n' # Missing value.
' }\n'
'}\n'
)
self.assertRaisesRegex(
text_format.ParseError,
'Invalid field value: }',
text_format.Parse,
malformed,
message,
allow_unknown_extension=True,
)
message = unittest_mset_pb2.TestMessageSetContainer()
malformed = (
'message_set {\n'
' [unknown_extension] {\n'
' str: "malformed string\n' # Missing closing quote.
' }\n'
'}\n'
)
self.assertRaisesRegex(
text_format.ParseError,
'Invalid field value: "',
text_format.Parse,
malformed,
message,
allow_unknown_extension=True,
)
message = unittest_mset_pb2.TestMessageSetContainer()
malformed = (
'message_set {\n'
' [unknown_extension] {\n'
' str: "malformed\n multiline\n string\n'
' }\n'
'}\n'
)
self.assertRaisesRegex(
text_format.ParseError,
'Invalid field value: "',
text_format.Parse,
malformed,
message,
allow_unknown_extension=True,
)
message = unittest_mset_pb2.TestMessageSetContainer()
malformed = (
'message_set {\n'
' [malformed_extension] <\n'
' i: -5\n'
' \n' # Missing '>' here.
'}\n'
)
self.assertRaisesRegex(
text_format.ParseError,
'5:1 : \'}\': Expected ">".',
text_format.Parse,
malformed,
message,
allow_unknown_extension=True,
)
# Don't allow unknown fields with allow_unknown_extension=True.
message = unittest_mset_pb2.TestMessageSetContainer()
malformed = 'message_set {\n unknown_field: true\n}\n'
self.assertRaisesRegex(
text_format.ParseError,
(
'2:3 : Message type '
'"proto2_wireformat_unittest.TestMessageSet" has no'
' field named "unknown_field".'
),
text_format.Parse,
malformed,
message,
allow_unknown_extension=True,
)
# Parse known extension correctly.
message = unittest_mset_pb2.TestMessageSetContainer()
text = (
'message_set {\n'
' [proto2_unittest.TestMessageSetExtension1] {\n'
' i: 23\n'
' }\n'
' [proto2_unittest.TestMessageSetExtension2] {\n'
' str: "foo"\n'
' }\n'
'}\n'
)
text_format.Parse(text, message, allow_unknown_extension=True)
ext1 = unittest_mset_pb2.TestMessageSetExtension1.message_set_extension
ext2 = unittest_mset_pb2.TestMessageSetExtension2.message_set_extension
self.assertEqual(23, message.message_set.Extensions[ext1].i)
self.assertEqual('foo', message.message_set.Extensions[ext2].str)
# Handle Any messages inside unknown extensions.
message = any_test_pb2.TestAny()
text = (
'any_value {\n'
' [type.googleapis.com/google.protobuf.internal.TestAny] {\n'
' [unknown_extension] {\n'
' str: "string"\n'
' any_value {\n'
' [type.googleapis.com/proto2_unittest.OneString] {\n'
' data: "string"\n'
' }\n'
' }\n'
' }\n'
' }\n'
'}\n'
'int32_value: 123'
)
text_format.Parse(text, message, allow_unknown_extension=True)
self.assertEqual(123, message.int32_value)
# Fail if invalid Any message type url inside unknown extensions.
message = any_test_pb2.TestAny()
text = (
'any_value {\n'
' [invalid@prefix/google.protobuf.internal.TestAny] {\n'
' [unknown_extension] {\n'
' str: "string"\n'
' any_value {\n'
' [type.googleapis.com/proto2_unittest.OneString] {\n'
' data: "string"\n'
' }\n'
' }\n'
' }\n'
' }\n'
'}\n'
'int32_value: 123'
)
with self.assertRaisesRegex(
text_format.ParseError,
'[invalid@prefix/google.protobuf.internal.TestAny]',
):
text_format.Parse(text, message, allow_unknown_extension=True)
def testParseBadIdentifier(self):
message = unittest_pb2.TestAllTypes()
text = 'optional_nested_message { "bb": 1 }'
with self.assertRaises(text_format.ParseError) as e:
text_format.Parse(text, message)
self.assertEqual(
str(e.exception),
'1:27 : \'optional_nested_message { "bb": 1 }\': '
'Expected identifier or number, got "bb".',
)
def testParseBadExtension(self):
message = unittest_pb2.TestAllExtensions()
text = '[unknown_extension]: 8\n'
self.assertRaisesRegex(
text_format.ParseError,
'1:2 : Extension "unknown_extension" not registered.',
text_format.Parse,
text,
message,
)
message = unittest_pb2.TestAllTypes()
self.assertRaisesRegex(
text_format.ParseError,
(
'1:2 : Message type "proto2_unittest.TestAllTypes" does not have '
'extensions.'
),
text_format.Parse,
text,
message,
)
def testParseNumericUnknownEnum(self):
message = unittest_pb2.TestAllTypes()
text = 'optional_nested_enum: 100'
self.assertRaisesRegex(
text_format.ParseError,
(
r'1:23 : \'optional_nested_enum: 100\': '
r'Enum type "\w+.TestAllTypes.NestedEnum" '
r'has no value with number 100.'
),
text_format.Parse,
text,
message,
)
def testMergeDuplicateExtensionScalars(self):
message = unittest_pb2.TestAllExtensions()
text = (
'[proto2_unittest.optional_int32_extension]: 42 '
'[proto2_unittest.optional_int32_extension]: 67'
)
text_format.Merge(text, message)
self.assertEqual(
67, message.Extensions[unittest_pb2.optional_int32_extension]
)
def testParseDuplicateExtensionScalars(self):
message = unittest_pb2.TestAllExtensions()
text = (
'[proto2_unittest.optional_int32_extension]: 42 '
'[proto2_unittest.optional_int32_extension]: 67'
)
self.assertRaisesRegex(
text_format.ParseError,
(
'1:92 : Message type "proto2_unittest.TestAllExtensions" '
'should not have multiple '
'"proto2_unittest.optional_int32_extension" extensions.'
),
text_format.Parse,
text,
message,
)
def testParseDuplicateExtensionMessages(self):
message = unittest_pb2.TestAllExtensions()
text = (
'[proto2_unittest.optional_nested_message_extension]: {} '
'[proto2_unittest.optional_nested_message_extension]: {}'
)
self.assertRaisesRegex(
text_format.ParseError,
(
'1:110 : Message type "proto2_unittest.TestAllExtensions" '
'should not have multiple '
'"proto2_unittest.optional_nested_message_extension" extensions.'
),
text_format.Parse,
text,
message,
)
def testParseGroupNotClosed(self):
message = unittest_pb2.TestAllTypes()
text = 'RepeatedGroup: <'
self.assertRaisesRegex(
text_format.ParseError,
'1:16 : Expected ">".',
text_format.Parse,
text,
message,
)
text = 'RepeatedGroup: {'
self.assertRaisesRegex(
text_format.ParseError,
'1:16 : Expected "}".',
text_format.Parse,
text,
message,
)
def testParseEmptyGroup(self):
message = unittest_pb2.TestAllTypes()
text = 'OptionalGroup: {}'
text_format.Parse(text, message)
self.assertTrue(message.HasField('optionalgroup'))
message.Clear()
message = unittest_pb2.TestAllTypes()
text = 'OptionalGroup: <>'
text_format.Parse(text, message)
self.assertTrue(message.HasField('optionalgroup'))
# Maps aren't really proto2-only, but our test schema only has maps for
# proto2.
def testParseMap(self):
text = (
'map_int32_int32 {\n'
' key: -123\n'
' value: -456\n'
'}\n'
'map_int64_int64 {\n'
' key: -8589934592\n'
' value: -17179869184\n'
'}\n'
'map_uint32_uint32 {\n'
' key: 123\n'
' value: 456\n'
'}\n'
'map_uint64_uint64 {\n'
' key: 8589934592\n'
' value: 17179869184\n'
'}\n'
'map_string_string {\n'
' key: "abc"\n'
' value: "123"\n'
'}\n'
'map_int32_foreign_message {\n'
' key: 111\n'
' value {\n'
' c: 5\n'
' }\n'
'}\n'
)
message = map_unittest_pb2.TestMap()
text_format.Parse(text, message)
self.assertEqual(-456, message.map_int32_int32[-123])
self.assertEqual(-(2**34), message.map_int64_int64[-(2**33)])
self.assertEqual(456, message.map_uint32_uint32[123])
self.assertEqual(2**34, message.map_uint64_uint64[2**33])
self.assertEqual('123', message.map_string_string['abc'])
self.assertEqual(5, message.map_int32_foreign_message[111].c)
class Proto3Tests(parameterized.TestCase):
def testPrintMessageExpandAny(self):
packed_message = unittest_pb2.OneString()
packed_message.data = 'string'
message = any_test_pb2.TestAny()
message.any_value.Pack(packed_message)
self.assertEqual(
text_format.MessageToString(
message, descriptor_pool=descriptor_pool.Default()
),
'any_value {\n'
' [type.googleapis.com/proto2_unittest.OneString] {\n'
' data: "string"\n'
' }\n'
'}\n',
)
def testPrintStructInAny(self):
packed_message = struct_pb2.Struct()
packed_message['name'] = 'Jim'
message = any_test_pb2.TestAny()
message.any_value.Pack(packed_message)
print(
text_format.MessageToString(
message, descriptor_pool=descriptor_pool.Default()
)
)
self.assertEqual(
text_format.MessageToString(
message, descriptor_pool=descriptor_pool.Default()
),
'any_value {\n'
' [type.googleapis.com/google.protobuf.Struct] {\n'
' fields {\n'
' key: "name"\n'
' value {\n'
' string_value: "Jim"\n'
' }\n'
' }\n'
' }\n'
'}\n',
)
def testTopAnyMessage(self):
packed_msg = unittest_pb2.OneString()
msg = any_pb2.Any()
msg.Pack(packed_msg)
text = text_format.MessageToString(msg)
other_msg = text_format.Parse(text, any_pb2.Any())
self.assertEqual(msg, other_msg)
def testPrintMessageExpandAnyRepeated(self):
packed_message = unittest_pb2.OneString()
message = any_test_pb2.TestAny()
packed_message.data = 'string0'
message.repeated_any_value.add().Pack(packed_message)
packed_message.data = 'string1'
message.repeated_any_value.add().Pack(packed_message)
self.assertEqual(
text_format.MessageToString(message),
'repeated_any_value {\n'
' [type.googleapis.com/proto2_unittest.OneString] {\n'
' data: "string0"\n'
' }\n'
'}\n'
'repeated_any_value {\n'
' [type.googleapis.com/proto2_unittest.OneString] {\n'
' data: "string1"\n'
' }\n'
'}\n',
)
def testPrintMessageExpandAnyDescriptorPoolMissingType(self):
packed_message = unittest_pb2.OneString()
packed_message.data = 'string'
message = any_test_pb2.TestAny()
message.any_value.Pack(packed_message)
empty_pool = descriptor_pool.DescriptorPool()
self.assertEqual(
text_format.MessageToString(message, descriptor_pool=empty_pool),
'any_value {\n'
' type_url: "type.googleapis.com/proto2_unittest.OneString"\n'
' value: "\\n\\006string"\n'
'}\n',
)
def testPrintMessageExpandAnyPointyBrackets(self):
packed_message = unittest_pb2.OneString()
packed_message.data = 'string'
message = any_test_pb2.TestAny()
message.any_value.Pack(packed_message)
self.assertEqual(
text_format.MessageToString(message, pointy_brackets=True),
'any_value <\n'
' [type.googleapis.com/proto2_unittest.OneString] <\n'
' data: "string"\n'
' >\n'
'>\n',
)
def testPrintMessageExpandAnyAsOneLine(self):
packed_message = unittest_pb2.OneString()
packed_message.data = 'string'
message = any_test_pb2.TestAny()
message.any_value.Pack(packed_message)
self.assertEqual(
text_format.MessageToString(message, as_one_line=True),
'any_value {'
' [type.googleapis.com/proto2_unittest.OneString]'
' { data: "string" } '
'}',
)
def testPrintMessageExpandAnyAsOneLinePointyBrackets(self):
packed_message = unittest_pb2.OneString()
packed_message.data = 'string'
message = any_test_pb2.TestAny()
message.any_value.Pack(packed_message)
self.assertEqual(
text_format.MessageToString(
message,
as_one_line=True,
pointy_brackets=True,
descriptor_pool=descriptor_pool.Default(),
),
'any_value <'
' [type.googleapis.com/proto2_unittest.OneString]'
' < data: "string" > '
'>',
)
def testPrintAndParseMessageInvalidAny(self):
packed_message = unittest_pb2.OneString()
packed_message.data = 'string'
message = any_test_pb2.TestAny()
message.any_value.Pack(packed_message)
# Only include string after last '/' in type_url.
message.any_value.type_url = message.any_value.TypeName()
text = text_format.MessageToString(message)
self.assertEqual(
text,
'any_value {\n'
' type_url: "proto2_unittest.OneString"\n'
' value: "\\n\\006string"\n'
'}\n',
)
parsed_message = any_test_pb2.TestAny()
text_format.Parse(text, parsed_message)
self.assertEqual(message, parsed_message)
def testUnknownEnums(self):
message = unittest_proto3_arena_pb2.TestAllTypes()
message2 = unittest_proto3_arena_pb2.TestAllTypes()
message.optional_nested_enum = 999
text_string = text_format.MessageToString(message)
text_format.Parse(text_string, message2)
self.assertEqual(999, message2.optional_nested_enum)
def testMergeExpandedAny(self):
message = any_test_pb2.TestAny()
text = (
'any_value {\n'
' [type.googleapis.com/proto2_unittest.OneString] {\n'
' data: "string"\n'
' }\n'
'}\n'
)
text_format.Merge(text, message)
packed_message = unittest_pb2.OneString()
message.any_value.Unpack(packed_message)
self.assertEqual('string', packed_message.data)
message.Clear()
text_format.Parse(text, message)
packed_message = unittest_pb2.OneString()
message.any_value.Unpack(packed_message)
self.assertEqual('string', packed_message.data)
def testMergeExpandedAnyRepeated(self):
message = any_test_pb2.TestAny()
text = (
'repeated_any_value {\n'
' [type.googleapis.com/proto2_unittest.OneString] {\n'
' data: "string0"\n'
' }\n'
'}\n'
'repeated_any_value {\n'
' [type.googleapis.com/proto2_unittest.OneString] {\n'
' data: "string1"\n'
' }\n'
'}\n'
)
text_format.Merge(text, message)
packed_message = unittest_pb2.OneString()
message.repeated_any_value[0].Unpack(packed_message)
self.assertEqual('string0', packed_message.data)
message.repeated_any_value[1].Unpack(packed_message)
self.assertEqual('string1', packed_message.data)
def testMergeExpandedAnyPointyBrackets(self):
message = any_test_pb2.TestAny()
text = (
'any_value {\n'
' [type.googleapis.com/proto2_unittest.OneString] <\n'
' data: "string"\n'
' >\n'
'}\n'
)
text_format.Merge(text, message)
packed_message = unittest_pb2.OneString()
message.any_value.Unpack(packed_message)
self.assertEqual('string', packed_message.data)
@parameterized.parameters(
{
'any_name': '[domain.com/proto2_unittest.OneString]',
'type_url': 'domain.com/proto2_unittest.OneString',
},
# Multiple slashes in prefix
{
'any_name': '[domain.com/path/proto2_unittest.OneString]',
'type_url': 'domain.com/path/proto2_unittest.OneString',
},
{
'any_name': '[domain.com///path//proto2_unittest.OneString]',
'type_url': 'domain.com///path//proto2_unittest.OneString',
},
# Special characters in prefix
{
'any_name': '[domain.com/-.~_!$&()*+,;=/proto2_unittest.OneString]',
'type_url': 'domain.com/-.~_!$&()*+,;=/proto2_unittest.OneString',
},
# Percent escapes in prefix
{
'any_name': (
'[percent.escapes/%0a%1B%2c%3D%4e%F5%A6%b7%C8%f9/proto2_unittest.OneString]'
),
'type_url': (
'percent.escapes/%0a%1B%2c%3D%4e%F5%A6%b7%C8%f9/proto2_unittest.OneString'
),
},
# Whitespace and comments (should be ignored between [])
{
'any_name': '[ domain . com / proto2_ unittest. One String ]',
'type_url': 'domain.com/proto2_unittest.OneString',
},
{
'any_name': (
'[ \t\n\r\f\v domain.com/pr \t\n\r\f\v oto2_unittest.OneString'
' \t\n\r\f\v ]'
),
'type_url': 'domain.com/proto2_unittest.OneString',
},
{
'any_name': (
'[ # comment\n domain.com/pr # comment\n oto2_unittest.One String'
' # comment\n ]'
),
'type_url': 'domain.com/proto2_unittest.OneString',
},
)
def testMergeExpandedAnyTypeUrls(self, *, any_name, type_url):
message = any_test_pb2.TestAny()
text = f'any_value {{\n {any_name} {{\n data: "string"\n }}\n }}'
text_format.Merge(text, message)
self.assertEqual(type_url, message.any_value.type_url)
@parameterized.parameters(
# General error cases
{
'any_name': '[',
'error_msg': '2:4 : \' [ {\': Expected "]"',
},
{
'any_name': '[]',
'error_msg': '2:5 : \' [] {\': Type URL does not contain "/"',
},
{
'any_name': '[.type]',
'error_msg': '2:10 : \' [.type] {\': Type URL does not contain "/"',
},
# Prefix error cases
{
'any_name': '[/]',
'error_msg': "2:6 : ' [/] {': Type URL prefix is empty.",
},
{
'any_name': '[/proto2_unittest.OneString]',
'error_msg': (
"2:31 : ' [/proto2_unittest.OneString] {': "
'Type URL prefix is empty'
),
},
{
'any_name': '[/domain.com/proto2_unittest.OneString]',
'error_msg': (
"2:42 : ' [/domain.com/proto2_unittest.OneString] {': "
'Type URL prefix starts with "/"'
),
},
# Special characters in prefix
{
'any_name': '[domain.com/?/proto2_unittest.OneString]',
'error_msg': (
"2:14 : ' [domain.com/?/proto2_unittest.OneString] {':"
' Expected "]"'
),
},
{
'any_name': '[domain.com/:/proto2_unittest.OneString]',
'error_msg': (
"2:14 : ' [domain.com/:/proto2_unittest.OneString] {':"
' Expected "]"'
),
},
{
'any_name': '[domain.com/@/proto2_unittest.OneString]',
'error_msg': (
"2:14 : ' [domain.com/@/proto2_unittest.OneString] {': "
'Expected "]".'
),
},
{
'any_name': '[domain.com/@/proto2_unittest.OneString]',
'error_msg': (
"2:14 : ' [domain.com/@/proto2_unittest.OneString] {':"
' Expected "]"'
),
},
# Percent escapes in prefix
{
'any_name': '[percent.escapes/%/proto2_unittest.OneString]',
'error_msg': (
"2:48 : ' [percent.escapes/%/proto2_unittest.OneString] {':"
' Invalid percent escape, got "%".'
),
},
{
'any_name': '[percent.escapes/%G/proto2_unittest.OneString]',
'error_msg': (
"2:49 : ' [percent.escapes/%G/proto2_unittest.OneString] {':"
' Invalid percent escape, got "%G".'
),
},
{
'any_name': '[percent.escapes/%aG/proto2_unittest.OneString]',
'error_msg': (
"2:50 : ' [percent.escapes/%aG/proto2_unittest.OneString] {':"
' Invalid percent escape, got "%aG".'
),
},
# Invalid type names
{
'any_name': '[domain.com/]',
'error_msg': (
'2:16 : \' [domain.com/] {\': Expected type name, got "".'
),
},
{
'any_name': '[domain.com/.]',
'error_msg': (
'2:17 : \' [domain.com/.] {\': Expected type name, got ".".'
),
},
{
'any_name': '[domain.com/.OneString]',
'error_msg': (
"2:26 : ' [domain.com/.OneString] {': "
'Expected type name, got ".OneString".'
),
},
{
'any_name': '[domain.com/proto2_unittest.]',
'error_msg': (
"2:32 : ' [domain.com/proto2_unittest.] {': "
'Expected type name, got "proto2_unittest.".'
),
},
{
'any_name': '[domain.com/5type]',
'error_msg': (
"2:21 : ' [domain.com/5type] {': "
'Expected type name, got "5type".'
),
},
{
'any_name': '[domain.com/!]',
'error_msg': (
'2:17 : \' [domain.com/!] {\': Expected type name, got "!".'
),
},
{
'any_name': '[domain.com/my_?_type]',
'error_msg': '2:17 : \' [domain.com/my_?_type] {\': Expected "]".',
},
{
'any_name': '[domain.com/my.:type]',
'error_msg': '2:17 : \' [domain.com/my.:type] {\': Expected "]".',
},
{
'any_name': '[domain.com/my.type@]',
'error_msg': '2:21 : \' [domain.com/my.type@] {\': Expected "]".',
},
)
def testMergeFailsOnInvalidExpandedAnyTypeUrls(self, *, any_name, error_msg):
message = any_test_pb2.TestAny()
text = 'any_value {\n %s {\n data: "string"\n }\n }' % any_name
with self.assertRaises(text_format.ParseError) as e:
text_format.Merge(text, message)
self.assertIn(error_msg, str(e.exception))
def testMergeExpandedAnyDescriptorPoolMissingType(self):
message = any_test_pb2.TestAny()
text = (
'any_value {\n'
' [type.googleapis.com/proto2_unittest.OneString] {\n'
' data: "string"\n'
' }\n'
'}\n'
)
with self.assertRaises(text_format.ParseError) as e:
empty_pool = descriptor_pool.DescriptorPool()
text_format.Merge(text, message, descriptor_pool=empty_pool)
self.assertEqual(
str(e.exception),
'Type proto2_unittest.OneString not found in descriptor pool',
)
def testMergeUnexpandedAny(self):
text = (
'any_value {\n'
' type_url: "type.googleapis.com/proto2_unittest.OneString"\n'
' value: "\\n\\006string"\n'
'}\n'
)
message = any_test_pb2.TestAny()
text_format.Merge(text, message)
packed_message = unittest_pb2.OneString()
message.any_value.Unpack(packed_message)
self.assertEqual('string', packed_message.data)
def testMergeMissingAnyEndToken(self):
message = any_test_pb2.TestAny()
text = (
'any_value {\n'
' [type.googleapis.com/proto2_unittest.OneString] {\n'
' data: "string"\n'
)
with self.assertRaises(text_format.ParseError) as e:
text_format.Merge(text, message)
self.assertEqual(str(e.exception), '3:11 : Expected "}".')
def testParseExpandedAnyListValue(self):
any_msg = any_pb2.Any()
any_msg.Pack(struct_pb2.ListValue())
msg = any_test_pb2.TestAny(any_value=any_msg)
text = (
'any_value {\n [type.googleapis.com/google.protobuf.ListValue] {}\n}\n'
)
parsed_msg = text_format.Parse(text, any_test_pb2.TestAny())
self.assertEqual(msg, parsed_msg)
def testProto3Optional(self):
msg = test_proto3_optional_pb2.TestProto3Optional()
self.assertEqual(text_format.MessageToString(msg), '')
msg.optional_int32 = 0
msg.optional_float = 0.0
msg.optional_string = ''
msg.optional_nested_message.bb = 0
text = (
'optional_int32: 0\n'
'optional_float: 0.0\n'
'optional_string: ""\n'
'optional_nested_message {\n'
' bb: 0\n'
'}\n'
)
self.assertEqual(text_format.MessageToString(msg), text)
msg2 = test_proto3_optional_pb2.TestProto3Optional()
text_format.Parse(text, msg2)
self.assertEqual(text_format.MessageToString(msg2), text)
def _CreateConsumeLiteralToken(expected_literal):
def _Consume(tokenizer):
tokenizer.Consume(expected_literal)
return expected_literal
return (_Consume, expected_literal)
class TokenizerTest(parameterized.TestCase):
@parameterized.named_parameters([
dict(
testcase_name='_string_double_quotes',
text='identifier1:"string1"\n',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'identifier1'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeString, 'string1'),
],
),
dict(
testcase_name='_integer',
text='identifier2 : \n \n123 ',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'identifier2'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeInteger, 123),
],
),
dict(
testcase_name='_string_single_quotes',
text="\n identifier3:'string'\n",
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'identifier3'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeString, 'string'),
],
),
dict(
testcase_name='_float_exponent',
text='identifiER_4 : 1.1e+2 ',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'identifiER_4'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeFloat, 1.1e2),
],
),
dict(
testcase_name='_float',
text='ID5:-0.23',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'ID5'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeFloat, -0.23),
],
),
dict(
testcase_name='_escape_single_quote',
text="ID6:'aaaa\\'bbbb'\n",
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'ID6'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeString, "aaaa'bbbb"),
],
),
dict(
testcase_name='_escape_double_quote',
text='ID7 : "aa\\"bb"\n\n\n\n ',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'ID7'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeString, 'aa"bb'),
],
),
dict(
testcase_name='_submessage',
text='ID8: {A:inf B:-inf C:true D:false}\n',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'ID8'),
_CreateConsumeLiteralToken(':'),
_CreateConsumeLiteralToken('{'),
(text_format.Tokenizer.ConsumeIdentifier, 'A'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeFloat, float('inf')),
(text_format.Tokenizer.ConsumeIdentifier, 'B'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeFloat, float('-inf')),
(text_format.Tokenizer.ConsumeIdentifier, 'C'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeBool, True),
(text_format.Tokenizer.ConsumeIdentifier, 'D'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeBool, False),
_CreateConsumeLiteralToken('}'),
],
),
dict(
testcase_name='_large_negative_integer',
text='ID10: -111111111111111111 ',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'ID10'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeInteger, -111111111111111111),
],
),
dict(
testcase_name='_negative_integer',
text='ID11: -22\n',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'ID11'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeInteger, -22),
],
),
dict(
testcase_name='_large_integer',
text='ID12: 2222222222222222222 ',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'ID12'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeInteger, 2222222222222222222),
],
),
dict(
testcase_name='_float_suffix',
text='ID13: 1.23456f ',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'ID13'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeFloat, 1.23456),
],
),
dict(
testcase_name='_float_capital_suffix',
text='ID13: 1.23456F ',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'ID13'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeFloat, 1.23456),
],
),
dict(
testcase_name='_float_exponent_suffix',
text='ID14: 1.2e+2f ',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'ID14'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeFloat, 1.2e2),
],
),
dict(
testcase_name='_bool_zero',
text='false_bool: 0 ',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'false_bool'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeBool, False),
],
),
dict(
testcase_name='_bool_t',
text='true_BOOL:t ',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'true_BOOL'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeBool, True),
],
),
dict(
testcase_name='_bool_one',
text='true_bool1: 1 ',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'true_bool1'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeBool, True),
],
),
dict(
testcase_name='_bool_f',
text='false_BOOL1:f ',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'false_BOOL1'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeBool, False),
],
),
dict(
testcase_name='_bool_false',
text='False_bool: False ',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'False_bool'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeBool, False),
],
),
dict(
testcase_name='_bool_true',
text='True_bool: True ',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'True_bool'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeBool, True),
],
),
dict(
testcase_name='_float_inf',
text='X:iNf ',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'X'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeFloat, float('inf')),
],
),
dict(
testcase_name='_float_negative_inf',
text='Y:-inF ',
expected=[
(text_format.Tokenizer.ConsumeIdentifier, 'Y'),
_CreateConsumeLiteralToken(':'),
(text_format.Tokenizer.ConsumeFloat, float('-inf')),
],
),
])
def testSimpleTokenCases(self, text, expected):
consume_functions, expected_tokens = zip(*expected)
tokenizer = text_format.Tokenizer(text.splitlines())
tokens = [consume(tokenizer) for consume in consume_functions]
self.assertTrue(tokenizer.AtEnd())
self.assertEqual(tokens, [token for token in expected_tokens])
def testConsumeNan(self):
tokenizer = text_format.Tokenizer(['nAN'])
token = tokenizer.ConsumeFloat()
self.assertTrue(math.isnan(token), 'Expected NaN, got %s' % token)
def testConsumeAbstractIntegers(self):
# This test only tests the failures in the integer parsing methods as well
# as the '0' special cases.
int64_max = (1 << 63) - 1
uint32_max = (1 << 32) - 1
text = '-1 %d %d' % (uint32_max + 1, int64_max + 1)
tokenizer = text_format.Tokenizer(text.splitlines())
self.assertEqual(-1, tokenizer.ConsumeInteger())
self.assertEqual(uint32_max + 1, tokenizer.ConsumeInteger())
self.assertEqual(int64_max + 1, tokenizer.ConsumeInteger())
self.assertTrue(tokenizer.AtEnd())
text = '-0 0 0 1.2'
tokenizer = text_format.Tokenizer(text.splitlines())
self.assertEqual(0, tokenizer.ConsumeInteger())
self.assertEqual(0, tokenizer.ConsumeInteger())
self.assertEqual(True, tokenizer.TryConsumeInteger())
self.assertEqual(False, tokenizer.TryConsumeInteger())
with self.assertRaises(text_format.ParseError):
tokenizer.ConsumeInteger()
self.assertEqual(1.2, tokenizer.ConsumeFloat())
self.assertTrue(tokenizer.AtEnd())
def testConsumeIntegers(self):
# This test only tests the failures in the integer parsing methods as well
# as the '0' special cases.
int64_max = (1 << 63) - 1
uint32_max = (1 << 32) - 1
text = '-1 %d %d' % (uint32_max + 1, int64_max + 1)
tokenizer = text_format.Tokenizer(text.splitlines())
self.assertRaises(
text_format.ParseError, text_format._ConsumeUint32, tokenizer
)
self.assertRaises(
text_format.ParseError, text_format._ConsumeUint64, tokenizer
)
self.assertEqual(-1, text_format._ConsumeInt32(tokenizer))
self.assertRaises(
text_format.ParseError, text_format._ConsumeUint32, tokenizer
)
self.assertRaises(
text_format.ParseError, text_format._ConsumeInt32, tokenizer
)
self.assertEqual(uint32_max + 1, text_format._ConsumeInt64(tokenizer))
self.assertRaises(
text_format.ParseError, text_format._ConsumeInt64, tokenizer
)
self.assertEqual(int64_max + 1, text_format._ConsumeUint64(tokenizer))
self.assertTrue(tokenizer.AtEnd())
text = '-0 -0 0 0'
tokenizer = text_format.Tokenizer(text.splitlines())
self.assertEqual(0, text_format._ConsumeUint32(tokenizer))
self.assertEqual(0, text_format._ConsumeUint64(tokenizer))
self.assertEqual(0, text_format._ConsumeUint32(tokenizer))
self.assertEqual(0, text_format._ConsumeUint64(tokenizer))
self.assertTrue(tokenizer.AtEnd())
def testConsumeOctalIntegers(self):
"""Test support for C style octal integers."""
text = '00 -00 04 0755 -010 007 -0033 08 -09 01'
tokenizer = text_format.Tokenizer(text.splitlines())
self.assertEqual(0, tokenizer.ConsumeInteger())
self.assertEqual(0, tokenizer.ConsumeInteger())
self.assertEqual(4, tokenizer.ConsumeInteger())
self.assertEqual(0o755, tokenizer.ConsumeInteger())
self.assertEqual(-0o10, tokenizer.ConsumeInteger())
self.assertEqual(7, tokenizer.ConsumeInteger())
self.assertEqual(-0o033, tokenizer.ConsumeInteger())
with self.assertRaises(text_format.ParseError):
tokenizer.ConsumeInteger() # 08
tokenizer.NextToken()
with self.assertRaises(text_format.ParseError):
tokenizer.ConsumeInteger() # -09
tokenizer.NextToken()
self.assertEqual(1, tokenizer.ConsumeInteger())
self.assertTrue(tokenizer.AtEnd())
@parameterized.parameters('00', '09', '01.123', '-00', '-09', '-01.234')
def testConsumeOctalFloats(self, text):
"""Test rejection of for octal-formatted floats."""
tokenizer = text_format.Tokenizer([text])
self.assertRaisesRegex(
text_format.ParseError,
'Invalid octal float: %s' % text,
tokenizer.ConsumeFloat,
)
def testConsumeByteString(self):
text = '"string1\''
tokenizer = text_format.Tokenizer(text.splitlines())
self.assertRaises(text_format.ParseError, tokenizer.ConsumeByteString)
text = 'string1"'
tokenizer = text_format.Tokenizer(text.splitlines())
self.assertRaises(text_format.ParseError, tokenizer.ConsumeByteString)
text = '\n"\\xt"'
tokenizer = text_format.Tokenizer(text.splitlines())
self.assertRaises(text_format.ParseError, tokenizer.ConsumeByteString)
text = '\n"\\"'
tokenizer = text_format.Tokenizer(text.splitlines())
self.assertRaises(text_format.ParseError, tokenizer.ConsumeByteString)
text = '\n"\\x"'
tokenizer = text_format.Tokenizer(text.splitlines())
self.assertRaises(text_format.ParseError, tokenizer.ConsumeByteString)
def testConsumeBool(self):
text = 'not-a-bool'
tokenizer = text_format.Tokenizer(text.splitlines())
self.assertRaises(text_format.ParseError, tokenizer.ConsumeBool)
def testSkipComment(self):
tokenizer = text_format.Tokenizer('# some comment'.splitlines())
self.assertTrue(tokenizer.AtEnd())
self.assertRaises(text_format.ParseError, tokenizer.ConsumeComment)
def testConsumeComment(self):
tokenizer = text_format.Tokenizer(
'# some comment'.splitlines(), skip_comments=False
)
self.assertFalse(tokenizer.AtEnd())
self.assertEqual('# some comment', tokenizer.ConsumeComment())
self.assertTrue(tokenizer.AtEnd())
def testConsumeTwoComments(self):
text = '# some comment\n# another comment'
tokenizer = text_format.Tokenizer(text.splitlines(), skip_comments=False)
self.assertEqual('# some comment', tokenizer.ConsumeComment())
self.assertFalse(tokenizer.AtEnd())
self.assertEqual('# another comment', tokenizer.ConsumeComment())
self.assertTrue(tokenizer.AtEnd())
def testConsumeTrailingComment(self):
text = 'some_number: 4\n# some comment'
tokenizer = text_format.Tokenizer(text.splitlines(), skip_comments=False)
self.assertRaises(text_format.ParseError, tokenizer.ConsumeComment)
self.assertEqual('some_number', tokenizer.ConsumeIdentifier())
self.assertEqual(tokenizer.token, ':')
tokenizer.NextToken()
self.assertRaises(text_format.ParseError, tokenizer.ConsumeComment)
self.assertEqual(4, tokenizer.ConsumeInteger())
self.assertFalse(tokenizer.AtEnd())
self.assertEqual('# some comment', tokenizer.ConsumeComment())
self.assertTrue(tokenizer.AtEnd())
def testConsumeLineComment(self):
tokenizer = text_format.Tokenizer(
'# some comment'.splitlines(), skip_comments=False
)
self.assertFalse(tokenizer.AtEnd())
self.assertEqual(
(False, '# some comment'), tokenizer.ConsumeCommentOrTrailingComment()
)
self.assertTrue(tokenizer.AtEnd())
def testConsumeTwoLineComments(self):
text = '# some comment\n# another comment'
tokenizer = text_format.Tokenizer(text.splitlines(), skip_comments=False)
self.assertEqual(
(False, '# some comment'), tokenizer.ConsumeCommentOrTrailingComment()
)
self.assertFalse(tokenizer.AtEnd())
self.assertEqual(
(False, '# another comment'),
tokenizer.ConsumeCommentOrTrailingComment(),
)
self.assertTrue(tokenizer.AtEnd())
def testConsumeAndCheckTrailingComment(self):
text = 'some_number: 4 # some comment' # trailing comment on the same line
tokenizer = text_format.Tokenizer(text.splitlines(), skip_comments=False)
self.assertRaises(
text_format.ParseError, tokenizer.ConsumeCommentOrTrailingComment
)
self.assertEqual('some_number', tokenizer.ConsumeIdentifier())
self.assertEqual(tokenizer.token, ':')
tokenizer.NextToken()
self.assertRaises(
text_format.ParseError, tokenizer.ConsumeCommentOrTrailingComment
)
self.assertEqual(4, tokenizer.ConsumeInteger())
self.assertFalse(tokenizer.AtEnd())
self.assertEqual(
(True, '# some comment'), tokenizer.ConsumeCommentOrTrailingComment()
)
self.assertTrue(tokenizer.AtEnd())
def testHashinComment(self):
text = 'some_number: 4 # some comment # not a new comment'
tokenizer = text_format.Tokenizer(text.splitlines(), skip_comments=False)
self.assertEqual('some_number', tokenizer.ConsumeIdentifier())
self.assertEqual(tokenizer.token, ':')
tokenizer.NextToken()
self.assertEqual(4, tokenizer.ConsumeInteger())
self.assertEqual(
(True, '# some comment # not a new comment'),
tokenizer.ConsumeCommentOrTrailingComment(),
)
self.assertTrue(tokenizer.AtEnd())
def testHugeString(self):
# With pathologic backtracking, fails with Forge OOM.
text = '"' + 'a' * (10 * 1024 * 1024) + '"'
tokenizer = text_format.Tokenizer(text.splitlines(), skip_comments=False)
tokenizer.ConsumeString()
def testGroupName(self):
grp = unittest_pb2.TestGroupExtension()
grp.Extensions[
unittest_pb2.TestNestedExtension.optionalgroup_extension
].a = 6
self.assertEqual(
'[proto2_unittest.TestNestedExtension.optionalgroup_extension] {\n a:'
' 6\n}\n',
str(grp),
)
msg = unittest_pb2.TestAllTypes(
repeatedgroup=[unittest_pb2.TestAllTypes.RepeatedGroup(a=1)]
)
if api_implementation.Type() == 'upb':
self.assertEqual('repeatedgroup {\n a: 1\n}\n', str(msg))
else:
self.assertEqual('RepeatedGroup {\n a: 1\n}\n', str(msg))
def testPrintGroupLikeDelimited(self):
msg = unittest_delimited_pb2.TestDelimited(
grouplike=unittest_delimited_pb2.TestDelimited.GroupLike(a=1)
)
if api_implementation.Type() == 'upb':
self.assertEqual(str(msg), 'grouplike {\n a: 1\n}\n')
else:
self.assertEqual(str(msg), 'GroupLike {\n a: 1\n}\n')
def testPrintGroupLikeDelimitedExtension(self):
msg = unittest_delimited_pb2.TestDelimited()
msg.Extensions[unittest_delimited_pb2.grouplikefilescope].b = 5
self.assertEqual(
str(msg), '[editions_unittest.grouplikefilescope] {\n b: 5\n}\n'
)
def testPrintGroupLikeNotDelimited(self):
msg = unittest_delimited_pb2.TestDelimited(
lengthprefixed=unittest_delimited_pb2.TestDelimited.LengthPrefixed(b=9)
)
self.assertEqual(str(msg), 'lengthprefixed {\n b: 9\n}\n')
def testPrintGroupLikeMismatchedName(self):
msg = unittest_delimited_pb2.TestDelimited(
notgrouplike=unittest_delimited_pb2.TestDelimited.GroupLike(b=2)
)
self.assertEqual(str(msg), 'notgrouplike {\n b: 2\n}\n')
def testPrintGroupLikeExtensionMismatchedName(self):
msg = unittest_delimited_pb2.TestDelimited()
msg.Extensions[unittest_delimited_pb2.not_group_like_scope].b = 5
self.assertEqual(
str(msg), '[editions_unittest.not_group_like_scope] {\n b: 5\n}\n'
)
def testPrintGroupLikeMismatchedScope(self):
msg = unittest_delimited_pb2.TestDelimited(
notgrouplikescope=unittest_delimited_pb2.NotGroupLikeScope(b=9)
)
self.assertEqual(str(msg), 'notgrouplikescope {\n b: 9\n}\n')
def testPrintGroupLikeExtensionMismatchedScope(self):
msg = unittest_delimited_pb2.TestDelimited()
msg.Extensions[unittest_delimited_pb2.grouplike].b = 1
self.assertEqual(str(msg), '[editions_unittest.grouplike] {\n b: 1\n}\n')
def testPrintGroupLikeMismatchedFile(self):
msg = unittest_delimited_pb2.TestDelimited(
messageimport=unittest_delimited_import_pb2.MessageImport(b=9)
)
self.assertEqual(str(msg), 'messageimport {\n b: 9\n}\n')
def testParseDelimitedGroupLikeType(self):
msg = unittest_delimited_pb2.TestDelimited()
text_format.Parse('GroupLike { a: 1 }', msg)
self.assertEqual(msg.grouplike.a, 1)
self.assertFalse(msg.HasField('notgrouplike'))
def testParseDelimitedGroupLikeField(self):
msg = unittest_delimited_pb2.TestDelimited()
text_format.Parse('grouplike { a: 2 }', msg)
self.assertEqual(msg.grouplike.a, 2)
self.assertFalse(msg.HasField('notgrouplike'))
def testParseDelimitedGroupLikeExtension(self):
msg = unittest_delimited_pb2.TestDelimited()
text_format.Parse('[editions_unittest.grouplike] { a: 2 }', msg)
self.assertEqual(msg.Extensions[unittest_delimited_pb2.grouplike].a, 2)
def testParseDelimitedGroupLikeInvalid(self):
msg = unittest_delimited_pb2.TestDelimited()
with self.assertRaises(text_format.ParseError):
text_format.Parse('GROUPlike { b:1 }', msg)
def testParseDelimitedGroupLikeInvalidExtension(self):
msg = unittest_delimited_pb2.TestDelimited()
with self.assertRaises(text_format.ParseError):
text_format.Parse('[editions_unittest.GroupLike] { a: 2 }', msg)
def testParseDelimited(self):
msg = unittest_delimited_pb2.TestDelimited()
text_format.Parse('notgrouplike { b: 1 }', msg)
self.assertEqual(msg.notgrouplike.b, 1)
self.assertFalse(msg.HasField('grouplike'))
def testParseDelimitedInvalid(self):
msg = unittest_delimited_pb2.TestDelimited()
with self.assertRaises(text_format.ParseError):
text_format.Parse('NotGroupLike { b:1 }', msg)
def testParseDelimitedInvalidScope(self):
msg = unittest_delimited_pb2.TestDelimited()
with self.assertRaises(text_format.ParseError):
text_format.Parse('NotGroupLikeScope { b:1 }', msg)
# Tests for pretty printer functionality.
@parameterized.parameters((unittest_pb2), (unittest_proto3_arena_pb2))
class PrettyPrinterTest(TextFormatBase):
def testPrettyPrintNoMatch(self, message_module):
def printer(message, indent, as_one_line):
del message, indent, as_one_line
return None
message = message_module.TestAllTypes()
msg = message.repeated_nested_message.add()
msg.bb = 42
self.CompareToGoldenText(
text_format.MessageToString(
message, as_one_line=True, message_formatter=printer
),
'repeated_nested_message { bb: 42 }',
)
def testPrettyPrintOneLine(self, message_module):
def printer(m, indent, as_one_line):
del indent, as_one_line
if m.DESCRIPTOR == message_module.TestAllTypes.NestedMessage.DESCRIPTOR:
return 'My lucky number is %s' % m.bb
message = message_module.TestAllTypes()
msg = message.repeated_nested_message.add()
msg.bb = 42
self.CompareToGoldenText(
text_format.MessageToString(
message, as_one_line=True, message_formatter=printer
),
'repeated_nested_message { My lucky number is 42 }',
)
def testPrettyPrintMultiLine(self, message_module):
def printer(m, indent, as_one_line):
if m.DESCRIPTOR == message_module.TestAllTypes.NestedMessage.DESCRIPTOR:
line_deliminator = (' ' if as_one_line else '\n') + ' ' * indent
return 'My lucky number is:%s%s' % (line_deliminator, m.bb)
return None
message = message_module.TestAllTypes()
msg = message.repeated_nested_message.add()
msg.bb = 42
self.CompareToGoldenText(
text_format.MessageToString(
message, as_one_line=True, message_formatter=printer
),
'repeated_nested_message { My lucky number is: 42 }',
)
self.CompareToGoldenText(
text_format.MessageToString(
message, as_one_line=False, message_formatter=printer
),
'repeated_nested_message {\n My lucky number is:\n 42\n}\n',
)
def testPrettyPrintEntireMessage(self, message_module):
def printer(m, indent, as_one_line):
del indent, as_one_line
if m.DESCRIPTOR == message_module.TestAllTypes.DESCRIPTOR:
return 'The is the message!'
return None
message = message_module.TestAllTypes()
self.CompareToGoldenText(
text_format.MessageToString(
message, as_one_line=False, message_formatter=printer
),
'The is the message!\n',
)
self.CompareToGoldenText(
text_format.MessageToString(
message, as_one_line=True, message_formatter=printer
),
'The is the message!',
)
def testPrettyPrintMultipleParts(self, message_module):
def printer(m, indent, as_one_line):
del indent, as_one_line
if m.DESCRIPTOR == message_module.TestAllTypes.NestedMessage.DESCRIPTOR:
return 'My lucky number is %s' % m.bb
return None
message = message_module.TestAllTypes()
message.optional_int32 = 61
msg = message.repeated_nested_message.add()
msg.bb = 42
msg = message.repeated_nested_message.add()
msg.bb = 99
msg = message.optional_nested_message
msg.bb = 1
self.CompareToGoldenText(
text_format.MessageToString(
message, as_one_line=True, message_formatter=printer
),
(
'optional_int32: 61 '
'optional_nested_message { My lucky number is 1 } '
'repeated_nested_message { My lucky number is 42 } '
'repeated_nested_message { My lucky number is 99 }'
),
)
out = text_format.TextWriter(False)
text_format.PrintField(
message_module.TestAllTypes.DESCRIPTOR.fields_by_name[
'optional_nested_message'
],
message.optional_nested_message,
out,
message_formatter=printer,
)
self.assertEqual(
'optional_nested_message {\n My lucky number is 1\n}\n', out.getvalue()
)
out.close()
out = text_format.TextWriter(False)
text_format.PrintFieldValue(
message_module.TestAllTypes.DESCRIPTOR.fields_by_name[
'optional_nested_message'
],
message.optional_nested_message,
out,
message_formatter=printer,
)
self.assertEqual('{\n My lucky number is 1\n}', out.getvalue())
out.close()
class WhitespaceTest(TextFormatBase):
def setUp(self):
self.out = text_format.TextWriter(False)
self.addCleanup(self.out.close)
self.message = unittest_pb2.NestedTestAllTypes()
self.message.child.payload.optional_string = 'value'
self.field = self.message.DESCRIPTOR.fields_by_name['child']
self.value = self.message.child
def testMessageToString(self):
self.CompareToGoldenText(
text_format.MessageToString(self.message),
textwrap.dedent("""\
child {
payload {
optional_string: "value"
}
}
"""),
)
def testPrintMessage(self):
text_format.PrintMessage(self.message, self.out)
self.CompareToGoldenText(
self.out.getvalue(),
textwrap.dedent("""\
child {
payload {
optional_string: "value"
}
}
"""),
)
def testPrintField(self):
text_format.PrintField(self.field, self.value, self.out)
self.CompareToGoldenText(
self.out.getvalue(),
textwrap.dedent("""\
child {
payload {
optional_string: "value"
}
}
"""),
)
def testPrintFieldValue(self):
text_format.PrintFieldValue(self.field, self.value, self.out)
self.CompareToGoldenText(
self.out.getvalue(),
textwrap.dedent("""\
{
payload {
optional_string: "value"
}
}"""),
)
class OptionalColonMessageToStringTest(unittest.TestCase):
def testForcePrintOptionalColon(self):
packed_message = unittest_pb2.OneString()
packed_message.data = 'string'
message = any_test_pb2.TestAny()
message.any_value.Pack(packed_message)
output = text_format.MessageToString(message, force_colon=True)
expected = (
'any_value: {\n'
' [type.googleapis.com/proto2_unittest.OneString]: {\n'
' data: "string"\n'
' }\n'
'}\n'
)
self.assertEqual(expected, output)
def testPrintShortFormatRepeatedFields(self):
message = unittest_pb2.TestAllTypes()
message.repeated_int32.append(1)
output = text_format.MessageToString(
message, use_short_repeated_primitives=True, force_colon=True
)
self.assertEqual('repeated_int32: [1]\n', output)
if __name__ == '__main__':
unittest.main()
Reference in a new issue View git blame Copy permalink