| #!/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 protorpc.messages.""" |
| import six |
| |
| __author__ = 'rafek@google.com (Rafe Kaplan)' |
| |
| |
| import pickle |
| import re |
| import sys |
| import types |
| import unittest |
| |
| from protorpc import descriptor |
| from protorpc import message_types |
| from protorpc import messages |
| from protorpc import test_util |
| |
| |
| 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.assertEquals('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.assertEquals('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 |
| 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.assertEquals( |
| set(['BLUE', 'GREEN', 'INDIGO', 'ORANGE', 'RED', 'VIOLET', 'YELLOW']), |
| set(Color.names())) |
| |
| def testNumbers(self): |
| """Tests that numbers iterates of enum numbers.""" |
| self.assertEquals(set([2, 4, 5, 20, 40, 50, 80]), set(Color.numbers())) |
| |
| def testIterate(self): |
| """Test that __iter__ iterates over all enum values.""" |
| self.assertEquals(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.assertEquals([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.assertEquals(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.assertEquals(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.assertEquals(Color.RED, Color('RED')) |
| self.assertEquals(Color.RED, Color(u'RED')) |
| self.assertEquals(Color.RED, Color(20)) |
| if six.PY2: |
| self.assertEquals(Color.RED, Color(long(20))) |
| self.assertEquals(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.assertEquals(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.assertEquals('Wednesday', WeekDay(3).name) |
| self.assertEquals(6, WeekDay('Friday').number) |
| self.assertEquals(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.assertEquals(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.assertEquals('RED', str(Color.RED)) |
| self.assertEquals('ORANGE', str(Color.ORANGE)) |
| |
| def testInt(self): |
| """Test converting to int.""" |
| self.assertEquals(20, int(Color.RED)) |
| self.assertEquals(2, int(Color.ORANGE)) |
| |
| def testRepr(self): |
| """Test enum representation.""" |
| self.assertEquals('Color(RED, 20)', repr(Color.RED)) |
| self.assertEquals('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.assertEquals('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.assertEquals('%s.Color' % module_name, Color.definition_name()) |
| self.assertEquals(module_name, Color.outer_definition_name()) |
| self.assertEquals(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.assertEquals('my.package.Color', Color.definition_name()) |
| self.assertEquals('my.package', Color.outer_definition_name()) |
| self.assertEquals('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.assertEquals('Enum1', Enum1.definition_name()) |
| self.assertEquals(None, Enum1.outer_definition_name()) |
| self.assertEquals(None, Enum1.definition_package()) |
| self.assertEquals(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.assertEquals('%s.MyMessage.NestedEnum' % module_name, |
| MyMessage.NestedEnum.definition_name()) |
| self.assertEquals('%s.MyMessage' % module_name, |
| MyMessage.NestedEnum.outer_definition_name()) |
| self.assertEquals(module_name, |
| MyMessage.NestedEnum.definition_package()) |
| |
| self.assertEquals('%s.MyMessage.NestedMessage.NestedEnum' % module_name, |
| MyMessage.NestedMessage.NestedEnum.definition_name()) |
| self.assertEquals( |
| '%s.MyMessage.NestedMessage' % module_name, |
| MyMessage.NestedMessage.NestedEnum.outer_definition_name()) |
| self.assertEquals(module_name, |
| MyMessage.NestedMessage.NestedEnum.definition_package()) |
| |
| def testMessageDefinition(self): |
| """Test that enumeration knows its enclosing message definition.""" |
| class OuterEnum(messages.Enum): |
| pass |
| |
| self.assertEquals(None, OuterEnum.message_definition()) |
| |
| class OuterMessage(messages.Message): |
| |
| class InnerEnum(messages.Enum): |
| pass |
| |
| self.assertEquals(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.assertEquals(Enum1.VAL1, Enum1.VAL1) |
| self.assertNotEquals(Enum1.VAL1, Enum1.VAL2) |
| self.assertNotEquals(Enum1.VAL1, Enum2.VAL1) |
| self.assertNotEquals(Enum1.VAL1, 'VAL1') |
| self.assertNotEquals(Enum1.VAL1, 1) |
| self.assertNotEquals(Enum1.VAL1, 2) |
| self.assertNotEquals(Enum1.VAL1, None) |
| self.assertNotEquals(Enum1.VAL1, Enum2.VAL1) |
| |
| self.assertTrue(Enum1.VAL1 < Enum1.VAL2) |
| self.assertTrue(Enum1.VAL2 > Enum1.VAL1) |
| |
| self.assertNotEquals(1, Enum2.VAL1) |
| |
| def testPickle(self): |
| """Testing pickling and unpickling of Enum instances.""" |
| colors = list(Color) |
| unpickled = pickle.loads(pickle.dumps(colors)) |
| self.assertEquals(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.assertEquals([1, 2, 3], |
| messages.FieldList(self.integer_field, [1, 2, 3])) |
| self.assertEquals([1, 2, 3], |
| messages.FieldList(self.integer_field, (1, 2, 3))) |
| self.assertEquals([], 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.assertEquals([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.assertEquals([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.assertEquals([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.assertEquals([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.assertEquals([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 disconnected FieldList instances.""" |
| field_list = messages.FieldList(self.integer_field, [1, 2, 3, 4, 5]) |
| unpickled = pickle.loads(pickle.dumps(field_list)) |
| self.assertEquals(field_list, unpickled) |
| self.assertIsInstance(unpickled.field, messages.IntegerField) |
| self.assertEquals(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. |
| """ |
| for field_class in (messages.IntegerField, |
| messages.FloatField, |
| messages.BooleanField, |
| messages.BytesField, |
| messages.StringField, |
| ): |
| 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.assertEquals(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.assertEquals(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-ASCII string .*: " |
| r"'ascii' codec can't decode byte 0x89 in position 0: " |
| r"ordinal not in range", |
| messages.StringField, 1, default=b'\x89') |
| |
| 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.assertEquals(Symbol.ALPHA, field.default) |
| |
| def testDefaultFields_EnumStringDelayedResolution(self): |
| """Test that enum fields resolve default strings.""" |
| field = messages.EnumField('protorpc.descriptor.FieldDescriptor.Label', |
| 1, |
| default='OPTIONAL') |
| |
| self.assertEquals(descriptor.FieldDescriptor.Label.OPTIONAL, field.default) |
| |
| def testDefaultFields_EnumIntDelayedResolution(self): |
| """Test that enum fields resolve default integers.""" |
| field = messages.EnumField('protorpc.descriptor.FieldDescriptor.Label', |
| 1, |
| default=2) |
| |
| self.assertEquals(descriptor.FieldDescriptor.Label.REQUIRED, field.default) |
| |
| def testDefaultFields_EnumOkIfTypeKnown(self): |
| """Test that enum fields accept valid default values when type is known.""" |
| field = messages.EnumField(descriptor.FieldDescriptor.Label, |
| 1, |
| default='REPEATED') |
| |
| self.assertEquals(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('protorpc.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: 1, |
| 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, |
| messages.FloatField: 1.5, |
| messages.BooleanField: False, |
| messages.BytesField: 'abc', |
| messages.StringField: u'abc', |
| } |
| def action(field_class): |
| # Optional. |
| field = field_class(1) |
| field.validate_element(values[field_class]) |
| |
| # Required. |
| field = field_class(1, required=True) |
| field.validate_element(values[field_class]) |
| |
| # Repeated. |
| field = field_class(1, repeated=True) |
| self.assertRaises(message.VAlidationError, |
| field.validate_element, |
| []) |
| self.assertRaises(message.VAlidationError, |
| field.validate_element, |
| ()) |
| field.validate_element(values[field_class]) |
| field.validate_element(values[field_class]) |
| |
| # Right value, but repeated. |
| self.assertRaises(messages.ValidationError, |
| field.validate_element, |
| [values[field_class]]) |
| self.assertRaises(messages.ValidationError, |
| field.validate_element, |
| (values[field_class],)) |
| |
| 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.assertEquals(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.assertEquals(MyMessage, |
| MyMessage.field_by_name('self_reference').type) |
| |
| self.assertEquals(ForwardMessage, |
| MyMessage.field_by_name('forward').type) |
| |
| self.assertEquals(ForwardMessage.NestedMessage, |
| MyMessage.field_by_name('nested').type) |
| |
| self.assertEquals(MyMessage.Inner, |
| MyMessage.field_by_name('inner').type) |
| |
| self.assertEquals(MyMessage.Sibling, |
| MyMessage.Inner.field_by_name('sibling').type) |
| finally: |
| try: |
| del MyMessage |
| del ForwardMessage |
| 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.assertEquals(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.assertEquals(ForwardEnum, |
| MyMessage.field_by_name('forward').type) |
| |
| self.assertEquals(ForwardMessage.NestedEnum, |
| MyMessage.field_by_name('nested').type) |
| |
| self.assertEquals(MyMessage.Inner, |
| MyMessage.field_by_name('inner').type) |
| finally: |
| try: |
| del MyMessage |
| del ForwardEnum |
| del ForwardMessage |
| 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.assertEquals(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.assertEquals(m1, m2) |
| |
| def testNonAsciiStr(self): |
| """Test validation fails for non-ascii StringField values.""" |
| class Thing(messages.Message): |
| string_field = messages.StringField(2) |
| |
| thing = Thing() |
| self.assertRaisesWithRegexpMatch( |
| messages.ValidationError, |
| 'Field string_field encountered non-ASCII 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.assertEquals(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.assertEquals(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 that attribute assignment on Message classes are 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.assertEquals('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.assertEquals(3, len(fields)) |
| self.assertEquals('a3', fields[0].name) |
| self.assertEquals('b1', fields[1].name) |
| self.assertEquals('c2', fields[2].name) |
| |
| def testFieldByName(self): |
| """Test getting field by name.""" |
| ComplexMessage = self.CreateMessageClass() |
| |
| self.assertEquals(3, ComplexMessage.field_by_name('a3').number) |
| self.assertEquals(1, ComplexMessage.field_by_name('b1').number) |
| self.assertEquals(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.assertEquals('a3', ComplexMessage.field_by_number(3).name) |
| self.assertEquals('b1', ComplexMessage.field_by_number(1).name) |
| self.assertEquals('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.assertEquals(None, message.get_assigned_value('a_value')) |
| |
| message.a_value = u'a string' |
| self.assertEquals(u'a string', message.get_assigned_value('a_value')) |
| |
| message.a_value = u'a default' |
| self.assertEquals(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.assertEquals(u'a default', message.a_value) |
| message.reset('a_value') |
| self.assertEquals(u'a default', message.a_value) |
| |
| message.a_value = u'a new value' |
| self.assertEquals(u'a new value', message.a_value) |
| message.reset('a_value') |
| self.assertEquals(u'a default', message.a_value) |
| |
| message.repeated = [1, 2, 3] |
| self.assertEquals([1, 2, 3], message.repeated) |
| saved = message.repeated |
| message.reset('repeated') |
| self.assertEquals([], message.repeated) |
| self.assertIsInstance(message.repeated, messages.FieldList) |
| self.assertEquals([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.assertEquals(['Currency', 'Duration'], Trade.__enums__) |
| |
| # Message definition will now be set on Enumerated objects. |
| self.assertEquals(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.assertEquals(['Agent', 'Lot'], Trade.__messages__) |
| self.assertEquals(Trade, Trade.Agent.message_definition()) |
| self.assertEquals(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.assertNotEquals('hi', message1) |
| self.assertNotEquals(AnotherMessage(), message1) |
| self.assertEquals(message1, message1) |
| |
| message2 = MyMessage() |
| |
| self.assertEquals(message1, message2) |
| |
| message1.field1 = 10 |
| self.assertNotEquals(message1, message2) |
| |
| message2.field1 = 20 |
| self.assertNotEquals(message1, message2) |
| |
| message2.field1 = 10 |
| self.assertEquals(message1, message2) |
| |
| message1.field2 = MyEnum.val1 |
| self.assertNotEquals(message1, message2) |
| |
| message2.field2 = MyEnum.val2 |
| self.assertNotEquals(message1, message2) |
| |
| message2.field2 = MyEnum.val1 |
| self.assertEquals(message1, message2) |
| |
| message1.field3 = AnotherMessage() |
| message1.field3.string = 'value1' |
| self.assertNotEquals(message1, message2) |
| |
| message2.field3 = AnotherMessage() |
| message2.field3.string = 'value2' |
| self.assertNotEquals(message1, message2) |
| |
| message2.field3.string = 'value1' |
| self.assertEquals(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.assertEquals(message1, message2) |
| message1.set_unrecognized_field('unknown1', 'value1', |
| messages.Variant.STRING) |
| self.assertEquals(message1, message2) |
| |
| message1.set_unrecognized_field('unknown2', ['asdf', 3], |
| messages.Variant.STRING) |
| message1.set_unrecognized_field('unknown3', 4.7, |
| messages.Variant.DOUBLE) |
| self.assertEquals(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.assertEquals('%s.MyMessage' % module_name, |
| MyMessage.definition_name()) |
| self.assertEquals(module_name, MyMessage.outer_definition_name()) |
| self.assertEquals(module_name, MyMessage.definition_package()) |
| |
| self.assertEquals(six.text_type, type(MyMessage.definition_name())) |
| self.assertEquals(six.text_type, type(MyMessage.outer_definition_name())) |
| self.assertEquals(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.assertEquals('my.package.MyMessage', MyMessage.definition_name()) |
| self.assertEquals('my.package', MyMessage.outer_definition_name()) |
| self.assertEquals('my.package', MyMessage.definition_package()) |
| |
| self.assertEquals(six.text_type, type(MyMessage.definition_name())) |
| self.assertEquals(six.text_type, type(MyMessage.outer_definition_name())) |
| self.assertEquals(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.assertEquals('MyMessage', MyMessage.definition_name()) |
| self.assertEquals(None, MyMessage.outer_definition_name()) |
| self.assertEquals(None, MyMessage.definition_package()) |
| |
| self.assertEquals(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.assertEquals('%s.MyMessage.NestedMessage' % module_name, |
| MyMessage.NestedMessage.definition_name()) |
| self.assertEquals('%s.MyMessage' % module_name, |
| MyMessage.NestedMessage.outer_definition_name()) |
| self.assertEquals(module_name, |
| MyMessage.NestedMessage.definition_package()) |
| |
| self.assertEquals('%s.MyMessage.NestedMessage.NestedMessage' % module_name, |
| MyMessage.NestedMessage.NestedMessage.definition_name()) |
| self.assertEquals( |
| '%s.MyMessage.NestedMessage' % module_name, |
| MyMessage.NestedMessage.NestedMessage.outer_definition_name()) |
| self.assertEquals( |
| 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.assertEquals(None, OuterMessage.message_definition()) |
| self.assertEquals(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.assertEquals(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.assertEquals([], 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.assertEquals(message1, message2) |
| |
| def testUnknownValues(self): |
| """Test message class equality with unknown fields.""" |
| class MyMessage(messages.Message): |
| field1 = messages.IntegerField(1) |
| |
| message = MyMessage() |
| self.assertEquals([], message.all_unrecognized_fields()) |
| self.assertEquals((None, None), |
| message.get_unrecognized_field_info('doesntexist')) |
| self.assertEquals((None, None), |
| message.get_unrecognized_field_info( |
| 'doesntexist', None, None)) |
| self.assertEquals(('defaultvalue', 'defaultwire'), |
| message.get_unrecognized_field_info( |
| 'doesntexist', 'defaultvalue', 'defaultwire')) |
| self.assertEquals((3, None), |
| message.get_unrecognized_field_info( |
| 'doesntexist', value_default=3)) |
| |
| message.set_unrecognized_field('exists', 9.5, messages.Variant.DOUBLE) |
| self.assertEquals(1, len(message.all_unrecognized_fields())) |
| self.assertTrue('exists' in message.all_unrecognized_fields()) |
| self.assertEquals((9.5, messages.Variant.DOUBLE), |
| message.get_unrecognized_field_info('exists')) |
| self.assertEquals((9.5, messages.Variant.DOUBLE), |
| message.get_unrecognized_field_info('exists', 'type', |
| 1234)) |
| self.assertEquals((1234, None), |
| message.get_unrecognized_field_info('doesntexist', 1234)) |
| |
| message.set_unrecognized_field('another', 'value', messages.Variant.STRING) |
| self.assertEquals(2, len(message.all_unrecognized_fields())) |
| self.assertTrue('exists' in message.all_unrecognized_fields()) |
| self.assertTrue('another' in message.all_unrecognized_fields()) |
| self.assertEquals((9.5, messages.Variant.DOUBLE), |
| message.get_unrecognized_field_info('exists')) |
| self.assertEquals(('value', messages.Variant.STRING), |
| message.get_unrecognized_field_info('another')) |
| |
| message.set_unrecognized_field('typetest1', ['list', 0, ('test',)], |
| messages.Variant.STRING) |
| self.assertEquals((['list', 0, ('test',)], messages.Variant.STRING), |
| message.get_unrecognized_field_info('typetest1')) |
| message.set_unrecognized_field('typetest2', '', messages.Variant.STRING) |
| self.assertEquals(('', 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.assertEquals(message, unpickled) |
| self.assertTrue(AnotherMessage.string is unpickled.field3.string.field) |
| self.assertTrue('exists' in message.all_unrecognized_fields()) |
| self.assertEquals(('value', messages.Variant.STRING), |
| message.get_unrecognized_field_info('exists')) |
| self.assertEquals((['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={}, 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. |
| """ |
| # 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 |
| |
| 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() |
| |
| 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.assertEquals(A, messages.find_definition('a.b.c.A', |
| importer=self.Importer)) |
| B = self.DefineMessage('a.b.c', 'B', {'C':{}}) |
| self.assertEquals(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.assertEquals(A, messages.find_definition( |
| 'A', a, importer=self.Importer)) |
| self.assertEquals(B, messages.find_definition( |
| 'b.B', a, importer=self.Importer)) |
| self.assertEquals(C, messages.find_definition( |
| 'b.c.C', a, importer=self.Importer)) |
| self.assertEquals(D, messages.find_definition( |
| 'b.d.D', a, importer=self.Importer)) |
| |
| # Find A, B, C and D relative to b. |
| self.assertEquals(A, messages.find_definition( |
| 'A', b, importer=self.Importer)) |
| self.assertEquals(B, messages.find_definition( |
| 'B', b, importer=self.Importer)) |
| self.assertEquals(C, messages.find_definition( |
| 'c.C', b, importer=self.Importer)) |
| self.assertEquals(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.assertEquals(A, messages.find_definition( |
| 'A', c, importer=self.Importer)) |
| self.assertEquals(B, messages.find_definition( |
| 'B', c, importer=self.Importer)) |
| self.assertEquals(C, messages.find_definition( |
| 'C', c, importer=self.Importer)) |
| self.assertEquals(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.assertEquals(A, messages.find_definition( |
| 'A', A, importer=self.Importer)) |
| self.assertEquals(B, messages.find_definition( |
| 'B', A, importer=self.Importer)) |
| self.assertEquals(C, messages.find_definition( |
| 'B.C', A, importer=self.Importer)) |
| self.assertEquals(D, messages.find_definition( |
| 'B.D', A, importer=self.Importer)) |
| |
| # Find relative to B. |
| self.assertEquals(A, messages.find_definition( |
| 'A', B, importer=self.Importer)) |
| self.assertEquals(B, messages.find_definition( |
| 'B', B, importer=self.Importer)) |
| self.assertEquals(C, messages.find_definition( |
| 'C', B, importer=self.Importer)) |
| self.assertEquals(D, messages.find_definition( |
| 'D', B, importer=self.Importer)) |
| |
| # Find relative to C. |
| self.assertEquals(A, messages.find_definition( |
| 'A', C, importer=self.Importer)) |
| self.assertEquals(B, messages.find_definition( |
| 'B', C, importer=self.Importer)) |
| self.assertEquals(C, messages.find_definition( |
| 'C', C, importer=self.Importer)) |
| self.assertEquals(D, messages.find_definition( |
| 'D', C, importer=self.Importer)) |
| |
| # Find relative to C searching from c. |
| self.assertEquals(A, messages.find_definition( |
| 'b.A', C, importer=self.Importer)) |
| self.assertEquals(B, messages.find_definition( |
| 'b.A.B', C, importer=self.Importer)) |
| self.assertEquals(C, messages.find_definition( |
| 'b.A.B.C', C, importer=self.Importer)) |
| self.assertEquals(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.assertEquals(aA, messages.find_definition('.a.A', None, |
| importer=self.Importer)) |
| self.assertEquals(aA, messages.find_definition('.a.A', a, |
| importer=self.Importer)) |
| self.assertEquals(aA, messages.find_definition('.a.A', aA, |
| importer=self.Importer)) |
| self.assertEquals(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.assertEquals( |
| Color, |
| messages.find_definition('Color', A, importer=self.Importer)) |
| |
| def testFalseScope(self): |
| """Test that 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.assertEquals(A, messages.find_definition( |
| 'a.A', None, importer=self.Importer)) |
| |
| |
| class FindDefinitionUnicodeTests(test_util.TestCase): |
| |
| # TODO(craigcitro): Fix this test and re-enable it. |
| def notatestUnicodeString(self): |
| """Test using unicode names.""" |
| from protorpc import registry |
| self.assertEquals('ServiceMapping', |
| messages.find_definition( |
| u'protorpc.registry.ServiceMapping', |
| None).__name__) |
| |
| |
| def main(): |
| unittest.main() |
| |
| |
| if __name__ == '__main__': |
| main() |
