Coverage for oarepo_c4gh / crypt4gh / rawio.py: 100%

1"""This module provides a wrapper on top of any Proto4GH-compatible 

2object with RawIO protocol for the linear stream of cleartext data 

3from the Crypt4GH container. 

4 

5""" 

6 

7from io import RawIOBase 

8from .common.proto4gh import Proto4GH 

9 

10 

11class Crypt4GHRawIO(RawIOBase): 

12 """RawIO-compatible read-only wrapper around Proto4GH. Implements 

13 only the `readinto` method - the rest of functionality must be 

14 provided by BufferedIOBase and TextIOBase wrappers. 

15 

16 """ 

17 

18 def __init__(self, container: Proto4GH) -> None: 

19 """Initializes the container wrapper and sets internal block 

20 caching up. 

21 

22 Parameters: 

23 container: opened Crypt4GH container providing the underlying 

24 data blocks 

25 

26 """ 

27 self._container = container 

28 self._data_blocks = None 

29 self._current_block = None 

30 self._current_pos = 0 

31 self._finished = False 

32 

33 def readinto(self, b: bytearray) -> int: 

34 """As required by RawIO, read bytes into a pre-allocated, 

35 writable bytes-like object b, and return the number of bytes 

36 read. 

37 

38 Parameters: 

39 b: buffer to read the data into 

40 

41 Returns: 

42 The number of bytes read. 

43 """ 

44 if self._finished: 

45 return 0 

46 if self._data_blocks is None: 

47 self._data_blocks = self._container.data_blocks 

48 self._edit_list = self._container.header.edit_list.copy() 

49 self._edit_skipping = True 

50 blen = len(b) 

51 bpos = 0 

52 while bpos < blen: 

53 # Load next block if the current one is completely used. 

54 if self._current_block is None or self._current_pos >= len( 

55 self._current_block 

56 ): 

57 try: 

58 nxt = next(self._data_blocks) 

59 except StopIteration: 

60 # Cannot read more blocks, end immediately. 

61 self._finished = True 

62 return bpos 

63 self._current_pos = 0 

64 if not nxt.is_deciphered: 

65 # The error raised is similar to failed read from 

66 # a disk which resembles this situation the most. 

67 raise OSError 

68 self._current_block = nxt.cleartext 

69 # Check for skipping first 

70 if self._edit_skipping and (len(self._edit_list) > 0): 

71 # How much is left for skipping and how much data in 

72 # the current block remains so that it can be skipped 

73 # immediately 

74 skip_req = self._edit_list[0] 

75 avail = len(self._current_block) - self._current_pos 

76 to_skip = min(skip_req, avail) 

77 self._edit_list[0] = self._edit_list[0] - to_skip 

78 self._current_pos = self._current_pos + to_skip 

79 if self._edit_list[0] == 0: 

80 # Skipped enough, flip the flag, advance to next 

81 # count and continue 

82 self._edit_list = self._edit_list[1:] 

83 self._edit_skipping = False 

84 else: 

85 # Get how much can be copied first. 

86 avail = len(self._current_block) - self._current_pos 

87 to_copy = min(blen - bpos, avail) 

88 # Check whether there are still some edit list counts 

89 # left and apply it if necessary. 

90 if len(self._edit_list) > 0: 

91 # Adjust to_copy to maximum 

92 max_copy = self._edit_list[0] 

93 to_copy = min(to_copy, max_copy) 

94 b[bpos : bpos + to_copy] = self._current_block[ 

95 self._current_pos : self._current_pos + to_copy 

96 ] 

97 self._current_pos = self._current_pos + to_copy 

98 bpos = bpos + to_copy 

99 if len(self._edit_list) > 0: 

100 # Reduce current not-skipping counter 

101 self._edit_list[0] = self._edit_list[0] - to_copy 

102 if self._edit_list[0] == 0: 

103 # Finished the not-skipping part 

104 self._edit_list = self._edit_list[1:] 

105 if len(self._edit_list) > 0: 

106 # Something left, therefore set the 

107 # skipping flag again. 

108 self._edit_skipping = True 

109 else: 

110 # Nothing left 

111 self._finished = True 

112 return bpos 

113 return bpos 

114 

115 def writable(self) -> bool: 

116 """According to RawIO specification this method returning 

117 always False ensures no write-like methods can be used as this 

118 implementation provides read-only access. 

119 

120 Returns: 

121 Always False. 

122 """ 

123 return False 

124 

125 def readable(self) -> bool: 

126 """According to RawIO specification this method returning 

127 always True ensures read-like methods can be used. 

128 

129 Returns: 

130 Always True. 

131 """ 

132 return True