Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/google/protobuf/unknown_fields.py: 7%

1# Protocol Buffers - Google's data interchange format 

2# Copyright 2008 Google Inc. All rights reserved. 

3# 

4# Use of this source code is governed by a BSD-style 

5# license that can be found in the LICENSE file or at 

6# https://developers.google.com/open-source/licenses/bsd 

7 

8"""Contains Unknown Fields APIs. 

9 

10Simple usage example: 

11 unknown_field_set = UnknownFieldSet(message) 

12 for unknown_field in unknown_field_set: 

13 wire_type = unknown_field.wire_type 

14 field_number = unknown_field.field_number 

15 data = unknown_field.data 

16""" 

17 

18 

19from google.protobuf.internal import api_implementation 

20 

21if api_implementation._c_module is not None: # pylint: disable=protected-access 

22 UnknownFieldSet = api_implementation._c_module.UnknownFieldSet # pylint: disable=protected-access 

23else: 

24 from google.protobuf.internal import decoder # pylint: disable=g-import-not-at-top 

25 from google.protobuf.internal import wire_format # pylint: disable=g-import-not-at-top 

26 

27 class UnknownField: 

28 """A parsed unknown field.""" 

29 

30 # Disallows assignment to other attributes. 

31 __slots__ = ['_field_number', '_wire_type', '_data'] 

32 

33 def __init__(self, field_number, wire_type, data): 

34 self._field_number = field_number 

35 self._wire_type = wire_type 

36 self._data = data 

37 return 

38 

39 @property 

40 def field_number(self): 

41 return self._field_number 

42 

43 @property 

44 def wire_type(self): 

45 return self._wire_type 

46 

47 @property 

48 def data(self): 

49 return self._data 

50 

51 class UnknownFieldSet: 

52 """UnknownField container.""" 

53 

54 # Disallows assignment to other attributes. 

55 __slots__ = ['_values'] 

56 

57 def __init__(self, msg): 

58 

59 def InternalAdd(field_number, wire_type, data): 

60 unknown_field = UnknownField(field_number, wire_type, data) 

61 self._values.append(unknown_field) 

62 

63 self._values = [] 

64 msg_des = msg.DESCRIPTOR 

65 # pylint: disable=protected-access 

66 unknown_fields = msg._unknown_fields 

67 if (msg_des.has_options and 

68 msg_des.GetOptions().message_set_wire_format): 

69 local_decoder = decoder.UnknownMessageSetItemDecoder() 

70 for _, buffer in unknown_fields: 

71 (field_number, data) = local_decoder(memoryview(buffer)) 

72 InternalAdd(field_number, wire_format.WIRETYPE_LENGTH_DELIMITED, data) 

73 else: 

74 for tag_bytes, buffer in unknown_fields: 

75 # pylint: disable=protected-access 

76 (tag, _) = decoder._DecodeVarint(tag_bytes, 0) 

77 field_number, wire_type = wire_format.UnpackTag(tag) 

78 if field_number == 0: 

79 raise RuntimeError('Field number 0 is illegal.') 

80 (data, _) = decoder._DecodeUnknownField( 

81 memoryview(buffer), 0, wire_type) 

82 InternalAdd(field_number, wire_type, data) 

83 

84 def __getitem__(self, index): 

85 size = len(self._values) 

86 if index < 0: 

87 index += size 

88 if index < 0 or index >= size: 

89 raise IndexError('index %d out of range'.index) 

90 

91 return self._values[index] 

92 

93 def __len__(self): 

94 return len(self._values) 

95 

96 def __iter__(self): 

97 return iter(self._values)