Index: tools/telemetry/third_party/gsutil/third_party/protorpc/protorpc/protobuf_test.py |
diff --git a/tools/telemetry/third_party/gsutil/third_party/protorpc/protorpc/protobuf_test.py b/tools/telemetry/third_party/gsutil/third_party/protorpc/protorpc/protobuf_test.py |
deleted file mode 100644 |
index 9a65824f72ccefa2ada11aacc1a1883374b28039..0000000000000000000000000000000000000000 |
--- a/tools/telemetry/third_party/gsutil/third_party/protorpc/protorpc/protobuf_test.py |
+++ /dev/null |
@@ -1,299 +0,0 @@ |
-#!/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.protobuf.""" |
- |
-__author__ = 'rafek@google.com (Rafe Kaplan)' |
- |
- |
-import datetime |
-import unittest |
- |
-from protorpc import message_types |
-from protorpc import messages |
-from protorpc import protobuf |
-from protorpc import protorpc_test_pb2 |
-from protorpc import test_util |
-from protorpc import util |
- |
-# TODO: Add DateTimeFields to protorpc_test.proto when definition.py |
-# supports date time fields. |
-class HasDateTimeMessage(messages.Message): |
- value = message_types.DateTimeField(1) |
- |
-class NestedDateTimeMessage(messages.Message): |
- value = messages.MessageField(message_types.DateTimeMessage, 1) |
- |
- |
-class ModuleInterfaceTest(test_util.ModuleInterfaceTest, |
- test_util.TestCase): |
- |
- MODULE = protobuf |
- |
- |
-class EncodeMessageTest(test_util.TestCase, |
- test_util.ProtoConformanceTestBase): |
- """Test message to protocol buffer encoding.""" |
- |
- PROTOLIB = protobuf |
- |
- def assertErrorIs(self, exception, message, function, *params, **kwargs): |
- try: |
- function(*params, **kwargs) |
- self.fail('Expected to raise exception %s but did not.' % exception) |
- except exception as err: |
- self.assertEquals(message, str(err)) |
- |
- @property |
- def encoded_partial(self): |
- proto = protorpc_test_pb2.OptionalMessage() |
- proto.double_value = 1.23 |
- proto.int64_value = -100000000000 |
- proto.int32_value = 1020 |
- proto.string_value = u'a string' |
- proto.enum_value = protorpc_test_pb2.OptionalMessage.VAL2 |
- |
- return proto.SerializeToString() |
- |
- @property |
- def encoded_full(self): |
- proto = protorpc_test_pb2.OptionalMessage() |
- proto.double_value = 1.23 |
- proto.float_value = -2.5 |
- proto.int64_value = -100000000000 |
- proto.uint64_value = 102020202020 |
- proto.int32_value = 1020 |
- proto.bool_value = True |
- proto.string_value = u'a string\u044f' |
- proto.bytes_value = b'a bytes\xff\xfe' |
- proto.enum_value = protorpc_test_pb2.OptionalMessage.VAL2 |
- |
- return proto.SerializeToString() |
- |
- @property |
- def encoded_repeated(self): |
- proto = protorpc_test_pb2.RepeatedMessage() |
- proto.double_value.append(1.23) |
- proto.double_value.append(2.3) |
- proto.float_value.append(-2.5) |
- proto.float_value.append(0.5) |
- proto.int64_value.append(-100000000000) |
- proto.int64_value.append(20) |
- proto.uint64_value.append(102020202020) |
- proto.uint64_value.append(10) |
- proto.int32_value.append(1020) |
- proto.int32_value.append(718) |
- proto.bool_value.append(True) |
- proto.bool_value.append(False) |
- proto.string_value.append(u'a string\u044f') |
- proto.string_value.append(u'another string') |
- proto.bytes_value.append(b'a bytes\xff\xfe') |
- proto.bytes_value.append(b'another bytes') |
- proto.enum_value.append(protorpc_test_pb2.RepeatedMessage.VAL2) |
- proto.enum_value.append(protorpc_test_pb2.RepeatedMessage.VAL1) |
- |
- return proto.SerializeToString() |
- |
- @property |
- def encoded_nested(self): |
- proto = protorpc_test_pb2.HasNestedMessage() |
- proto.nested.a_value = 'a string' |
- |
- return proto.SerializeToString() |
- |
- @property |
- def encoded_repeated_nested(self): |
- proto = protorpc_test_pb2.HasNestedMessage() |
- proto.repeated_nested.add().a_value = 'a string' |
- proto.repeated_nested.add().a_value = 'another string' |
- |
- return proto.SerializeToString() |
- |
- unexpected_tag_message = ( |
- chr((15 << protobuf._WIRE_TYPE_BITS) | protobuf._Encoder.NUMERIC) + |
- chr(5)) |
- |
- @property |
- def encoded_default_assigned(self): |
- proto = protorpc_test_pb2.HasDefault() |
- proto.a_value = test_util.HasDefault.a_value.default |
- return proto.SerializeToString() |
- |
- @property |
- def encoded_nested_empty(self): |
- proto = protorpc_test_pb2.HasOptionalNestedMessage() |
- proto.nested.Clear() |
- return proto.SerializeToString() |
- |
- @property |
- def encoded_repeated_nested_empty(self): |
- proto = protorpc_test_pb2.HasOptionalNestedMessage() |
- proto.repeated_nested.add() |
- proto.repeated_nested.add() |
- return proto.SerializeToString() |
- |
- @property |
- def encoded_extend_message(self): |
- proto = protorpc_test_pb2.RepeatedMessage() |
- proto.add_int64_value(400) |
- proto.add_int64_value(50) |
- proto.add_int64_value(6000) |
- return proto.SerializeToString() |
- |
- @property |
- def encoded_string_types(self): |
- proto = protorpc_test_pb2.OptionalMessage() |
- proto.string_value = u'Latin' |
- return proto.SerializeToString() |
- |
- @property |
- def encoded_invalid_enum(self): |
- encoder = protobuf._Encoder() |
- field_num = test_util.OptionalMessage.enum_value.number |
- tag = (field_num << protobuf._WIRE_TYPE_BITS) | encoder.NUMERIC |
- encoder.putVarInt32(tag) |
- encoder.putVarInt32(1000) |
- return encoder.buffer().tostring() |
- |
- def testDecodeWrongWireFormat(self): |
- """Test what happens when wrong wire format found in protobuf.""" |
- class ExpectedProto(messages.Message): |
- value = messages.StringField(1) |
- |
- class WrongVariant(messages.Message): |
- value = messages.IntegerField(1) |
- |
- original = WrongVariant() |
- original.value = 10 |
- self.assertErrorIs(messages.DecodeError, |
- 'Expected wire type STRING but found NUMERIC', |
- protobuf.decode_message, |
- ExpectedProto, |
- protobuf.encode_message(original)) |
- |
- def testDecodeBadWireType(self): |
- """Test what happens when non-existant wire type found in protobuf.""" |
- # Message has tag 1, type 3 which does not exist. |
- bad_wire_type_message = chr((1 << protobuf._WIRE_TYPE_BITS) | 3) |
- |
- self.assertErrorIs(messages.DecodeError, |
- 'No such wire type 3', |
- protobuf.decode_message, |
- test_util.OptionalMessage, |
- bad_wire_type_message) |
- |
- def testUnexpectedTagBelowOne(self): |
- """Test that completely invalid tags generate an error.""" |
- # Message has tag 0, type NUMERIC. |
- invalid_tag_message = chr(protobuf._Encoder.NUMERIC) |
- |
- self.assertErrorIs(messages.DecodeError, |
- 'Invalid tag value 0', |
- protobuf.decode_message, |
- test_util.OptionalMessage, |
- invalid_tag_message) |
- |
- def testProtocolBufferDecodeError(self): |
- """Test what happens when there a ProtocolBufferDecodeError. |
- |
- This is what happens when the underlying ProtocolBuffer library raises |
- it's own decode error. |
- """ |
- # Message has tag 1, type DOUBLE, missing value. |
- truncated_message = ( |
- chr((1 << protobuf._WIRE_TYPE_BITS) | protobuf._Encoder.DOUBLE)) |
- |
- self.assertErrorIs(messages.DecodeError, |
- 'Decoding error: truncated', |
- protobuf.decode_message, |
- test_util.OptionalMessage, |
- truncated_message) |
- |
- def testProtobufUnrecognizedField(self): |
- """Test that unrecognized fields are serialized and can be accessed.""" |
- decoded = protobuf.decode_message(test_util.OptionalMessage, |
- self.unexpected_tag_message) |
- self.assertEquals(1, len(decoded.all_unrecognized_fields())) |
- self.assertEquals(15, decoded.all_unrecognized_fields()[0]) |
- self.assertEquals((5, messages.Variant.INT64), |
- decoded.get_unrecognized_field_info(15)) |
- |
- def testUnrecognizedFieldWrongFormat(self): |
- """Test that unrecognized fields in the wrong format are skipped.""" |
- |
- class SimpleMessage(messages.Message): |
- value = messages.IntegerField(1) |
- |
- message = SimpleMessage(value=3) |
- message.set_unrecognized_field('from_json', 'test', messages.Variant.STRING) |
- |
- encoded = protobuf.encode_message(message) |
- expected = ( |
- chr((1 << protobuf._WIRE_TYPE_BITS) | protobuf._Encoder.NUMERIC) + |
- chr(3)) |
- self.assertEquals(encoded, expected) |
- |
- def testProtobufDecodeDateTimeMessage(self): |
- """Test what happens when decoding a DateTimeMessage.""" |
- |
- nested = NestedDateTimeMessage() |
- nested.value = message_types.DateTimeMessage(milliseconds=2500) |
- value = protobuf.decode_message(HasDateTimeMessage, |
- protobuf.encode_message(nested)).value |
- self.assertEqual(datetime.datetime(1970, 1, 1, 0, 0, 2, 500000), value) |
- |
- def testProtobufDecodeDateTimeMessageWithTimeZone(self): |
- """Test what happens when decoding a DateTimeMessage with a time zone.""" |
- nested = NestedDateTimeMessage() |
- nested.value = message_types.DateTimeMessage(milliseconds=12345678, |
- time_zone_offset=60) |
- value = protobuf.decode_message(HasDateTimeMessage, |
- protobuf.encode_message(nested)).value |
- self.assertEqual(datetime.datetime(1970, 1, 1, 3, 25, 45, 678000, |
- tzinfo=util.TimeZoneOffset(60)), |
- value) |
- |
- def testProtobufEncodeDateTimeMessage(self): |
- """Test what happens when encoding a DateTimeField.""" |
- mine = HasDateTimeMessage(value=datetime.datetime(1970, 1, 1)) |
- nested = NestedDateTimeMessage() |
- nested.value = message_types.DateTimeMessage(milliseconds=0) |
- |
- my_encoded = protobuf.encode_message(mine) |
- encoded = protobuf.encode_message(nested) |
- self.assertEquals(my_encoded, encoded) |
- |
- def testProtobufEncodeDateTimeMessageWithTimeZone(self): |
- """Test what happens when encoding a DateTimeField with a time zone.""" |
- for tz_offset in (30, -30, 8 * 60, 0): |
- mine = HasDateTimeMessage(value=datetime.datetime( |
- 1970, 1, 1, tzinfo=util.TimeZoneOffset(tz_offset))) |
- nested = NestedDateTimeMessage() |
- nested.value = message_types.DateTimeMessage( |
- milliseconds=0, time_zone_offset=tz_offset) |
- |
- my_encoded = protobuf.encode_message(mine) |
- encoded = protobuf.encode_message(nested) |
- self.assertEquals(my_encoded, encoded) |
- |
- |
-def main(): |
- unittest.main() |
- |
- |
-if __name__ == '__main__': |
- main() |