我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用google.protobuf.descriptor_pb2.FileDescriptorProto()。
def testMakeDescriptorWithUnsignedIntField(self): file_descriptor_proto = descriptor_pb2.FileDescriptorProto() file_descriptor_proto.name = 'Foo' message_type = file_descriptor_proto.message_type.add() message_type.name = file_descriptor_proto.name enum_type = message_type.enum_type.add() enum_type.name = 'FOO' enum_type_val = enum_type.value.add() enum_type_val.name = 'BAR' enum_type_val.number = 3 field = message_type.field.add() field.number = 1 field.name = 'uint64_field' field.label = descriptor.FieldDescriptor.LABEL_REQUIRED field.type = descriptor.FieldDescriptor.TYPE_UINT64 enum_field = message_type.field.add() enum_field.number = 2 enum_field.name = 'bar_field' enum_field.label = descriptor.FieldDescriptor.LABEL_REQUIRED enum_field.type = descriptor.FieldDescriptor.TYPE_ENUM enum_field.type_name = 'Foo.FOO' result = descriptor.MakeDescriptor(message_type) self.assertEqual(result.fields[0].cpp_type, descriptor.FieldDescriptor.CPPTYPE_UINT64)
def testEnumDefaultValue(self): """Test the default value of enums which don't start at zero.""" def _CheckDefaultValue(file_descriptor): default_value = (file_descriptor .message_types_by_name['DescriptorPoolTest1'] .fields_by_name['nested_enum'] .default_value) self.assertEqual(default_value, descriptor_pool_test1_pb2.DescriptorPoolTest1.BETA) # First check what the generated descriptor contains. _CheckDefaultValue(descriptor_pool_test1_pb2.DESCRIPTOR) # Then check the generated pool. Normally this is the same descriptor. file_descriptor = symbol_database.Default().pool.FindFileByName( 'google/protobuf/internal/descriptor_pool_test1.proto') self.assertIs(file_descriptor, descriptor_pool_test1_pb2.DESCRIPTOR) _CheckDefaultValue(file_descriptor) # Then check the dynamic pool and its internal DescriptorDatabase. descriptor_proto = descriptor_pb2.FileDescriptorProto.FromString( descriptor_pool_test1_pb2.DESCRIPTOR.serialized_pb) self.pool.Add(descriptor_proto) # And do the same check as above file_descriptor = self.pool.FindFileByName( 'google/protobuf/internal/descriptor_pool_test1.proto') _CheckDefaultValue(file_descriptor)
def testParsingNestedClass(self): """Test that the generated class can parse a nested message.""" file_descriptor = descriptor_pb2.FileDescriptorProto() file_descriptor.ParseFromString(self._GetSerializedFileDescriptor('C')) msg_descriptor = descriptor.MakeDescriptor( file_descriptor.message_type[0]) msg_class = reflection.MakeClass(msg_descriptor) msg = msg_class() msg_str = ( 'bar {' ' baz {' ' deep: 4' ' }' '}') text_format.Merge(msg_str, msg) self.assertEqual(msg.bar.baz.deep, 4)
def setUp(self): # TODO(jieluo): Should make the pool which is created by # serialized_pb same with generated pool. # TODO(jieluo): More test coverage for the generated pool. self.pool = descriptor_pool.DescriptorPool() self.factory_test1_fd = descriptor_pb2.FileDescriptorProto.FromString( factory_test1_pb2.DESCRIPTOR.serialized_pb) self.factory_test2_fd = descriptor_pb2.FileDescriptorProto.FromString( factory_test2_pb2.DESCRIPTOR.serialized_pb) self.pool.Add(self.factory_test1_fd) self.pool.Add(self.factory_test2_fd) self.pool.Add(descriptor_pb2.FileDescriptorProto.FromString( unittest_import_public_pb2.DESCRIPTOR.serialized_pb)) self.pool.Add(descriptor_pb2.FileDescriptorProto.FromString( unittest_import_pb2.DESCRIPTOR.serialized_pb)) self.pool.Add(descriptor_pb2.FileDescriptorProto.FromString( unittest_pb2.DESCRIPTOR.serialized_pb))
def testFileDescriptorOptionsWithCustomDescriptorPool(self): # Create a descriptor pool, and add a new FileDescriptorProto to it. pool = descriptor_pool.DescriptorPool() file_name = 'file_descriptor_options_with_custom_descriptor_pool.proto' file_descriptor_proto = descriptor_pb2.FileDescriptorProto(name=file_name) extension_id = file_options_test_pb2.foo_options file_descriptor_proto.options.Extensions[extension_id].foo_name = 'foo' pool.Add(file_descriptor_proto) # The options set on the FileDescriptorProto should be available in the # descriptor even if they contain extensions that cannot be deserialized # using the pool. file_descriptor = pool.FindFileByName(file_name) options = file_descriptor.GetOptions() self.assertEqual('foo', options.Extensions[extension_id].foo_name) # The object returned by GetOptions() is cached. self.assertIs(options, file_descriptor.GetOptions())
def testParsingFlatClassWithExplicitClassDeclaration(self): """Test that the generated class can parse a flat message.""" # TODO(xiaofeng): This test fails with cpp implemetnation in the call # of six.with_metaclass(). The other two callsites of with_metaclass # in this file are both excluded from cpp test, so it might be expected # to fail. Need someone more familiar with the python code to take a # look at this. if api_implementation.Type() != 'python': return file_descriptor = descriptor_pb2.FileDescriptorProto() file_descriptor.ParseFromString(self._GetSerializedFileDescriptor('A')) msg_descriptor = descriptor.MakeDescriptor( file_descriptor.message_type[0]) class MessageClass(six.with_metaclass(reflection.GeneratedProtocolMessageType, message.Message)): DESCRIPTOR = msg_descriptor msg = MessageClass() msg_str = ( 'flat: 0 ' 'flat: 1 ' 'flat: 2 ') text_format.Merge(msg_str, msg) self.assertEqual(msg.flat, [0, 1, 2])
def CopyToProto(self, proto): """Copies this to a descriptor_pb2.FileDescriptorProto. Args: proto: An empty descriptor_pb2.FileDescriptorProto. """ proto.ParseFromString(self.serialized_pb)
def _MakeFileDescriptorProto(proto_file_name, full_name, field_items): """Populate FileDescriptorProto for MessageFactory's DescriptorPool.""" package, name = full_name.rsplit('.', 1) file_proto = descriptor_pb2.FileDescriptorProto() file_proto.name = os.path.join(package.replace('.', '/'), proto_file_name) file_proto.package = package desc_proto = file_proto.message_type.add() desc_proto.name = name for f_number, (f_name, f_type) in enumerate(field_items, 1): field_proto = desc_proto.field.add() field_proto.name = f_name field_proto.number = f_number field_proto.label = descriptor_pb2.FieldDescriptorProto.LABEL_OPTIONAL field_proto.type = f_type return file_proto
def get_service_identify_from_pb_file(stubs, pb2): method_mapping = dict() p = descriptor_pb2.FileDescriptorProto() pb2.DESCRIPTOR.CopyToProto(p) for s in p.service: for m in s.method: if not method_mapping.get(m.name): method_mapping[m.name] = dict() for k, v in m.options._fields.items(): if hasattr(v, '_fields'): for i, j in v._fields.items(): method_mapping[m.name][i.name] = j return method_mapping
def setUp(self): file_proto = descriptor_pb2.FileDescriptorProto( name='some/filename/some.proto', package='protobuf_unittest') message_proto = file_proto.message_type.add( name='NestedMessage') message_proto.field.add( name='bb', number=1, type=descriptor_pb2.FieldDescriptorProto.TYPE_INT32, label=descriptor_pb2.FieldDescriptorProto.LABEL_OPTIONAL) enum_proto = message_proto.enum_type.add( name='ForeignEnum') enum_proto.value.add(name='FOREIGN_FOO', number=4) enum_proto.value.add(name='FOREIGN_BAR', number=5) enum_proto.value.add(name='FOREIGN_BAZ', number=6) file_proto.message_type.add(name='ResponseMessage') service_proto = file_proto.service.add( name='Service') method_proto = service_proto.method.add( name='CallMethod', input_type='.protobuf_unittest.NestedMessage', output_type='.protobuf_unittest.ResponseMessage') # Note: Calling DescriptorPool.Add() multiple times with the same file only # works if the input is canonical; in particular, all type names must be # fully qualified. self.pool = self.GetDescriptorPool() self.pool.Add(file_proto) self.my_file = self.pool.FindFileByName(file_proto.name) self.my_message = self.my_file.message_types_by_name[message_proto.name] self.my_enum = self.my_message.enum_types_by_name[enum_proto.name] self.my_service = self.my_file.services_by_name[service_proto.name] self.my_method = self.my_service.methods_by_name[method_proto.name]
def testComplexNesting(self): more_messages_desc = descriptor_pb2.FileDescriptorProto.FromString( more_messages_pb2.DESCRIPTOR.serialized_pb) test1_desc = descriptor_pb2.FileDescriptorProto.FromString( descriptor_pool_test1_pb2.DESCRIPTOR.serialized_pb) test2_desc = descriptor_pb2.FileDescriptorProto.FromString( descriptor_pool_test2_pb2.DESCRIPTOR.serialized_pb) self.pool.Add(more_messages_desc) self.pool.Add(test1_desc) self.pool.Add(test2_desc) TEST1_FILE.CheckFile(self, self.pool) TEST2_FILE.CheckFile(self, self.pool)
def testDefaultValueForCustomMessages(self): """Check the value returned by non-existent fields.""" def _CheckValueAndType(value, expected_value, expected_type): self.assertEqual(value, expected_value) self.assertIsInstance(value, expected_type) def _CheckDefaultValues(msg): try: int64 = long except NameError: # Python3 int64 = int try: unicode_type = unicode except NameError: # Python3 unicode_type = str _CheckValueAndType(msg.optional_int32, 0, int) _CheckValueAndType(msg.optional_uint64, 0, (int64, int)) _CheckValueAndType(msg.optional_float, 0, (float, int)) _CheckValueAndType(msg.optional_double, 0, (float, int)) _CheckValueAndType(msg.optional_bool, False, bool) _CheckValueAndType(msg.optional_string, u'', unicode_type) _CheckValueAndType(msg.optional_bytes, b'', bytes) _CheckValueAndType(msg.optional_nested_enum, msg.FOO, int) # First for the generated message _CheckDefaultValues(unittest_pb2.TestAllTypes()) # Then for a message built with from the DescriptorPool. pool = descriptor_pool.DescriptorPool() pool.Add(descriptor_pb2.FileDescriptorProto.FromString( unittest_import_public_pb2.DESCRIPTOR.serialized_pb)) pool.Add(descriptor_pb2.FileDescriptorProto.FromString( unittest_import_pb2.DESCRIPTOR.serialized_pb)) pool.Add(descriptor_pb2.FileDescriptorProto.FromString( unittest_pb2.DESCRIPTOR.serialized_pb)) message_class = message_factory.MessageFactory(pool).GetPrototype( pool.FindMessageTypeByName( unittest_pb2.TestAllTypes.DESCRIPTOR.full_name)) _CheckDefaultValues(message_class())
def testEmptyDescriptorPool(self): # Check that an empty DescriptorPool() contains no messages. pool = descriptor_pool.DescriptorPool() proto_file_name = descriptor_pb2.DESCRIPTOR.name self.assertRaises(KeyError, pool.FindFileByName, proto_file_name) # Add the above file to the pool file_descriptor = descriptor_pb2.FileDescriptorProto() descriptor_pb2.DESCRIPTOR.CopyToProto(file_descriptor) pool.Add(file_descriptor) # Now it exists. self.assertTrue(pool.FindFileByName(proto_file_name))
def testCustomDescriptorPool(self): # Create a new pool, and add a file descriptor. pool = descriptor_pool.DescriptorPool() file_desc = descriptor_pb2.FileDescriptorProto( name='some/file.proto', package='package') file_desc.message_type.add(name='Message') pool.Add(file_desc) self.assertEqual(pool.FindFileByName('some/file.proto').name, 'some/file.proto') self.assertEqual(pool.FindMessageTypeByName('package.Message').name, 'Message')
def testAddFileDescriptor(self): # pylint: disable=g-import-not-at-top from google.protobuf.pyext import _message pool = _message.default_pool file_desc = descriptor_pb2.FileDescriptorProto(name='some/file.proto') pool.Add(file_desc) pool.AddSerializedFile(file_desc.SerializeToString())
def setUp(self): self.factory_test1_fd = descriptor_pb2.FileDescriptorProto.FromString( factory_test1_pb2.DESCRIPTOR.serialized_pb) self.factory_test2_fd = descriptor_pb2.FileDescriptorProto.FromString( factory_test2_pb2.DESCRIPTOR.serialized_pb)
def setUpClass(cls): # At the moment, reference cycles between DescriptorPool and Message classes # are not detected and these objects are never freed. # To avoid errors with ReferenceLeakChecker, we create the class only once. file_desc = """ name: "f/f.msg2" package: "f" message_type { name: "msg1" field { name: "payload" number: 1 label: LABEL_OPTIONAL type: TYPE_STRING } } message_type { name: "msg2" field { name: "field" number: 1 label: LABEL_OPTIONAL type: TYPE_MESSAGE type_name: "msg1" } } """ pool = descriptor_pool.DescriptorPool() desc = descriptor_pb2.FileDescriptorProto() text_format.Parse(file_desc, desc) pool.Add(desc) cls.proto_cls = message_factory.MessageFactory(pool).GetPrototype( pool.FindMessageTypeByName('f.msg2'))
def testFileDescriptorErrors(self): file_name = 'test_file_descriptor_errors.proto' package_name = 'test_file_descriptor_errors.proto' file_descriptor_proto = descriptor_pb2.FileDescriptorProto() file_descriptor_proto.name = file_name file_descriptor_proto.package = package_name m1 = file_descriptor_proto.message_type.add() m1.name = 'msg1' # Compiles the proto into the C++ descriptor pool descriptor.FileDescriptor( file_name, package_name, serialized_pb=file_descriptor_proto.SerializeToString()) # Add a FileDescriptorProto that has duplicate symbols another_file_name = 'another_test_file_descriptor_errors.proto' file_descriptor_proto.name = another_file_name m2 = file_descriptor_proto.message_type.add() m2.name = 'msg2' with self.assertRaises(TypeError) as cm: descriptor.FileDescriptor( another_file_name, package_name, serialized_pb=file_descriptor_proto.SerializeToString()) self.assertTrue(hasattr(cm, 'exception'), '%s not raised' % getattr(cm.expected, '__name__', cm.expected)) self.assertIn('test_file_descriptor_errors.proto', str(cm.exception)) # Error message will say something about this definition being a # duplicate, though we don't check the message exactly to avoid a # dependency on the C++ logging code. self.assertIn('test_file_descriptor_errors.msg1', str(cm.exception))
def testParsingFlatClass(self): """Test that the generated class can parse a flat message.""" file_descriptor = descriptor_pb2.FileDescriptorProto() file_descriptor.ParseFromString(self._GetSerializedFileDescriptor('B')) msg_descriptor = descriptor.MakeDescriptor( file_descriptor.message_type[0]) msg_class = reflection.MakeClass(msg_descriptor) msg = msg_class() msg_str = ( 'flat: 0 ' 'flat: 1 ' 'flat: 2 ') text_format.Merge(msg_str, msg) self.assertEqual(msg.flat, [0, 1, 2])
def testMakeDescriptorWithNestedFields(self): file_descriptor_proto = descriptor_pb2.FileDescriptorProto() file_descriptor_proto.name = 'Foo2' message_type = file_descriptor_proto.message_type.add() message_type.name = file_descriptor_proto.name nested_type = message_type.nested_type.add() nested_type.name = 'Sub' enum_type = nested_type.enum_type.add() enum_type.name = 'FOO' enum_type_val = enum_type.value.add() enum_type_val.name = 'BAR' enum_type_val.number = 3 field = message_type.field.add() field.number = 1 field.name = 'uint64_field' field.label = descriptor.FieldDescriptor.LABEL_REQUIRED field.type = descriptor.FieldDescriptor.TYPE_UINT64 field = message_type.field.add() field.number = 2 field.name = 'nested_message_field' field.label = descriptor.FieldDescriptor.LABEL_REQUIRED field.type = descriptor.FieldDescriptor.TYPE_MESSAGE field.type_name = 'Sub' enum_field = nested_type.field.add() enum_field.number = 2 enum_field.name = 'bar_field' enum_field.label = descriptor.FieldDescriptor.LABEL_REQUIRED enum_field.type = descriptor.FieldDescriptor.TYPE_ENUM enum_field.type_name = 'Foo2.Sub.FOO' result = descriptor.MakeDescriptor(message_type) self.assertEqual(result.fields[0].cpp_type, descriptor.FieldDescriptor.CPPTYPE_UINT64) self.assertEqual(result.fields[1].cpp_type, descriptor.FieldDescriptor.CPPTYPE_MESSAGE) self.assertEqual(result.fields[1].message_type.containing_type, result) self.assertEqual(result.nested_types[0].fields[0].full_name, 'Foo2.Sub.bar_field') self.assertEqual(result.nested_types[0].fields[0].enum_type, result.nested_types[0].enum_types[0])
def testParsingFlatClassWithExplicitClassDeclaration(self): """Test that the generated class can parse a flat message.""" file_descriptor = descriptor_pb2.FileDescriptorProto() file_descriptor.ParseFromString(self._GetSerializedFileDescriptor('A')) msg_descriptor = descriptor.MakeDescriptor( file_descriptor.message_type[0]) class MessageClass(message.Message): __metaclass__ = reflection.GeneratedProtocolMessageType DESCRIPTOR = msg_descriptor msg = MessageClass() msg_str = ( 'flat: 0 ' 'flat: 1 ' 'flat: 2 ') text_format.Merge(msg_str, msg) self.assertEqual(msg.flat, [0, 1, 2])
def testCopyToProto_FileDescriptor(self): UNITTEST_IMPORT_FILE_DESCRIPTOR_ASCII = (""" name: 'google/protobuf/unittest_import.proto' package: 'protobuf_unittest_import' dependency: 'google/protobuf/unittest_import_public.proto' message_type: < name: 'ImportMessage' field: < name: 'd' number: 1 label: 1 # Optional type: 5 # TYPE_INT32 > > """ + """enum_type: < name: 'ImportEnum' value: < name: 'IMPORT_FOO' number: 7 > value: < name: 'IMPORT_BAR' number: 8 > value: < name: 'IMPORT_BAZ' number: 9 > > options: < java_package: 'com.google.protobuf.test' optimize_for: 1 # SPEED > public_dependency: 0 """) self._InternalTestCopyToProto( unittest_import_pb2.DESCRIPTOR, descriptor_pb2.FileDescriptorProto, UNITTEST_IMPORT_FILE_DESCRIPTOR_ASCII)
def testMakeDescriptorWithUnsignedIntField(self): file_descriptor_proto = descriptor_pb2.FileDescriptorProto() file_descriptor_proto.name = 'Foo' message_type = file_descriptor_proto.message_type.add() message_type.name = file_descriptor_proto.name field = message_type.field.add() field.number = 1 field.name = 'uint64_field' field.label = descriptor.FieldDescriptor.LABEL_REQUIRED field.type = descriptor.FieldDescriptor.TYPE_UINT64 result = descriptor.MakeDescriptor(message_type) self.assertEqual(result.fields[0].cpp_type, descriptor.FieldDescriptor.CPPTYPE_UINT64)
