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

1"""Miscellaneous helper functions for Crypt4GH stream processing. 

2 

3""" 

4 

5import io 

6 

7 

8def read_crypt4gh_stream_le_uint32( 

9 istream: io.RawIOBase, name: str = "number" 

10) -> int: 

11 """Reads little-endian integer from given stream handling read 

12 errors with customizable error message. 

13 

14 Parameters: 

15 istream: the container input stream 

16 name: optional name of the number in the error message 

17 

18 Raises: 

19 ValueError: if not enough data can be read 

20 

21 """ 

22 number_bytes = istream.read(4) 

23 return parse_crypt4gh_bytes_le_uint(number_bytes, name, 4) 

24 

25 

26def read_crypt4gh_bytes_le_uint32( 

27 ibytes: bytes, offset: int, name: str = "number" 

28) -> int: 

29 """Extracts little-endian integer from given bytes object handling 

30 errors with customizable message. 

31 

32 Parameters: 

33 ibytes: bytes with the binary structure 

34 offset: starting byte of the encoded number 

35 name: optional name of the number in the error message 

36 

37 Raises: 

38 ValueError: if not enough data given 

39 

40 """ 

41 number_bytes = ibytes[offset : offset + 4] 

42 return parse_crypt4gh_bytes_le_uint(number_bytes, name, 4) 

43 

44 

45def parse_crypt4gh_bytes_le_uint( 

46 number_bytes: bytes, name: str, size: int 

47) -> int: 

48 """Parses size-byte little-endian binary number from given bytes 

49 handling insufficient data errors with customizable error message. 

50 

51 Parameters: 

52 number_bytes: the bytes to parse 

53 name: optional name of the number in the error message 

54 size: number of bytes the encoding should contain 

55 

56 Raises: 

57 ValueError: if the bytes given are too short 

58 

59 """ 

60 number_bytes_len = len(number_bytes) 

61 if number_bytes_len != size: 

62 raise ValueError( 

63 f"Only {number_bytes_len} bytes for reading le_uint({size}) {name}" 

64 ) 

65 return int.from_bytes(number_bytes, byteorder="little")