File: //proc/self/root/opt/gsutil/third_party/apitools/apitools/base/py/encoding_test.py
# -*- coding: utf-8 -*-
# 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.
import base64
import datetime
import json
import sys
import unittest
from apitools.base.protorpclite import message_types
from apitools.base.protorpclite import messages
from apitools.base.protorpclite import util
from apitools.base.py import encoding
from apitools.base.py import exceptions
from apitools.base.py import extra_types
class SimpleMessage(messages.Message):
field = messages.StringField(1)
repfield = messages.StringField(2, repeated=True)
class BytesMessage(messages.Message):
field = messages.BytesField(1)
repfield = messages.BytesField(2, repeated=True)
class TimeMessage(messages.Message):
timefield = message_types.DateTimeField(3)
@encoding.MapUnrecognizedFields('additionalProperties')
class AdditionalPropertiesMessage(messages.Message):
class AdditionalProperty(messages.Message):
key = messages.StringField(1)
value = messages.StringField(2)
additionalProperties = messages.MessageField(
'AdditionalProperty', 1, repeated=True)
@encoding.MapUnrecognizedFields('additionalProperties')
class AdditionalIntPropertiesMessage(messages.Message):
class AdditionalProperty(messages.Message):
key = messages.StringField(1)
value = messages.IntegerField(2)
additionalProperties = messages.MessageField(
'AdditionalProperty', 1, repeated=True)
@encoding.MapUnrecognizedFields('additionalProperties')
class UnrecognizedEnumMessage(messages.Message):
class ThisEnum(messages.Enum):
VALUE_ONE = 1
VALUE_TWO = 2
class AdditionalProperty(messages.Message):
key = messages.StringField(1)
value = messages.EnumField('UnrecognizedEnumMessage.ThisEnum', 2)
additionalProperties = messages.MessageField(
AdditionalProperty, 1, repeated=True)
class CompoundPropertyType(messages.Message):
index = messages.IntegerField(1)
name = messages.StringField(2)
class MessageWithEnum(messages.Message):
class ThisEnum(messages.Enum):
VALUE_ONE = 1
VALUE_TWO = 2
field_one = messages.EnumField(ThisEnum, 1)
field_two = messages.EnumField(ThisEnum, 2, default=ThisEnum.VALUE_TWO)
field_three = messages.EnumField(ThisEnum, 3, repeated=True)
ignored_field = messages.EnumField(ThisEnum, 4)
@encoding.MapUnrecognizedFields('additionalProperties')
class AdditionalMessagePropertiesMessage(messages.Message):
class AdditionalProperty(messages.Message):
key = messages.StringField(1)
value = messages.MessageField(CompoundPropertyType, 2)
additionalProperties = messages.MessageField(
'AdditionalProperty', 1, repeated=True)
@encoding.MapUnrecognizedFields('additionalProperties')
class MapToMessageWithEnum(messages.Message):
class AdditionalProperty(messages.Message):
key = messages.StringField(1)
value = messages.MessageField(MessageWithEnum, 2)
additionalProperties = messages.MessageField(
'AdditionalProperty', 1, repeated=True)
@encoding.MapUnrecognizedFields('additionalProperties')
class NestedAdditionalPropertiesWithEnumMessage(messages.Message):
class AdditionalProperty(messages.Message):
key = messages.StringField(1)
value = messages.MessageField(
MapToMessageWithEnum, 2)
additionalProperties = messages.MessageField(
'AdditionalProperty', 1, repeated=True)
@encoding.MapUnrecognizedFields('additionalProperties')
class AdditionalPropertiesWithEnumMessage(messages.Message):
class AdditionalProperty(messages.Message):
key = messages.StringField(1)
value = messages.MessageField(MessageWithEnum, 2)
additionalProperties = messages.MessageField(
'AdditionalProperty', 1, repeated=True)
class NestedMapMessage(messages.Message):
msg_field = messages.MessageField(AdditionalPropertiesWithEnumMessage, 1)
class RepeatedNestedMapMessage(messages.Message):
map_field = messages.MessageField(NestedMapMessage, 1, repeated=True)
class NestedWithEnumMessage(messages.Message):
class ThisEnum(messages.Enum):
VALUE_ONE = 1
VALUE_TWO = 2
msg_field = messages.MessageField(MessageWithEnum, 1)
enum_field = messages.EnumField(ThisEnum, 2)
class RepeatedNestedMessage(messages.Message):
msg_field = messages.MessageField(SimpleMessage, 1, repeated=True)
@encoding.MapUnrecognizedFields('additionalProperties')
class MapToBytesValue(messages.Message):
class AdditionalProperty(messages.Message):
key = messages.StringField(1)
value = messages.BytesField(2)
additionalProperties = messages.MessageField('AdditionalProperty', 1,
repeated=True)
@encoding.MapUnrecognizedFields('additionalProperties')
class MapToDateTimeValue(messages.Message):
class AdditionalProperty(messages.Message):
key = messages.StringField(1)
value = message_types.DateTimeField(2)
additionalProperties = messages.MessageField('AdditionalProperty', 1,
repeated=True)
class HasNestedMessage(messages.Message):
nested = messages.MessageField(AdditionalPropertiesMessage, 1)
nested_list = messages.StringField(2, repeated=True)
class ExtraNestedMessage(messages.Message):
nested = messages.MessageField(HasNestedMessage, 1)
class MessageWithRemappings(messages.Message):
class SomeEnum(messages.Enum):
enum_value = 1
second_value = 2
enum_field = messages.EnumField(SomeEnum, 1)
double_encoding = messages.EnumField(SomeEnum, 2)
another_field = messages.StringField(3)
repeated_enum = messages.EnumField(SomeEnum, 4, repeated=True)
repeated_field = messages.StringField(5, repeated=True)
class MessageWithPackageAndRemappings(messages.Message):
class SomeEnum(messages.Enum):
enum_value = 1
second_value = 2
enum_field = messages.EnumField(SomeEnum, 1)
another_field = messages.StringField(2)
@encoding.MapUnrecognizedFields('additionalProperties')
class RepeatedJsonValueMessage(messages.Message):
class AdditionalProperty(messages.Message):
key = messages.StringField(1)
value = messages.MessageField(extra_types.JsonValue, 2, repeated=True)
additionalProperties = messages.MessageField('AdditionalProperty', 1,
repeated=True)
encoding.AddCustomJsonEnumMapping(MessageWithRemappings.SomeEnum,
'enum_value', 'wire_name')
encoding.AddCustomJsonFieldMapping(MessageWithRemappings,
'double_encoding', 'doubleEncoding')
encoding.AddCustomJsonFieldMapping(MessageWithRemappings,
'another_field', 'anotherField')
encoding.AddCustomJsonFieldMapping(MessageWithRemappings,
'repeated_field', 'repeatedField')
class EncodingTest(unittest.TestCase):
def testCopyProtoMessage(self):
msg = SimpleMessage(field='abc')
new_msg = encoding.CopyProtoMessage(msg)
self.assertEqual(msg.field, new_msg.field)
msg.field = 'def'
self.assertNotEqual(msg.field, new_msg.field)
def testCopyProtoMessageInvalidEnum(self):
json_msg = '{"field_one": "BAD_VALUE"}'
orig_msg = encoding.JsonToMessage(MessageWithEnum, json_msg)
new_msg = encoding.CopyProtoMessage(orig_msg)
for msg in (orig_msg, new_msg):
self.assertEqual(msg.all_unrecognized_fields(), ['field_one'])
self.assertEqual(
msg.get_unrecognized_field_info('field_one',
value_default=None),
('BAD_VALUE', messages.Variant.ENUM))
def testCopyProtoMessageInvalidRepeatedEnum(self):
json_msg = '{"field_three": ["VALUE_ONE", "BAD_VALUE"]}'
orig_msg = encoding.JsonToMessage(MessageWithEnum, json_msg)
new_msg = encoding.CopyProtoMessage(orig_msg)
for msg in (orig_msg, new_msg):
self.assertEqual(msg.all_unrecognized_fields(), ['field_three'])
self.assertEqual(
msg.get_unrecognized_field_info('field_three',
value_default=None),
(['VALUE_ONE', 'BAD_VALUE'], messages.Variant.ENUM))
def testCopyProtoMessageAdditionalProperties(self):
msg = AdditionalPropertiesMessage(additionalProperties=[
AdditionalPropertiesMessage.AdditionalProperty(
key='key', value='value')])
new_msg = encoding.CopyProtoMessage(msg)
self.assertEqual(len(new_msg.additionalProperties), 1)
self.assertEqual(new_msg.additionalProperties[0].key, 'key')
self.assertEqual(new_msg.additionalProperties[0].value, 'value')
def testCopyProtoMessageMappingInvalidEnum(self):
json_msg = '{"key_one": {"field_one": "BAD_VALUE"}}'
orig_msg = encoding.JsonToMessage(MapToMessageWithEnum, json_msg)
new_msg = encoding.CopyProtoMessage(orig_msg)
for msg in (orig_msg, new_msg):
self.assertEqual(
msg.additionalProperties[0].value.all_unrecognized_fields(),
['field_one'])
self.assertEqual(
msg.additionalProperties[0].value.get_unrecognized_field_info(
'field_one', value_default=None),
('BAD_VALUE', messages.Variant.ENUM))
def testCopyProtoMessageMappingInvalidRepeatedEnum(self):
json_msg = '{"key_one": {"field_three": ["VALUE_ONE", "BAD_VALUE"]}}'
orig_msg = encoding.JsonToMessage(MapToMessageWithEnum, json_msg)
new_msg = encoding.CopyProtoMessage(orig_msg)
for msg in (orig_msg, new_msg):
self.assertEqual(
msg.additionalProperties[0].value.all_unrecognized_fields(),
['field_three'])
self.assertEqual(
msg.additionalProperties[0].value.get_unrecognized_field_info(
'field_three', value_default=None),
(['VALUE_ONE', 'BAD_VALUE'], messages.Variant.ENUM))
def testBytesEncoding(self):
b64_str = 'AAc+'
b64_msg = '{"field": "%s"}' % b64_str
urlsafe_b64_str = 'AAc-'
urlsafe_b64_msg = '{"field": "%s"}' % urlsafe_b64_str
data = base64.b64decode(b64_str)
msg = BytesMessage(field=data)
self.assertEqual(
msg, encoding.JsonToMessage(BytesMessage, urlsafe_b64_msg))
self.assertEqual(msg, encoding.JsonToMessage(BytesMessage, b64_msg))
self.assertEqual(urlsafe_b64_msg, encoding.MessageToJson(msg))
enc_rep_msg = '{"repfield": ["%(b)s", "%(b)s"]}' % {
'b': urlsafe_b64_str}
rep_msg = BytesMessage(repfield=[data, data])
self.assertEqual(
rep_msg, encoding.JsonToMessage(BytesMessage, enc_rep_msg))
self.assertEqual(enc_rep_msg, encoding.MessageToJson(rep_msg))
def testBase64RoundtripForMapFields(self):
raw_data = b'\xFF\x0F\x80'
encoded_data = '/w+A' # Has url-unsafe base64 characters
safe_encoded_data = base64.urlsafe_b64encode(raw_data).decode("utf-8")
self.assertEqual(raw_data, base64.b64decode(encoded_data))
# Use unsafe encoding, make sure we can load it.
json_data = '{"1st": "%s"}' % encoded_data
msg = encoding.JsonToMessage(MapToBytesValue, json_data)
self.assertEqual(raw_data, msg.additionalProperties[0].value)
# Now back to json and again to message
from_msg_json_data = encoding.MessageToJson(msg)
# Make sure now it is safe url encoded
self.assertEqual(safe_encoded_data,
json.loads(from_msg_json_data)['1st'])
# Make sure we can also load url safe encoded bytes.
redone_msg = encoding.JsonToMessage(MapToBytesValue,
from_msg_json_data)
# Still matches
self.assertEqual(raw_data, redone_msg.additionalProperties[0].value)
def testBytesEncodingInAMap(self):
# Leading bit is 1 should not be interpreted as unicode.
data1 = b'\xF0\x11\x0F'
data2 = b'\xFF\xFF\xFF'
msg = MapToBytesValue(
additionalProperties=[
MapToBytesValue.AdditionalProperty(key='1st', value=data1),
MapToBytesValue.AdditionalProperty(key='2nd', value=data2)
])
self.assertEqual(
'{"1st": "%s", "2nd": "%s"}' % (
base64.b64encode(data1, b'-_').decode("utf-8"),
base64.b64encode(data2, b'-_').decode("utf-8")),
encoding.MessageToJson(msg))
def testDateTimeEncodingInAMap(self):
msg = MapToDateTimeValue(
additionalProperties=[
MapToDateTimeValue.AdditionalProperty(
key='1st',
value=datetime.datetime(
2014, 7, 2, 23, 33, 25, 541000,
tzinfo=util.TimeZoneOffset(datetime.timedelta(0)))),
MapToDateTimeValue.AdditionalProperty(
key='2nd',
value=datetime.datetime(
2015, 7, 2, 23, 33, 25, 541000,
tzinfo=util.TimeZoneOffset(datetime.timedelta(0))))
])
self.assertEqual(
'{"1st": "2014-07-02T23:33:25.541000+00:00",'
' "2nd": "2015-07-02T23:33:25.541000+00:00"}',
encoding.MessageToJson(msg))
def testInvalidEnumEncodingInAMap(self):
json_msg = '{"key_one": {"field_one": "BAD_VALUE"}}'
msg = encoding.JsonToMessage(MapToMessageWithEnum, json_msg)
new_msg = encoding.MessageToJson(msg)
self.assertEqual('{"key_one": {"field_one": "BAD_VALUE"}}', new_msg)
def testInvalidRepeatedEnumEncodingInAMap(self):
json_msg = '{"key_one": {"field_three": ["VALUE_ONE", "BAD_VALUE"]}}'
msg = encoding.JsonToMessage(MapToMessageWithEnum, json_msg)
new_msg = encoding.MessageToJson(msg)
self.assertEqual(
'{"key_one": {"field_three": ["VALUE_ONE", "BAD_VALUE"]}}', new_msg)
def testIncludeFields(self):
msg = SimpleMessage()
self.assertEqual('{}', encoding.MessageToJson(msg))
self.assertEqual(
'{"field": null}',
encoding.MessageToJson(msg, include_fields=['field']))
self.assertEqual(
'{"repfield": []}',
encoding.MessageToJson(msg, include_fields=['repfield']))
def testNestedIncludeFields(self):
msg = HasNestedMessage(
nested=AdditionalPropertiesMessage(
additionalProperties=[]))
self.assertEqual(
'{"nested": null}',
encoding.MessageToJson(msg, include_fields=['nested']))
self.assertEqual(
'{"nested": {"additionalProperties": []}}',
encoding.MessageToJson(
msg, include_fields=['nested.additionalProperties']))
msg = ExtraNestedMessage(nested=msg)
self.assertEqual(
'{"nested": {"nested": null}}',
encoding.MessageToJson(msg, include_fields=['nested.nested']))
# When clearing 'nested.nested_list', its sibling ('nested.nested')
# should remain unaffected.
self.assertIn(
encoding.MessageToJson(msg, include_fields=['nested.nested_list']),
['{"nested": {"nested": {}, "nested_list": []}}',
'{"nested": {"nested_list": [], "nested": {}}}'])
self.assertEqual(
'{"nested": {"nested": {"additionalProperties": []}}}',
encoding.MessageToJson(
msg, include_fields=['nested.nested.additionalProperties']))
def testAdditionalPropertyMapping(self):
msg = AdditionalPropertiesMessage()
msg.additionalProperties = [
AdditionalPropertiesMessage.AdditionalProperty(
key='key_one', value='value_one'),
AdditionalPropertiesMessage.AdditionalProperty(
key=u'key_twð', value='value_two'),
]
encoded_msg = encoding.MessageToJson(msg)
self.assertEqual(
{'key_one': 'value_one', u'key_twð': 'value_two'},
json.loads(encoded_msg))
new_msg = encoding.JsonToMessage(type(msg), encoded_msg)
self.assertEqual(
set(('key_one', u'key_twð')),
set([x.key for x in new_msg.additionalProperties]))
self.assertIsNot(msg, new_msg)
new_msg.additionalProperties.pop()
self.assertEqual(1, len(new_msg.additionalProperties))
self.assertEqual(2, len(msg.additionalProperties))
def testNumericPropertyName(self):
json_msg = '{"nested": {"123": "def"}}'
msg = encoding.JsonToMessage(HasNestedMessage, json_msg)
self.assertEqual(1, len(msg.nested.additionalProperties))
def testNumericPropertyValue(self):
json_msg = '{"key_one": "123"}'
msg = encoding.JsonToMessage(AdditionalIntPropertiesMessage, json_msg)
self.assertEqual(
AdditionalIntPropertiesMessage(
additionalProperties=[
AdditionalIntPropertiesMessage.AdditionalProperty(
key='key_one', value=123)]),
msg)
def testAdditionalMessageProperties(self):
json_msg = '{"input": {"index": 0, "name": "output"}}'
result = encoding.JsonToMessage(
AdditionalMessagePropertiesMessage, json_msg)
self.assertEqual(1, len(result.additionalProperties))
self.assertEqual(0, result.additionalProperties[0].value.index)
def testUnrecognizedEnum(self):
json_msg = '{"input": "VALUE_ONE"}'
result = encoding.JsonToMessage(
UnrecognizedEnumMessage, json_msg)
self.assertEqual(1, len(result.additionalProperties))
self.assertEqual(UnrecognizedEnumMessage.ThisEnum.VALUE_ONE,
result.additionalProperties[0].value)
def testNestedFieldMapping(self):
nested_msg = AdditionalPropertiesMessage()
nested_msg.additionalProperties = [
AdditionalPropertiesMessage.AdditionalProperty(
key='key_one', value='value_one'),
AdditionalPropertiesMessage.AdditionalProperty(
key='key_two', value='value_two'),
]
msg = HasNestedMessage(nested=nested_msg)
encoded_msg = encoding.MessageToJson(msg)
self.assertEqual(
{'nested': {'key_one': 'value_one', 'key_two': 'value_two'}},
json.loads(encoded_msg))
new_msg = encoding.JsonToMessage(type(msg), encoded_msg)
self.assertEqual(
set(('key_one', 'key_two')),
set([x.key for x in new_msg.nested.additionalProperties]))
new_msg.nested.additionalProperties.pop()
self.assertEqual(1, len(new_msg.nested.additionalProperties))
self.assertEqual(2, len(msg.nested.additionalProperties))
def testValidEnums(self):
message_json = '{"field_one": "VALUE_ONE"}'
message = encoding.JsonToMessage(MessageWithEnum, message_json)
self.assertEqual(MessageWithEnum.ThisEnum.VALUE_ONE, message.field_one)
self.assertEqual(MessageWithEnum.ThisEnum.VALUE_TWO, message.field_two)
self.assertEqual(json.loads(message_json),
json.loads(encoding.MessageToJson(message)))
def testIgnoredEnums(self):
json_with_typo = '{"field_one": "VALUE_OEN"}'
message = encoding.JsonToMessage(MessageWithEnum, json_with_typo)
self.assertEqual(None, message.field_one)
self.assertEqual(('VALUE_OEN', messages.Variant.ENUM),
message.get_unrecognized_field_info('field_one'))
self.assertEqual(json.loads(json_with_typo),
json.loads(encoding.MessageToJson(message)))
empty_json = ''
message = encoding.JsonToMessage(MessageWithEnum, empty_json)
self.assertEqual(None, message.field_one)
def testIgnoredEnumsWithDefaults(self):
json_with_typo = '{"field_two": "VALUE_OEN"}'
message = encoding.JsonToMessage(MessageWithEnum, json_with_typo)
self.assertEqual(MessageWithEnum.ThisEnum.VALUE_TWO, message.field_two)
self.assertEqual(json.loads(json_with_typo),
json.loads(encoding.MessageToJson(message)))
def testUnknownNestedRoundtrip(self):
json_message = '{"field": "abc", "submessage": {"a": 1, "b": "foo"}}'
message = encoding.JsonToMessage(SimpleMessage, json_message)
self.assertEqual(json.loads(json_message),
json.loads(encoding.MessageToJson(message)))
def testUnknownEnumNestedRoundtrip(self):
json_with_typo = ('{"outer_key": {"key_one": {"field_one": '
'"VALUE_OEN", "field_two": "VALUE_OEN", '
'"field_three": ["VALUE_ONE", "BAD_VALUE"]}}}')
msg = encoding.JsonToMessage(NestedAdditionalPropertiesWithEnumMessage,
json_with_typo)
self.assertEqual(json.loads(json_with_typo),
json.loads(encoding.MessageToJson(msg)))
def testJsonDatetime(self):
msg = TimeMessage(timefield=datetime.datetime(
2014, 7, 2, 23, 33, 25, 541000,
tzinfo=util.TimeZoneOffset(datetime.timedelta(0))))
self.assertEqual(
'{"timefield": "2014-07-02T23:33:25.541000+00:00"}',
encoding.MessageToJson(msg))
def testEnumRemapping(self):
msg = MessageWithRemappings(
enum_field=MessageWithRemappings.SomeEnum.enum_value)
json_message = encoding.MessageToJson(msg)
self.assertEqual('{"enum_field": "wire_name"}', json_message)
self.assertEqual(
msg, encoding.JsonToMessage(MessageWithRemappings, json_message))
def testRepeatedEnumRemapping(self):
msg = MessageWithRemappings(
repeated_enum=[
MessageWithRemappings.SomeEnum.enum_value,
MessageWithRemappings.SomeEnum.second_value,
])
json_message = encoding.MessageToJson(msg)
self.assertEqual('{"repeated_enum": ["wire_name", "second_value"]}',
json_message)
self.assertEqual(
msg, encoding.JsonToMessage(MessageWithRemappings, json_message))
def testFieldRemapping(self):
msg = MessageWithRemappings(another_field='abc')
json_message = encoding.MessageToJson(msg)
self.assertEqual('{"anotherField": "abc"}', json_message)
self.assertEqual(
msg, encoding.JsonToMessage(MessageWithRemappings, json_message))
def testFieldRemappingWithPackage(self):
this_module = sys.modules[__name__]
package_name = 'my_package'
try:
setattr(this_module, 'package', package_name)
encoding.AddCustomJsonFieldMapping(
MessageWithPackageAndRemappings,
'another_field', 'wire_field_name', package=package_name)
msg = MessageWithPackageAndRemappings(another_field='my value')
json_message = encoding.MessageToJson(msg)
self.assertEqual('{"wire_field_name": "my value"}', json_message)
self.assertEqual(
msg,
encoding.JsonToMessage(MessageWithPackageAndRemappings,
json_message))
finally:
delattr(this_module, 'package')
def testEnumRemappingWithPackage(self):
this_module = sys.modules[__name__]
package_name = 'my_package'
try:
setattr(this_module, 'package', package_name)
encoding.AddCustomJsonEnumMapping(
MessageWithPackageAndRemappings.SomeEnum,
'enum_value', 'other_wire_name', package=package_name)
msg = MessageWithPackageAndRemappings(
enum_field=MessageWithPackageAndRemappings.SomeEnum.enum_value)
json_message = encoding.MessageToJson(msg)
self.assertEqual('{"enum_field": "other_wire_name"}', json_message)
self.assertEqual(
msg,
encoding.JsonToMessage(MessageWithPackageAndRemappings,
json_message))
finally:
delattr(this_module, 'package')
def testRepeatedFieldRemapping(self):
msg = MessageWithRemappings(repeated_field=['abc', 'def'])
json_message = encoding.MessageToJson(msg)
self.assertEqual('{"repeatedField": ["abc", "def"]}', json_message)
self.assertEqual(
msg, encoding.JsonToMessage(MessageWithRemappings, json_message))
def testMultipleRemapping(self):
msg = MessageWithRemappings(
double_encoding=MessageWithRemappings.SomeEnum.enum_value)
json_message = encoding.MessageToJson(msg)
self.assertEqual('{"doubleEncoding": "wire_name"}', json_message)
self.assertEqual(
msg, encoding.JsonToMessage(MessageWithRemappings, json_message))
def testRepeatedRemapping(self):
# Should allow remapping if the mapping remains the same.
encoding.AddCustomJsonEnumMapping(MessageWithRemappings.SomeEnum,
'enum_value', 'wire_name')
encoding.AddCustomJsonFieldMapping(MessageWithRemappings,
'double_encoding', 'doubleEncoding')
encoding.AddCustomJsonFieldMapping(MessageWithRemappings,
'another_field', 'anotherField')
encoding.AddCustomJsonFieldMapping(MessageWithRemappings,
'repeated_field', 'repeatedField')
# Should raise errors if the remapping changes the mapping.
self.assertRaises(
exceptions.InvalidDataError,
encoding.AddCustomJsonFieldMapping,
MessageWithRemappings, 'double_encoding', 'something_else')
self.assertRaises(
exceptions.InvalidDataError,
encoding.AddCustomJsonFieldMapping,
MessageWithRemappings, 'enum_field', 'anotherField')
self.assertRaises(
exceptions.InvalidDataError,
encoding.AddCustomJsonEnumMapping,
MessageWithRemappings.SomeEnum, 'enum_value', 'another_name')
self.assertRaises(
exceptions.InvalidDataError,
encoding.AddCustomJsonEnumMapping,
MessageWithRemappings.SomeEnum, 'second_value', 'wire_name')
def testMessageToRepr(self):
# Using the same string returned by MessageToRepr, with the
# module names fixed.
# pylint: disable=bad-whitespace
msg = SimpleMessage(field='field', repfield=['field', 'field', ],)
# pylint: enable=bad-whitespace
self.assertEqual(
encoding.MessageToRepr(msg),
r"%s.SimpleMessage(field='field',repfield=['field','field',],)" % (
__name__,))
self.assertEqual(
encoding.MessageToRepr(msg, no_modules=True),
r"SimpleMessage(field='field',repfield=['field','field',],)")
def testMessageToReprWithTime(self):
msg = TimeMessage(timefield=datetime.datetime(
2014, 7, 2, 23, 33, 25, 541000,
tzinfo=util.TimeZoneOffset(datetime.timedelta(0))))
self.assertEqual(
encoding.MessageToRepr(msg, multiline=True),
('%s.TimeMessage(\n '
'timefield=datetime.datetime(2014, 7, 2, 23, 33, 25, 541000, '
'tzinfo=apitools.base.protorpclite.util.TimeZoneOffset('
'datetime.timedelta(0))),\n)') % __name__)
self.assertEqual(
encoding.MessageToRepr(msg, multiline=True, no_modules=True),
'TimeMessage(\n '
'timefield=datetime.datetime(2014, 7, 2, 23, 33, 25, 541000, '
'tzinfo=TimeZoneOffset(datetime.timedelta(0))),\n)')
def testRepeatedJsonValuesAsRepeatedProperty(self):
encoded_msg = '{"a": [{"one": 1}]}'
msg = encoding.JsonToMessage(RepeatedJsonValueMessage, encoded_msg)
self.assertEqual(encoded_msg, encoding.MessageToJson(msg))
def testDictToAdditionalPropertyMessage(self):
dict_ = {'key': 'value'}
encoded_msg = encoding.DictToAdditionalPropertyMessage(
dict_, AdditionalPropertiesMessage)
expected_msg = AdditionalPropertiesMessage()
expected_msg.additionalProperties = [
AdditionalPropertiesMessage.AdditionalProperty(
key='key', value='value')
]
self.assertEqual(encoded_msg, expected_msg)
def testDictToAdditionalPropertyMessageSorted(self):
tuples = [('key{0:02}'.format(i), 'value') for i in range(100)]
dict_ = dict(tuples)
encoded_msg = encoding.DictToAdditionalPropertyMessage(
dict_, AdditionalPropertiesMessage, sort_items=True)
expected_msg = AdditionalPropertiesMessage()
expected_msg.additionalProperties = [
AdditionalPropertiesMessage.AdditionalProperty(
key=key, value=value)
for key, value in tuples
]
self.assertEqual(encoded_msg, expected_msg)
def testDictToAdditionalPropertyMessageNumeric(self):
dict_ = {'key': 1}
encoded_msg = encoding.DictToAdditionalPropertyMessage(
dict_, AdditionalIntPropertiesMessage)
expected_msg = AdditionalIntPropertiesMessage()
expected_msg.additionalProperties = [
AdditionalIntPropertiesMessage.AdditionalProperty(
key='key', value=1)
]
self.assertEqual(encoded_msg, expected_msg)
def testUnrecognizedFieldIter(self):
m = encoding.DictToMessage({
'nested': {
'nested': {'a': 'b'},
'nested_list': ['foo'],
'extra_field': 'foo',
}
}, ExtraNestedMessage)
results = list(encoding.UnrecognizedFieldIter(m))
self.assertEqual(1, len(results))
edges, fields = results[0]
expected_edge = encoding.ProtoEdge(
encoding.EdgeType.SCALAR, 'nested', None)
self.assertEqual((expected_edge,), edges)
self.assertEqual(['extra_field'], fields)
def testUnrecognizedFieldIterRepeated(self):
m = encoding.DictToMessage({
'msg_field': [
{'field': 'foo'},
{'not_a_field': 'bar'}
]
}, RepeatedNestedMessage)
results = list(encoding.UnrecognizedFieldIter(m))
self.assertEqual(1, len(results))
edges, fields = results[0]
expected_edge = encoding.ProtoEdge(
encoding.EdgeType.REPEATED, 'msg_field', 1)
self.assertEqual((expected_edge,), edges)
self.assertEqual(['not_a_field'], fields)
def testUnrecognizedFieldIterNestedMap(self):
m = encoding.DictToMessage({
'map_field': [{
'msg_field': {
'foo': {'field_one': 1},
'bar': {'not_a_field': 1},
}
}]
}, RepeatedNestedMapMessage)
results = list(encoding.UnrecognizedFieldIter(m))
self.assertEqual(1, len(results))
edges, fields = results[0]
expected_edges = (
encoding.ProtoEdge(encoding.EdgeType.REPEATED, 'map_field', 0),
encoding.ProtoEdge(encoding.EdgeType.MAP, 'msg_field', 'bar'),
)
self.assertEqual(expected_edges, edges)
self.assertEqual(['not_a_field'], fields)
def testUnrecognizedFieldIterAbortAfterFirstError(self):
m = encoding.DictToMessage({
'msg_field': {'field_one': 3},
'enum_field': 3,
}, NestedWithEnumMessage)
self.assertEqual(1, len(list(encoding.UnrecognizedFieldIter(m))))