Repository navigation
Expand file tree
/
Copy pathdecoder_test.py
More file actions
155 lines (130 loc) · 4.67 KB
/
Copy pathdecoder_test.py
File metadata and controls
155 lines (130 loc) · 4.67 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
# -*- coding: utf-8 -*-
# Protocol Buffers - Google's data interchange format
# Copyright 2008 Google Inc. All rights reserved.
#
# Use of this source code is governed by a BSD-style
# license that can be found in the LICENSE file or at
# https://developers.google.com/open-source/licenses/bsd
"""Test decoder."""
import io
import unittest
from google.protobuf import message
from google.protobuf.internal import api_implementation
from google.protobuf.internal import decoder
from google.protobuf.internal import message_set_extensions_pb2
from google.protobuf.internal import testing_refleaks
from google.protobuf.internal import wire_format
from absl.testing import parameterized
_INPUT_BYTES = b'\x84r\x12'
_EXPECTED = (14596, 18)
@testing_refleaks.TestCase
class DecoderTest(parameterized.TestCase):
def test_decode_varint_bytes(self):
size, pos = decoder._DecodeVarint(_INPUT_BYTES, 0)
self.assertEqual(size, _EXPECTED[0])
self.assertEqual(pos, 2)
size, pos = decoder._DecodeVarint(_INPUT_BYTES, 2)
self.assertEqual(size, _EXPECTED[1])
self.assertEqual(pos, 3)
def test_decode_varint_bytes_empty(self):
with self.assertRaises(IndexError) as context:
decoder._DecodeVarint(b'', 0)
self.assertIn('index out of range', str(context.exception))
def test_decode_varint_bytesio(self):
index = 0
input_io = io.BytesIO(_INPUT_BYTES)
while True:
size = decoder._DecodeVarint(input_io)
if size is None:
break
self.assertEqual(size, _EXPECTED[index])
index += 1
self.assertEqual(index, len(_EXPECTED))
def test_decode_varint_bytesio_empty(self):
input_io = io.BytesIO(b'')
size = decoder._DecodeVarint(input_io)
self.assertIsNone(size)
def test_decode_unknown_group_field(self):
data = memoryview(b'\013\020\003\014\040\005')
parsed, pos = decoder._DecodeUnknownField(
data, 1, len(data), 1, wire_format.WIRETYPE_START_GROUP
)
self.assertEqual(pos, 4)
self.assertEqual(len(parsed), 1)
self.assertEqual(parsed[0].field_number, 2)
self.assertEqual(parsed[0].data, 3)
def test_decode_unknown_group_field_nested(self):
data = memoryview(b'\013\023\013\030\004\014\024\014\050\006')
parsed, pos = decoder._DecodeUnknownField(
data, 1, len(data), 1, wire_format.WIRETYPE_START_GROUP
)
self.assertEqual(pos, 8)
self.assertEqual(len(parsed), 1)
self.assertEqual(parsed[0].field_number, 2)
self.assertEqual(len(parsed[0].data), 1)
self.assertEqual(parsed[0].data[0].field_number, 1)
self.assertEqual(len(parsed[0].data[0].data), 1)
self.assertEqual(parsed[0].data[0].data[0].field_number, 3)
self.assertEqual(parsed[0].data[0].data[0].data, 4)
def test_decode_unknown_group_field_too_many_levels(self):
data = memoryview(b'\023' * 5_000_000)
self.assertRaisesRegex(
message.DecodeError,
'Error parsing message',
decoder._DecodeUnknownField,
data,
1,
len(data),
1,
wire_format.WIRETYPE_START_GROUP,
)
def test_decode_unknown_mismatched_end_group(self):
self.assertRaisesRegex(
message.DecodeError,
'Missing group end tag.*',
decoder._DecodeUnknownField,
memoryview(b'\013\024'),
1,
2,
1,
wire_format.WIRETYPE_START_GROUP,
)
def test_decode_unknown_mismatched_end_group_nested(self):
self.assertRaisesRegex(
message.DecodeError,
'Missing group end tag.*',
decoder._DecodeUnknownField,
memoryview(b'\013\023\034\024\014'),
1,
5,
1,
wire_format.WIRETYPE_START_GROUP,
)
def test_decode_message_set_unknown_mismatched_end_group(self):
proto = message_set_extensions_pb2.TestMessageSet()
self.assertRaisesRegex(
message.DecodeError,
'Unexpected end-group tag.'
if api_implementation.Type() == 'python'
else '.*',
proto.ParseFromString,
b'\013\054\014',
)
def test_unknown_message_set_decoder_mismatched_end_group(self):
# This behavior isn't actually reachable in practice, but it's good to
# test anyway.
decode = decoder.UnknownMessageSetItemDecoder()
self.assertRaisesRegex(
message.DecodeError,
'Unexpected end-group tag.',
decode,
memoryview(b'\054\014'),
)
@parameterized.parameters(int(0), float(0.0), False, '')
def test_default_scalar(self, value):
self.assertTrue(decoder.IsDefaultScalarValue(value))
@parameterized.parameters(int(1), float(-0.0), float(1.0), True, 'a')
def test_not_default_scalar(self, value):
self.assertFalse(decoder.IsDefaultScalarValue(value))
if __name__ == '__main__':
unittest.main()