mirror of
https://github.com/protocolbuffers/protobuf
synced 2026-08-26 02:23:14 -04:00
230 lines
9.2 KiB
Python
Executable file
230 lines
9.2 KiB
Python
Executable file
# Protocol Buffers - Google's data interchange format
|
|
# Copyright 2008 Google Inc.
|
|
# http://code.google.com/p/protobuf/
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
"""Test for google.protobuf.internal.decoder."""
|
|
|
|
__author__ = 'robinson@google.com (Will Robinson)'
|
|
|
|
import struct
|
|
import unittest
|
|
from google.protobuf.internal import wire_format
|
|
from google.protobuf.internal import encoder
|
|
from google.protobuf.internal import decoder
|
|
import logging
|
|
import mox
|
|
from google.protobuf.internal import input_stream
|
|
from google.protobuf import message
|
|
|
|
|
|
class DecoderTest(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.mox = mox.Mox()
|
|
self.mock_stream = self.mox.CreateMock(input_stream.InputStream)
|
|
self.mock_message = self.mox.CreateMock(message.Message)
|
|
|
|
def testReadFieldNumberAndWireType(self):
|
|
# Test field numbers that will require various varint sizes.
|
|
for expected_field_number in (1, 15, 16, 2047, 2048):
|
|
for expected_wire_type in range(6): # Highest-numbered wiretype is 5.
|
|
e = encoder.Encoder()
|
|
e._AppendTag(expected_field_number, expected_wire_type)
|
|
s = e.ToString()
|
|
d = decoder.Decoder(s)
|
|
field_number, wire_type = d.ReadFieldNumberAndWireType()
|
|
self.assertEqual(expected_field_number, field_number)
|
|
self.assertEqual(expected_wire_type, wire_type)
|
|
|
|
def ReadScalarTestHelper(self, test_name, decoder_method, expected_result,
|
|
expected_stream_method_name,
|
|
stream_method_return, *args):
|
|
"""Helper for testReadScalars below.
|
|
|
|
Calls one of the Decoder.Read*() methods and ensures that the results are
|
|
as expected.
|
|
|
|
Args:
|
|
test_name: Name of this test, used for logging only.
|
|
decoder_method: Unbound decoder.Decoder method to call.
|
|
expected_result: Value we expect returned from decoder_method().
|
|
expected_stream_method_name: (string) Name of the InputStream
|
|
method we expect Decoder to call to actually read the value
|
|
on the wire.
|
|
stream_method_return: Value our mocked-out stream method should
|
|
return to the decoder.
|
|
args: Additional arguments that we expect to be passed to the
|
|
stream method.
|
|
"""
|
|
logging.info('Testing %s scalar input.\n'
|
|
'Calling %r(), and expecting that to call the '
|
|
'stream method %s(%r), which will return %r. Finally, '
|
|
'expecting the Decoder method to return %r'% (
|
|
test_name, decoder_method,
|
|
expected_stream_method_name, args, stream_method_return,
|
|
expected_result))
|
|
|
|
d = decoder.Decoder('')
|
|
d._stream = self.mock_stream
|
|
if decoder_method in (decoder.Decoder.ReadString,
|
|
decoder.Decoder.ReadBytes):
|
|
self.mock_stream.ReadVarUInt32().AndReturn(len(stream_method_return))
|
|
# We have to use names instead of methods to work around some
|
|
# mox weirdness. (ResetAll() is overzealous).
|
|
expected_stream_method = getattr(self.mock_stream,
|
|
expected_stream_method_name)
|
|
expected_stream_method(*args).AndReturn(stream_method_return)
|
|
|
|
self.mox.ReplayAll()
|
|
self.assertEqual(expected_result, decoder_method(d))
|
|
self.mox.VerifyAll()
|
|
self.mox.ResetAll()
|
|
|
|
def testReadScalars(self):
|
|
test_string = 'I can feel myself getting sutpider.'
|
|
scalar_tests = [
|
|
['int32', decoder.Decoder.ReadInt32, 0, 'ReadVarint32', 0],
|
|
['int64', decoder.Decoder.ReadInt64, 0, 'ReadVarint64', 0],
|
|
['uint32', decoder.Decoder.ReadUInt32, 0, 'ReadVarUInt32', 0],
|
|
['uint64', decoder.Decoder.ReadUInt64, 0, 'ReadVarUInt64', 0],
|
|
['fixed32', decoder.Decoder.ReadFixed32, 0xffffffff,
|
|
'ReadLittleEndian32', 0xffffffff],
|
|
['fixed64', decoder.Decoder.ReadFixed64, 0xffffffffffffffff,
|
|
'ReadLittleEndian64', 0xffffffffffffffff],
|
|
['sfixed32', decoder.Decoder.ReadSFixed32, -1,
|
|
'ReadLittleEndian32', 0xffffffff],
|
|
['sfixed64', decoder.Decoder.ReadSFixed64, -1,
|
|
'ReadLittleEndian64', 0xffffffffffffffff],
|
|
['float', decoder.Decoder.ReadFloat, 0.0,
|
|
'ReadString', struct.pack('f', 0.0), 4],
|
|
['double', decoder.Decoder.ReadDouble, 0.0,
|
|
'ReadString', struct.pack('d', 0.0), 8],
|
|
['bool', decoder.Decoder.ReadBool, True, 'ReadVarUInt32', 1],
|
|
['enum', decoder.Decoder.ReadEnum, 23, 'ReadVarUInt32', 23],
|
|
['string', decoder.Decoder.ReadString,
|
|
test_string, 'ReadString', test_string, len(test_string)],
|
|
['bytes', decoder.Decoder.ReadBytes,
|
|
test_string, 'ReadString', test_string, len(test_string)],
|
|
# We test zigzag decoding routines more extensively below.
|
|
['sint32', decoder.Decoder.ReadSInt32, -1, 'ReadVarUInt32', 1],
|
|
['sint64', decoder.Decoder.ReadSInt64, -1, 'ReadVarUInt64', 1],
|
|
]
|
|
# Ensure that we're testing different Decoder methods and using
|
|
# different test names in all test cases above.
|
|
self.assertEqual(len(scalar_tests), len(set(t[0] for t in scalar_tests)))
|
|
self.assertEqual(len(scalar_tests), len(set(t[1] for t in scalar_tests)))
|
|
for args in scalar_tests:
|
|
self.ReadScalarTestHelper(*args)
|
|
|
|
def testReadMessageInto(self):
|
|
length = 23
|
|
def Test(simulate_error):
|
|
d = decoder.Decoder('')
|
|
d._stream = self.mock_stream
|
|
self.mock_stream.ReadVarUInt32().AndReturn(length)
|
|
sub_buffer = object()
|
|
self.mock_stream.GetSubBuffer(length).AndReturn(sub_buffer)
|
|
|
|
if simulate_error:
|
|
self.mock_message.MergeFromString(sub_buffer).AndReturn(length - 1)
|
|
self.mox.ReplayAll()
|
|
self.assertRaises(
|
|
message.DecodeError, d.ReadMessageInto, self.mock_message)
|
|
else:
|
|
self.mock_message.MergeFromString(sub_buffer).AndReturn(length)
|
|
self.mock_stream.SkipBytes(length)
|
|
self.mox.ReplayAll()
|
|
d.ReadMessageInto(self.mock_message)
|
|
|
|
self.mox.VerifyAll()
|
|
self.mox.ResetAll()
|
|
|
|
Test(simulate_error=False)
|
|
Test(simulate_error=True)
|
|
|
|
def testReadGroupInto_Success(self):
|
|
# Test both the empty and nonempty cases.
|
|
for num_bytes in (5, 0):
|
|
field_number = expected_field_number = 10
|
|
d = decoder.Decoder('')
|
|
d._stream = self.mock_stream
|
|
sub_buffer = object()
|
|
self.mock_stream.GetSubBuffer().AndReturn(sub_buffer)
|
|
self.mock_message.MergeFromString(sub_buffer).AndReturn(num_bytes)
|
|
self.mock_stream.SkipBytes(num_bytes)
|
|
self.mock_stream.ReadVarUInt32().AndReturn(wire_format.PackTag(
|
|
field_number, wire_format.WIRETYPE_END_GROUP))
|
|
self.mox.ReplayAll()
|
|
d.ReadGroupInto(expected_field_number, self.mock_message)
|
|
self.mox.VerifyAll()
|
|
self.mox.ResetAll()
|
|
|
|
def ReadGroupInto_FailureTestHelper(self, bytes_read):
|
|
d = decoder.Decoder('')
|
|
d._stream = self.mock_stream
|
|
sub_buffer = object()
|
|
self.mock_stream.GetSubBuffer().AndReturn(sub_buffer)
|
|
self.mock_message.MergeFromString(sub_buffer).AndReturn(bytes_read)
|
|
return d
|
|
|
|
def testReadGroupInto_NegativeBytesReported(self):
|
|
expected_field_number = 10
|
|
d = self.ReadGroupInto_FailureTestHelper(bytes_read=-1)
|
|
self.mox.ReplayAll()
|
|
self.assertRaises(message.DecodeError,
|
|
d.ReadGroupInto, expected_field_number,
|
|
self.mock_message)
|
|
self.mox.VerifyAll()
|
|
|
|
def testReadGroupInto_NoEndGroupTag(self):
|
|
field_number = expected_field_number = 10
|
|
num_bytes = 5
|
|
d = self.ReadGroupInto_FailureTestHelper(bytes_read=num_bytes)
|
|
self.mock_stream.SkipBytes(num_bytes)
|
|
# Right field number, wrong wire type.
|
|
self.mock_stream.ReadVarUInt32().AndReturn(wire_format.PackTag(
|
|
field_number, wire_format.WIRETYPE_LENGTH_DELIMITED))
|
|
self.mox.ReplayAll()
|
|
self.assertRaises(message.DecodeError,
|
|
d.ReadGroupInto, expected_field_number,
|
|
self.mock_message)
|
|
self.mox.VerifyAll()
|
|
|
|
def testReadGroupInto_WrongFieldNumberInEndGroupTag(self):
|
|
expected_field_number = 10
|
|
field_number = expected_field_number + 1
|
|
num_bytes = 5
|
|
d = self.ReadGroupInto_FailureTestHelper(bytes_read=num_bytes)
|
|
self.mock_stream.SkipBytes(num_bytes)
|
|
# Wrong field number, right wire type.
|
|
self.mock_stream.ReadVarUInt32().AndReturn(wire_format.PackTag(
|
|
field_number, wire_format.WIRETYPE_END_GROUP))
|
|
self.mox.ReplayAll()
|
|
self.assertRaises(message.DecodeError,
|
|
d.ReadGroupInto, expected_field_number,
|
|
self.mock_message)
|
|
self.mox.VerifyAll()
|
|
|
|
def testSkipBytes(self):
|
|
d = decoder.Decoder('')
|
|
num_bytes = 1024
|
|
self.mock_stream.SkipBytes(num_bytes)
|
|
d._stream = self.mock_stream
|
|
self.mox.ReplayAll()
|
|
d.SkipBytes(num_bytes)
|
|
self.mox.VerifyAll()
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|