Index: tools/telemetry/third_party/gsutil/third_party/apitools/apitools/base/py/encoding_test.py |
diff --git a/tools/telemetry/third_party/gsutil/third_party/apitools/apitools/base/py/encoding_test.py b/tools/telemetry/third_party/gsutil/third_party/apitools/apitools/base/py/encoding_test.py |
deleted file mode 100644 |
index 0d10d8b6590a2bd29c88fd215a70343699a3e429..0000000000000000000000000000000000000000 |
--- a/tools/telemetry/third_party/gsutil/third_party/apitools/apitools/base/py/encoding_test.py |
+++ /dev/null |
@@ -1,344 +0,0 @@ |
-import base64 |
-import datetime |
-import json |
- |
-from protorpc import message_types |
-from protorpc import messages |
-from protorpc import util |
-import unittest2 |
- |
-from apitools.base.py import encoding |
-from apitools.base.py import exceptions |
- |
- |
-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('additional_properties') |
-class AdditionalPropertiesMessage(messages.Message): |
- |
- class AdditionalProperty(messages.Message): |
- key = messages.StringField(1) |
- value = messages.StringField(2) |
- |
- additional_properties = 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) |
- ignored_field = messages.EnumField(ThisEnum, 3) |
- |
- |
-@encoding.MapUnrecognizedFields('additional_properties') |
-class AdditionalMessagePropertiesMessage(messages.Message): |
- |
- class AdditionalProperty(messages.Message): |
- key = messages.StringField(1) |
- value = messages.MessageField(CompoundPropertyType, 2) |
- |
- additional_properties = 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) |
- |
- |
-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(unittest2.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 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 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( |
- additional_properties=[])) |
- self.assertEqual( |
- '{"nested": null}', |
- encoding.MessageToJson(msg, include_fields=['nested'])) |
- self.assertEqual( |
- '{"nested": {"additional_properties": []}}', |
- encoding.MessageToJson( |
- msg, include_fields=['nested.additional_properties'])) |
- msg = ExtraNestedMessage(nested=msg) |
- self.assertEqual( |
- '{"nested": {"nested": null}}', |
- encoding.MessageToJson(msg, include_fields=['nested.nested'])) |
- self.assertEqual( |
- '{"nested": {"nested_list": []}}', |
- encoding.MessageToJson(msg, include_fields=['nested.nested_list'])) |
- self.assertEqual( |
- '{"nested": {"nested": {"additional_properties": []}}}', |
- encoding.MessageToJson( |
- msg, include_fields=['nested.nested.additional_properties'])) |
- |
- def testAdditionalPropertyMapping(self): |
- msg = AdditionalPropertiesMessage() |
- msg.additional_properties = [ |
- AdditionalPropertiesMessage.AdditionalProperty( |
- key='key_one', value='value_one'), |
- AdditionalPropertiesMessage.AdditionalProperty( |
- key='key_two', value='value_two'), |
- ] |
- |
- encoded_msg = encoding.MessageToJson(msg) |
- self.assertEqual( |
- {'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.additional_properties])) |
- self.assertIsNot(msg, new_msg) |
- |
- new_msg.additional_properties.pop() |
- self.assertEqual(1, len(new_msg.additional_properties)) |
- self.assertEqual(2, len(msg.additional_properties)) |
- |
- def testAdditionalMessageProperties(self): |
- json_msg = '{"input": {"index": 0, "name": "output"}}' |
- result = encoding.JsonToMessage( |
- AdditionalMessagePropertiesMessage, json_msg) |
- self.assertEqual(1, len(result.additional_properties)) |
- self.assertEqual(0, result.additional_properties[0].value.index) |
- |
- def testNestedFieldMapping(self): |
- nested_msg = AdditionalPropertiesMessage() |
- nested_msg.additional_properties = [ |
- 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.additional_properties])) |
- |
- new_msg.nested.additional_properties.pop() |
- self.assertEqual(1, len(new_msg.nested.additional_properties)) |
- self.assertEqual(2, len(msg.nested.additional_properties)) |
- |
- 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 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 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 testNoRepeatedRemapping(self): |
- 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=protorpc.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)') |