Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-06-25 08:26
📄 __init__.py # # Copyright 2015 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Shared __init__.py for apitools.""" from pkgutil import extend_path __path__ = extend_path(__path__, __name__)	693 B	2025-03-11 18:02
📄 descriptor.py #!/usr/bin/env python # # Copyright 2010 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # """Services descriptor definitions. Contains message definitions and functions for converting service classes into transmittable message format. Describing an Enum instance, Enum class, Field class or Message class will generate an appropriate descriptor object that describes that class. This message can itself be used to transmit information to clients wishing to know the description of an enum value, enum, field or message without needing to download the source code. This format is also compatible with other, non-Python languages. The descriptors are modeled to be binary compatible with https://github.com/google/protobuf NOTE: The names of types and fields are not always the same between these descriptors and the ones defined in descriptor.proto. This was done in order to make source code files that use these descriptors easier to read. For example, it is not necessary to prefix TYPE to all the values in FieldDescriptor.Variant as is done in descriptor.proto FieldDescriptorProto.Type. Example: class Pixel(messages.Message): x = messages.IntegerField(1, required=True) y = messages.IntegerField(2, required=True) color = messages.BytesField(3) # Describe Pixel class using message descriptor. fields = [] field = FieldDescriptor() field.name = 'x' field.number = 1 field.label = FieldDescriptor.Label.REQUIRED field.variant = FieldDescriptor.Variant.INT64 fields.append(field) field = FieldDescriptor() field.name = 'y' field.number = 2 field.label = FieldDescriptor.Label.REQUIRED field.variant = FieldDescriptor.Variant.INT64 fields.append(field) field = FieldDescriptor() field.name = 'color' field.number = 3 field.label = FieldDescriptor.Label.OPTIONAL field.variant = FieldDescriptor.Variant.BYTES fields.append(field) message = MessageDescriptor() message.name = 'Pixel' message.fields = fields # Describing is the equivalent of building the above message. message == describe_message(Pixel) Public Classes: EnumValueDescriptor: Describes Enum values. EnumDescriptor: Describes Enum classes. FieldDescriptor: Describes field instances. FileDescriptor: Describes a single 'file' unit. FileSet: Describes a collection of file descriptors. MessageDescriptor: Describes Message classes. Public Functions: describe_enum_value: Describe an individual enum-value. describe_enum: Describe an Enum class. describe_field: Describe a Field definition. describe_file: Describe a 'file' unit from a Python module or object. describe_file_set: Describe a file set from a list of modules or objects. describe_message: Describe a Message definition. """ import codecs import types import six from apitools.base.protorpclite import messages from apitools.base.protorpclite import util __all__ = [ 'EnumDescriptor', 'EnumValueDescriptor', 'FieldDescriptor', 'MessageDescriptor', 'FileDescriptor', 'FileSet', 'DescriptorLibrary', 'describe_enum', 'describe_enum_value', 'describe_field', 'describe_message', 'describe_file', 'describe_file_set', 'describe', 'import_descriptor_loader', ] # NOTE: MessageField is missing because message fields cannot have # a default value at this time. # TODO(rafek): Support default message values. # # Map to functions that convert default values of fields of a given type # to a string. The function must return a value that is compatible with # FieldDescriptor.default_value and therefore a unicode string. _DEFAULT_TO_STRING_MAP = { messages.IntegerField: six.text_type, messages.FloatField: six.text_type, messages.BooleanField: lambda value: value and u'true' or u'false', messages.BytesField: lambda value: codecs.escape_encode(value)[0], messages.StringField: lambda value: value, messages.EnumField: lambda value: six.text_type(value.number), } _DEFAULT_FROM_STRING_MAP = { messages.IntegerField: int, messages.FloatField: float, messages.BooleanField: lambda value: value == u'true', messages.BytesField: lambda value: codecs.escape_decode(value)[0], messages.StringField: lambda value: value, messages.EnumField: int, } class EnumValueDescriptor(messages.Message): """Enum value descriptor. Fields: name: Name of enumeration value. number: Number of enumeration value. """ # TODO(rafek): Why are these listed as optional in descriptor.proto. # Harmonize? name = messages.StringField(1, required=True) number = messages.IntegerField(2, required=True, variant=messages.Variant.INT32) class EnumDescriptor(messages.Message): """Enum class descriptor. Fields: name: Name of Enum without any qualification. values: Values defined by Enum class. """ name = messages.StringField(1) values = messages.MessageField(EnumValueDescriptor, 2, repeated=True) class FieldDescriptor(messages.Message): """Field definition descriptor. Enums: Variant: Wire format hint sub-types for field. Label: Values for optional, required and repeated fields. Fields: name: Name of field. number: Number of field. variant: Variant of field. type_name: Type name for message and enum fields. default_value: String representation of default value. """ Variant = messages.Variant # pylint:disable=invalid-name class Label(messages.Enum): """Field label.""" OPTIONAL = 1 REQUIRED = 2 REPEATED = 3 name = messages.StringField(1, required=True) number = messages.IntegerField(3, required=True, variant=messages.Variant.INT32) label = messages.EnumField(Label, 4, default=Label.OPTIONAL) variant = messages.EnumField(Variant, 5) type_name = messages.StringField(6) # For numeric types, contains the original text representation of # the value. # For booleans, "true" or "false". # For strings, contains the default text contents (not escaped in any # way). # For bytes, contains the C escaped value. All bytes < 128 are that are # traditionally considered unprintable are also escaped. default_value = messages.StringField(7) class MessageDescriptor(messages.Message): """Message definition descriptor. Fields: name: Name of Message without any qualification. fields: Fields defined for message. message_types: Nested Message classes defined on message. enum_types: Nested Enum classes defined on message. """ name = messages.StringField(1) fields = messages.MessageField(FieldDescriptor, 2, repeated=True) message_types = messages.MessageField( 'apitools.base.protorpclite.descriptor.MessageDescriptor', 3, repeated=True) enum_types = messages.MessageField(EnumDescriptor, 4, repeated=True) class FileDescriptor(messages.Message): """Description of file containing protobuf definitions. Fields: package: Fully qualified name of package that definitions belong to. message_types: Message definitions contained in file. enum_types: Enum definitions contained in file. """ package = messages.StringField(2) # TODO(rafek): Add dependency field message_types = messages.MessageField(MessageDescriptor, 4, repeated=True) enum_types = messages.MessageField(EnumDescriptor, 5, repeated=True) class FileSet(messages.Message): """A collection of FileDescriptors. Fields: files: Files in file-set. """ files = messages.MessageField(FileDescriptor, 1, repeated=True) def describe_enum_value(enum_value): """Build descriptor for Enum instance. Args: enum_value: Enum value to provide descriptor for. Returns: Initialized EnumValueDescriptor instance describing the Enum instance. """ enum_value_descriptor = EnumValueDescriptor() enum_value_descriptor.name = six.text_type(enum_value.name) enum_value_descriptor.number = enum_value.number return enum_value_descriptor def describe_enum(enum_definition): """Build descriptor for Enum class. Args: enum_definition: Enum class to provide descriptor for. Returns: Initialized EnumDescriptor instance describing the Enum class. """ enum_descriptor = EnumDescriptor() enum_descriptor.name = enum_definition.definition_name().split('.')[-1] values = [] for number in sorted(enum_definition.numbers()): value = enum_definition.lookup_by_number(number) values.append(describe_enum_value(value)) if values: enum_descriptor.values = values return enum_descriptor def describe_field(field_definition): """Build descriptor for Field instance. Args: field_definition: Field instance to provide descriptor for. Returns: Initialized FieldDescriptor instance describing the Field instance. """ field_descriptor = FieldDescriptor() field_descriptor.name = field_definition.name field_descriptor.number = field_definition.number field_descriptor.variant = field_definition.variant if isinstance(field_definition, messages.EnumField): field_descriptor.type_name = field_definition.type.definition_name() if isinstance(field_definition, messages.MessageField): field_descriptor.type_name = ( field_definition.message_type.definition_name()) if field_definition.default is not None: field_descriptor.default_value = _DEFAULT_TO_STRING_MAP[ type(field_definition)](field_definition.default) # Set label. if field_definition.repeated: field_descriptor.label = FieldDescriptor.Label.REPEATED elif field_definition.required: field_descriptor.label = FieldDescriptor.Label.REQUIRED else: field_descriptor.label = FieldDescriptor.Label.OPTIONAL return field_descriptor def describe_message(message_definition): """Build descriptor for Message class. Args: message_definition: Message class to provide descriptor for. Returns: Initialized MessageDescriptor instance describing the Message class. """ message_descriptor = MessageDescriptor() message_descriptor.name = message_definition.definition_name().split( '.')[-1] fields = sorted(message_definition.all_fields(), key=lambda v: v.number) if fields: message_descriptor.fields = [describe_field(field) for field in fields] try: nested_messages = message_definition.__messages__ except AttributeError: pass else: message_descriptors = [] for name in nested_messages: value = getattr(message_definition, name) message_descriptors.append(describe_message(value)) message_descriptor.message_types = message_descriptors try: nested_enums = message_definition.__enums__ except AttributeError: pass else: enum_descriptors = [] for name in nested_enums: value = getattr(message_definition, name) enum_descriptors.append(describe_enum(value)) message_descriptor.enum_types = enum_descriptors return message_descriptor def describe_file(module): """Build a file from a specified Python module. Args: module: Python module to describe. Returns: Initialized FileDescriptor instance describing the module. """ descriptor = FileDescriptor() descriptor.package = util.get_package_for_module(module) if not descriptor.package: descriptor.package = None message_descriptors = [] enum_descriptors = [] # Need to iterate over all top level attributes of the module looking for # message and enum definitions. Each definition must be itself described. for name in sorted(dir(module)): value = getattr(module, name) if isinstance(value, type): if issubclass(value, messages.Message): message_descriptors.append(describe_message(value)) elif issubclass(value, messages.Enum): enum_descriptors.append(describe_enum(value)) if message_descriptors: descriptor.message_types = message_descriptors if enum_descriptors: descriptor.enum_types = enum_descriptors return descriptor def describe_file_set(modules): """Build a file set from a specified Python modules. Args: modules: Iterable of Python module to describe. Returns: Initialized FileSet instance describing the modules. """ descriptor = FileSet() file_descriptors = [] for module in modules: file_descriptors.append(describe_file(module)) if file_descriptors: descriptor.files = file_descriptors return descriptor def describe(value): """Describe any value as a descriptor. Helper function for describing any object with an appropriate descriptor object. Args: value: Value to describe as a descriptor. Returns: Descriptor message class if object is describable as a descriptor, else None. """ if isinstance(value, types.ModuleType): return describe_file(value) elif isinstance(value, messages.Field): return describe_field(value) elif isinstance(value, messages.Enum): return describe_enum_value(value) elif isinstance(value, type): if issubclass(value, messages.Message): return describe_message(value) elif issubclass(value, messages.Enum): return describe_enum(value) return None @util.positional(1) def import_descriptor_loader(definition_name, importer=__import__): """Find objects by importing modules as needed. A definition loader is a function that resolves a definition name to a descriptor. The import finder resolves definitions to their names by importing modules when necessary. Args: definition_name: Name of definition to find. importer: Import function used for importing new modules. Returns: Appropriate descriptor for any describable type located by name. Raises: DefinitionNotFoundError when a name does not refer to either a definition or a module. """ # Attempt to import descriptor as a module. if definition_name.startswith('.'): definition_name = definition_name[1:] if not definition_name.startswith('.'): leaf = definition_name.split('.')[-1] if definition_name: try: module = importer(definition_name, '', '', [leaf]) except ImportError: pass else: return describe(module) try: # Attempt to use messages.find_definition to find item. return describe(messages.find_definition(definition_name, importer=__import__)) except messages.DefinitionNotFoundError as err: # There are things that find_definition will not find, but if # the parent is loaded, its children can be searched for a # match. split_name = definition_name.rsplit('.', 1) if len(split_name) > 1: parent, child = split_name try: parent_definition = import_descriptor_loader( parent, importer=importer) except messages.DefinitionNotFoundError: # Fall through to original error. pass else: # Check the parent definition for a matching descriptor. if isinstance(parent_definition, EnumDescriptor): search_list = parent_definition.values or [] elif isinstance(parent_definition, MessageDescriptor): search_list = parent_definition.fields or [] else: search_list = [] for definition in search_list: if definition.name == child: return definition # Still didn't find. Reraise original exception. raise err class DescriptorLibrary(object): """A descriptor library is an object that contains known definitions. A descriptor library contains a cache of descriptor objects mapped by definition name. It contains all types of descriptors except for file sets. When a definition name is requested that the library does not know about it can be provided with a descriptor loader which attempt to resolve the missing descriptor. """ @util.positional(1) def __init__(self, descriptors=None, descriptor_loader=import_descriptor_loader): """Constructor. Args: descriptors: A dictionary or dictionary-like object that can be used to store and cache descriptors by definition name. definition_loader: A function used for resolving missing descriptors. The function takes a definition name as its parameter and returns an appropriate descriptor. It may raise DefinitionNotFoundError. """ self.__descriptor_loader = descriptor_loader self.__descriptors = descriptors or {} def lookup_descriptor(self, definition_name): """Lookup descriptor by name. Get descriptor from library by name. If descriptor is not found will attempt to find via descriptor loader if provided. Args: definition_name: Definition name to find. Returns: Descriptor that describes definition name. Raises: DefinitionNotFoundError if not descriptor exists for definition name. """ try: return self.__descriptors[definition_name] except KeyError: pass if self.__descriptor_loader: definition = self.__descriptor_loader(definition_name) self.__descriptors[definition_name] = definition return definition else: raise messages.DefinitionNotFoundError( 'Could not find definition for %s' % definition_name) def lookup_package(self, definition_name): """Determines the package name for any definition. Determine the package that any definition name belongs to. May check parent for package name and will resolve missing descriptors if provided descriptor loader. Args: definition_name: Definition name to find package for. """ while True: descriptor = self.lookup_descriptor(definition_name) if isinstance(descriptor, FileDescriptor): return descriptor.package else: index = definition_name.rfind('.') if index < 0: return None definition_name = definition_name[:index]	19.25 KB	2025-03-11 18:02
📄 descriptor_test.py #!/usr/bin/env python # # Copyright 2010 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # """Tests for apitools.base.protorpclite.descriptor.""" import platform import types import unittest import six from apitools.base.protorpclite import descriptor from apitools.base.protorpclite import message_types from apitools.base.protorpclite import messages from apitools.base.protorpclite import test_util RUSSIA = u'\u0420\u043e\u0441\u0441\u0438\u044f' class ModuleInterfaceTest(test_util.ModuleInterfaceTest, test_util.TestCase): MODULE = descriptor class DescribeEnumValueTest(test_util.TestCase): def testDescribe(self): class MyEnum(messages.Enum): MY_NAME = 10 expected = descriptor.EnumValueDescriptor() expected.name = 'MY_NAME' expected.number = 10 described = descriptor.describe_enum_value(MyEnum.MY_NAME) described.check_initialized() self.assertEqual(expected, described) class DescribeEnumTest(test_util.TestCase): def testEmptyEnum(self): class EmptyEnum(messages.Enum): pass expected = descriptor.EnumDescriptor() expected.name = 'EmptyEnum' described = descriptor.describe_enum(EmptyEnum) described.check_initialized() self.assertEqual(expected, described) def testNestedEnum(self): class MyScope(messages.Message): class NestedEnum(messages.Enum): pass expected = descriptor.EnumDescriptor() expected.name = 'NestedEnum' described = descriptor.describe_enum(MyScope.NestedEnum) described.check_initialized() self.assertEqual(expected, described) @unittest.skipIf('PyPy' in platform.python_implementation(), 'todo: reenable this') def testEnumWithItems(self): class EnumWithItems(messages.Enum): A = 3 B = 1 C = 2 expected = descriptor.EnumDescriptor() expected.name = 'EnumWithItems' a = descriptor.EnumValueDescriptor() a.name = 'A' a.number = 3 b = descriptor.EnumValueDescriptor() b.name = 'B' b.number = 1 c = descriptor.EnumValueDescriptor() c.name = 'C' c.number = 2 expected.values = [b, c, a] described = descriptor.describe_enum(EnumWithItems) described.check_initialized() self.assertEqual(expected, described) class DescribeFieldTest(test_util.TestCase): def testLabel(self): for repeated, required, expected_label in ( (True, False, descriptor.FieldDescriptor.Label.REPEATED), (False, True, descriptor.FieldDescriptor.Label.REQUIRED), (False, False, descriptor.FieldDescriptor.Label.OPTIONAL)): field = messages.IntegerField( 10, required=required, repeated=repeated) field.name = 'a_field' expected = descriptor.FieldDescriptor() expected.name = 'a_field' expected.number = 10 expected.label = expected_label expected.variant = descriptor.FieldDescriptor.Variant.INT64 described = descriptor.describe_field(field) described.check_initialized() self.assertEqual(expected, described) def testDefault(self): test_cases = ( (messages.IntegerField, 200, '200'), (messages.FloatField, 1.5, '1.5'), (messages.FloatField, 1e6, '1000000.0'), (messages.BooleanField, True, 'true'), (messages.BooleanField, False, 'false'), (messages.BytesField, b''.join([six.int2byte(x) for x in (31, 32, 33)]), b'\\x1f !'), (messages.StringField, RUSSIA, RUSSIA), ) for field_class, default, expected_default in test_cases: field = field_class(10, default=default) field.name = u'a_field' expected = descriptor.FieldDescriptor() expected.name = u'a_field' expected.number = 10 expected.label = descriptor.FieldDescriptor.Label.OPTIONAL expected.variant = field_class.DEFAULT_VARIANT expected.default_value = expected_default described = descriptor.describe_field(field) described.check_initialized() self.assertEqual(expected, described) def testDefault_EnumField(self): class MyEnum(messages.Enum): VAL = 1 module_name = test_util.get_module_name(MyEnum) field = messages.EnumField(MyEnum, 10, default=MyEnum.VAL) field.name = 'a_field' expected = descriptor.FieldDescriptor() expected.name = 'a_field' expected.number = 10 expected.label = descriptor.FieldDescriptor.Label.OPTIONAL expected.variant = messages.EnumField.DEFAULT_VARIANT expected.type_name = '%s.MyEnum' % module_name expected.default_value = '1' described = descriptor.describe_field(field) self.assertEqual(expected, described) def testMessageField(self): field = messages.MessageField(descriptor.FieldDescriptor, 10) field.name = 'a_field' expected = descriptor.FieldDescriptor() expected.name = 'a_field' expected.number = 10 expected.label = descriptor.FieldDescriptor.Label.OPTIONAL expected.variant = messages.MessageField.DEFAULT_VARIANT expected.type_name = ( 'apitools.base.protorpclite.descriptor.FieldDescriptor') described = descriptor.describe_field(field) described.check_initialized() self.assertEqual(expected, described) def testDateTimeField(self): field = message_types.DateTimeField(20) field.name = 'a_timestamp' expected = descriptor.FieldDescriptor() expected.name = 'a_timestamp' expected.number = 20 expected.label = descriptor.FieldDescriptor.Label.OPTIONAL expected.variant = messages.MessageField.DEFAULT_VARIANT expected.type_name = ( 'apitools.base.protorpclite.message_types.DateTimeMessage') described = descriptor.describe_field(field) described.check_initialized() self.assertEqual(expected, described) class DescribeMessageTest(test_util.TestCase): def testEmptyDefinition(self): class MyMessage(messages.Message): pass expected = descriptor.MessageDescriptor() expected.name = 'MyMessage' described = descriptor.describe_message(MyMessage) described.check_initialized() self.assertEqual(expected, described) def testDefinitionWithFields(self): class MessageWithFields(messages.Message): field1 = messages.IntegerField(10) field2 = messages.StringField(30) field3 = messages.IntegerField(20) expected = descriptor.MessageDescriptor() expected.name = 'MessageWithFields' expected.fields = [ descriptor.describe_field( MessageWithFields.field_by_name('field1')), descriptor.describe_field( MessageWithFields.field_by_name('field3')), descriptor.describe_field( MessageWithFields.field_by_name('field2')), ] described = descriptor.describe_message(MessageWithFields) described.check_initialized() self.assertEqual(expected, described) def testNestedEnum(self): class MessageWithEnum(messages.Message): class Mood(messages.Enum): GOOD = 1 BAD = 2 UGLY = 3 class Music(messages.Enum): CLASSIC = 1 JAZZ = 2 BLUES = 3 expected = descriptor.MessageDescriptor() expected.name = 'MessageWithEnum' expected.enum_types = [descriptor.describe_enum(MessageWithEnum.Mood), descriptor.describe_enum(MessageWithEnum.Music)] described = descriptor.describe_message(MessageWithEnum) described.check_initialized() self.assertEqual(expected, described) def testNestedMessage(self): class MessageWithMessage(messages.Message): class Nesty(messages.Message): pass expected = descriptor.MessageDescriptor() expected.name = 'MessageWithMessage' expected.message_types = [ descriptor.describe_message(MessageWithMessage.Nesty)] described = descriptor.describe_message(MessageWithMessage) described.check_initialized() self.assertEqual(expected, described) class DescribeFileTest(test_util.TestCase): """Test describing modules.""" def LoadModule(self, module_name, source): result = { '__name__': module_name, 'messages': messages, } exec(source, result) module = types.ModuleType(module_name) for name, value in result.items(): setattr(module, name, value) return module def testEmptyModule(self): """Test describing an empty file.""" module = types.ModuleType('my.package.name') expected = descriptor.FileDescriptor() expected.package = 'my.package.name' described = descriptor.describe_file(module) described.check_initialized() self.assertEqual(expected, described) def testNoPackageName(self): """Test describing a module with no module name.""" module = types.ModuleType('') expected = descriptor.FileDescriptor() described = descriptor.describe_file(module) described.check_initialized() self.assertEqual(expected, described) def testPackageName(self): """Test using the 'package' module attribute.""" module = types.ModuleType('my.module.name') module.package = 'my.package.name' expected = descriptor.FileDescriptor() expected.package = 'my.package.name' described = descriptor.describe_file(module) described.check_initialized() self.assertEqual(expected, described) def testMain(self): """Test using the 'package' module attribute.""" module = types.ModuleType('__main__') module.__file__ = '/blim/blam/bloom/my_package.py' expected = descriptor.FileDescriptor() expected.package = 'my_package' described = descriptor.describe_file(module) described.check_initialized() self.assertEqual(expected, described) def testMessages(self): """Test that messages are described.""" module = self.LoadModule('my.package', 'class Message1(messages.Message): pass\n' 'class Message2(messages.Message): pass\n') message1 = descriptor.MessageDescriptor() message1.name = 'Message1' message2 = descriptor.MessageDescriptor() message2.name = 'Message2' expected = descriptor.FileDescriptor() expected.package = 'my.package' expected.message_types = [message1, message2] described = descriptor.describe_file(module) described.check_initialized() self.assertEqual(expected, described) def testEnums(self): """Test that enums are described.""" module = self.LoadModule('my.package', 'class Enum1(messages.Enum): pass\n' 'class Enum2(messages.Enum): pass\n') enum1 = descriptor.EnumDescriptor() enum1.name = 'Enum1' enum2 = descriptor.EnumDescriptor() enum2.name = 'Enum2' expected = descriptor.FileDescriptor() expected.package = 'my.package' expected.enum_types = [enum1, enum2] described = descriptor.describe_file(module) described.check_initialized() self.assertEqual(expected, described) class DescribeFileSetTest(test_util.TestCase): """Test describing multiple modules.""" def testNoModules(self): """Test what happens when no modules provided.""" described = descriptor.describe_file_set([]) described.check_initialized() # The described FileSet.files will be None. self.assertEqual(descriptor.FileSet(), described) def testWithModules(self): """Test what happens when no modules provided.""" modules = [types.ModuleType('package1'), types.ModuleType('package1')] file1 = descriptor.FileDescriptor() file1.package = 'package1' file2 = descriptor.FileDescriptor() file2.package = 'package2' expected = descriptor.FileSet() expected.files = [file1, file1] described = descriptor.describe_file_set(modules) described.check_initialized() self.assertEqual(expected, described) class DescribeTest(test_util.TestCase): def testModule(self): self.assertEqual(descriptor.describe_file(test_util), descriptor.describe(test_util)) def testField(self): self.assertEqual( descriptor.describe_field(test_util.NestedMessage.a_value), descriptor.describe(test_util.NestedMessage.a_value)) def testEnumValue(self): self.assertEqual( descriptor.describe_enum_value( test_util.OptionalMessage.SimpleEnum.VAL1), descriptor.describe(test_util.OptionalMessage.SimpleEnum.VAL1)) def testMessage(self): self.assertEqual(descriptor.describe_message(test_util.NestedMessage), descriptor.describe(test_util.NestedMessage)) def testEnum(self): self.assertEqual( descriptor.describe_enum(test_util.OptionalMessage.SimpleEnum), descriptor.describe(test_util.OptionalMessage.SimpleEnum)) def testUndescribable(self): class NonService(object): def fn(self): pass for value in (NonService, NonService.fn, 1, 'string', 1.2, None): self.assertEqual(None, descriptor.describe(value)) class ModuleFinderTest(test_util.TestCase): def testFindMessage(self): self.assertEqual( descriptor.describe_message(descriptor.FileSet), descriptor.import_descriptor_loader( 'apitools.base.protorpclite.descriptor.FileSet')) def testFindField(self): self.assertEqual( descriptor.describe_field(descriptor.FileSet.files), descriptor.import_descriptor_loader( 'apitools.base.protorpclite.descriptor.FileSet.files')) def testFindEnumValue(self): self.assertEqual( descriptor.describe_enum_value( test_util.OptionalMessage.SimpleEnum.VAL1), descriptor.import_descriptor_loader( 'apitools.base.protorpclite.test_util.' 'OptionalMessage.SimpleEnum.VAL1')) class DescriptorLibraryTest(test_util.TestCase): def setUp(self): self.packageless = descriptor.MessageDescriptor() self.packageless.name = 'Packageless' self.library = descriptor.DescriptorLibrary( descriptors={ 'not.real.Packageless': self.packageless, 'Packageless': self.packageless, }) def testLookupPackage(self): self.assertEqual('csv', self.library.lookup_package('csv')) self.assertEqual( 'apitools.base.protorpclite', self.library.lookup_package('apitools.base.protorpclite')) def testLookupNonPackages(self): lib = 'apitools.base.protorpclite.descriptor.DescriptorLibrary' for name in ('', 'a', lib): self.assertRaisesWithRegexpMatch( messages.DefinitionNotFoundError, 'Could not find definition for %s' % name, self.library.lookup_package, name) def testNoPackage(self): self.assertRaisesWithRegexpMatch( messages.DefinitionNotFoundError, 'Could not find definition for not.real', self.library.lookup_package, 'not.real.Packageless') self.assertEqual(None, self.library.lookup_package('Packageless')) if __name__ == '__main__': unittest.main()	16.7 KB	2025-03-11 18:02
📄 message_types.py #!/usr/bin/env python # # Copyright 2010 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # """Simple protocol message types. Includes new message and field types that are outside what is defined by the protocol buffers standard. """ import datetime from apitools.base.protorpclite import messages from apitools.base.protorpclite import util __all__ = [ 'DateTimeField', 'DateTimeMessage', 'VoidMessage', ] class VoidMessage(messages.Message): """Empty message.""" class DateTimeMessage(messages.Message): """Message to store/transmit a DateTime. Fields: milliseconds: Milliseconds since Jan 1st 1970 local time. time_zone_offset: Optional time zone offset, in minutes from UTC. """ milliseconds = messages.IntegerField(1, required=True) time_zone_offset = messages.IntegerField(2) class DateTimeField(messages.MessageField): """Field definition for datetime values. Stores a python datetime object as a field. If time zone information is included in the datetime object, it will be included in the encoded data when this is encoded/decoded. """ type = datetime.datetime message_type = DateTimeMessage @util.positional(3) def __init__(self, number, kwargs): super(DateTimeField, self).__init__(self.message_type, number, kwargs) def value_from_message(self, message): """Convert DateTimeMessage to a datetime. Args: A DateTimeMessage instance. Returns: A datetime instance. """ message = super(DateTimeField, self).value_from_message(message) if message.time_zone_offset is None: return datetime.datetime.fromtimestamp( message.milliseconds / 1000.0, tz=datetime.timezone.utc).replace(tzinfo=None) # Need to subtract the time zone offset, because when we call # datetime.fromtimestamp, it will add the time zone offset to the # value we pass. milliseconds = (message.milliseconds - 60000 * message.time_zone_offset) timezone = util.TimeZoneOffset(message.time_zone_offset) return datetime.datetime.fromtimestamp(milliseconds / 1000.0, tz=timezone) def value_to_message(self, value): value = super(DateTimeField, self).value_to_message(value) # First, determine the delta from the epoch, so we can fill in # DateTimeMessage's milliseconds field. if value.tzinfo is None: time_zone_offset = 0 local_epoch = datetime.datetime.fromtimestamp(0, tz=datetime.timezone.utc).replace(tzinfo=None) else: time_zone_offset = util.total_seconds( value.tzinfo.utcoffset(value)) # Determine Jan 1, 1970 local time. local_epoch = datetime.datetime.fromtimestamp(-time_zone_offset, tz=value.tzinfo) delta = value - local_epoch # Create and fill in the DateTimeMessage, including time zone if # one was specified. message = DateTimeMessage() message.milliseconds = int(util.total_seconds(delta) * 1000) if value.tzinfo is not None: utc_offset = value.tzinfo.utcoffset(value) if utc_offset is not None: message.time_zone_offset = int( util.total_seconds(value.tzinfo.utcoffset(value)) / 60) return message	4.05 KB	2025-03-11 18:02
📄 message_types_test.py #!/usr/bin/env python # # Copyright 2013 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # """Tests for apitools.base.protorpclite.message_types.""" import datetime import unittest from apitools.base.protorpclite import message_types from apitools.base.protorpclite import messages from apitools.base.protorpclite import test_util from apitools.base.protorpclite import util class ModuleInterfaceTest(test_util.ModuleInterfaceTest, test_util.TestCase): MODULE = message_types class DateTimeFieldTest(test_util.TestCase): def testValueToMessage(self): field = message_types.DateTimeField(1) message = field.value_to_message( datetime.datetime(2033, 2, 4, 11, 22, 10)) self.assertEqual( message_types.DateTimeMessage(milliseconds=1991128930000), message) def testValueToMessageBadValue(self): field = message_types.DateTimeField(1) self.assertRaisesWithRegexpMatch( messages.EncodeError, 'Expected type datetime, got int: 20', field.value_to_message, 20) def testValueToMessageWithTimeZone(self): time_zone = util.TimeZoneOffset(60 * 10) field = message_types.DateTimeField(1) message = field.value_to_message( datetime.datetime(2033, 2, 4, 11, 22, 10, tzinfo=time_zone)) self.assertEqual( message_types.DateTimeMessage(milliseconds=1991128930000, time_zone_offset=600), message) def testValueFromMessage(self): message = message_types.DateTimeMessage(milliseconds=1991128000000) field = message_types.DateTimeField(1) timestamp = field.value_from_message(message) self.assertEqual(datetime.datetime(2033, 2, 4, 11, 6, 40), timestamp) def testValueFromMessageBadValue(self): field = message_types.DateTimeField(1) self.assertRaisesWithRegexpMatch( messages.DecodeError, 'Expected type DateTimeMessage, got VoidMessage: <VoidMessage>', field.value_from_message, message_types.VoidMessage()) def testValueFromMessageWithTimeZone(self): message = message_types.DateTimeMessage(milliseconds=1991128000000, time_zone_offset=300) field = message_types.DateTimeField(1) timestamp = field.value_from_message(message) time_zone = util.TimeZoneOffset(60 * 5) self.assertEqual( datetime.datetime(2033, 2, 4, 11, 6, 40, tzinfo=time_zone), timestamp) if __name__ == '__main__': unittest.main()	3.12 KB	2025-03-11 18:02
📄 messages.py #!/usr/bin/env python # # Copyright 2010 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # # pylint: disable=too-many-lines """Stand-alone implementation of in memory protocol messages. Public Classes: Enum: Represents an enumerated type. Variant: Hint for wire format to determine how to serialize. Message: Base class for user defined messages. IntegerField: Field for integer values. FloatField: Field for float values. BooleanField: Field for boolean values. BytesField: Field for binary string values. StringField: Field for UTF-8 string values. MessageField: Field for other message type values. EnumField: Field for enumerated type values. Public Exceptions (indentation indications class hierarchy): EnumDefinitionError: Raised when enumeration is incorrectly defined. FieldDefinitionError: Raised when field is incorrectly defined. InvalidVariantError: Raised when variant is not compatible with field type. InvalidDefaultError: Raised when default is not compatiable with field. InvalidNumberError: Raised when field number is out of range or reserved. MessageDefinitionError: Raised when message is incorrectly defined. DuplicateNumberError: Raised when field has duplicate number with another. ValidationError: Raised when a message or field is not valid. DefinitionNotFoundError: Raised when definition not found. """ import types import weakref import six from apitools.base.protorpclite import util __all__ = [ 'MAX_ENUM_VALUE', 'MAX_FIELD_NUMBER', 'FIRST_RESERVED_FIELD_NUMBER', 'LAST_RESERVED_FIELD_NUMBER', 'Enum', 'Field', 'FieldList', 'Variant', 'Message', 'IntegerField', 'FloatField', 'BooleanField', 'BytesField', 'StringField', 'MessageField', 'EnumField', 'find_definition', 'Error', 'DecodeError', 'EncodeError', 'EnumDefinitionError', 'FieldDefinitionError', 'InvalidVariantError', 'InvalidDefaultError', 'InvalidNumberError', 'MessageDefinitionError', 'DuplicateNumberError', 'ValidationError', 'DefinitionNotFoundError', ] # pylint:disable=attribute-defined-outside-init # pylint:disable=protected-access # TODO(rafek): Add extended module test to ensure all exceptions # in services extends Error. Error = util.Error class EnumDefinitionError(Error): """Enumeration definition error.""" class FieldDefinitionError(Error): """Field definition error.""" class InvalidVariantError(FieldDefinitionError): """Invalid variant provided to field.""" class InvalidDefaultError(FieldDefinitionError): """Invalid default provided to field.""" class InvalidNumberError(FieldDefinitionError): """Invalid number provided to field.""" class MessageDefinitionError(Error): """Message definition error.""" class DuplicateNumberError(Error): """Duplicate number assigned to field.""" class DefinitionNotFoundError(Error): """Raised when definition is not found.""" class DecodeError(Error): """Error found decoding message from encoded form.""" class EncodeError(Error): """Error found when encoding message.""" class ValidationError(Error): """Invalid value for message error.""" def __str__(self): """Prints string with field name if present on exception.""" return Error.__str__(self) # Attributes that are reserved by a class definition that # may not be used by either Enum or Message class definitions. _RESERVED_ATTRIBUTE_NAMES = frozenset( ['__module__', '__doc__', '__qualname__', '__static_attributes__', '__firstlineno__']) _POST_INIT_FIELD_ATTRIBUTE_NAMES = frozenset( ['name', '_message_definition', '_MessageField__type', '_EnumField__type', '_EnumField__resolved_default']) _POST_INIT_ATTRIBUTE_NAMES = frozenset( ['_message_definition']) # Maximum enumeration value as defined by the protocol buffers standard. # All enum values must be less than or equal to this value. MAX_ENUM_VALUE = (2 29) - 1 # Maximum field number as defined by the protocol buffers standard. # All field numbers must be less than or equal to this value. MAX_FIELD_NUMBER = (2 29) - 1 # Field numbers between 19000 and 19999 inclusive are reserved by the # protobuf protocol and may not be used by fields. FIRST_RESERVED_FIELD_NUMBER = 19000 LAST_RESERVED_FIELD_NUMBER = 19999 # pylint: disable=no-value-for-parameter class _DefinitionClass(type): """Base meta-class used for definition meta-classes. The Enum and Message definition classes share some basic functionality. Both of these classes may be contained by a Message definition. After initialization, neither class may have attributes changed except for the protected _message_definition attribute, and that attribute may change only once. """ __initialized = False # pylint:disable=invalid-name def __init__(cls, name, bases, dct): """Constructor.""" type.__init__(cls, name, bases, dct) # Base classes may never be initialized. if cls.__bases__ != (object,): cls.__initialized = True def message_definition(cls): """Get outer Message definition that contains this definition. Returns: Containing Message definition if definition is contained within one, else None. """ try: return cls._message_definition() except AttributeError: return None def __setattr__(cls, name, value): """Overridden to avoid setting variables after init. Setting attributes on a class must work during the period of initialization to set the enumation value class variables and build the name/number maps. Once __init__ has set the __initialized flag to True prohibits setting any more values on the class. The class is in effect frozen. Args: name: Name of value to set. value: Value to set. """ if cls.__initialized and name not in _POST_INIT_ATTRIBUTE_NAMES: raise AttributeError('May not change values: %s' % name) else: type.__setattr__(cls, name, value) def __delattr__(cls, name): """Overridden so that cannot delete varaibles on definition classes.""" raise TypeError('May not delete attributes on definition class') def definition_name(cls): """Helper method for creating definition name. Names will be generated to include the classes package name, scope (if the class is nested in another definition) and class name. By default, the package name for a definition is derived from its module name. However, this value can be overriden by placing a 'package' attribute in the module that contains the definition class. For example: package = 'some.alternate.package' class MyMessage(Message): ... >>> MyMessage.definition_name() some.alternate.package.MyMessage Returns: Dot-separated fully qualified name of definition. """ outer_definition_name = cls.outer_definition_name() if outer_definition_name is None: return six.text_type(cls.__name__) return u'%s.%s' % (outer_definition_name, cls.__name__) def outer_definition_name(cls): """Helper method for creating outer definition name. Returns: If definition is nested, will return the outer definitions name, else the package name. """ outer_definition = cls.message_definition() if not outer_definition: return util.get_package_for_module(cls.__module__) return outer_definition.definition_name() def definition_package(cls): """Helper method for creating creating the package of a definition. Returns: Name of package that definition belongs to. """ outer_definition = cls.message_definition() if not outer_definition: return util.get_package_for_module(cls.__module__) return outer_definition.definition_package() class _EnumClass(_DefinitionClass): """Meta-class used for defining the Enum base class. Meta-class enables very specific behavior for any defined Enum class. All attributes defined on an Enum sub-class must be integers. Each attribute defined on an Enum sub-class is translated into an instance of that sub-class, with the name of the attribute as its name, and the number provided as its value. It also ensures that only one level of Enum class hierarchy is possible. In other words it is not possible to delcare sub-classes of sub-classes of Enum. This class also defines some functions in order to restrict the behavior of the Enum class and its sub-classes. It is not possible to change the behavior of the Enum class in later classes since any new classes may be defined with only integer values, and no methods. """ def __init__(cls, name, bases, dct): # Can only define one level of sub-classes below Enum. if not (bases == (object,) or bases == (Enum,)): raise EnumDefinitionError( 'Enum type %s may only inherit from Enum' % name) cls.__by_number = {} cls.__by_name = {} # Enum base class does not need to be initialized or locked. if bases != (object,): # Replace integer with number. for attribute, value in dct.items(): # Module will be in every enum class. if attribute in _RESERVED_ATTRIBUTE_NAMES: continue # Reject anything that is not an int. if not isinstance(value, six.integer_types): raise EnumDefinitionError( 'May only use integers in Enum definitions. ' 'Found: %s = %s' % (attribute, value)) # Protocol buffer standard recommends non-negative values. # Reject negative values. if value < 0: raise EnumDefinitionError( 'Must use non-negative enum values. Found: %s = %d' % (attribute, value)) if value > MAX_ENUM_VALUE: raise EnumDefinitionError( 'Must use enum values less than or equal %d. ' 'Found: %s = %d' % (MAX_ENUM_VALUE, attribute, value)) if value in cls.__by_number: raise EnumDefinitionError( 'Value for %s = %d is already defined: %s' % (attribute, value, cls.__by_number[value].name)) # Create enum instance and list in new Enum type. instance = object.__new__(cls) # pylint:disable=non-parent-init-called cls.__init__(instance, attribute, value) cls.__by_name[instance.name] = instance cls.__by_number[instance.number] = instance setattr(cls, attribute, instance) _DefinitionClass.__init__(cls, name, bases, dct) def __iter__(cls): """Iterate over all values of enum. Yields: Enumeration instances of the Enum class in arbitrary order. """ return iter(cls.__by_number.values()) def names(cls): """Get all names for Enum. Returns: An iterator for names of the enumeration in arbitrary order. """ return cls.__by_name.keys() def numbers(cls): """Get all numbers for Enum. Returns: An iterator for all numbers of the enumeration in arbitrary order. """ return cls.__by_number.keys() def lookup_by_name(cls, name): """Look up Enum by name. Args: name: Name of enum to find. Returns: Enum sub-class instance of that value. """ return cls.__by_name[name] def lookup_by_number(cls, number): """Look up Enum by number. Args: number: Number of enum to find. Returns: Enum sub-class instance of that value. """ return cls.__by_number[number] def __len__(cls): return len(cls.__by_name) class Enum(six.with_metaclass(_EnumClass, object)): """Base class for all enumerated types.""" __slots__ = set(('name', 'number')) def __new__(cls, index): """Acts as look-up routine after class is initialized. The purpose of overriding __new__ is to provide a way to treat Enum subclasses as casting types, similar to how the int type functions. A program can pass a string or an integer and this method with "convert" that value in to an appropriate Enum instance. Args: index: Name or number to look up. During initialization this is always the name of the new enum value. Raises: TypeError: When an inappropriate index value is passed provided. """ # If is enum type of this class, return it. if isinstance(index, cls): return index # If number, look up by number. if isinstance(index, six.integer_types): try: return cls.lookup_by_number(index) except KeyError: pass # If name, look up by name. if isinstance(index, six.string_types): try: return cls.lookup_by_name(index) except KeyError: pass raise TypeError('No such value for %s in Enum %s' % (index, cls.__name__)) def __init__(self, name, number=None): """Initialize new Enum instance. Since this should only be called during class initialization any calls that happen after the class is frozen raises an exception. """ # Immediately return if __init__ was called after _Enum.__init__(). # It means that casting operator version of the class constructor # is being used. if getattr(type(self), '_DefinitionClass__initialized'): return object.__setattr__(self, 'name', name) object.__setattr__(self, 'number', number) def __setattr__(self, name, value): raise TypeError('May not change enum values') def __str__(self): return self.name def __int__(self): return self.number def __repr__(self): return '%s(%s, %d)' % (type(self).__name__, self.name, self.number) def __reduce__(self): """Enable pickling. Returns: A 2-tuple containing the class and __new__ args to be used for restoring a pickled instance. """ return self.__class__, (self.number,) def __cmp__(self, other): """Order is by number.""" if isinstance(other, type(self)): return cmp(self.number, other.number) return NotImplemented def __lt__(self, other): """Order is by number.""" if isinstance(other, type(self)): return self.number < other.number return NotImplemented def __le__(self, other): """Order is by number.""" if isinstance(other, type(self)): return self.number <= other.number return NotImplemented def __eq__(self, other): """Order is by number.""" if isinstance(other, type(self)): return self.number == other.number return NotImplemented def __ne__(self, other): """Order is by number.""" if isinstance(other, type(self)): return self.number != other.number return NotImplemented def __ge__(self, other): """Order is by number.""" if isinstance(other, type(self)): return self.number >= other.number return NotImplemented def __gt__(self, other): """Order is by number.""" if isinstance(other, type(self)): return self.number > other.number return NotImplemented def __hash__(self): """Hash by number.""" return hash(self.number) @classmethod def to_dict(cls): """Make dictionary version of enumerated class. Dictionary created this way can be used with def_num. Returns: A dict (name) -> number """ return dict((item.name, item.number) for item in iter(cls)) @staticmethod def def_enum(dct, name): """Define enum class from dictionary. Args: dct: Dictionary of enumerated values for type. name: Name of enum. """ return type(name, (Enum,), dct) # TODO(rafek): Determine to what degree this enumeration should be compatible # with FieldDescriptor.Type in https://github.com/google/protobuf. class Variant(Enum): """Wire format variant. Used by the 'protobuf' wire format to determine how to transmit a single piece of data. May be used by other formats. See: http://code.google.com/apis/protocolbuffers/docs/encoding.html Values: DOUBLE: 64-bit floating point number. FLOAT: 32-bit floating point number. INT64: 64-bit signed integer. UINT64: 64-bit unsigned integer. INT32: 32-bit signed integer. BOOL: Boolean value (True or False). STRING: String of UTF-8 encoded text. MESSAGE: Embedded message as byte string. BYTES: String of 8-bit bytes. UINT32: 32-bit unsigned integer. ENUM: Enum value as integer. SINT32: 32-bit signed integer. Uses "zig-zag" encoding. SINT64: 64-bit signed integer. Uses "zig-zag" encoding. """ DOUBLE = 1 FLOAT = 2 INT64 = 3 UINT64 = 4 INT32 = 5 BOOL = 8 STRING = 9 MESSAGE = 11 BYTES = 12 UINT32 = 13 ENUM = 14 SINT32 = 17 SINT64 = 18 class _MessageClass(_DefinitionClass): """Meta-class used for defining the Message base class. For more details about Message classes, see the Message class docstring. Information contained there may help understanding this class. Meta-class enables very specific behavior for any defined Message class. All attributes defined on an Message sub-class must be field instances, Enum class definitions or other Message class definitions. Each field attribute defined on an Message sub-class is added to the set of field definitions and the attribute is translated in to a slot. It also ensures that only one level of Message class hierarchy is possible. In other words it is not possible to declare sub-classes of sub-classes of Message. This class also defines some functions in order to restrict the behavior of the Message class and its sub-classes. It is not possible to change the behavior of the Message class in later classes since any new classes may be defined with only field, Enums and Messages, and no methods. """ # pylint:disable=bad-mcs-classmethod-argument def __new__(cls, name, bases, dct): """Create new Message class instance. The __new__ method of the _MessageClass type is overridden so as to allow the translation of Field instances to slots. """ by_number = {} by_name = {} variant_map = {} # pylint:disable=unused-variable if bases != (object,): # Can only define one level of sub-classes below Message. if bases != (Message,): raise MessageDefinitionError( 'Message types may only inherit from Message') enums = [] messages = [] # Must not use iteritems because this loop will change the state of # dct. for key, field in dct.items(): if key in _RESERVED_ATTRIBUTE_NAMES: continue if isinstance(field, type) and issubclass(field, Enum): enums.append(key) continue if (isinstance(field, type) and issubclass(field, Message) and field is not Message): messages.append(key) continue # Reject anything that is not a field. # pylint:disable=unidiomatic-typecheck if type(field) is Field or not isinstance(field, Field): raise MessageDefinitionError( 'May only use fields in message definitions. ' 'Found: %s = %s' % (key, field)) if field.number in by_number: raise DuplicateNumberError( 'Field with number %d declared more than once in %s' % (field.number, name)) field.name = key # Place in name and number maps. by_name[key] = field by_number[field.number] = field # Add enums if any exist. if enums: dct['__enums__'] = sorted(enums) # Add messages if any exist. if messages: dct['__messages__'] = sorted(messages) dct['_Message__by_number'] = by_number dct['_Message__by_name'] = by_name return _DefinitionClass.__new__(cls, name, bases, dct) def __init__(cls, name, bases, dct): """Initializer required to assign references to new class.""" if bases != (object,): for v in dct.values(): if isinstance(v, _DefinitionClass) and v is not Message: v._message_definition = weakref.ref(cls) for field in cls.all_fields(): field._message_definition = weakref.ref(cls) _DefinitionClass.__init__(cls, name, bases, dct) class Message(six.with_metaclass(_MessageClass, object)): """Base class for user defined message objects. Used to define messages for efficient transmission across network or process space. Messages are defined using the field classes (IntegerField, FloatField, EnumField, etc.). Messages are more restricted than normal classes in that they may only contain field attributes and other Message and Enum definitions. These restrictions are in place because the structure of the Message class is intentended to itself be transmitted across network or process space and used directly by clients or even other servers. As such methods and non-field attributes could not be transmitted with the structural information causing discrepancies between different languages and implementations. Initialization and validation: A Message object is considered to be initialized if it has all required fields and any nested messages are also initialized. Calling 'check_initialized' will raise a ValidationException if it is not initialized; 'is_initialized' returns a boolean value indicating if it is valid. Validation automatically occurs when Message objects are created and populated. Validation that a given value will be compatible with a field that it is assigned to can be done through the Field instances validate() method. The validate method used on a message will check that all values of a message and its sub-messages are valid. Assingning an invalid value to a field will raise a ValidationException. Example: # Trade type. class TradeType(Enum): BUY = 1 SELL = 2 SHORT = 3 CALL = 4 class Lot(Message): price = IntegerField(1, required=True) quantity = IntegerField(2, required=True) class Order(Message): symbol = StringField(1, required=True) total_quantity = IntegerField(2, required=True) trade_type = EnumField(TradeType, 3, required=True) lots = MessageField(Lot, 4, repeated=True) limit = IntegerField(5) order = Order(symbol='GOOG', total_quantity=10, trade_type=TradeType.BUY) lot1 = Lot(price=304, quantity=7) lot2 = Lot(price = 305, quantity=3) order.lots = [lot1, lot2] # Now object is initialized! order.check_initialized() """ def __init__(self, kwargs): """Initialize internal messages state. Args: A message can be initialized via the constructor by passing in keyword arguments corresponding to fields. For example: class Date(Message): day = IntegerField(1) month = IntegerField(2) year = IntegerField(3) Invoking: date = Date(day=6, month=6, year=1911) is the same as doing: date = Date() date.day = 6 date.month = 6 date.year = 1911 """ # Tag being an essential implementation detail must be private. self.__tags = {} self.__unrecognized_fields = {} assigned = set() for name, value in kwargs.items(): setattr(self, name, value) assigned.add(name) # initialize repeated fields. for field in self.all_fields(): if field.repeated and field.name not in assigned: setattr(self, field.name, []) def check_initialized(self): """Check class for initialization status. Check that all required fields are initialized Raises: ValidationError: If message is not initialized. """ for name, field in self.__by_name.items(): value = getattr(self, name) if value is None: if field.required: raise ValidationError( "Message %s is missing required field %s" % (type(self).__name__, name)) else: try: if (isinstance(field, MessageField) and issubclass(field.message_type, Message)): if field.repeated: for item in value: item_message_value = field.value_to_message( item) item_message_value.check_initialized() else: message_value = field.value_to_message(value) message_value.check_initialized() except ValidationError as err: if not hasattr(err, 'message_name'): err.message_name = type(self).__name__ raise def is_initialized(self): """Get initialization status. Returns: True if message is valid, else False. """ try: self.check_initialized() except ValidationError: return False else: return True @classmethod def all_fields(cls): """Get all field definition objects. Ordering is arbitrary. Returns: Iterator over all values in arbitrary order. """ return cls.__by_name.values() @classmethod def field_by_name(cls, name): """Get field by name. Returns: Field object associated with name. Raises: KeyError if no field found by that name. """ return cls.__by_name[name] @classmethod def field_by_number(cls, number): """Get field by number. Returns: Field object associated with number. Raises: KeyError if no field found by that number. """ return cls.__by_number[number] def get_assigned_value(self, name): """Get the assigned value of an attribute. Get the underlying value of an attribute. If value has not been set, will not return the default for the field. Args: name: Name of attribute to get. Returns: Value of attribute, None if it has not been set. """ message_type = type(self) try: field = message_type.field_by_name(name) except KeyError: raise AttributeError('Message %s has no field %s' % ( message_type.__name__, name)) return self.__tags.get(field.number) def reset(self, name): """Reset assigned value for field. Resetting a field will return it to its default value or None. Args: name: Name of field to reset. """ message_type = type(self) try: field = message_type.field_by_name(name) except KeyError: if name not in message_type.__by_name: raise AttributeError('Message %s has no field %s' % ( message_type.__name__, name)) if field.repeated: self.__tags[field.number] = FieldList(field, []) else: self.__tags.pop(field.number, None) def all_unrecognized_fields(self): """Get the names of all unrecognized fields in this message.""" return list(self.__unrecognized_fields.keys()) def get_unrecognized_field_info(self, key, value_default=None, variant_default=None): """Get the value and variant of an unknown field in this message. Args: key: The name or number of the field to retrieve. value_default: Value to be returned if the key isn't found. variant_default: Value to be returned as variant if the key isn't found. Returns: (value, variant), where value and variant are whatever was passed to set_unrecognized_field. """ value, variant = self.__unrecognized_fields.get(key, (value_default, variant_default)) return value, variant def set_unrecognized_field(self, key, value, variant): """Set an unrecognized field, used when decoding a message. Args: key: The name or number used to refer to this unknown value. value: The value of the field. variant: Type information needed to interpret the value or re-encode it. Raises: TypeError: If the variant is not an instance of messages.Variant. """ if not isinstance(variant, Variant): raise TypeError('Variant type %s is not valid.' % variant) self.__unrecognized_fields[key] = value, variant def __setattr__(self, name, value): """Change set behavior for messages. Messages may only be assigned values that are fields. Does not try to validate field when set. Args: name: Name of field to assign to. value: Value to assign to field. Raises: AttributeError when trying to assign value that is not a field. """ if name in self.__by_name or name.startswith('_Message__'): object.__setattr__(self, name, value) else: raise AttributeError("May not assign arbitrary value %s " "to message %s" % (name, type(self).__name__)) def __repr__(self): """Make string representation of message. Example: class MyMessage(messages.Message): integer_value = messages.IntegerField(1) string_value = messages.StringField(2) my_message = MyMessage() my_message.integer_value = 42 my_message.string_value = u'A string' print my_message >>> <MyMessage ... integer_value: 42 ... string_value: u'A string'> Returns: String representation of message, including the values of all fields and repr of all sub-messages. """ body = ['<', type(self).__name__] for field in sorted(self.all_fields(), key=lambda f: f.number): attribute = field.name value = self.get_assigned_value(field.name) if value is not None: body.append('\n %s: %s' % (attribute, repr(value))) body.append('>') return ''.join(body) def __eq__(self, other): """Equality operator. Does field by field comparison with other message. For equality, must be same type and values of all fields must be equal. Messages not required to be initialized for comparison. Does not attempt to determine equality for values that have default values that are not set. In other words: class HasDefault(Message): attr1 = StringField(1, default='default value') message1 = HasDefault() message2 = HasDefault() message2.attr1 = 'default value' message1 != message2 Does not compare unknown values. Args: other: Other message to compare with. """ # TODO(rafek): Implement "equivalent" which does comparisons # taking default values in to consideration. if self is other: return True if type(self) is not type(other): return False return self.__tags == other.__tags def __ne__(self, other): """Not equals operator. Does field by field comparison with other message. For non-equality, must be different type or any value of a field must be non-equal to the same field in the other instance. Messages not required to be initialized for comparison. Args: other: Other message to compare with. """ return not self.__eq__(other) class FieldList(list): """List implementation that validates field values. This list implementation overrides all methods that add values in to a list in order to validate those new elements. Attempting to add or set list values that are not of the correct type will raise ValidationError. """ def __init__(self, field_instance, sequence): """Constructor. Args: field_instance: Instance of field that validates the list. sequence: List or tuple to construct list from. """ if not field_instance.repeated: raise FieldDefinitionError( 'FieldList may only accept repeated fields') self.__field = field_instance self.__field.validate(sequence) list.__init__(self, sequence) def __getstate__(self): """Enable pickling. The assigned field instance can't be pickled if it belongs to a Message definition (message_definition uses a weakref), so the Message class and field number are returned in that case. Returns: A 3-tuple containing: - The field instance, or None if it belongs to a Message class. - The Message class that the field instance belongs to, or None. - The field instance number of the Message class it belongs to, or None. """ message_class = self.__field.message_definition() if message_class is None: return self.__field, None, None return None, message_class, self.__field.number def __setstate__(self, state): """Enable unpickling. Args: state: A 3-tuple containing: - The field instance, or None if it belongs to a Message class. - The Message class that the field instance belongs to, or None. - The field instance number of the Message class it belongs to, or None. """ field_instance, message_class, number = state if field_instance is None: self.__field = message_class.field_by_number(number) else: self.__field = field_instance @property def field(self): """Field that validates list.""" return self.__field def __setslice__(self, i, j, sequence): """Validate slice assignment to list.""" self.__field.validate(sequence) list.__setslice__(self, i, j, sequence) def __setitem__(self, index, value): """Validate item assignment to list.""" if isinstance(index, slice): self.__field.validate(value) else: self.__field.validate_element(value) list.__setitem__(self, index, value) def append(self, value): """Validate item appending to list.""" if getattr(self, '_FieldList__field', None): self.__field.validate_element(value) return list.append(self, value) def extend(self, sequence): """Validate extension of list.""" if getattr(self, '_FieldList__field', None): self.__field.validate(sequence) return list.extend(self, sequence) def insert(self, index, value): """Validate item insertion to list.""" self.__field.validate_element(value) return list.insert(self, index, value) class _FieldMeta(type): def __init__(cls, name, bases, dct): getattr(cls, '_Field__variant_to_type').update( (variant, cls) for variant in dct.get('VARIANTS', [])) type.__init__(cls, name, bases, dct) # TODO(rafek): Prevent additional field subclasses. class Field(six.with_metaclass(_FieldMeta, object)): """Definition for message field.""" __initialized = False # pylint:disable=invalid-name __variant_to_type = {} # pylint:disable=invalid-name @util.positional(2) def __init__(self, number, required=False, repeated=False, variant=None, default=None): """Constructor. The required and repeated parameters are mutually exclusive. Setting both to True will raise a FieldDefinitionError. Sub-class Attributes: Each sub-class of Field must define the following: VARIANTS: Set of variant types accepted by that field. DEFAULT_VARIANT: Default variant type if not specified in constructor. Args: number: Number of field. Must be unique per message class. required: Whether or not field is required. Mutually exclusive with 'repeated'. repeated: Whether or not field is repeated. Mutually exclusive with 'required'. variant: Wire-format variant hint. default: Default value for field if not found in stream. Raises: InvalidVariantError when invalid variant for field is provided. InvalidDefaultError when invalid default for field is provided. FieldDefinitionError when invalid number provided or mutually exclusive fields are used. InvalidNumberError when the field number is out of range or reserved. """ if not isinstance(number, int) or not 1 <= number <= MAX_FIELD_NUMBER: raise InvalidNumberError( 'Invalid number for field: %s\n' 'Number must be 1 or greater and %d or less' % (number, MAX_FIELD_NUMBER)) if FIRST_RESERVED_FIELD_NUMBER <= number <= LAST_RESERVED_FIELD_NUMBER: raise InvalidNumberError('Tag number %d is a reserved number.\n' 'Numbers %d to %d are reserved' % (number, FIRST_RESERVED_FIELD_NUMBER, LAST_RESERVED_FIELD_NUMBER)) if repeated and required: raise FieldDefinitionError('Cannot set both repeated and required') if variant is None: variant = self.DEFAULT_VARIANT if repeated and default is not None: raise FieldDefinitionError('Repeated fields may not have defaults') if variant not in self.VARIANTS: raise InvalidVariantError( 'Invalid variant: %s\nValid variants for %s are %r' % (variant, type(self).__name__, sorted(self.VARIANTS))) self.number = number self.required = required self.repeated = repeated self.variant = variant if default is not None: try: self.validate_default(default) except ValidationError as err: try: name = self.name except AttributeError: # For when raising error before name initialization. raise InvalidDefaultError( 'Invalid default value for %s: %r: %s' % (self.__class__.__name__, default, err)) else: raise InvalidDefaultError( 'Invalid default value for field %s: ' '%r: %s' % (name, default, err)) self.__default = default self.__initialized = True def __setattr__(self, name, value): """Setter overidden to prevent assignment to fields after creation. Args: name: Name of attribute to set. value: Value to assign. """ # Special case post-init names. They need to be set after constructor. if name in _POST_INIT_FIELD_ATTRIBUTE_NAMES: object.__setattr__(self, name, value) return # All other attributes must be set before __initialized. if not self.__initialized: # Not initialized yet, allow assignment. object.__setattr__(self, name, value) else: raise AttributeError('Field objects are read-only') def __set__(self, message_instance, value): """Set value on message. Args: message_instance: Message instance to set value on. value: Value to set on message. """ # Reaches in to message instance directly to assign to private tags. if value is None: if self.repeated: raise ValidationError( 'May not assign None to repeated field %s' % self.name) else: message_instance._Message__tags.pop(self.number, None) else: if self.repeated: value = FieldList(self, value) else: value = self.validate(value) message_instance._Message__tags[self.number] = value def __get__(self, message_instance, message_class): if message_instance is None: return self result = message_instance._Message__tags.get(self.number) if result is None: return self.default return result def validate_element(self, value): """Validate single element of field. This is different from validate in that it is used on individual values of repeated fields. Args: value: Value to validate. Returns: The value casted in the expected type. Raises: ValidationError if value is not expected type. """ if not isinstance(value, self.type): # Authorize int values as float. if isinstance(value, six.integer_types) and self.type == float: return float(value) if value is None: if self.required: raise ValidationError('Required field is missing') else: try: name = self.name except AttributeError: raise ValidationError('Expected type %s for %s, ' 'found %s (type %s)' % (self.type, self.__class__.__name__, value, type(value))) else: raise ValidationError( 'Expected type %s for field %s, found %s (type %s)' % (self.type, name, value, type(value))) return value def __validate(self, value, validate_element): """Internal validation function. Validate an internal value using a function to validate individual elements. Args: value: Value to validate. validate_element: Function to use to validate individual elements. Raises: ValidationError if value is not expected type. """ if not self.repeated: return validate_element(value) else: # Must be a list or tuple, may not be a string. if isinstance(value, (list, tuple)): result = [] for element in value: if element is None: try: name = self.name except AttributeError: raise ValidationError( 'Repeated values for %s ' 'may not be None' % self.__class__.__name__) else: raise ValidationError( 'Repeated values for field %s ' 'may not be None' % name) result.append(validate_element(element)) return result elif value is not None: try: name = self.name except AttributeError: raise ValidationError('%s is repeated. Found: %s' % ( self.__class__.__name__, value)) else: raise ValidationError( 'Field %s is repeated. Found: %s' % (name, value)) return value def validate(self, value): """Validate value assigned to field. Args: value: Value to validate. Returns: the value in casted in the correct type. Raises: ValidationError if value is not expected type. """ return self.__validate(value, self.validate_element) def validate_default_element(self, value): """Validate value as assigned to field default field. Some fields may allow for delayed resolution of default types necessary in the case of circular definition references. In this case, the default value might be a place holder that is resolved when needed after all the message classes are defined. Args: value: Default value to validate. Returns: the value in casted in the correct type. Raises: ValidationError if value is not expected type. """ return self.validate_element(value) def validate_default(self, value): """Validate default value assigned to field. Args: value: Value to validate. Returns: the value in casted in the correct type. Raises: ValidationError if value is not expected type. """ return self.__validate(value, self.validate_default_element) def message_definition(self): """Get Message definition that contains this Field definition. Returns: Containing Message definition for Field. Will return None if for some reason Field is defined outside of a Message class. """ try: return self._message_definition() except AttributeError: return None @property def default(self): """Get default value for field.""" return self.__default @classmethod def lookup_field_type_by_variant(cls, variant): return cls.__variant_to_type[variant] class IntegerField(Field): """Field definition for integer values.""" VARIANTS = frozenset([ Variant.INT32, Variant.INT64, Variant.UINT32, Variant.UINT64, Variant.SINT32, Variant.SINT64, ]) DEFAULT_VARIANT = Variant.INT64 type = six.integer_types class FloatField(Field): """Field definition for float values.""" VARIANTS = frozenset([ Variant.FLOAT, Variant.DOUBLE, ]) DEFAULT_VARIANT = Variant.DOUBLE type = float class BooleanField(Field): """Field definition for boolean values.""" VARIANTS = frozenset([Variant.BOOL]) DEFAULT_VARIANT = Variant.BOOL type = bool class BytesField(Field): """Field definition for byte string values.""" VARIANTS = frozenset([Variant.BYTES]) DEFAULT_VARIANT = Variant.BYTES type = bytes class StringField(Field): """Field definition for unicode string values.""" VARIANTS = frozenset([Variant.STRING]) DEFAULT_VARIANT = Variant.STRING type = six.text_type def validate_element(self, value): """Validate StringField allowing for str and unicode. Raises: ValidationError if a str value is not UTF-8. """ # If value is str is it considered valid. Satisfies "required=True". if isinstance(value, bytes): try: six.text_type(value, 'UTF-8') except UnicodeDecodeError as err: try: _ = self.name except AttributeError: validation_error = ValidationError( 'Field encountered non-UTF-8 string %r: %s' % (value, err)) else: validation_error = ValidationError( 'Field %s encountered non-UTF-8 string %r: %s' % ( self.name, value, err)) validation_error.field_name = self.name raise validation_error else: return super(StringField, self).validate_element(value) return value class MessageField(Field): """Field definition for sub-message values. Message fields contain instance of other messages. Instances stored on messages stored on message fields are considered to be owned by the containing message instance and should not be shared between owning instances. Message fields must be defined to reference a single type of message. Normally message field are defined by passing the referenced message class in to the constructor. It is possible to define a message field for a type that does not yet exist by passing the name of the message in to the constructor instead of a message class. Resolution of the actual type of the message is deferred until it is needed, for example, during message verification. Names provided to the constructor must refer to a class within the same python module as the class that is using it. Names refer to messages relative to the containing messages scope. For example, the two fields of OuterMessage refer to the same message type: class Outer(Message): inner_relative = MessageField('Inner', 1) inner_absolute = MessageField('Outer.Inner', 2) class Inner(Message): ... When resolving an actual type, MessageField will traverse the entire scope of nested messages to match a message name. This makes it easy for siblings to reference siblings: class Outer(Message): class Inner(Message): sibling = MessageField('Sibling', 1) class Sibling(Message): ... """ VARIANTS = frozenset([Variant.MESSAGE]) DEFAULT_VARIANT = Variant.MESSAGE @util.positional(3) def __init__(self, message_type, number, required=False, repeated=False, variant=None): """Constructor. Args: message_type: Message type for field. Must be subclass of Message. number: Number of field. Must be unique per message class. required: Whether or not field is required. Mutually exclusive to 'repeated'. repeated: Whether or not field is repeated. Mutually exclusive to 'required'. variant: Wire-format variant hint. Raises: FieldDefinitionError when invalid message_type is provided. """ valid_type = (isinstance(message_type, six.string_types) or (message_type is not Message and isinstance(message_type, type) and issubclass(message_type, Message))) if not valid_type: raise FieldDefinitionError( 'Invalid message class: %s' % message_type) if isinstance(message_type, six.string_types): self.__type_name = message_type self.__type = None else: self.__type = message_type super(MessageField, self).__init__(number, required=required, repeated=repeated, variant=variant) def __set__(self, message_instance, value): """Set value on message. Args: message_instance: Message instance to set value on. value: Value to set on message. """ t = self.type if isinstance(t, type) and issubclass(t, Message): if self.repeated: if value and isinstance(value, (list, tuple)): value = [(t(v) if isinstance(v, dict) else v) for v in value] elif isinstance(value, dict): value = t(value) super(MessageField, self).__set__(message_instance, value) @property def type(self): """Message type used for field.""" if self.__type is None: message_type = find_definition( self.__type_name, self.message_definition()) if not (message_type is not Message and isinstance(message_type, type) and issubclass(message_type, Message)): raise FieldDefinitionError( 'Invalid message class: %s' % message_type) self.__type = message_type return self.__type @property def message_type(self): """Underlying message type used for serialization. Will always be a sub-class of Message. This is different from type which represents the python value that message_type is mapped to for use by the user. """ return self.type def value_from_message(self, message): """Convert a message to a value instance. Used by deserializers to convert from underlying messages to value of expected user type. Args: message: A message instance of type self.message_type. Returns: Value of self.message_type. """ if not isinstance(message, self.message_type): raise DecodeError('Expected type %s, got %s: %r' % (self.message_type.__name__, type(message).__name__, message)) return message def value_to_message(self, value): """Convert a value instance to a message. Used by serializers to convert Python user types to underlying messages for transmission. Args: value: A value of type self.type. Returns: An instance of type self.message_type. """ if not isinstance(value, self.type): raise EncodeError('Expected type %s, got %s: %r' % (self.type.__name__, type(value).__name__, value)) return value class EnumField(Field): """Field definition for enum values. Enum fields may have default values that are delayed until the associated enum type is resolved. This is necessary to support certain circular references. For example: class Message1(Message): class Color(Enum): RED = 1 GREEN = 2 BLUE = 3 # This field default value will be validated when default is accessed. animal = EnumField('Message2.Animal', 1, default='HORSE') class Message2(Message): class Animal(Enum): DOG = 1 CAT = 2 HORSE = 3 # This fields default value will be validated right away since Color # is already fully resolved. color = EnumField(Message1.Color, 1, default='RED') """ VARIANTS = frozenset([Variant.ENUM]) DEFAULT_VARIANT = Variant.ENUM def __init__(self, enum_type, number, kwargs): """Constructor. Args: enum_type: Enum type for field. Must be subclass of Enum. number: Number of field. Must be unique per message class. required: Whether or not field is required. Mutually exclusive to 'repeated'. repeated: Whether or not field is repeated. Mutually exclusive to 'required'. variant: Wire-format variant hint. default: Default value for field if not found in stream. Raises: FieldDefinitionError when invalid enum_type is provided. """ valid_type = (isinstance(enum_type, six.string_types) or (enum_type is not Enum and isinstance(enum_type, type) and issubclass(enum_type, Enum))) if not valid_type: raise FieldDefinitionError('Invalid enum type: %s' % enum_type) if isinstance(enum_type, six.string_types): self.__type_name = enum_type self.__type = None else: self.__type = enum_type super(EnumField, self).__init__(number, **kwargs) def validate_default_element(self, value): """Validate default element of Enum field. Enum fields allow for delayed resolution of default values when the type of the field has not been resolved. The default value of a field may be a string or an integer. If the Enum type of the field has been resolved, the default value is validated against that type. Args: value: Value to validate. Raises: ValidationError if value is not expected message type. """ if isinstance(value, (six.string_types, six.integer_types)): # Validation of the value does not happen for delayed resolution # enumerated types. Ignore if type is not yet resolved. if self.__type: self.__type(value) return value return super(EnumField, self).validate_default_element(value) @property def type(self): """Enum type used for field.""" if self.__type is None: found_type = find_definition( self.__type_name, self.message_definition()) if not (found_type is not Enum and isinstance(found_type, type) and issubclass(found_type, Enum)): raise FieldDefinitionError( 'Invalid enum type: %s' % found_type) self.__type = found_type return self.__type @property def default(self): """Default for enum field. Will cause resolution of Enum type and unresolved default value. """ try: return self.__resolved_default except AttributeError: resolved_default = super(EnumField, self).default if isinstance(resolved_default, (six.string_types, six.integer_types)): # pylint:disable=not-callable resolved_default = self.type(resolved_default) self.__resolved_default = resolved_default return self.__resolved_default @util.positional(2) def find_definition(name, relative_to=None, importer=__import__): """Find definition by name in module-space. The find algorthm will look for definitions by name relative to a message definition or by fully qualfied name. If no definition is found relative to the relative_to parameter it will do the same search against the container of relative_to. If relative_to is a nested Message, it will search its message_definition(). If that message has no message_definition() it will search its module. If relative_to is a module, it will attempt to look for the containing module and search relative to it. If the module is a top-level module, it will look for the a message using a fully qualified name. If no message is found then, the search fails and DefinitionNotFoundError is raised. For example, when looking for any definition 'foo.bar.ADefinition' relative to an actual message definition abc.xyz.SomeMessage: find_definition('foo.bar.ADefinition', SomeMessage) It is like looking for the following fully qualified names: abc.xyz.SomeMessage. foo.bar.ADefinition abc.xyz. foo.bar.ADefinition abc. foo.bar.ADefinition foo.bar.ADefinition When resolving the name relative to Message definitions and modules, the algorithm searches any Messages or sub-modules found in its path. Non-Message values are not searched. A name that begins with '.' is considered to be a fully qualified name. The name is always searched for from the topmost package. For example, assume two message types: abc.xyz.SomeMessage xyz.SomeMessage Searching for '.xyz.SomeMessage' relative to 'abc' will resolve to 'xyz.SomeMessage' and not 'abc.xyz.SomeMessage'. For this kind of name, the relative_to parameter is effectively ignored and always set to None. For more information about package name resolution, please see: http://code.google.com/apis/protocolbuffers/docs/proto.html#packages Args: name: Name of definition to find. May be fully qualified or relative name. relative_to: Search for definition relative to message definition or module. None will cause a fully qualified name search. importer: Import function to use for resolving modules. Returns: Enum or Message class definition associated with name. Raises: DefinitionNotFoundError if no definition is found in any search path. """ # Check parameters. if not (relative_to is None or isinstance(relative_to, types.ModuleType) or isinstance(relative_to, type) and issubclass(relative_to, Message)): raise TypeError( 'relative_to must be None, Message definition or module.' ' Found: %s' % relative_to) name_path = name.split('.') # Handle absolute path reference. if not name_path[0]: relative_to = None name_path = name_path[1:] def search_path(): """Performs a single iteration searching the path from relative_to. This is the function that searches up the path from a relative object. fully.qualified.object . relative.or.nested.Definition ----------------------------> ^ \| this part of search --+ Returns: Message or Enum at the end of name_path, else None. """ next_part = relative_to for node in name_path: # Look for attribute first. attribute = getattr(next_part, node, None) if attribute is not None: next_part = attribute else: # If module, look for sub-module. if (next_part is None or isinstance(next_part, types.ModuleType)): if next_part is None: module_name = node else: module_name = '%s.%s' % (next_part.__name__, node) try: fromitem = module_name.split('.')[-1] next_part = importer(module_name, '', '', [str(fromitem)]) except ImportError: return None else: return None if not isinstance(next_part, types.ModuleType): if not (isinstance(next_part, type) and issubclass(next_part, (Message, Enum))): return None return next_part while True: found = search_path() if isinstance(found, type) and issubclass(found, (Enum, Message)): return found else: # Find next relative_to to search against. # # fully.qualified.object . relative.or.nested.Definition # <--------------------- # ^ # \| # does this part of search if relative_to is None: # Fully qualified search was done. Nothing found. Fail. raise DefinitionNotFoundError( 'Could not find definition for %s' % name) else: if isinstance(relative_to, types.ModuleType): # Find parent module. module_path = relative_to.__name__.split('.')[:-1] if not module_path: relative_to = None else: # Should not raise ImportError. If it does... # weird and unexpected. Propagate. relative_to = importer( '.'.join(module_path), '', '', [module_path[-1]]) elif (isinstance(relative_to, type) and issubclass(relative_to, Message)): parent = relative_to.message_definition() if parent is None: last_module_name = relative_to.__module__.split( '.')[-1] relative_to = importer( relative_to.__module__, '', '', [last_module_name]) else: relative_to = parent	66.13 KB	2025-03-11 18:02
📄 messages_test.py #!/usr/bin/env python # # Copyright 2010 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # """Tests for apitools.base.protorpclite.messages.""" import pickle import re import sys import types import unittest import six from apitools.base.protorpclite import descriptor from apitools.base.protorpclite import message_types from apitools.base.protorpclite import messages from apitools.base.protorpclite import test_util # This package plays lots of games with modifying global variables inside # test cases. Hence: # pylint:disable=function-redefined # pylint:disable=global-variable-not-assigned # pylint:disable=global-variable-undefined # pylint:disable=redefined-outer-name # pylint:disable=undefined-variable # pylint:disable=unused-variable # pylint:disable=too-many-lines try: long # Python 2 except NameError: long = int # Python 3 class ModuleInterfaceTest(test_util.ModuleInterfaceTest, test_util.TestCase): MODULE = messages class ValidationErrorTest(test_util.TestCase): def testStr_NoFieldName(self): """Test string version of ValidationError when no name provided.""" self.assertEqual('Validation error', str(messages.ValidationError('Validation error'))) def testStr_FieldName(self): """Test string version of ValidationError when no name provided.""" validation_error = messages.ValidationError('Validation error') validation_error.field_name = 'a_field' self.assertEqual('Validation error', str(validation_error)) class EnumTest(test_util.TestCase): def setUp(self): """Set up tests.""" # Redefine Color class in case so that changes to it (an # error) in one test does not affect other tests. global Color # pylint:disable=global-variable-not-assigned # pylint:disable=unused-variable class Color(messages.Enum): RED = 20 ORANGE = 2 YELLOW = 40 GREEN = 4 BLUE = 50 INDIGO = 5 VIOLET = 80 def testNames(self): """Test that names iterates over enum names.""" self.assertEqual( set(['BLUE', 'GREEN', 'INDIGO', 'ORANGE', 'RED', 'VIOLET', 'YELLOW']), set(Color.names())) def testNumbers(self): """Tests that numbers iterates of enum numbers.""" self.assertEqual(set([2, 4, 5, 20, 40, 50, 80]), set(Color.numbers())) def testIterate(self): """Test that __iter__ iterates over all enum values.""" self.assertEqual(set(Color), set([Color.RED, Color.ORANGE, Color.YELLOW, Color.GREEN, Color.BLUE, Color.INDIGO, Color.VIOLET])) def testNaturalOrder(self): """Test that natural order enumeration is in numeric order.""" self.assertEqual([Color.ORANGE, Color.GREEN, Color.INDIGO, Color.RED, Color.YELLOW, Color.BLUE, Color.VIOLET], sorted(Color)) def testByName(self): """Test look-up by name.""" self.assertEqual(Color.RED, Color.lookup_by_name('RED')) self.assertRaises(KeyError, Color.lookup_by_name, 20) self.assertRaises(KeyError, Color.lookup_by_name, Color.RED) def testByNumber(self): """Test look-up by number.""" self.assertRaises(KeyError, Color.lookup_by_number, 'RED') self.assertEqual(Color.RED, Color.lookup_by_number(20)) self.assertRaises(KeyError, Color.lookup_by_number, Color.RED) def testConstructor(self): """Test that constructor look-up by name or number.""" self.assertEqual(Color.RED, Color('RED')) self.assertEqual(Color.RED, Color(u'RED')) self.assertEqual(Color.RED, Color(20)) if six.PY2: self.assertEqual(Color.RED, Color(long(20))) self.assertEqual(Color.RED, Color(Color.RED)) self.assertRaises(TypeError, Color, 'Not exists') self.assertRaises(TypeError, Color, 'Red') self.assertRaises(TypeError, Color, 100) self.assertRaises(TypeError, Color, 10.0) def testLen(self): """Test that len function works to count enums.""" self.assertEqual(7, len(Color)) def testNoSubclasses(self): """Test that it is not possible to sub-class enum classes.""" def declare_subclass(): class MoreColor(Color): pass self.assertRaises(messages.EnumDefinitionError, declare_subclass) def testClassNotMutable(self): """Test that enum classes themselves are not mutable.""" self.assertRaises(AttributeError, setattr, Color, 'something_new', 10) def testInstancesMutable(self): """Test that enum instances are not mutable.""" self.assertRaises(TypeError, setattr, Color.RED, 'something_new', 10) def testDefEnum(self): """Test def_enum works by building enum class from dict.""" WeekDay = messages.Enum.def_enum({'Monday': 1, 'Tuesday': 2, 'Wednesday': 3, 'Thursday': 4, 'Friday': 6, 'Saturday': 7, 'Sunday': 8}, 'WeekDay') self.assertEqual('Wednesday', WeekDay(3).name) self.assertEqual(6, WeekDay('Friday').number) self.assertEqual(WeekDay.Sunday, WeekDay('Sunday')) def testNonInt(self): """Test that non-integer values rejection by enum def.""" self.assertRaises(messages.EnumDefinitionError, messages.Enum.def_enum, {'Bad': '1'}, 'BadEnum') def testNegativeInt(self): """Test that negative numbers rejection by enum def.""" self.assertRaises(messages.EnumDefinitionError, messages.Enum.def_enum, {'Bad': -1}, 'BadEnum') def testLowerBound(self): """Test that zero is accepted by enum def.""" class NotImportant(messages.Enum): """Testing for value zero""" VALUE = 0 self.assertEqual(0, int(NotImportant.VALUE)) def testTooLargeInt(self): """Test that numbers too large are rejected.""" self.assertRaises(messages.EnumDefinitionError, messages.Enum.def_enum, {'Bad': (2 ** 29)}, 'BadEnum') def testRepeatedInt(self): """Test duplicated numbers are forbidden.""" self.assertRaises(messages.EnumDefinitionError, messages.Enum.def_enum, {'Ok': 1, 'Repeated': 1}, 'BadEnum') def testStr(self): """Test converting to string.""" self.assertEqual('RED', str(Color.RED)) self.assertEqual('ORANGE', str(Color.ORANGE)) def testInt(self): """Test converting to int.""" self.assertEqual(20, int(Color.RED)) self.assertEqual(2, int(Color.ORANGE)) def testRepr(self): """Test enum representation.""" self.assertEqual('Color(RED, 20)', repr(Color.RED)) self.assertEqual('Color(YELLOW, 40)', repr(Color.YELLOW)) def testDocstring(self): """Test that docstring is supported ok.""" class NotImportant(messages.Enum): """I have a docstring.""" VALUE1 = 1 self.assertEqual('I have a docstring.', NotImportant.__doc__) def testDeleteEnumValue(self): """Test that enum values cannot be deleted.""" self.assertRaises(TypeError, delattr, Color, 'RED') def testEnumName(self): """Test enum name.""" module_name = test_util.get_module_name(EnumTest) self.assertEqual('%s.Color' % module_name, Color.definition_name()) self.assertEqual(module_name, Color.outer_definition_name()) self.assertEqual(module_name, Color.definition_package()) def testDefinitionName_OverrideModule(self): """Test enum module is overriden by module package name.""" global package try: package = 'my.package' self.assertEqual('my.package.Color', Color.definition_name()) self.assertEqual('my.package', Color.outer_definition_name()) self.assertEqual('my.package', Color.definition_package()) finally: del package def testDefinitionName_NoModule(self): """Test what happens when there is no module for enum.""" class Enum1(messages.Enum): pass original_modules = sys.modules sys.modules = dict(sys.modules) try: del sys.modules[__name__] self.assertEqual('Enum1', Enum1.definition_name()) self.assertEqual(None, Enum1.outer_definition_name()) self.assertEqual(None, Enum1.definition_package()) self.assertEqual(six.text_type, type(Enum1.definition_name())) finally: sys.modules = original_modules def testDefinitionName_Nested(self): """Test nested Enum names.""" class MyMessage(messages.Message): class NestedEnum(messages.Enum): pass class NestedMessage(messages.Message): class NestedEnum(messages.Enum): pass module_name = test_util.get_module_name(EnumTest) self.assertEqual('%s.MyMessage.NestedEnum' % module_name, MyMessage.NestedEnum.definition_name()) self.assertEqual('%s.MyMessage' % module_name, MyMessage.NestedEnum.outer_definition_name()) self.assertEqual(module_name, MyMessage.NestedEnum.definition_package()) self.assertEqual( '%s.MyMessage.NestedMessage.NestedEnum' % module_name, MyMessage.NestedMessage.NestedEnum.definition_name()) self.assertEqual( '%s.MyMessage.NestedMessage' % module_name, MyMessage.NestedMessage.NestedEnum.outer_definition_name()) self.assertEqual( module_name, MyMessage.NestedMessage.NestedEnum.definition_package()) def testMessageDefinition(self): """Test that enumeration knows its enclosing message definition.""" class OuterEnum(messages.Enum): pass self.assertEqual(None, OuterEnum.message_definition()) class OuterMessage(messages.Message): class InnerEnum(messages.Enum): pass self.assertEqual( OuterMessage, OuterMessage.InnerEnum.message_definition()) def testComparison(self): """Test comparing various enums to different types.""" class Enum1(messages.Enum): VAL1 = 1 VAL2 = 2 class Enum2(messages.Enum): VAL1 = 1 self.assertEqual(Enum1.VAL1, Enum1.VAL1) self.assertNotEqual(Enum1.VAL1, Enum1.VAL2) self.assertNotEqual(Enum1.VAL1, Enum2.VAL1) self.assertNotEqual(Enum1.VAL1, 'VAL1') self.assertNotEqual(Enum1.VAL1, 1) self.assertNotEqual(Enum1.VAL1, 2) self.assertNotEqual(Enum1.VAL1, None) self.assertNotEqual(Enum1.VAL1, Enum2.VAL1) self.assertTrue(Enum1.VAL1 < Enum1.VAL2) self.assertTrue(Enum1.VAL2 > Enum1.VAL1) self.assertNotEqual(1, Enum2.VAL1) def testPickle(self): """Testing pickling and unpickling of Enum instances.""" colors = list(Color) unpickled = pickle.loads(pickle.dumps(colors)) self.assertEqual(colors, unpickled) # Unpickling shouldn't create new enum instances. for i, color in enumerate(colors): self.assertTrue(color is unpickled[i]) class FieldListTest(test_util.TestCase): def setUp(self): self.integer_field = messages.IntegerField(1, repeated=True) def testConstructor(self): self.assertEqual([1, 2, 3], messages.FieldList(self.integer_field, [1, 2, 3])) self.assertEqual([1, 2, 3], messages.FieldList(self.integer_field, (1, 2, 3))) self.assertEqual([], messages.FieldList(self.integer_field, [])) def testNone(self): self.assertRaises(TypeError, messages.FieldList, self.integer_field, None) def testDoNotAutoConvertString(self): string_field = messages.StringField(1, repeated=True) self.assertRaises(messages.ValidationError, messages.FieldList, string_field, 'abc') def testConstructorCopies(self): a_list = [1, 3, 6] field_list = messages.FieldList(self.integer_field, a_list) self.assertFalse(a_list is field_list) self.assertFalse(field_list is messages.FieldList(self.integer_field, field_list)) def testNonRepeatedField(self): self.assertRaisesWithRegexpMatch( messages.FieldDefinitionError, 'FieldList may only accept repeated fields', messages.FieldList, messages.IntegerField(1), []) def testConstructor_InvalidValues(self): self.assertRaisesWithRegexpMatch( messages.ValidationError, re.escape("Expected type %r " "for IntegerField, found 1 (type %r)" % (six.integer_types, str)), messages.FieldList, self.integer_field, ["1", "2", "3"]) def testConstructor_Scalars(self): self.assertRaisesWithRegexpMatch( messages.ValidationError, "IntegerField is repeated. Found: 3", messages.FieldList, self.integer_field, 3) self.assertRaisesWithRegexpMatch( messages.ValidationError, ("IntegerField is repeated. Found: " "<(list[_]?\|sequence)iterator object"), messages.FieldList, self.integer_field, iter([1, 2, 3])) def testSetSlice(self): field_list = messages.FieldList(self.integer_field, [1, 2, 3, 4, 5]) field_list[1:3] = [10, 20] self.assertEqual([1, 10, 20, 4, 5], field_list) def testSetSlice_InvalidValues(self): field_list = messages.FieldList(self.integer_field, [1, 2, 3, 4, 5]) def setslice(): field_list[1:3] = ['10', '20'] msg_re = re.escape("Expected type %r " "for IntegerField, found 10 (type %r)" % (six.integer_types, str)) self.assertRaisesWithRegexpMatch( messages.ValidationError, msg_re, setslice) def testSetItem(self): field_list = messages.FieldList(self.integer_field, [2]) field_list[0] = 10 self.assertEqual([10], field_list) def testSetItem_InvalidValues(self): field_list = messages.FieldList(self.integer_field, [2]) def setitem(): field_list[0] = '10' self.assertRaisesWithRegexpMatch( messages.ValidationError, re.escape("Expected type %r " "for IntegerField, found 10 (type %r)" % (six.integer_types, str)), setitem) def testAppend(self): field_list = messages.FieldList(self.integer_field, [2]) field_list.append(10) self.assertEqual([2, 10], field_list) def testAppend_InvalidValues(self): field_list = messages.FieldList(self.integer_field, [2]) field_list.name = 'a_field' def append(): field_list.append('10') self.assertRaisesWithRegexpMatch( messages.ValidationError, re.escape("Expected type %r " "for IntegerField, found 10 (type %r)" % (six.integer_types, str)), append) def testExtend(self): field_list = messages.FieldList(self.integer_field, [2]) field_list.extend([10]) self.assertEqual([2, 10], field_list) def testExtend_InvalidValues(self): field_list = messages.FieldList(self.integer_field, [2]) def extend(): field_list.extend(['10']) self.assertRaisesWithRegexpMatch( messages.ValidationError, re.escape("Expected type %r " "for IntegerField, found 10 (type %r)" % (six.integer_types, str)), extend) def testInsert(self): field_list = messages.FieldList(self.integer_field, [2, 3]) field_list.insert(1, 10) self.assertEqual([2, 10, 3], field_list) def testInsert_InvalidValues(self): field_list = messages.FieldList(self.integer_field, [2, 3]) def insert(): field_list.insert(1, '10') self.assertRaisesWithRegexpMatch( messages.ValidationError, re.escape("Expected type %r " "for IntegerField, found 10 (type %r)" % (six.integer_types, str)), insert) def testPickle(self): """Testing pickling and unpickling of FieldList instances.""" field_list = messages.FieldList(self.integer_field, [1, 2, 3, 4, 5]) unpickled = pickle.loads(pickle.dumps(field_list)) self.assertEqual(field_list, unpickled) self.assertIsInstance(unpickled.field, messages.IntegerField) self.assertEqual(1, unpickled.field.number) self.assertTrue(unpickled.field.repeated) class FieldTest(test_util.TestCase): def ActionOnAllFieldClasses(self, action): """Test all field classes except Message and Enum. Message and Enum require separate tests. Args: action: Callable that takes the field class as a parameter. """ classes = (messages.IntegerField, messages.FloatField, messages.BooleanField, messages.BytesField, messages.StringField) for field_class in classes: action(field_class) def testNumberAttribute(self): """Test setting the number attribute.""" def action(field_class): # Check range. self.assertRaises(messages.InvalidNumberError, field_class, 0) self.assertRaises(messages.InvalidNumberError, field_class, -1) self.assertRaises(messages.InvalidNumberError, field_class, messages.MAX_FIELD_NUMBER + 1) # Check reserved. self.assertRaises(messages.InvalidNumberError, field_class, messages.FIRST_RESERVED_FIELD_NUMBER) self.assertRaises(messages.InvalidNumberError, field_class, messages.LAST_RESERVED_FIELD_NUMBER) self.assertRaises(messages.InvalidNumberError, field_class, '1') # This one should work. field_class(number=1) self.ActionOnAllFieldClasses(action) def testRequiredAndRepeated(self): """Test setting the required and repeated fields.""" def action(field_class): field_class(1, required=True) field_class(1, repeated=True) self.assertRaises(messages.FieldDefinitionError, field_class, 1, required=True, repeated=True) self.ActionOnAllFieldClasses(action) def testInvalidVariant(self): """Test field with invalid variants.""" def action(field_class): if field_class is not message_types.DateTimeField: self.assertRaises(messages.InvalidVariantError, field_class, 1, variant=messages.Variant.ENUM) self.ActionOnAllFieldClasses(action) def testDefaultVariant(self): """Test that default variant is used when not set.""" def action(field_class): field = field_class(1) self.assertEqual(field_class.DEFAULT_VARIANT, field.variant) self.ActionOnAllFieldClasses(action) def testAlternateVariant(self): """Test that default variant is used when not set.""" field = messages.IntegerField(1, variant=messages.Variant.UINT32) self.assertEqual(messages.Variant.UINT32, field.variant) def testDefaultFields_Single(self): """Test default field is correct type (single).""" defaults = { messages.IntegerField: 10, messages.FloatField: 1.5, messages.BooleanField: False, messages.BytesField: b'abc', messages.StringField: u'abc', } def action(field_class): field_class(1, default=defaults[field_class]) self.ActionOnAllFieldClasses(action) # Run defaults test again checking for str/unicode compatiblity. defaults[messages.StringField] = 'abc' self.ActionOnAllFieldClasses(action) def testStringField_BadUnicodeInDefault(self): """Test binary values in string field.""" self.assertRaisesWithRegexpMatch( messages.InvalidDefaultError, r"Invalid default value for StringField:.: " r"Field encountered non-UTF-8 string .: " r"'utf.?8' codec can't decode byte 0xc3 in position 0: " r"invalid continuation byte", messages.StringField, 1, default=b'\xc3\x28') def testDefaultFields_InvalidSingle(self): """Test default field is correct type (invalid single).""" def action(field_class): self.assertRaises(messages.InvalidDefaultError, field_class, 1, default=object()) self.ActionOnAllFieldClasses(action) def testDefaultFields_InvalidRepeated(self): """Test default field does not accept defaults.""" self.assertRaisesWithRegexpMatch( messages.FieldDefinitionError, 'Repeated fields may not have defaults', messages.StringField, 1, repeated=True, default=[1, 2, 3]) def testDefaultFields_None(self): """Test none is always acceptable.""" def action(field_class): field_class(1, default=None) field_class(1, required=True, default=None) field_class(1, repeated=True, default=None) self.ActionOnAllFieldClasses(action) def testDefaultFields_Enum(self): """Test the default for enum fields.""" class Symbol(messages.Enum): ALPHA = 1 BETA = 2 GAMMA = 3 field = messages.EnumField(Symbol, 1, default=Symbol.ALPHA) self.assertEqual(Symbol.ALPHA, field.default) def testDefaultFields_EnumStringDelayedResolution(self): """Test that enum fields resolve default strings.""" field = messages.EnumField( 'apitools.base.protorpclite.descriptor.FieldDescriptor.Label', 1, default='OPTIONAL') self.assertEqual( descriptor.FieldDescriptor.Label.OPTIONAL, field.default) def testDefaultFields_EnumIntDelayedResolution(self): """Test that enum fields resolve default integers.""" field = messages.EnumField( 'apitools.base.protorpclite.descriptor.FieldDescriptor.Label', 1, default=2) self.assertEqual( descriptor.FieldDescriptor.Label.REQUIRED, field.default) def testDefaultFields_EnumOkIfTypeKnown(self): """Test enum fields accept valid default values when type is known.""" field = messages.EnumField(descriptor.FieldDescriptor.Label, 1, default='REPEATED') self.assertEqual( descriptor.FieldDescriptor.Label.REPEATED, field.default) def testDefaultFields_EnumForceCheckIfTypeKnown(self): """Test that enum fields validate default values if type is known.""" self.assertRaisesWithRegexpMatch(TypeError, 'No such value for NOT_A_LABEL in ' 'Enum Label', messages.EnumField, descriptor.FieldDescriptor.Label, 1, default='NOT_A_LABEL') def testDefaultFields_EnumInvalidDelayedResolution(self): """Test that enum fields raise errors upon delayed resolution error.""" field = messages.EnumField( 'apitools.base.protorpclite.descriptor.FieldDescriptor.Label', 1, default=200) self.assertRaisesWithRegexpMatch(TypeError, 'No such value for 200 in Enum Label', getattr, field, 'default') def testValidate_Valid(self): """Test validation of valid values.""" values = { messages.IntegerField: 10, messages.FloatField: 1.5, messages.BooleanField: False, messages.BytesField: b'abc', messages.StringField: u'abc', } def action(field_class): # Optional. field = field_class(1) field.validate(values[field_class]) # Required. field = field_class(1, required=True) field.validate(values[field_class]) # Repeated. field = field_class(1, repeated=True) field.validate([]) field.validate(()) field.validate([values[field_class]]) field.validate((values[field_class],)) # Right value, but not repeated. self.assertRaises(messages.ValidationError, field.validate, values[field_class]) self.assertRaises(messages.ValidationError, field.validate, values[field_class]) self.ActionOnAllFieldClasses(action) def testValidate_Invalid(self): """Test validation of valid values.""" values = { messages.IntegerField: "10", messages.FloatField: "blah", messages.BooleanField: 0, messages.BytesField: 10.20, messages.StringField: 42, } def action(field_class): # Optional. field = field_class(1) self.assertRaises(messages.ValidationError, field.validate, values[field_class]) # Required. field = field_class(1, required=True) self.assertRaises(messages.ValidationError, field.validate, values[field_class]) # Repeated. field = field_class(1, repeated=True) self.assertRaises(messages.ValidationError, field.validate, [values[field_class]]) self.assertRaises(messages.ValidationError, field.validate, (values[field_class],)) self.ActionOnAllFieldClasses(action) def testValidate_None(self): """Test that None is valid for non-required fields.""" def action(field_class): # Optional. field = field_class(1) field.validate(None) # Required. field = field_class(1, required=True) self.assertRaisesWithRegexpMatch(messages.ValidationError, 'Required field is missing', field.validate, None) # Repeated. field = field_class(1, repeated=True) field.validate(None) self.assertRaisesWithRegexpMatch( messages.ValidationError, 'Repeated values for %s may ' 'not be None' % field_class.__name__, field.validate, [None]) self.assertRaises(messages.ValidationError, field.validate, (None,)) self.ActionOnAllFieldClasses(action) def testValidateElement(self): """Test validation of valid values.""" values = { messages.IntegerField: (10, -1, 0), messages.FloatField: (1.5, -1.5, 3), # for json it is all a number messages.BooleanField: (True, False), messages.BytesField: (b'abc',), messages.StringField: (u'abc',), } def action(field_class): # Optional. field = field_class(1) for value in values[field_class]: field.validate_element(value) # Required. field = field_class(1, required=True) for value in values[field_class]: field.validate_element(value) # Repeated. field = field_class(1, repeated=True) self.assertRaises(messages.ValidationError, field.validate_element, []) self.assertRaises(messages.ValidationError, field.validate_element, ()) for value in values[field_class]: field.validate_element(value) # Right value, but repeated. self.assertRaises(messages.ValidationError, field.validate_element, list(values[field_class])) # testing list self.assertRaises(messages.ValidationError, field.validate_element, values[field_class]) # testing tuple self.ActionOnAllFieldClasses(action) def testValidateCastingElement(self): field = messages.FloatField(1) self.assertEqual(type(field.validate_element(12)), float) self.assertEqual(type(field.validate_element(12.0)), float) field = messages.IntegerField(1) self.assertEqual(type(field.validate_element(12)), int) self.assertRaises(messages.ValidationError, field.validate_element, 12.0) # should fails from float to int def testReadOnly(self): """Test that objects are all read-only.""" def action(field_class): field = field_class(10) self.assertRaises(AttributeError, setattr, field, 'number', 20) self.assertRaises(AttributeError, setattr, field, 'anything_else', 'whatever') self.ActionOnAllFieldClasses(action) def testMessageField(self): """Test the construction of message fields.""" self.assertRaises(messages.FieldDefinitionError, messages.MessageField, str, 10) self.assertRaises(messages.FieldDefinitionError, messages.MessageField, messages.Message, 10) class MyMessage(messages.Message): pass field = messages.MessageField(MyMessage, 10) self.assertEqual(MyMessage, field.type) def testMessageField_ForwardReference(self): """Test the construction of forward reference message fields.""" global MyMessage global ForwardMessage try: class MyMessage(messages.Message): self_reference = messages.MessageField('MyMessage', 1) forward = messages.MessageField('ForwardMessage', 2) nested = messages.MessageField( 'ForwardMessage.NestedMessage', 3) inner = messages.MessageField('Inner', 4) class Inner(messages.Message): sibling = messages.MessageField('Sibling', 1) class Sibling(messages.Message): pass class ForwardMessage(messages.Message): class NestedMessage(messages.Message): pass self.assertEqual(MyMessage, MyMessage.field_by_name('self_reference').type) self.assertEqual(ForwardMessage, MyMessage.field_by_name('forward').type) self.assertEqual(ForwardMessage.NestedMessage, MyMessage.field_by_name('nested').type) self.assertEqual(MyMessage.Inner, MyMessage.field_by_name('inner').type) self.assertEqual(MyMessage.Sibling, MyMessage.Inner.field_by_name('sibling').type) finally: try: del MyMessage del ForwardMessage except: # pylint:disable=bare-except pass def testMessageField_WrongType(self): """Test that forward referencing the wrong type raises an error.""" global AnEnum try: class AnEnum(messages.Enum): pass class AnotherMessage(messages.Message): a_field = messages.MessageField('AnEnum', 1) self.assertRaises(messages.FieldDefinitionError, getattr, AnotherMessage.field_by_name('a_field'), 'type') finally: del AnEnum def testMessageFieldValidate(self): """Test validation on message field.""" class MyMessage(messages.Message): pass class AnotherMessage(messages.Message): pass field = messages.MessageField(MyMessage, 10) field.validate(MyMessage()) self.assertRaises(messages.ValidationError, field.validate, AnotherMessage()) def testMessageFieldMessageType(self): """Test message_type property.""" class MyMessage(messages.Message): pass class HasMessage(messages.Message): field = messages.MessageField(MyMessage, 1) self.assertEqual(HasMessage.field.type, HasMessage.field.message_type) def testMessageFieldValueFromMessage(self): class MyMessage(messages.Message): pass class HasMessage(messages.Message): field = messages.MessageField(MyMessage, 1) instance = MyMessage() self.assertTrue( instance is HasMessage.field.value_from_message(instance)) def testMessageFieldValueFromMessageWrongType(self): class MyMessage(messages.Message): pass class HasMessage(messages.Message): field = messages.MessageField(MyMessage, 1) self.assertRaisesWithRegexpMatch( messages.DecodeError, 'Expected type MyMessage, got int: 10', HasMessage.field.value_from_message, 10) def testMessageFieldValueToMessage(self): class MyMessage(messages.Message): pass class HasMessage(messages.Message): field = messages.MessageField(MyMessage, 1) instance = MyMessage() self.assertTrue( instance is HasMessage.field.value_to_message(instance)) def testMessageFieldValueToMessageWrongType(self): class MyMessage(messages.Message): pass class MyOtherMessage(messages.Message): pass class HasMessage(messages.Message): field = messages.MessageField(MyMessage, 1) instance = MyOtherMessage() self.assertRaisesWithRegexpMatch( messages.EncodeError, 'Expected type MyMessage, got MyOtherMessage: <MyOtherMessage>', HasMessage.field.value_to_message, instance) def testIntegerField_AllowLong(self): """Test that the integer field allows for longs.""" if six.PY2: messages.IntegerField(10, default=long(10)) def testMessageFieldValidate_Initialized(self): """Test validation on message field.""" class MyMessage(messages.Message): field1 = messages.IntegerField(1, required=True) field = messages.MessageField(MyMessage, 10) # Will validate messages where is_initialized() is False. message = MyMessage() field.validate(message) message.field1 = 20 field.validate(message) def testEnumField(self): """Test the construction of enum fields.""" self.assertRaises(messages.FieldDefinitionError, messages.EnumField, str, 10) self.assertRaises(messages.FieldDefinitionError, messages.EnumField, messages.Enum, 10) class Color(messages.Enum): RED = 1 GREEN = 2 BLUE = 3 field = messages.EnumField(Color, 10) self.assertEqual(Color, field.type) class Another(messages.Enum): VALUE = 1 self.assertRaises(messages.InvalidDefaultError, messages.EnumField, Color, 10, default=Another.VALUE) def testEnumField_ForwardReference(self): """Test the construction of forward reference enum fields.""" global MyMessage global ForwardEnum global ForwardMessage try: class MyMessage(messages.Message): forward = messages.EnumField('ForwardEnum', 1) nested = messages.EnumField('ForwardMessage.NestedEnum', 2) inner = messages.EnumField('Inner', 3) class Inner(messages.Enum): pass class ForwardEnum(messages.Enum): pass class ForwardMessage(messages.Message): class NestedEnum(messages.Enum): pass self.assertEqual(ForwardEnum, MyMessage.field_by_name('forward').type) self.assertEqual(ForwardMessage.NestedEnum, MyMessage.field_by_name('nested').type) self.assertEqual(MyMessage.Inner, MyMessage.field_by_name('inner').type) finally: try: del MyMessage del ForwardEnum del ForwardMessage except: # pylint:disable=bare-except pass def testEnumField_WrongType(self): """Test that forward referencing the wrong type raises an error.""" global AMessage try: class AMessage(messages.Message): pass class AnotherMessage(messages.Message): a_field = messages.EnumField('AMessage', 1) self.assertRaises(messages.FieldDefinitionError, getattr, AnotherMessage.field_by_name('a_field'), 'type') finally: del AMessage def testMessageDefinition(self): """Test that message definition is set on fields.""" class MyMessage(messages.Message): my_field = messages.StringField(1) self.assertEqual( MyMessage, MyMessage.field_by_name('my_field').message_definition()) def testNoneAssignment(self): """Test that assigning None does not change comparison.""" class MyMessage(messages.Message): my_field = messages.StringField(1) m1 = MyMessage() m2 = MyMessage() m2.my_field = None self.assertEqual(m1, m2) def testNonUtf8Str(self): """Test validation fails for non-UTF-8 StringField values.""" class Thing(messages.Message): string_field = messages.StringField(2) thing = Thing() self.assertRaisesWithRegexpMatch( messages.ValidationError, 'Field string_field encountered non-UTF-8 string', setattr, thing, 'string_field', test_util.BINARY) class MessageTest(test_util.TestCase): """Tests for message class.""" def CreateMessageClass(self): """Creates a simple message class with 3 fields. Fields are defined in alphabetical order but with conflicting numeric order. """ class ComplexMessage(messages.Message): a3 = messages.IntegerField(3) b1 = messages.StringField(1) c2 = messages.StringField(2) return ComplexMessage def testSameNumbers(self): """Test that cannot assign two fields with same numbers.""" def action(): class BadMessage(messages.Message): f1 = messages.IntegerField(1) f2 = messages.IntegerField(1) self.assertRaises(messages.DuplicateNumberError, action) def testStrictAssignment(self): """Tests that cannot assign to unknown or non-reserved attributes.""" class SimpleMessage(messages.Message): field = messages.IntegerField(1) simple_message = SimpleMessage() self.assertRaises(AttributeError, setattr, simple_message, 'does_not_exist', 10) def testListAssignmentDoesNotCopy(self): class SimpleMessage(messages.Message): repeated = messages.IntegerField(1, repeated=True) message = SimpleMessage() original = message.repeated message.repeated = [] self.assertFalse(original is message.repeated) def testValidate_Optional(self): """Tests validation of optional fields.""" class SimpleMessage(messages.Message): non_required = messages.IntegerField(1) simple_message = SimpleMessage() simple_message.check_initialized() simple_message.non_required = 10 simple_message.check_initialized() def testValidate_Required(self): """Tests validation of required fields.""" class SimpleMessage(messages.Message): required = messages.IntegerField(1, required=True) simple_message = SimpleMessage() self.assertRaises(messages.ValidationError, simple_message.check_initialized) simple_message.required = 10 simple_message.check_initialized() def testValidate_Repeated(self): """Tests validation of repeated fields.""" class SimpleMessage(messages.Message): repeated = messages.IntegerField(1, repeated=True) simple_message = SimpleMessage() # Check valid values. for valid_value in [], [10], [10, 20], (), (10,), (10, 20): simple_message.repeated = valid_value simple_message.check_initialized() # Check cleared. simple_message.repeated = [] simple_message.check_initialized() # Check invalid values. for invalid_value in 10, ['10', '20'], [None], (None,): self.assertRaises( messages.ValidationError, setattr, simple_message, 'repeated', invalid_value) def testIsInitialized(self): """Tests is_initialized.""" class SimpleMessage(messages.Message): required = messages.IntegerField(1, required=True) simple_message = SimpleMessage() self.assertFalse(simple_message.is_initialized()) simple_message.required = 10 self.assertTrue(simple_message.is_initialized()) def testIsInitializedNestedField(self): """Tests is_initialized for nested fields.""" class SimpleMessage(messages.Message): required = messages.IntegerField(1, required=True) class NestedMessage(messages.Message): simple = messages.MessageField(SimpleMessage, 1) simple_message = SimpleMessage() self.assertFalse(simple_message.is_initialized()) nested_message = NestedMessage(simple=simple_message) self.assertFalse(nested_message.is_initialized()) simple_message.required = 10 self.assertTrue(simple_message.is_initialized()) self.assertTrue(nested_message.is_initialized()) def testInitializeNestedFieldFromDict(self): """Tests initializing nested fields from dict.""" class SimpleMessage(messages.Message): required = messages.IntegerField(1, required=True) class NestedMessage(messages.Message): simple = messages.MessageField(SimpleMessage, 1) class RepeatedMessage(messages.Message): simple = messages.MessageField(SimpleMessage, 1, repeated=True) nested_message1 = NestedMessage(simple={'required': 10}) self.assertTrue(nested_message1.is_initialized()) self.assertTrue(nested_message1.simple.is_initialized()) nested_message2 = NestedMessage() nested_message2.simple = {'required': 10} self.assertTrue(nested_message2.is_initialized()) self.assertTrue(nested_message2.simple.is_initialized()) repeated_values = [{}, {'required': 10}, SimpleMessage(required=20)] repeated_message1 = RepeatedMessage(simple=repeated_values) self.assertEqual(3, len(repeated_message1.simple)) self.assertFalse(repeated_message1.is_initialized()) repeated_message1.simple[0].required = 0 self.assertTrue(repeated_message1.is_initialized()) repeated_message2 = RepeatedMessage() repeated_message2.simple = repeated_values self.assertEqual(3, len(repeated_message2.simple)) self.assertFalse(repeated_message2.is_initialized()) repeated_message2.simple[0].required = 0 self.assertTrue(repeated_message2.is_initialized()) def testNestedMethodsNotAllowed(self): """Test that method definitions on Message classes are not allowed.""" def action(): class WithMethods(messages.Message): def not_allowed(self): pass self.assertRaises(messages.MessageDefinitionError, action) def testNestedAttributesNotAllowed(self): """Test attribute assignment on Message classes is not allowed.""" def int_attribute(): class WithMethods(messages.Message): not_allowed = 1 def string_attribute(): class WithMethods(messages.Message): not_allowed = 'not allowed' def enum_attribute(): class WithMethods(messages.Message): not_allowed = Color.RED for action in (int_attribute, string_attribute, enum_attribute): self.assertRaises(messages.MessageDefinitionError, action) def testNameIsSetOnFields(self): """Make sure name is set on fields after Message class init.""" class HasNamedFields(messages.Message): field = messages.StringField(1) self.assertEqual('field', HasNamedFields.field_by_number(1).name) def testSubclassingMessageDisallowed(self): """Not permitted to create sub-classes of message classes.""" class SuperClass(messages.Message): pass def action(): class SubClass(SuperClass): pass self.assertRaises(messages.MessageDefinitionError, action) def testAllFields(self): """Test all_fields method.""" ComplexMessage = self.CreateMessageClass() fields = list(ComplexMessage.all_fields()) # Order does not matter, so sort now. fields = sorted(fields, key=lambda f: f.name) self.assertEqual(3, len(fields)) self.assertEqual('a3', fields[0].name) self.assertEqual('b1', fields[1].name) self.assertEqual('c2', fields[2].name) def testFieldByName(self): """Test getting field by name.""" ComplexMessage = self.CreateMessageClass() self.assertEqual(3, ComplexMessage.field_by_name('a3').number) self.assertEqual(1, ComplexMessage.field_by_name('b1').number) self.assertEqual(2, ComplexMessage.field_by_name('c2').number) self.assertRaises(KeyError, ComplexMessage.field_by_name, 'unknown') def testFieldByNumber(self): """Test getting field by number.""" ComplexMessage = self.CreateMessageClass() self.assertEqual('a3', ComplexMessage.field_by_number(3).name) self.assertEqual('b1', ComplexMessage.field_by_number(1).name) self.assertEqual('c2', ComplexMessage.field_by_number(2).name) self.assertRaises(KeyError, ComplexMessage.field_by_number, 4) def testGetAssignedValue(self): """Test getting the assigned value of a field.""" class SomeMessage(messages.Message): a_value = messages.StringField(1, default=u'a default') message = SomeMessage() self.assertEqual(None, message.get_assigned_value('a_value')) message.a_value = u'a string' self.assertEqual(u'a string', message.get_assigned_value('a_value')) message.a_value = u'a default' self.assertEqual(u'a default', message.get_assigned_value('a_value')) self.assertRaisesWithRegexpMatch( AttributeError, 'Message SomeMessage has no field no_such_field', message.get_assigned_value, 'no_such_field') def testReset(self): """Test resetting a field value.""" class SomeMessage(messages.Message): a_value = messages.StringField(1, default=u'a default') repeated = messages.IntegerField(2, repeated=True) message = SomeMessage() self.assertRaises(AttributeError, message.reset, 'unknown') self.assertEqual(u'a default', message.a_value) message.reset('a_value') self.assertEqual(u'a default', message.a_value) message.a_value = u'a new value' self.assertEqual(u'a new value', message.a_value) message.reset('a_value') self.assertEqual(u'a default', message.a_value) message.repeated = [1, 2, 3] self.assertEqual([1, 2, 3], message.repeated) saved = message.repeated message.reset('repeated') self.assertEqual([], message.repeated) self.assertIsInstance(message.repeated, messages.FieldList) self.assertEqual([1, 2, 3], saved) def testAllowNestedEnums(self): """Test allowing nested enums in a message definition.""" class Trade(messages.Message): class Duration(messages.Enum): GTC = 1 DAY = 2 class Currency(messages.Enum): USD = 1 GBP = 2 INR = 3 # Sorted by name order seems to be the only feasible option. self.assertEqual(['Currency', 'Duration'], Trade.__enums__) # Message definition will now be set on Enumerated objects. self.assertEqual(Trade, Trade.Duration.message_definition()) def testAllowNestedMessages(self): """Test allowing nested messages in a message definition.""" class Trade(messages.Message): class Lot(messages.Message): pass class Agent(messages.Message): pass # Sorted by name order seems to be the only feasible option. self.assertEqual(['Agent', 'Lot'], Trade.__messages__) self.assertEqual(Trade, Trade.Agent.message_definition()) self.assertEqual(Trade, Trade.Lot.message_definition()) # But not Message itself. def action(): class Trade(messages.Message): NiceTry = messages.Message self.assertRaises(messages.MessageDefinitionError, action) def testDisallowClassAssignments(self): """Test setting class attributes may not happen.""" class MyMessage(messages.Message): pass self.assertRaises(AttributeError, setattr, MyMessage, 'x', 'do not assign') def testEquality(self): """Test message class equality.""" # Comparison against enums must work. class MyEnum(messages.Enum): val1 = 1 val2 = 2 # Comparisons against nested messages must work. class AnotherMessage(messages.Message): string = messages.StringField(1) class MyMessage(messages.Message): field1 = messages.IntegerField(1) field2 = messages.EnumField(MyEnum, 2) field3 = messages.MessageField(AnotherMessage, 3) message1 = MyMessage() self.assertNotEqual('hi', message1) self.assertNotEqual(AnotherMessage(), message1) self.assertEqual(message1, message1) message2 = MyMessage() self.assertEqual(message1, message2) message1.field1 = 10 self.assertNotEqual(message1, message2) message2.field1 = 20 self.assertNotEqual(message1, message2) message2.field1 = 10 self.assertEqual(message1, message2) message1.field2 = MyEnum.val1 self.assertNotEqual(message1, message2) message2.field2 = MyEnum.val2 self.assertNotEqual(message1, message2) message2.field2 = MyEnum.val1 self.assertEqual(message1, message2) message1.field3 = AnotherMessage() message1.field3.string = 'value1' self.assertNotEqual(message1, message2) message2.field3 = AnotherMessage() message2.field3.string = 'value2' self.assertNotEqual(message1, message2) message2.field3.string = 'value1' self.assertEqual(message1, message2) def testEqualityWithUnknowns(self): """Test message class equality with unknown fields.""" class MyMessage(messages.Message): field1 = messages.IntegerField(1) message1 = MyMessage() message2 = MyMessage() self.assertEqual(message1, message2) message1.set_unrecognized_field('unknown1', 'value1', messages.Variant.STRING) self.assertEqual(message1, message2) message1.set_unrecognized_field('unknown2', ['asdf', 3], messages.Variant.STRING) message1.set_unrecognized_field('unknown3', 4.7, messages.Variant.DOUBLE) self.assertEqual(message1, message2) def testUnrecognizedFieldInvalidVariant(self): class MyMessage(messages.Message): field1 = messages.IntegerField(1) message1 = MyMessage() self.assertRaises( TypeError, message1.set_unrecognized_field, 'unknown4', {'unhandled': 'type'}, None) self.assertRaises( TypeError, message1.set_unrecognized_field, 'unknown4', {'unhandled': 'type'}, 123) def testRepr(self): """Test represtation of Message object.""" class MyMessage(messages.Message): integer_value = messages.IntegerField(1) string_value = messages.StringField(2) unassigned = messages.StringField(3) unassigned_with_default = messages.StringField( 4, default=u'a default') my_message = MyMessage() my_message.integer_value = 42 my_message.string_value = u'A string' pat = re.compile(r"<MyMessage\n integer_value: 42\n" " string_value: [u]?'A string'>") self.assertTrue(pat.match(repr(my_message)) is not None) def testValidation(self): """Test validation of message values.""" # Test optional. class SubMessage(messages.Message): pass class Message(messages.Message): val = messages.MessageField(SubMessage, 1) message = Message() message_field = messages.MessageField(Message, 1) message_field.validate(message) message.val = SubMessage() message_field.validate(message) self.assertRaises(messages.ValidationError, setattr, message, 'val', [SubMessage()]) # Test required. class Message(messages.Message): val = messages.MessageField(SubMessage, 1, required=True) message = Message() message_field = messages.MessageField(Message, 1) message_field.validate(message) message.val = SubMessage() message_field.validate(message) self.assertRaises(messages.ValidationError, setattr, message, 'val', [SubMessage()]) # Test repeated. class Message(messages.Message): val = messages.MessageField(SubMessage, 1, repeated=True) message = Message() message_field = messages.MessageField(Message, 1) message_field.validate(message) self.assertRaisesWithRegexpMatch( messages.ValidationError, "Field val is repeated. Found: <SubMessage>", setattr, message, 'val', SubMessage()) message.val = [SubMessage()] message_field.validate(message) def testDefinitionName(self): """Test message name.""" class MyMessage(messages.Message): pass module_name = test_util.get_module_name(FieldTest) self.assertEqual('%s.MyMessage' % module_name, MyMessage.definition_name()) self.assertEqual(module_name, MyMessage.outer_definition_name()) self.assertEqual(module_name, MyMessage.definition_package()) self.assertEqual(six.text_type, type(MyMessage.definition_name())) self.assertEqual(six.text_type, type( MyMessage.outer_definition_name())) self.assertEqual(six.text_type, type(MyMessage.definition_package())) def testDefinitionName_OverrideModule(self): """Test message module is overriden by module package name.""" class MyMessage(messages.Message): pass global package package = 'my.package' try: self.assertEqual('my.package.MyMessage', MyMessage.definition_name()) self.assertEqual('my.package', MyMessage.outer_definition_name()) self.assertEqual('my.package', MyMessage.definition_package()) self.assertEqual(six.text_type, type(MyMessage.definition_name())) self.assertEqual(six.text_type, type( MyMessage.outer_definition_name())) self.assertEqual(six.text_type, type( MyMessage.definition_package())) finally: del package def testDefinitionName_NoModule(self): """Test what happens when there is no module for message.""" class MyMessage(messages.Message): pass original_modules = sys.modules sys.modules = dict(sys.modules) try: del sys.modules[__name__] self.assertEqual('MyMessage', MyMessage.definition_name()) self.assertEqual(None, MyMessage.outer_definition_name()) self.assertEqual(None, MyMessage.definition_package()) self.assertEqual(six.text_type, type(MyMessage.definition_name())) finally: sys.modules = original_modules def testDefinitionName_Nested(self): """Test nested message names.""" class MyMessage(messages.Message): class NestedMessage(messages.Message): class NestedMessage(messages.Message): pass module_name = test_util.get_module_name(MessageTest) self.assertEqual('%s.MyMessage.NestedMessage' % module_name, MyMessage.NestedMessage.definition_name()) self.assertEqual('%s.MyMessage' % module_name, MyMessage.NestedMessage.outer_definition_name()) self.assertEqual(module_name, MyMessage.NestedMessage.definition_package()) self.assertEqual( '%s.MyMessage.NestedMessage.NestedMessage' % module_name, MyMessage.NestedMessage.NestedMessage.definition_name()) self.assertEqual( '%s.MyMessage.NestedMessage' % module_name, MyMessage.NestedMessage.NestedMessage.outer_definition_name()) self.assertEqual( module_name, MyMessage.NestedMessage.NestedMessage.definition_package()) def testMessageDefinition(self): """Test that enumeration knows its enclosing message definition.""" class OuterMessage(messages.Message): class InnerMessage(messages.Message): pass self.assertEqual(None, OuterMessage.message_definition()) self.assertEqual(OuterMessage, OuterMessage.InnerMessage.message_definition()) def testConstructorKwargs(self): """Test kwargs via constructor.""" class SomeMessage(messages.Message): name = messages.StringField(1) number = messages.IntegerField(2) expected = SomeMessage() expected.name = 'my name' expected.number = 200 self.assertEqual(expected, SomeMessage(name='my name', number=200)) def testConstructorNotAField(self): """Test kwargs via constructor with wrong names.""" class SomeMessage(messages.Message): pass self.assertRaisesWithRegexpMatch( AttributeError, ('May not assign arbitrary value does_not_exist to message ' 'SomeMessage'), SomeMessage, does_not_exist=10) def testGetUnsetRepeatedValue(self): class SomeMessage(messages.Message): repeated = messages.IntegerField(1, repeated=True) instance = SomeMessage() self.assertEqual([], instance.repeated) self.assertTrue(isinstance(instance.repeated, messages.FieldList)) def testCompareAutoInitializedRepeatedFields(self): class SomeMessage(messages.Message): repeated = messages.IntegerField(1, repeated=True) message1 = SomeMessage(repeated=[]) message2 = SomeMessage() self.assertEqual(message1, message2) def testUnknownValues(self): """Test message class equality with unknown fields.""" class MyMessage(messages.Message): field1 = messages.IntegerField(1) message = MyMessage() self.assertEqual([], message.all_unrecognized_fields()) self.assertEqual((None, None), message.get_unrecognized_field_info('doesntexist')) self.assertEqual((None, None), message.get_unrecognized_field_info( 'doesntexist', None, None)) self.assertEqual(('defaultvalue', 'defaultwire'), message.get_unrecognized_field_info( 'doesntexist', 'defaultvalue', 'defaultwire')) self.assertEqual((3, None), message.get_unrecognized_field_info( 'doesntexist', value_default=3)) message.set_unrecognized_field('exists', 9.5, messages.Variant.DOUBLE) self.assertEqual(1, len(message.all_unrecognized_fields())) self.assertTrue('exists' in message.all_unrecognized_fields()) self.assertEqual((9.5, messages.Variant.DOUBLE), message.get_unrecognized_field_info('exists')) self.assertEqual((9.5, messages.Variant.DOUBLE), message.get_unrecognized_field_info('exists', 'type', 1234)) self.assertEqual( (1234, None), message.get_unrecognized_field_info('doesntexist', 1234)) message.set_unrecognized_field( 'another', 'value', messages.Variant.STRING) self.assertEqual(2, len(message.all_unrecognized_fields())) self.assertTrue('exists' in message.all_unrecognized_fields()) self.assertTrue('another' in message.all_unrecognized_fields()) self.assertEqual((9.5, messages.Variant.DOUBLE), message.get_unrecognized_field_info('exists')) self.assertEqual(('value', messages.Variant.STRING), message.get_unrecognized_field_info('another')) message.set_unrecognized_field('typetest1', ['list', 0, ('test',)], messages.Variant.STRING) self.assertEqual((['list', 0, ('test',)], messages.Variant.STRING), message.get_unrecognized_field_info('typetest1')) message.set_unrecognized_field( 'typetest2', '', messages.Variant.STRING) self.assertEqual(('', messages.Variant.STRING), message.get_unrecognized_field_info('typetest2')) def testPickle(self): """Testing pickling and unpickling of Message instances.""" global MyEnum global AnotherMessage global MyMessage class MyEnum(messages.Enum): val1 = 1 val2 = 2 class AnotherMessage(messages.Message): string = messages.StringField(1, repeated=True) class MyMessage(messages.Message): field1 = messages.IntegerField(1) field2 = messages.EnumField(MyEnum, 2) field3 = messages.MessageField(AnotherMessage, 3) message = MyMessage(field1=1, field2=MyEnum.val2, field3=AnotherMessage(string=['a', 'b', 'c'])) message.set_unrecognized_field( 'exists', 'value', messages.Variant.STRING) message.set_unrecognized_field('repeated', ['list', 0, ('test',)], messages.Variant.STRING) unpickled = pickle.loads(pickle.dumps(message)) self.assertEqual(message, unpickled) self.assertTrue(AnotherMessage.string is unpickled.field3.string.field) self.assertTrue('exists' in message.all_unrecognized_fields()) self.assertEqual(('value', messages.Variant.STRING), message.get_unrecognized_field_info('exists')) self.assertEqual((['list', 0, ('test',)], messages.Variant.STRING), message.get_unrecognized_field_info('repeated')) class FindDefinitionTest(test_util.TestCase): """Test finding definitions relative to various definitions and modules.""" def setUp(self): """Set up module-space. Starts off empty.""" self.modules = {} def DefineModule(self, name): """Define a module and its parents in module space. Modules that are already defined in self.modules are not re-created. Args: name: Fully qualified name of modules to create. Returns: Deepest nested module. For example: DefineModule('a.b.c') # Returns c. """ name_path = name.split('.') full_path = [] for node in name_path: full_path.append(node) full_name = '.'.join(full_path) self.modules.setdefault(full_name, types.ModuleType(full_name)) return self.modules[name] def DefineMessage(self, module, name, children=None, add_to_module=True): """Define a new Message class in the context of a module. Used for easily describing complex Message hierarchy. Message is defined including all child definitions. Args: module: Fully qualified name of module to place Message class in. name: Name of Message to define within module. children: Define any level of nesting of children definitions. To define a message, map the name to another dictionary. The dictionary can itself contain additional definitions, and so on. To map to an Enum, define the Enum class separately and map it by name. add_to_module: If True, new Message class is added to module. If False, new Message is not added. """ children = children or {} # Make sure module exists. module_instance = self.DefineModule(module) # Recursively define all child messages. for attribute, value in children.items(): if isinstance(value, dict): children[attribute] = self.DefineMessage( module, attribute, value, False) # Override default __module__ variable. children['__module__'] = module # Instantiate and possibly add to module. message_class = type(name, (messages.Message,), dict(children)) if add_to_module: setattr(module_instance, name, message_class) return message_class # pylint:disable=unused-argument # pylint:disable=redefined-builtin def Importer(self, module, globals='', locals='', fromlist=None): """Importer function. Acts like __import__. Only loads modules from self.modules. Does not try to load real modules defined elsewhere. Does not try to handle relative imports. Args: module: Fully qualified name of module to load from self.modules. """ if fromlist is None: module = module.split('.')[0] try: return self.modules[module] except KeyError: raise ImportError() # pylint:disable=unused-argument def testNoSuchModule(self): """Test searching for definitions that do no exist.""" self.assertRaises(messages.DefinitionNotFoundError, messages.find_definition, 'does.not.exist', importer=self.Importer) def testRefersToModule(self): """Test that referring to a module does not return that module.""" self.DefineModule('i.am.a.module') self.assertRaises(messages.DefinitionNotFoundError, messages.find_definition, 'i.am.a.module', importer=self.Importer) def testNoDefinition(self): """Test not finding a definition in an existing module.""" self.DefineModule('i.am.a.module') self.assertRaises(messages.DefinitionNotFoundError, messages.find_definition, 'i.am.a.module.MyMessage', importer=self.Importer) def testNotADefinition(self): """Test trying to fetch something that is not a definition.""" module = self.DefineModule('i.am.a.module') setattr(module, 'A', 'a string') self.assertRaises(messages.DefinitionNotFoundError, messages.find_definition, 'i.am.a.module.A', importer=self.Importer) def testGlobalFind(self): """Test finding definitions from fully qualified module names.""" A = self.DefineMessage('a.b.c', 'A', {}) self.assertEqual(A, messages.find_definition('a.b.c.A', importer=self.Importer)) B = self.DefineMessage('a.b.c', 'B', {'C': {}}) self.assertEqual( B.C, messages.find_definition('a.b.c.B.C', importer=self.Importer)) def testRelativeToModule(self): """Test finding definitions relative to modules.""" # Define modules. a = self.DefineModule('a') b = self.DefineModule('a.b') c = self.DefineModule('a.b.c') # Define messages. A = self.DefineMessage('a', 'A') B = self.DefineMessage('a.b', 'B') C = self.DefineMessage('a.b.c', 'C') D = self.DefineMessage('a.b.d', 'D') # Find A, B, C and D relative to a. self.assertEqual(A, messages.find_definition( 'A', a, importer=self.Importer)) self.assertEqual(B, messages.find_definition( 'b.B', a, importer=self.Importer)) self.assertEqual(C, messages.find_definition( 'b.c.C', a, importer=self.Importer)) self.assertEqual(D, messages.find_definition( 'b.d.D', a, importer=self.Importer)) # Find A, B, C and D relative to b. self.assertEqual(A, messages.find_definition( 'A', b, importer=self.Importer)) self.assertEqual(B, messages.find_definition( 'B', b, importer=self.Importer)) self.assertEqual(C, messages.find_definition( 'c.C', b, importer=self.Importer)) self.assertEqual(D, messages.find_definition( 'd.D', b, importer=self.Importer)) # Find A, B, C and D relative to c. Module d is the same case as c. self.assertEqual(A, messages.find_definition( 'A', c, importer=self.Importer)) self.assertEqual(B, messages.find_definition( 'B', c, importer=self.Importer)) self.assertEqual(C, messages.find_definition( 'C', c, importer=self.Importer)) self.assertEqual(D, messages.find_definition( 'd.D', c, importer=self.Importer)) def testRelativeToMessages(self): """Test finding definitions relative to Message definitions.""" A = self.DefineMessage('a.b', 'A', {'B': {'C': {}, 'D': {}}}) B = A.B C = A.B.C D = A.B.D # Find relative to A. self.assertEqual(A, messages.find_definition( 'A', A, importer=self.Importer)) self.assertEqual(B, messages.find_definition( 'B', A, importer=self.Importer)) self.assertEqual(C, messages.find_definition( 'B.C', A, importer=self.Importer)) self.assertEqual(D, messages.find_definition( 'B.D', A, importer=self.Importer)) # Find relative to B. self.assertEqual(A, messages.find_definition( 'A', B, importer=self.Importer)) self.assertEqual(B, messages.find_definition( 'B', B, importer=self.Importer)) self.assertEqual(C, messages.find_definition( 'C', B, importer=self.Importer)) self.assertEqual(D, messages.find_definition( 'D', B, importer=self.Importer)) # Find relative to C. self.assertEqual(A, messages.find_definition( 'A', C, importer=self.Importer)) self.assertEqual(B, messages.find_definition( 'B', C, importer=self.Importer)) self.assertEqual(C, messages.find_definition( 'C', C, importer=self.Importer)) self.assertEqual(D, messages.find_definition( 'D', C, importer=self.Importer)) # Find relative to C searching from c. self.assertEqual(A, messages.find_definition( 'b.A', C, importer=self.Importer)) self.assertEqual(B, messages.find_definition( 'b.A.B', C, importer=self.Importer)) self.assertEqual(C, messages.find_definition( 'b.A.B.C', C, importer=self.Importer)) self.assertEqual(D, messages.find_definition( 'b.A.B.D', C, importer=self.Importer)) def testAbsoluteReference(self): """Test finding absolute definition names.""" # Define modules. a = self.DefineModule('a') b = self.DefineModule('a.a') # Define messages. aA = self.DefineMessage('a', 'A') aaA = self.DefineMessage('a.a', 'A') # Always find a.A. self.assertEqual(aA, messages.find_definition('.a.A', None, importer=self.Importer)) self.assertEqual(aA, messages.find_definition('.a.A', a, importer=self.Importer)) self.assertEqual(aA, messages.find_definition('.a.A', aA, importer=self.Importer)) self.assertEqual(aA, messages.find_definition('.a.A', aaA, importer=self.Importer)) def testFindEnum(self): """Test that Enums are found.""" class Color(messages.Enum): pass A = self.DefineMessage('a', 'A', {'Color': Color}) self.assertEqual( Color, messages.find_definition('Color', A, importer=self.Importer)) def testFalseScope(self): """Test Message definitions nested in strange objects are hidden.""" global X class X(object): class A(messages.Message): pass self.assertRaises(TypeError, messages.find_definition, 'A', X) self.assertRaises(messages.DefinitionNotFoundError, messages.find_definition, 'X.A', sys.modules[__name__]) def testSearchAttributeFirst(self): """Make sure not faked out by module, but continues searching.""" A = self.DefineMessage('a', 'A') module_A = self.DefineModule('a.A') self.assertEqual(A, messages.find_definition( 'a.A', None, importer=self.Importer)) def main(): unittest.main() if __name__ == '__main__': main()	78.28 KB	2025-03-11 18:02
📄 protojson.py #!/usr/bin/env python # # Copyright 2010 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # """JSON support for message types. Public classes: MessageJSONEncoder: JSON encoder for message objects. Public functions: encode_message: Encodes a message in to a JSON string. decode_message: Merge from a JSON string in to a message. """ import base64 import binascii import logging import six from apitools.base.protorpclite import message_types from apitools.base.protorpclite import messages from apitools.base.protorpclite import util __all__ = [ 'ALTERNATIVE_CONTENT_TYPES', 'CONTENT_TYPE', 'MessageJSONEncoder', 'encode_message', 'decode_message', 'ProtoJson', ] def _load_json_module(): """Try to load a valid json module. There are more than one json modules that might be installed. They are mostly compatible with one another but some versions may be different. This function attempts to load various json modules in a preferred order. It does a basic check to guess if a loaded version of json is compatible. Returns: Compatible json module. Raises: ImportError if there are no json modules or the loaded json module is not compatible with ProtoRPC. """ first_import_error = None for module_name in ['json', 'simplejson']: try: module = __import__(module_name, {}, {}, 'json') if not hasattr(module, 'JSONEncoder'): message = ( 'json library "%s" is not compatible with ProtoRPC' % module_name) logging.warning(message) raise ImportError(message) else: return module except ImportError as err: if not first_import_error: first_import_error = err logging.error('Must use valid json library (json or simplejson)') raise first_import_error # pylint:disable=raising-bad-type json = _load_json_module() # TODO: Rename this to MessageJsonEncoder. class MessageJSONEncoder(json.JSONEncoder): """Message JSON encoder class. Extension of JSONEncoder that can build JSON from a message object. """ def __init__(self, protojson_protocol=None, kwargs): """Constructor. Args: protojson_protocol: ProtoJson instance. """ super(MessageJSONEncoder, self).__init__(kwargs) self.__protojson_protocol = ( protojson_protocol or ProtoJson.get_default()) def default(self, value): """Return dictionary instance from a message object. Args: value: Value to get dictionary for. If not encodable, will call superclasses default method. """ if isinstance(value, messages.Enum): return str(value) if six.PY3 and isinstance(value, bytes): return value.decode('utf8') if isinstance(value, messages.Message): result = {} for field in value.all_fields(): item = value.get_assigned_value(field.name) if item not in (None, [], ()): result[field.name] = ( self.__protojson_protocol.encode_field(field, item)) # Handle unrecognized fields, so they're included when a message is # decoded then encoded. for unknown_key in value.all_unrecognized_fields(): unrecognized_field, _ = value.get_unrecognized_field_info( unknown_key) # Unknown fields are not encoded as they should have been # processed before we get to here. result[unknown_key] = unrecognized_field return result return super(MessageJSONEncoder, self).default(value) class ProtoJson(object): """ProtoRPC JSON implementation class. Implementation of JSON based protocol used for serializing and deserializing message objects. Instances of remote.ProtocolConfig constructor or used with remote.Protocols.add_protocol. See the remote.py module for more details. """ CONTENT_TYPE = 'application/json' ALTERNATIVE_CONTENT_TYPES = [ 'application/x-javascript', 'text/javascript', 'text/x-javascript', 'text/x-json', 'text/json', ] def encode_field(self, field, value): """Encode a python field value to a JSON value. Args: field: A ProtoRPC field instance. value: A python value supported by field. Returns: A JSON serializable value appropriate for field. """ if isinstance(field, messages.BytesField): if field.repeated: value = [base64.b64encode(byte) for byte in value] else: value = base64.b64encode(value) elif isinstance(field, message_types.DateTimeField): # DateTimeField stores its data as a RFC 3339 compliant string. if field.repeated: value = [i.isoformat() for i in value] else: value = value.isoformat() return value def encode_message(self, message): """Encode Message instance to JSON string. Args: Message instance to encode in to JSON string. Returns: String encoding of Message instance in protocol JSON format. Raises: messages.ValidationError if message is not initialized. """ message.check_initialized() return json.dumps(message, cls=MessageJSONEncoder, protojson_protocol=self) def decode_message(self, message_type, encoded_message): """Merge JSON structure to Message instance. Args: message_type: Message to decode data to. encoded_message: JSON encoded version of message. Returns: Decoded instance of message_type. Raises: ValueError: If encoded_message is not valid JSON. messages.ValidationError if merged message is not initialized. """ encoded_message = six.ensure_str(encoded_message) if not encoded_message.strip(): return message_type() dictionary = json.loads(encoded_message) message = self.__decode_dictionary(message_type, dictionary) message.check_initialized() return message def __find_variant(self, value): """Find the messages.Variant type that describes this value. Args: value: The value whose variant type is being determined. Returns: The messages.Variant value that best describes value's type, or None if it's a type we don't know how to handle. """ if isinstance(value, bool): return messages.Variant.BOOL elif isinstance(value, six.integer_types): return messages.Variant.INT64 elif isinstance(value, float): return messages.Variant.DOUBLE elif isinstance(value, six.string_types): return messages.Variant.STRING elif isinstance(value, (list, tuple)): # Find the most specific variant that covers all elements. variant_priority = [None, messages.Variant.INT64, messages.Variant.DOUBLE, messages.Variant.STRING] chosen_priority = 0 for v in value: variant = self.__find_variant(v) try: priority = variant_priority.index(variant) except IndexError: priority = -1 if priority > chosen_priority: chosen_priority = priority return variant_priority[chosen_priority] # Unrecognized type. return None def __decode_dictionary(self, message_type, dictionary): """Merge dictionary in to message. Args: message: Message to merge dictionary in to. dictionary: Dictionary to extract information from. Dictionary is as parsed from JSON. Nested objects will also be dictionaries. """ message = message_type() for key, value in six.iteritems(dictionary): if value is None: try: message.reset(key) except AttributeError: pass # This is an unrecognized field, skip it. continue try: field = message.field_by_name(key) except KeyError: # Save unknown values. variant = self.__find_variant(value) if variant: message.set_unrecognized_field(key, value, variant) continue is_enum_field = isinstance(field, messages.EnumField) is_unrecognized_field = False if field.repeated: # This should be unnecessary? Or in fact become an error. if not isinstance(value, list): value = [value] valid_value = [] for item in value: try: v = self.decode_field(field, item) if is_enum_field and v is None: continue except messages.DecodeError: if not is_enum_field: raise is_unrecognized_field = True continue valid_value.append(v) setattr(message, field.name, valid_value) if is_unrecognized_field: variant = self.__find_variant(value) if variant: message.set_unrecognized_field(key, value, variant) continue # This is just for consistency with the old behavior. if value == []: continue try: setattr(message, field.name, self.decode_field(field, value)) except messages.DecodeError: # Save unknown enum values. if not is_enum_field: raise variant = self.__find_variant(value) if variant: message.set_unrecognized_field(key, value, variant) return message def decode_field(self, field, value): """Decode a JSON value to a python value. Args: field: A ProtoRPC field instance. value: A serialized JSON value. Return: A Python value compatible with field. """ if isinstance(field, messages.EnumField): try: return field.type(value) except TypeError: raise messages.DecodeError( 'Invalid enum value "%s"' % (value or '')) elif isinstance(field, messages.BytesField): try: return base64.b64decode(value) except (binascii.Error, TypeError) as err: raise messages.DecodeError('Base64 decoding error: %s' % err) elif isinstance(field, message_types.DateTimeField): try: return util.decode_datetime(value, truncate_time=True) except ValueError as err: raise messages.DecodeError(err) elif (isinstance(field, messages.MessageField) and issubclass(field.type, messages.Message)): return self.__decode_dictionary(field.type, value) elif (isinstance(field, messages.FloatField) and isinstance(value, (six.integer_types, six.string_types))): try: return float(value) except: # pylint:disable=bare-except pass elif (isinstance(field, messages.IntegerField) and isinstance(value, six.string_types)): try: return int(value) except: # pylint:disable=bare-except pass return value @staticmethod def get_default(): """Get default instanceof ProtoJson.""" try: return ProtoJson.__default except AttributeError: ProtoJson.__default = ProtoJson() return ProtoJson.__default @staticmethod def set_default(protocol): """Set the default instance of ProtoJson. Args: protocol: A ProtoJson instance. """ if not isinstance(protocol, ProtoJson): raise TypeError('Expected protocol of type ProtoJson') ProtoJson.__default = protocol CONTENT_TYPE = ProtoJson.CONTENT_TYPE ALTERNATIVE_CONTENT_TYPES = ProtoJson.ALTERNATIVE_CONTENT_TYPES encode_message = ProtoJson.get_default().encode_message decode_message = ProtoJson.get_default().decode_message	13.25 KB	2025-03-11 18:02
📄 protojson_test.py #!/usr/bin/env python # # Copyright 2010 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # """Tests for apitools.base.protorpclite.protojson.""" import datetime import json import unittest from apitools.base.protorpclite import message_types from apitools.base.protorpclite import messages from apitools.base.protorpclite import protojson from apitools.base.protorpclite import test_util class CustomField(messages.MessageField): """Custom MessageField class.""" type = int message_type = message_types.VoidMessage def __init__(self, number, kwargs): super(CustomField, self).__init__(self.message_type, number, kwargs) def value_to_message(self, value): return self.message_type() # pylint:disable=not-callable class MyMessage(messages.Message): """Test message containing various types.""" class Color(messages.Enum): RED = 1 GREEN = 2 BLUE = 3 class Nested(messages.Message): nested_value = messages.StringField(1) class NestedDatetime(messages.Message): nested_dt_value = message_types.DateTimeField(1) a_string = messages.StringField(2) an_integer = messages.IntegerField(3) a_float = messages.FloatField(4) a_boolean = messages.BooleanField(5) an_enum = messages.EnumField(Color, 6) a_nested = messages.MessageField(Nested, 7) a_repeated = messages.IntegerField(8, repeated=True) a_repeated_float = messages.FloatField(9, repeated=True) a_datetime = message_types.DateTimeField(10) a_repeated_datetime = message_types.DateTimeField(11, repeated=True) a_custom = CustomField(12) a_repeated_custom = CustomField(13, repeated=True) a_nested_datetime = messages.MessageField(NestedDatetime, 14) class ModuleInterfaceTest(test_util.ModuleInterfaceTest, test_util.TestCase): MODULE = protojson # TODO(rafek): Convert this test to the compliance test in test_util. class ProtojsonTest(test_util.TestCase, test_util.ProtoConformanceTestBase): """Test JSON encoding and decoding.""" PROTOLIB = protojson def CompareEncoded(self, expected_encoded, actual_encoded): """JSON encoding will be laundered to remove string differences.""" self.assertEqual(json.loads(expected_encoded), json.loads(actual_encoded)) encoded_empty_message = '{}' encoded_partial = """{ "double_value": 1.23, "int64_value": -100000000000, "int32_value": 1020, "string_value": "a string", "enum_value": "VAL2" } """ # pylint:disable=anomalous-unicode-escape-in-string encoded_full = """{ "double_value": 1.23, "float_value": -2.5, "int64_value": -100000000000, "uint64_value": 102020202020, "int32_value": 1020, "bool_value": true, "string_value": "a string\u044f", "bytes_value": "YSBieXRlc//+", "enum_value": "VAL2" } """ encoded_repeated = """{ "double_value": [1.23, 2.3], "float_value": [-2.5, 0.5], "int64_value": [-100000000000, 20], "uint64_value": [102020202020, 10], "int32_value": [1020, 718], "bool_value": [true, false], "string_value": ["a string\u044f", "another string"], "bytes_value": ["YSBieXRlc//+", "YW5vdGhlciBieXRlcw=="], "enum_value": ["VAL2", "VAL1"] } """ encoded_nested = """{ "nested": { "a_value": "a string" } } """ encoded_repeated_nested = """{ "repeated_nested": [{"a_value": "a string"}, {"a_value": "another string"}] } """ unexpected_tag_message = '{"unknown": "value"}' encoded_default_assigned = '{"a_value": "a default"}' encoded_nested_empty = '{"nested": {}}' encoded_repeated_nested_empty = '{"repeated_nested": [{}, {}]}' encoded_extend_message = '{"int64_value": [400, 50, 6000]}' encoded_string_types = '{"string_value": "Latin"}' encoded_invalid_enum = '{"enum_value": "undefined"}' encoded_invalid_repeated_enum = '{"enum_value": ["VAL1", "undefined"]}' def testConvertIntegerToFloat(self): """Test that integers passed in to float fields are converted. This is necessary because JSON outputs integers for numbers with 0 decimals. """ message = protojson.decode_message(MyMessage, '{"a_float": 10}') self.assertTrue(isinstance(message.a_float, float)) self.assertEqual(10.0, message.a_float) def testConvertStringToNumbers(self): """Test that strings passed to integer fields are converted.""" message = protojson.decode_message(MyMessage, """{"an_integer": "10", "a_float": "3.5", "a_repeated": ["1", "2"], "a_repeated_float": ["1.5", "2", 10] }""") self.assertEqual(MyMessage(an_integer=10, a_float=3.5, a_repeated=[1, 2], a_repeated_float=[1.5, 2.0, 10.0]), message) def testWrongTypeAssignment(self): """Test when wrong type is assigned to a field.""" self.assertRaises(messages.ValidationError, protojson.decode_message, MyMessage, '{"a_string": 10}') self.assertRaises(messages.ValidationError, protojson.decode_message, MyMessage, '{"an_integer": 10.2}') self.assertRaises(messages.ValidationError, protojson.decode_message, MyMessage, '{"an_integer": "10.2"}') def testNumericEnumeration(self): """Test that numbers work for enum values.""" message = protojson.decode_message(MyMessage, '{"an_enum": 2}') expected_message = MyMessage() expected_message.an_enum = MyMessage.Color.GREEN self.assertEqual(expected_message, message) def testNumericEnumerationNegativeTest(self): """Test with an invalid number for the enum value.""" # The message should successfully decode. message = protojson.decode_message(MyMessage, '{"an_enum": 89}') expected_message = MyMessage() self.assertEqual(expected_message, message) # The roundtrip should result in equivalent encoded # message. self.assertEqual('{"an_enum": 89}', protojson.encode_message(message)) def testAlphaEnumeration(self): """Test that alpha enum values work.""" message = protojson.decode_message(MyMessage, '{"an_enum": "RED"}') expected_message = MyMessage() expected_message.an_enum = MyMessage.Color.RED self.assertEqual(expected_message, message) def testAlphaEnumerationNegativeTest(self): """The alpha enum value is invalid.""" # The message should successfully decode. message = protojson.decode_message(MyMessage, '{"an_enum": "IAMINVALID"}') expected_message = MyMessage() self.assertEqual(expected_message, message) # The roundtrip should result in equivalent encoded message. self.assertEqual('{"an_enum": "IAMINVALID"}', protojson.encode_message(message)) def testEnumerationNegativeTestWithEmptyString(self): """The enum value is an empty string.""" # The message should successfully decode. message = protojson.decode_message(MyMessage, '{"an_enum": ""}') expected_message = MyMessage() self.assertEqual(expected_message, message) # The roundtrip should result in equivalent encoded message. self.assertEqual('{"an_enum": ""}', protojson.encode_message(message)) def testNullValues(self): """Test that null values overwrite existing values.""" self.assertEqual(MyMessage(), protojson.decode_message(MyMessage, ('{"an_integer": null,' ' "a_nested": null,' ' "an_enum": null' '}'))) def testEmptyList(self): """Test that empty lists are ignored.""" self.assertEqual(MyMessage(), protojson.decode_message(MyMessage, '{"a_repeated": []}')) def testNotJSON(self): """Test error when string is not valid JSON.""" self.assertRaises( ValueError, protojson.decode_message, MyMessage, '{this is not json}') def testDoNotEncodeStrangeObjects(self): """Test trying to encode a strange object. The main purpose of this test is to complete coverage. It ensures that the default behavior of the JSON encoder is preserved when someone tries to serialized an unexpected type. """ class BogusObject(object): def check_initialized(self): pass self.assertRaises(TypeError, protojson.encode_message, BogusObject()) def testMergeEmptyString(self): """Test merging the empty or space only string.""" message = protojson.decode_message(test_util.OptionalMessage, '') self.assertEqual(test_util.OptionalMessage(), message) message = protojson.decode_message(test_util.OptionalMessage, ' ') self.assertEqual(test_util.OptionalMessage(), message) def testProtojsonUnrecognizedFieldName(self): """Test that unrecognized fields are saved and can be accessed.""" decoded = protojson.decode_message( MyMessage, ('{"an_integer": 1, "unknown_val": 2}')) self.assertEqual(decoded.an_integer, 1) self.assertEqual(1, len(decoded.all_unrecognized_fields())) self.assertEqual('unknown_val', decoded.all_unrecognized_fields()[0]) self.assertEqual((2, messages.Variant.INT64), decoded.get_unrecognized_field_info('unknown_val')) def testProtojsonUnrecognizedFieldNumber(self): """Test that unrecognized fields are saved and can be accessed.""" decoded = protojson.decode_message( MyMessage, '{"an_integer": 1, "1001": "unknown", "-123": "negative", ' '"456_mixed": 2}') self.assertEqual(decoded.an_integer, 1) self.assertEqual(3, len(decoded.all_unrecognized_fields())) self.assertFalse(1001 in decoded.all_unrecognized_fields()) self.assertTrue('1001' in decoded.all_unrecognized_fields()) self.assertEqual(('unknown', messages.Variant.STRING), decoded.get_unrecognized_field_info('1001')) self.assertTrue('-123' in decoded.all_unrecognized_fields()) self.assertEqual(('negative', messages.Variant.STRING), decoded.get_unrecognized_field_info('-123')) self.assertTrue('456_mixed' in decoded.all_unrecognized_fields()) self.assertEqual((2, messages.Variant.INT64), decoded.get_unrecognized_field_info('456_mixed')) def testProtojsonUnrecognizedNull(self): """Test that unrecognized fields that are None are skipped.""" decoded = protojson.decode_message( MyMessage, '{"an_integer": 1, "unrecognized_null": null}') self.assertEqual(decoded.an_integer, 1) self.assertEqual(decoded.all_unrecognized_fields(), []) def testUnrecognizedFieldVariants(self): """Test that unrecognized fields are mapped to the right variants.""" for encoded, expected_variant in ( ('{"an_integer": 1, "unknown_val": 2}', messages.Variant.INT64), ('{"an_integer": 1, "unknown_val": 2.0}', messages.Variant.DOUBLE), ('{"an_integer": 1, "unknown_val": "string value"}', messages.Variant.STRING), ('{"an_integer": 1, "unknown_val": [1, 2, 3]}', messages.Variant.INT64), ('{"an_integer": 1, "unknown_val": [1, 2.0, 3]}', messages.Variant.DOUBLE), ('{"an_integer": 1, "unknown_val": [1, "foo", 3]}', messages.Variant.STRING), ('{"an_integer": 1, "unknown_val": true}', messages.Variant.BOOL)): decoded = protojson.decode_message(MyMessage, encoded) self.assertEqual(decoded.an_integer, 1) self.assertEqual(1, len(decoded.all_unrecognized_fields())) self.assertEqual( 'unknown_val', decoded.all_unrecognized_fields()[0]) _, decoded_variant = decoded.get_unrecognized_field_info( 'unknown_val') self.assertEqual(expected_variant, decoded_variant) def testDecodeDateTime(self): for datetime_string, datetime_vals in ( ('2012-09-30T15:31:50.262', (2012, 9, 30, 15, 31, 50, 262000)), ('2012-09-30T15:31:50', (2012, 9, 30, 15, 31, 50, 0))): message = protojson.decode_message( MyMessage, '{"a_datetime": "%s"}' % datetime_string) expected_message = MyMessage( a_datetime=datetime.datetime(datetime_vals)) self.assertEqual(expected_message, message) def testDecodeInvalidDateTime(self): self.assertRaises(messages.DecodeError, protojson.decode_message, MyMessage, '{"a_datetime": "invalid"}') def testDecodeInvalidMessage(self): encoded = """{ "a_nested_datetime": { "nested_dt_value": "invalid" } } """ self.assertRaises(messages.DecodeError, protojson.decode_message, MyMessage, encoded) def testEncodeDateTime(self): for datetime_string, datetime_vals in ( ('2012-09-30T15:31:50.262000', (2012, 9, 30, 15, 31, 50, 262000)), ('2012-09-30T15:31:50.262123', (2012, 9, 30, 15, 31, 50, 262123)), ('2012-09-30T15:31:50', (2012, 9, 30, 15, 31, 50, 0))): decoded_message = protojson.encode_message( MyMessage(a_datetime=datetime.datetime(datetime_vals))) expected_decoding = '{"a_datetime": "%s"}' % datetime_string self.CompareEncoded(expected_decoding, decoded_message) def testDecodeRepeatedDateTime(self): message = protojson.decode_message( MyMessage, '{"a_repeated_datetime": ["2012-09-30T15:31:50.262", ' '"2010-01-21T09:52:00", "2000-01-01T01:00:59.999999"]}') expected_message = MyMessage( a_repeated_datetime=[ datetime.datetime(2012, 9, 30, 15, 31, 50, 262000), datetime.datetime(2010, 1, 21, 9, 52), datetime.datetime(2000, 1, 1, 1, 0, 59, 999999)]) self.assertEqual(expected_message, message) def testDecodeCustom(self): message = protojson.decode_message(MyMessage, '{"a_custom": 1}') self.assertEqual(MyMessage(a_custom=1), message) def testDecodeInvalidCustom(self): self.assertRaises(messages.ValidationError, protojson.decode_message, MyMessage, '{"a_custom": "invalid"}') def testEncodeCustom(self): decoded_message = protojson.encode_message(MyMessage(a_custom=1)) self.CompareEncoded('{"a_custom": 1}', decoded_message) def testDecodeRepeatedCustom(self): message = protojson.decode_message( MyMessage, '{"a_repeated_custom": [1, 2, 3]}') self.assertEqual(MyMessage(a_repeated_custom=[1, 2, 3]), message) def testDecodeRepeatedEmpty(self): message = protojson.decode_message( MyMessage, '{"a_repeated": []}') self.assertEqual(MyMessage(a_repeated=[]), message) def testDecodeNone(self): message = protojson.decode_message( MyMessage, '{"an_integer": []}') self.assertEqual(MyMessage(an_integer=None), message) def testDecodeBadBase64BytesField(self): """Test decoding improperly encoded base64 bytes value.""" self.assertRaisesWithRegexpMatch( messages.DecodeError, 'Base64 decoding error', protojson.decode_message, test_util.OptionalMessage, '{"bytes_value": "abcdefghijklmnopq"}') class CustomProtoJson(protojson.ProtoJson): def encode_field(self, field, value): return '{encoded}' + value def decode_field(self, field, value): return '{decoded}' + value class CustomProtoJsonTest(test_util.TestCase): """Tests for serialization overriding functionality.""" def setUp(self): self.protojson = CustomProtoJson() def testEncode(self): self.assertEqual( '{"a_string": "{encoded}xyz"}', self.protojson.encode_message(MyMessage(a_string='xyz'))) def testDecode(self): self.assertEqual( MyMessage(a_string='{decoded}xyz'), self.protojson.decode_message(MyMessage, '{"a_string": "xyz"}')) def testDecodeEmptyMessage(self): self.assertEqual( MyMessage(a_string='{decoded}'), self.protojson.decode_message(MyMessage, '{"a_string": ""}')) def testDefault(self): self.assertTrue(protojson.ProtoJson.get_default(), protojson.ProtoJson.get_default()) instance = CustomProtoJson() protojson.ProtoJson.set_default(instance) self.assertTrue(instance is protojson.ProtoJson.get_default()) if __name__ == '__main__': unittest.main()	18.24 KB	2025-03-11 18:02
📄 test_util.py #!/usr/bin/env python # # Copyright 2010 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # """Test utilities for message testing. Includes module interface test to ensure that public parts of module are correctly declared in __all__. Includes message types that correspond to those defined in services_test.proto. Includes additional test utilities to make sure encoding/decoding libraries conform. """ import cgi import datetime import inspect import os import re import socket import types import unittest import six from six.moves import range # pylint: disable=redefined-builtin from apitools.base.protorpclite import message_types from apitools.base.protorpclite import messages from apitools.base.protorpclite import util # Unicode of the word "Russian" in cyrillic. RUSSIAN = u'\u0440\u0443\u0441\u0441\u043a\u0438\u0439' # All characters binary value interspersed with nulls. BINARY = b''.join(six.int2byte(value) + b'\0' for value in range(256)) class TestCase(unittest.TestCase): def assertRaisesWithRegexpMatch(self, exception, regexp, function, params, kwargs): """Check that exception is raised and text matches regular expression. Args: exception: Exception type that is expected. regexp: String regular expression that is expected in error message. function: Callable to test. params: Parameters to forward to function. kwargs: Keyword arguments to forward to function. """ try: function(params, *kwargs) self.fail('Expected exception %s was not raised' % exception.__name__) except exception as err: match = bool(re.match(regexp, str(err))) self.assertTrue(match, 'Expected match "%s", found "%s"' % (regexp, err)) def assertHeaderSame(self, header1, header2): """Check that two HTTP headers are the same. Args: header1: Header value string 1. header2: header value string 2. """ value1, params1 = cgi.parse_header(header1) value2, params2 = cgi.parse_header(header2) self.assertEqual(value1, value2) self.assertEqual(params1, params2) def assertIterEqual(self, iter1, iter2): """Check two iterators or iterables are equal independent of order. Similar to Python 2.7 assertItemsEqual. Named differently in order to avoid potential conflict. Args: iter1: An iterator or iterable. iter2: An iterator or iterable. """ list1 = list(iter1) list2 = list(iter2) unmatched1 = list() while list1: item1 = list1[0] del list1[0] for index in range(len(list2)): if item1 == list2[index]: del list2[index] break else: unmatched1.append(item1) error_message = [] for item in unmatched1: error_message.append( ' Item from iter1 not found in iter2: %r' % item) for item in list2: error_message.append( ' Item from iter2 not found in iter1: %r' % item) if error_message: self.fail('Collections not equivalent:\n' + '\n'.join(error_message)) class ModuleInterfaceTest(object): """Test to ensure module interface is carefully constructed. A module interface is the set of public objects listed in the module __all__ attribute. Modules that that are considered public should have this interface carefully declared. At all times, the __all__ attribute should have objects intended to be publically used and all other objects in the module should be considered unused. Protected attributes (those beginning with '_') and other imported modules should not be part of this set of variables. An exception is for variables that begin and end with '__' which are implicitly part of the interface (eg. __name__, __file__, __all__ itself, etc.). Modules that are imported in to the tested modules are an exception and may be left out of the __all__ definition. The test is done by checking the value of what would otherwise be a public name and not allowing it to be exported if it is an instance of a module. Modules that are explicitly exported are for the time being not permitted. To use this test class a module should define a new class that inherits first from ModuleInterfaceTest and then from test_util.TestCase. No other tests should be added to this test case, making the order of inheritance less important, but if setUp for some reason is overidden, it is important that ModuleInterfaceTest is first in the list so that its setUp method is invoked. Multiple inheritance is required so that ModuleInterfaceTest is not itself a test, and is not itself executed as one. The test class is expected to have the following class attributes defined: MODULE: A reference to the module that is being validated for interface correctness. Example: Module definition (hello.py): import sys __all__ = ['hello'] def _get_outputter(): return sys.stdout def hello(): _get_outputter().write('Hello\n') Test definition: import unittest from protorpc import test_util import hello class ModuleInterfaceTest(test_util.ModuleInterfaceTest, test_util.TestCase): MODULE = hello class HelloTest(test_util.TestCase): ... Test 'hello' module ... if __name__ == '__main__': unittest.main() """ def setUp(self): """Set up makes sure that MODULE and IMPORTED_MODULES is defined. This is a basic configuration test for the test itself so does not get it's own test case. """ if not hasattr(self, 'MODULE'): self.fail( "You must define 'MODULE' on ModuleInterfaceTest sub-class " "%s." % type(self).__name__) def testAllExist(self): """Test that all attributes defined in __all__ exist.""" missing_attributes = [] for attribute in self.MODULE.__all__: if not hasattr(self.MODULE, attribute): missing_attributes.append(attribute) if missing_attributes: self.fail('%s of __all__ are not defined in module.' % missing_attributes) def testAllExported(self): """Test that all public attributes not imported are in __all__.""" missing_attributes = [] for attribute in dir(self.MODULE): if not attribute.startswith('_'): if (attribute not in self.MODULE.__all__ and not isinstance(getattr(self.MODULE, attribute), types.ModuleType) and attribute != 'with_statement'): missing_attributes.append(attribute) if missing_attributes: self.fail('%s are not modules and not defined in __all__.' % missing_attributes) def testNoExportedProtectedVariables(self): """Test that there are no protected variables listed in __all__.""" protected_variables = [] for attribute in self.MODULE.__all__: if attribute.startswith('_'): protected_variables.append(attribute) if protected_variables: self.fail('%s are protected variables and may not be exported.' % protected_variables) def testNoExportedModules(self): """Test that no modules exist in __all__.""" exported_modules = [] for attribute in self.MODULE.__all__: try: value = getattr(self.MODULE, attribute) except AttributeError: # This is a different error case tested for in testAllExist. pass else: if isinstance(value, types.ModuleType): exported_modules.append(attribute) if exported_modules: self.fail('%s are modules and may not be exported.' % exported_modules) class NestedMessage(messages.Message): """Simple message that gets nested in another message.""" a_value = messages.StringField(1, required=True) class HasNestedMessage(messages.Message): """Message that has another message nested in it.""" nested = messages.MessageField(NestedMessage, 1) repeated_nested = messages.MessageField(NestedMessage, 2, repeated=True) class HasDefault(messages.Message): """Has a default value.""" a_value = messages.StringField(1, default=u'a default') class OptionalMessage(messages.Message): """Contains all message types.""" class SimpleEnum(messages.Enum): """Simple enumeration type.""" VAL1 = 1 VAL2 = 2 double_value = messages.FloatField(1, variant=messages.Variant.DOUBLE) float_value = messages.FloatField(2, variant=messages.Variant.FLOAT) int64_value = messages.IntegerField(3, variant=messages.Variant.INT64) uint64_value = messages.IntegerField(4, variant=messages.Variant.UINT64) int32_value = messages.IntegerField(5, variant=messages.Variant.INT32) bool_value = messages.BooleanField(6, variant=messages.Variant.BOOL) string_value = messages.StringField(7, variant=messages.Variant.STRING) bytes_value = messages.BytesField(8, variant=messages.Variant.BYTES) enum_value = messages.EnumField(SimpleEnum, 10) class RepeatedMessage(messages.Message): """Contains all message types as repeated fields.""" class SimpleEnum(messages.Enum): """Simple enumeration type.""" VAL1 = 1 VAL2 = 2 double_value = messages.FloatField(1, variant=messages.Variant.DOUBLE, repeated=True) float_value = messages.FloatField(2, variant=messages.Variant.FLOAT, repeated=True) int64_value = messages.IntegerField(3, variant=messages.Variant.INT64, repeated=True) uint64_value = messages.IntegerField(4, variant=messages.Variant.UINT64, repeated=True) int32_value = messages.IntegerField(5, variant=messages.Variant.INT32, repeated=True) bool_value = messages.BooleanField(6, variant=messages.Variant.BOOL, repeated=True) string_value = messages.StringField(7, variant=messages.Variant.STRING, repeated=True) bytes_value = messages.BytesField(8, variant=messages.Variant.BYTES, repeated=True) enum_value = messages.EnumField(SimpleEnum, 10, repeated=True) class HasOptionalNestedMessage(messages.Message): nested = messages.MessageField(OptionalMessage, 1) repeated_nested = messages.MessageField(OptionalMessage, 2, repeated=True) # pylint:disable=anomalous-unicode-escape-in-string class ProtoConformanceTestBase(object): """Protocol conformance test base class. Each supported protocol should implement two methods that support encoding and decoding of Message objects in that format: encode_message(message) - Serialize to encoding. encode_message(message, encoded_message) - Deserialize from encoding. Tests for the modules where these functions are implemented should extend this class in order to support basic behavioral expectations. This ensures that protocols correctly encode and decode message transparently to the caller. In order to support these test, the base class should also extend the TestCase class and implement the following class attributes which define the encoded version of certain protocol buffers: encoded_partial: <OptionalMessage double_value: 1.23 int64_value: -100000000000 string_value: u"a string" enum_value: OptionalMessage.SimpleEnum.VAL2 > encoded_full: <OptionalMessage double_value: 1.23 float_value: -2.5 int64_value: -100000000000 uint64_value: 102020202020 int32_value: 1020 bool_value: true string_value: u"a string\u044f" bytes_value: b"a bytes\xff\xfe" enum_value: OptionalMessage.SimpleEnum.VAL2 > encoded_repeated: <RepeatedMessage double_value: [1.23, 2.3] float_value: [-2.5, 0.5] int64_value: [-100000000000, 20] uint64_value: [102020202020, 10] int32_value: [1020, 718] bool_value: [true, false] string_value: [u"a string\u044f", u"another string"] bytes_value: [b"a bytes\xff\xfe", b"another bytes"] enum_value: [OptionalMessage.SimpleEnum.VAL2, OptionalMessage.SimpleEnum.VAL 1] > encoded_nested: <HasNestedMessage nested: <NestedMessage a_value: "a string" > > encoded_repeated_nested: <HasNestedMessage repeated_nested: [ <NestedMessage a_value: "a string">, <NestedMessage a_value: "another string"> ] > unexpected_tag_message: An encoded message that has an undefined tag or number in the stream. encoded_default_assigned: <HasDefault a_value: "a default" > encoded_nested_empty: <HasOptionalNestedMessage nested: <OptionalMessage> > encoded_invalid_enum: <OptionalMessage enum_value: (invalid value for serialization type) > encoded_invalid_repeated_enum: <RepeatedMessage enum_value: (invalid value for serialization type) > """ encoded_empty_message = '' def testEncodeInvalidMessage(self): message = NestedMessage() self.assertRaises(messages.ValidationError, self.PROTOLIB.encode_message, message) def CompareEncoded(self, expected_encoded, actual_encoded): """Compare two encoded protocol values. Can be overridden by sub-classes to special case comparison. For example, to eliminate white space from output that is not relevant to encoding. Args: expected_encoded: Expected string encoded value. actual_encoded: Actual string encoded value. """ self.assertEqual(expected_encoded, actual_encoded) def EncodeDecode(self, encoded, expected_message): message = self.PROTOLIB.decode_message(type(expected_message), encoded) self.assertEqual(expected_message, message) self.CompareEncoded(encoded, self.PROTOLIB.encode_message(message)) def testEmptyMessage(self): self.EncodeDecode(self.encoded_empty_message, OptionalMessage()) def testPartial(self): """Test message with a few values set.""" message = OptionalMessage() message.double_value = 1.23 message.int64_value = -100000000000 message.int32_value = 1020 message.string_value = u'a string' message.enum_value = OptionalMessage.SimpleEnum.VAL2 self.EncodeDecode(self.encoded_partial, message) def testFull(self): """Test all types.""" message = OptionalMessage() message.double_value = 1.23 message.float_value = -2.5 message.int64_value = -100000000000 message.uint64_value = 102020202020 message.int32_value = 1020 message.bool_value = True message.string_value = u'a string\u044f' message.bytes_value = b'a bytes\xff\xfe' message.enum_value = OptionalMessage.SimpleEnum.VAL2 self.EncodeDecode(self.encoded_full, message) def testRepeated(self): """Test repeated fields.""" message = RepeatedMessage() message.double_value = [1.23, 2.3] message.float_value = [-2.5, 0.5] message.int64_value = [-100000000000, 20] message.uint64_value = [102020202020, 10] message.int32_value = [1020, 718] message.bool_value = [True, False] message.string_value = [u'a string\u044f', u'another string'] message.bytes_value = [b'a bytes\xff\xfe', b'another bytes'] message.enum_value = [RepeatedMessage.SimpleEnum.VAL2, RepeatedMessage.SimpleEnum.VAL1] self.EncodeDecode(self.encoded_repeated, message) def testNested(self): """Test nested messages.""" nested_message = NestedMessage() nested_message.a_value = u'a string' message = HasNestedMessage() message.nested = nested_message self.EncodeDecode(self.encoded_nested, message) def testRepeatedNested(self): """Test repeated nested messages.""" nested_message1 = NestedMessage() nested_message1.a_value = u'a string' nested_message2 = NestedMessage() nested_message2.a_value = u'another string' message = HasNestedMessage() message.repeated_nested = [nested_message1, nested_message2] self.EncodeDecode(self.encoded_repeated_nested, message) def testStringTypes(self): """Test that encoding str on StringField works.""" message = OptionalMessage() message.string_value = 'Latin' self.EncodeDecode(self.encoded_string_types, message) def testEncodeUninitialized(self): """Test that cannot encode uninitialized message.""" required = NestedMessage() self.assertRaisesWithRegexpMatch(messages.ValidationError, "Message NestedMessage is missing " "required field a_value", self.PROTOLIB.encode_message, required) def testUnexpectedField(self): """Test decoding and encoding unexpected fields.""" loaded_message = self.PROTOLIB.decode_message( OptionalMessage, self.unexpected_tag_message) # Message should be equal to an empty message, since unknown # values aren't included in equality. self.assertEqual(OptionalMessage(), loaded_message) # Verify that the encoded message matches the source, including the # unknown value. self.assertEqual(self.unexpected_tag_message, self.PROTOLIB.encode_message(loaded_message)) def testDoNotSendDefault(self): """Test that default is not sent when nothing is assigned.""" self.EncodeDecode(self.encoded_empty_message, HasDefault()) def testSendDefaultExplicitlyAssigned(self): """Test that default is sent when explcitly assigned.""" message = HasDefault() message.a_value = HasDefault.a_value.default self.EncodeDecode(self.encoded_default_assigned, message) def testEncodingNestedEmptyMessage(self): """Test encoding a nested empty message.""" message = HasOptionalNestedMessage() message.nested = OptionalMessage() self.EncodeDecode(self.encoded_nested_empty, message) def testEncodingRepeatedNestedEmptyMessage(self): """Test encoding a nested empty message.""" message = HasOptionalNestedMessage() message.repeated_nested = [OptionalMessage(), OptionalMessage()] self.EncodeDecode(self.encoded_repeated_nested_empty, message) def testContentType(self): self.assertTrue(isinstance(self.PROTOLIB.CONTENT_TYPE, str)) def testDecodeInvalidEnumType(self): # Since protos need to be able to add new enums, a message should be # successfully decoded even if the enum value is invalid. Encoding the # decoded message should result in equivalence with the original # encoded message containing an invalid enum. decoded = self.PROTOLIB.decode_message(OptionalMessage, self.encoded_invalid_enum) message = OptionalMessage() self.assertEqual(message, decoded) encoded = self.PROTOLIB.encode_message(decoded) self.assertEqual(self.encoded_invalid_enum, encoded) def testDecodeInvalidRepeatedEnumType(self): # Since protos need to be able to add new enums, a message should be # successfully decoded even if the enum value is invalid. Encoding the # decoded message should result in equivalence with the original # encoded message containing an invalid enum. decoded = self.PROTOLIB.decode_message(RepeatedMessage, self.encoded_invalid_repeated_enum) message = RepeatedMessage() message.enum_value = [RepeatedMessage.SimpleEnum.VAL1] self.assertEqual(message, decoded) encoded = self.PROTOLIB.encode_message(decoded) self.assertEqual(self.encoded_invalid_repeated_enum, encoded) def testDateTimeNoTimeZone(self): """Test that DateTimeFields are encoded/decoded correctly.""" class MyMessage(messages.Message): value = message_types.DateTimeField(1) value = datetime.datetime(2013, 1, 3, 11, 36, 30, 123000) message = MyMessage(value=value) decoded = self.PROTOLIB.decode_message( MyMessage, self.PROTOLIB.encode_message(message)) self.assertEqual(decoded.value, value) def testDateTimeWithTimeZone(self): """Test DateTimeFields with time zones.""" class MyMessage(messages.Message): value = message_types.DateTimeField(1) value = datetime.datetime(2013, 1, 3, 11, 36, 30, 123000, util.TimeZoneOffset(8 60)) message = MyMessage(value=value) decoded = self.PROTOLIB.decode_message( MyMessage, self.PROTOLIB.encode_message(message)) self.assertEqual(decoded.value, value) def pick_unused_port(): """Find an unused port to use in tests. Derived from Damon Kohlers example: http://code.activestate.com/recipes/531822-pick-unused-port """ temp = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: temp.bind(('localhost', 0)) port = temp.getsockname()[1] finally: temp.close() return port def get_module_name(module_attribute): """Get the module name. Args: module_attribute: An attribute of the module. Returns: The fully qualified module name or simple module name where 'module_attribute' is defined if the module name is "__main__". """ if module_attribute.__module__ == '__main__': module_file = inspect.getfile(module_attribute) default = os.path.basename(module_file).split('.')[0] return default return module_attribute.__module__	23.88 KB	2025-03-11 18:02
📄 util.py #!/usr/bin/env python # # Copyright 2010 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # """Common utility library.""" from __future__ import with_statement import datetime import functools import inspect import logging import os import re import sys import six __all__ = [ 'Error', 'decode_datetime', 'get_package_for_module', 'positional', 'TimeZoneOffset', 'total_seconds', ] class Error(Exception): """Base class for protorpc exceptions.""" _TIME_ZONE_RE_STRING = r""" # Examples: # +01:00 # -05:30 # Z12:00 ((?P<z>Z) \| (?P<sign>[-+]) (?P<hours>\d\d) : (?P<minutes>\d\d))$ """ _TIME_ZONE_RE = re.compile(_TIME_ZONE_RE_STRING, re.IGNORECASE \| re.VERBOSE) def positional(max_positional_args): """A decorator that declares only the first N arguments may be positional. This decorator makes it easy to support Python 3 style keyword-only parameters. For example, in Python 3 it is possible to write: def fn(pos1, , kwonly1=None, kwonly1=None): ... All named parameters after must be a keyword: fn(10, 'kw1', 'kw2') # Raises exception. fn(10, kwonly1='kw1') # Ok. Example: To define a function like above, do: @positional(1) def fn(pos1, kwonly1=None, kwonly2=None): ... If no default value is provided to a keyword argument, it becomes a required keyword argument: @positional(0) def fn(required_kw): ... This must be called with the keyword parameter: fn() # Raises exception. fn(10) # Raises exception. fn(required_kw=10) # Ok. When defining instance or class methods always remember to account for 'self' and 'cls': class MyClass(object): @positional(2) def my_method(self, pos1, kwonly1=None): ... @classmethod @positional(2) def my_method(cls, pos1, kwonly1=None): ... One can omit the argument to 'positional' altogether, and then no arguments with default values may be passed positionally. This would be equivalent to placing a '' before the first argument with a default value in Python 3. If there are no arguments with default values, and no argument is given to 'positional', an error is raised. @positional def fn(arg1, arg2, required_kw1=None, required_kw2=0): ... fn(1, 3, 5) # Raises exception. fn(1, 3) # Ok. fn(1, 3, required_kw1=5) # Ok. Args: max_positional_arguments: Maximum number of positional arguments. All parameters after the this index must be keyword only. Returns: A decorator that prevents using arguments after max_positional_args from being used as positional parameters. Raises: TypeError if a keyword-only argument is provided as a positional parameter. ValueError if no maximum number of arguments is provided and the function has no arguments with default values. """ def positional_decorator(wrapped): """Creates a function wraper to enforce number of arguments.""" @functools.wraps(wrapped) def positional_wrapper(args, *kwargs): if len(args) > max_positional_args: plural_s = '' if max_positional_args != 1: plural_s = 's' raise TypeError('%s() takes at most %d positional argument%s ' '(%d given)' % (wrapped.__name__, max_positional_args, plural_s, len(args))) return wrapped(args, *kwargs) return positional_wrapper if isinstance(max_positional_args, six.integer_types): return positional_decorator else: args, _, _, defaults, _ = inspect.getfullargspec(max_positional_args) if defaults is None: raise ValueError( 'Functions with no keyword arguments must specify ' 'max_positional_args') return positional(len(args) - len(defaults))(max_positional_args) @positional(1) def get_package_for_module(module): """Get package name for a module. Helper calculates the package name of a module. Args: module: Module to get name for. If module is a string, try to find module in sys.modules. Returns: If module contains 'package' attribute, uses that as package name. Else, if module is not the '__main__' module, the module __name__. Else, the base name of the module file name. Else None. """ if isinstance(module, six.string_types): try: module = sys.modules[module] except KeyError: return None try: return six.text_type(module.package) except AttributeError: if module.__name__ == '__main__': try: file_name = module.__file__ except AttributeError: pass else: base_name = os.path.basename(file_name) split_name = os.path.splitext(base_name) if len(split_name) == 1: return six.text_type(base_name) return u'.'.join(split_name[:-1]) return six.text_type(module.__name__) def total_seconds(offset): """Backport of offset.total_seconds() from python 2.7+.""" seconds = offset.days * 24 * 60 * 60 + offset.seconds microseconds = seconds * 106 + offset.microseconds return microseconds / (106 * 1.0) class TimeZoneOffset(datetime.tzinfo): """Time zone information as encoded/decoded for DateTimeFields.""" def __init__(self, offset): """Initialize a time zone offset. Args: offset: Integer or timedelta time zone offset, in minutes from UTC. This can be negative. """ super(TimeZoneOffset, self).__init__() if isinstance(offset, datetime.timedelta): offset = total_seconds(offset) / 60 self.__offset = offset def utcoffset(self, _): """Get the a timedelta with the time zone's offset from UTC. Returns: The time zone offset from UTC, as a timedelta. """ return datetime.timedelta(minutes=self.__offset) def dst(self, _): """Get the daylight savings time offset. The formats that ProtoRPC uses to encode/decode time zone information don't contain any information about daylight savings time. So this always returns a timedelta of 0. Returns: A timedelta of 0. """ return datetime.timedelta(0) def decode_datetime(encoded_datetime, truncate_time=False): """Decode a DateTimeField parameter from a string to a python datetime. Args: encoded_datetime: A string in RFC 3339 format. truncate_time: If true, truncate time string with precision higher than microsecs. Returns: A datetime object with the date and time specified in encoded_datetime. Raises: ValueError: If the string is not in a recognized format. """ # Check if the string includes a time zone offset. Break out the # part that doesn't include time zone info. Convert to uppercase # because all our comparisons should be case-insensitive. time_zone_match = _TIME_ZONE_RE.search(encoded_datetime) if time_zone_match: time_string = encoded_datetime[:time_zone_match.start(1)].upper() else: time_string = encoded_datetime.upper() if '.' in time_string: format_string = '%Y-%m-%dT%H:%M:%S.%f' else: format_string = '%Y-%m-%dT%H:%M:%S' try: decoded_datetime = datetime.datetime.strptime(time_string, format_string) except ValueError: if truncate_time and '.' in time_string: datetime_string, decimal_secs = time_string.split('.') if len(decimal_secs) > 6: # datetime can handle only microsecs precision. truncated_time_string = '{}.{}'.format( datetime_string, decimal_secs[:6]) decoded_datetime = datetime.datetime.strptime( truncated_time_string, format_string) logging.warning( 'Truncating the datetime string from %s to %s', time_string, truncated_time_string) else: raise else: raise if not time_zone_match: return decoded_datetime # Time zone info was included in the parameter. Add a tzinfo # object to the datetime. Datetimes can't be changed after they're # created, so we'll need to create a new one. if time_zone_match.group('z'): offset_minutes = 0 else: sign = time_zone_match.group('sign') hours, minutes = [int(value) for value in time_zone_match.group('hours', 'minutes')] offset_minutes = hours * 60 + minutes if sign == '-': offset_minutes *= -1 return datetime.datetime(decoded_datetime.year, decoded_datetime.month, decoded_datetime.day, decoded_datetime.hour, decoded_datetime.minute, decoded_datetime.second, decoded_datetime.microsecond, TimeZoneOffset(offset_minutes))	9.98 KB	2025-03-11 18:02
📄 util_test.py #!/usr/bin/env python # # Copyright 2010 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # """Tests for apitools.base.protorpclite.util.""" import datetime import sys import types import unittest import six from apitools.base.protorpclite import test_util from apitools.base.protorpclite import util class ModuleInterfaceTest(test_util.ModuleInterfaceTest, test_util.TestCase): MODULE = util class UtilTest(test_util.TestCase): def testDecoratedFunction_LengthZero(self): @util.positional(0) def fn(kwonly=1): return [kwonly] self.assertEqual([1], fn()) self.assertEqual([2], fn(kwonly=2)) self.assertRaisesWithRegexpMatch(TypeError, r'fn\(\) takes at most 0 positional ' r'arguments \(1 given\)', fn, 1) def testDecoratedFunction_LengthOne(self): @util.positional(1) def fn(pos, kwonly=1): return [pos, kwonly] self.assertEqual([1, 1], fn(1)) self.assertEqual([2, 2], fn(2, kwonly=2)) self.assertRaisesWithRegexpMatch(TypeError, r'fn\(\) takes at most 1 positional ' r'argument \(2 given\)', fn, 2, 3) def testDecoratedFunction_LengthTwoWithDefault(self): @util.positional(2) def fn(pos1, pos2=1, kwonly=1): return [pos1, pos2, kwonly] self.assertEqual([1, 1, 1], fn(1)) self.assertEqual([2, 2, 1], fn(2, 2)) self.assertEqual([2, 3, 4], fn(2, 3, kwonly=4)) self.assertRaisesWithRegexpMatch(TypeError, r'fn\(\) takes at most 2 positional ' r'arguments \(3 given\)', fn, 2, 3, 4) def testDecoratedMethod(self): class MyClass(object): @util.positional(2) def meth(self, pos1, kwonly=1): return [pos1, kwonly] self.assertEqual([1, 1], MyClass().meth(1)) self.assertEqual([2, 2], MyClass().meth(2, kwonly=2)) self.assertRaisesWithRegexpMatch( TypeError, r'meth\(\) takes at most 2 positional arguments \(3 given\)', MyClass().meth, 2, 3) def testDefaultDecoration(self): @util.positional def fn(a, b, c=None): return a, b, c self.assertEqual((1, 2, 3), fn(1, 2, c=3)) self.assertEqual((3, 4, None), fn(3, b=4)) self.assertRaisesWithRegexpMatch(TypeError, r'fn\(\) takes at most 2 positional ' r'arguments \(3 given\)', fn, 2, 3, 4) def testDefaultDecorationNoKwdsFails(self): def fn(a): return a self.assertRaisesRegex( ValueError, ('Functions with no keyword arguments must specify ' 'max_positional_args'), util.positional, fn) def testDecoratedFunctionDocstring(self): @util.positional(0) def fn(kwonly=1): """fn docstring.""" return [kwonly] self.assertEqual('fn docstring.', fn.__doc__) class GetPackageForModuleTest(test_util.TestCase): def setUp(self): self.original_modules = dict(sys.modules) def tearDown(self): sys.modules.clear() sys.modules.update(self.original_modules) def CreateModule(self, name, file_name=None): if file_name is None: file_name = '%s.py' % name module = types.ModuleType(name) sys.modules[name] = module return module def assertPackageEquals(self, expected, actual): self.assertEqual(expected, actual) if actual is not None: self.assertTrue(isinstance(actual, six.text_type)) def testByString(self): module = self.CreateModule('service_module') module.package = 'my_package' self.assertPackageEquals('my_package', util.get_package_for_module('service_module')) def testModuleNameNotInSys(self): self.assertPackageEquals(None, util.get_package_for_module('service_module')) def testHasPackage(self): module = self.CreateModule('service_module') module.package = 'my_package' self.assertPackageEquals( 'my_package', util.get_package_for_module(module)) def testHasModuleName(self): module = self.CreateModule('service_module') self.assertPackageEquals('service_module', util.get_package_for_module(module)) def testIsMain(self): module = self.CreateModule('__main__') module.__file__ = '/bing/blam/bloom/blarm/my_file.py' self.assertPackageEquals( 'my_file', util.get_package_for_module(module)) def testIsMainCompiled(self): module = self.CreateModule('__main__') module.__file__ = '/bing/blam/bloom/blarm/my_file.pyc' self.assertPackageEquals( 'my_file', util.get_package_for_module(module)) def testNoExtension(self): module = self.CreateModule('__main__') module.__file__ = '/bing/blam/bloom/blarm/my_file' self.assertPackageEquals( 'my_file', util.get_package_for_module(module)) def testNoPackageAtAll(self): module = self.CreateModule('__main__') self.assertPackageEquals( '__main__', util.get_package_for_module(module)) class DateTimeTests(test_util.TestCase): def testDecodeDateTime(self): """Test that a RFC 3339 datetime string is decoded properly.""" for datetime_string, datetime_vals in ( ('2012-09-30T15:31:50.262', (2012, 9, 30, 15, 31, 50, 262000)), ('2012-09-30T15:31:50', (2012, 9, 30, 15, 31, 50, 0))): decoded = util.decode_datetime(datetime_string) expected = datetime.datetime(datetime_vals) self.assertEqual(expected, decoded) def testDecodeDateTimeWithTruncateTime(self): """Test that nanosec time is truncated with truncate_time flag.""" decoded = util.decode_datetime('2012-09-30T15:31:50.262343123', truncate_time=True) expected = datetime.datetime(2012, 9, 30, 15, 31, 50, 262343) self.assertEqual(expected, decoded) def testDateTimeTimeZones(self): """Test that a datetime string with a timezone is decoded correctly.""" tests = ( ('2012-09-30T15:31:50.262-06:00', (2012, 9, 30, 15, 31, 50, 262000, util.TimeZoneOffset(-360))), ('2012-09-30T15:31:50.262+01:30', (2012, 9, 30, 15, 31, 50, 262000, util.TimeZoneOffset(90))), ('2012-09-30T15:31:50+00:05', (2012, 9, 30, 15, 31, 50, 0, util.TimeZoneOffset(5))), ('2012-09-30T15:31:50+00:00', (2012, 9, 30, 15, 31, 50, 0, util.TimeZoneOffset(0))), ('2012-09-30t15:31:50-00:00', (2012, 9, 30, 15, 31, 50, 0, util.TimeZoneOffset(0))), ('2012-09-30t15:31:50z', (2012, 9, 30, 15, 31, 50, 0, util.TimeZoneOffset(0))), ('2012-09-30T15:31:50-23:00', (2012, 9, 30, 15, 31, 50, 0, util.TimeZoneOffset(-1380)))) for datetime_string, datetime_vals in tests: decoded = util.decode_datetime(datetime_string) expected = datetime.datetime(datetime_vals) self.assertEqual(expected, decoded) def testDecodeDateTimeInvalid(self): """Test that decoding malformed datetime strings raises execptions.""" for datetime_string in ('invalid', '2012-09-30T15:31:50.', '-08:00 2012-09-30T15:31:50.262', '2012-09-30T15:31', '2012-09-30T15:31Z', '2012-09-30T15:31:50ZZ', '2012-09-30T15:31:50.262 blah blah -08:00', '1000-99-99T25:99:99.999-99:99', '2012-09-30T15:31:50.262343123'): self.assertRaises( ValueError, util.decode_datetime, datetime_string) def testTimeZoneOffsetDelta(self): """Test that delta works with TimeZoneOffset.""" time_zone = util.TimeZoneOffset(datetime.timedelta(minutes=3)) epoch = time_zone.utcoffset(datetime.datetime.fromtimestamp(0, tz=datetime.timezone.utc).replace(tzinfo=None)) self.assertEqual(180, util.total_seconds(epoch)) def main(): unittest.main() if __name__ == '__main__': main()	9.23 KB	2025-03-11 18:02

Terminal