Coverage for apio/managers/unpacker.py: 83%

1"""DOC: TODO""" 

2 

3# -*- coding: utf-8 -*- 

4# -- This file is part of the Apio project 

5# -- (C) 2016-2019 FPGAwars 

6# -- Author Jesús Arroyo 

7# -- License GPLv2 

8# -- Derived from: 

9# ---- Platformio project 

10# ---- (C) 2014-2016 Ivan Kravets <me@ikravets.com> 

11# ---- License Apache v2 

12 

13from pathlib import Path 

14from tarfile import open as tarfile_open 

15from rich.progress import track 

16from apio.common.apio_console import console, cerror 

17from apio.utils import util 

18 

19 

20class ArchiveBase: 

21 """DOC: TODO""" 

22 

23 def __init__(self, arhfileobj, is_tar_file: bool): 

24 self._afo = arhfileobj 

25 self._is_tar_file = is_tar_file 

26 

27 def get_items(self): # pragma: no cover 

28 """DOC: TODO""" 

29 

30 raise NotImplementedError() 

31 

32 def extract_item(self, item, dest_dir): 

33 """DOC: TODO""" 

34 

35 if hasattr(item, "filename") and item.filename.endswith(".gitignore"): 35 ↛ 36line 35 didn't jump to line 36 because the condition on line 35 was never true

36 return 

37 if self._is_tar_file and util.get_python_ver_tuple() >= (3, 12, 0): 37 ↛ 43line 37 didn't jump to line 43 because the condition on line 37 was always true

38 # -- Special case for avoiding the tar deprecation warning. Search 

39 # -- 'extraction_filter' in the page 

40 # -- https://docs.python.org/3/library/tarfile.html 

41 self._afo.extract(item, dest_dir, filter="fully_trusted") 

42 else: 

43 self._afo.extract(item, dest_dir) 

44 self.after_extract(item, dest_dir) 

45 

46 def after_extract(self, item, dest_dir): 

47 """DOC: TODO""" 

48 

49 

50class TARArchive(ArchiveBase): 

51 """DOC: TODO""" 

52 

53 def __init__(self, archpath): 

54 # R1732: Consider using 'with' for resource-allocating operations 

55 # (consider-using-with) 

56 # pylint: disable=R1732 

57 ArchiveBase.__init__(self, tarfile_open(archpath), is_tar_file=True) 

58 

59 def get_items(self): 

60 return self._afo.getmembers() 

61 

62 

63# class ZIPArchive(ArchiveBase): 

64# """DOC: TODO""" 

65 

66# def __init__(self, archpath): 

67# # R1732: Consider using 'with' for resource-allocating operations 

68# # (consider-using-with) 

69# ArchiveBase.__init__(self, ZipFile(archpath), is_tar_file=False) 

70 

71# @staticmethod 

72# def preserve_permissions(item, dest_dir): 

73# """DOC: TODO""" 

74 

75# # -- Build the filename 

76# file = str(Path(dest_dir) / item.filename) 

77 

78# attrs = item.external_attr >> 16 

79# if attrs: 

80# chmod(file, attrs) 

81 

82# def get_items(self): 

83# """DOC: TODO""" 

84 

85# return self._afo.infolist() 

86 

87# def after_extract(self, item, dest_dir): 

88# """DOC: TODO""" 

89 

90# self.preserve_permissions(item, dest_dir) 

91 

92 

93class FileUnpacker: 

94 """Class for unpacking compressed files""" 

95 

96 def __init__(self, archpath: Path, dest_dir=Path(".")): 

97 """Initialize the unpacker object 

98 * INPUT: 

99 - archpath: filename with path to uncompress 

100 - des_dir: Destination folder 

101 """ 

102 

103 self._archpath = archpath 

104 self._dest_dir = dest_dir 

105 self._unpacker = None 

106 

107 # -- Get the file extension 

108 arch_ext = archpath.suffix 

109 

110 # -- Select the unpacker... according to the file extension 

111 # -- tar zip file 

112 if arch_ext in (".tgz"): 112 ↛ 120line 112 didn't jump to line 120 because the condition on line 112 was always true

113 self._unpacker = TARArchive(archpath) 

114 

115 # -- Zip file 

116 # elif arch_ext == ".zip": 

117 # self._unpacker = ZIPArchive(archpath) 

118 

119 # -- Fatal error. Unknown extension. 

120 if not self._unpacker: 120 ↛ 121line 120 didn't jump to line 121 because the condition on line 120 was never true

121 cerror(f"Can not unpack file '{archpath}'") 

122 raise util.ApioException() 

123 

124 def start(self) -> bool: 

125 """Start unpacking the file""" 

126 

127 # -- Build an array with all the files inside the tarball 

128 items = self._unpacker.get_items() 

129 

130 # -- Unpack while displaying a progress bar. 

131 for i in track( 

132 range(len(items)), 

133 description="Unpacking ", 

134 console=console(), 

135 ): 

136 self._unpacker.extract_item(items[i], self._dest_dir) 

137 

138 return True