Coverage for dlabel/util.py: 64%

1import fnmatch 

2import io 

3import tarfile 

4from collections.abc import Collection 

5from logging import getLogger 

6from pathlib import Path 

7 

8import docker 

9 

10_log = getLogger(__name__) 

11 

12# https://pkg.go.dev/io/fs#ModeDir 

13modebits = { 

14 "dir": 1 << 31, 

15 "append": 1 << 30, 

16 "exclusive": 1 << 29, 

17 "temporary": 1 << 28, 

18 "symlink": 1 << 27, 

19 "device": 1 << 26, 

20 "namedpipe": 1 << 25, 

21 "socket": 1 << 24, 

22 "setuid": 1 << 23, 

23 "setgid": 1 << 22, 

24 "chardev": 1 << 21, 

25 "sticky": 1 << 20, 

26 "irregular": 1 << 19, 

27} 

28nonreg = {"dir", "device", "namedpipe", "socket", "chardev", "irregular"} 

29 

30 

31def special_modes(mode: int) -> tuple[set[str], int]: 

32 res: set[str] = set() 

33 for k, v in modebits.items(): 

34 if (mode & v) != 0: 34 ↛ 35line 34 didn't jump to line 35 because the condition on line 34 was never true

35 res.add(k) 

36 return res, (mode & 0o777) 

37 

38 

39def download_files(ctn: docker.models.containers.Container, filename: str): 

40 bins, stat = ctn.get_archive(filename) 

41 is_dir = "dir" in special_modes(stat["mode"])[0] 

42 _log.debug("download %s: %s is_dir=%s", filename, stat, is_dir) 

43 fp = io.BytesIO() 

44 for chunk in bins: 

45 fp.write(chunk) 

46 fp.seek(0) 

47 with tarfile.open(fileobj=fp, mode="r|") as tar: 

48 for member in tar: 

49 if member.isfile(): 49 ↛ 48line 49 didn't jump to line 48 because the condition on line 49 was always true

50 _log.debug("extract %s", member.name) 

51 tf = tar.extractfile(member) 

52 if tf is not None: 52 ↛ 48line 52 didn't jump to line 48 because the condition on line 52 was always true

53 yield is_dir, member, tf.read() 

54 

55 

56def get_archives( 

57 container: docker.models.containers.Container, 

58 names: set[str], 

59 ignore: Collection[str], 

60 mode: str = "w:gz", 

61): 

62 if not names: 

63 return 

64 ofp = io.BytesIO() 

65 with tarfile.open(mode=mode, fileobj=ofp) as outarchive: 

66 for fn in sorted(names): 

67 _log.debug("extract: %s", fn) 

68 for is_dir, tinfo, bin in download_files(container, fn): 

69 if is_dir: 

70 tinfo.name = str(Path(fn) / tinfo.name).lstrip("/") 

71 else: 

72 tinfo.name = fn.lstrip("/") 

73 if is_match(ignore, tinfo.name): 

74 _log.debug("ignore: %s", tinfo.name) 

75 continue 

76 _log.debug( 

77 "add file: %s (%s bytes) is_dir=%s", tinfo.name, len(bin), is_dir 

78 ) 

79 outarchive.addfile(tinfo, io.BytesIO(bin)) 

80 return ofp.getvalue() 

81 

82 

83def is_match(patterns: Collection[str], target: str) -> bool: 

84 for p in patterns: 84 ↛ 85line 84 didn't jump to line 85 because the loop on line 84 never started

85 if fnmatch.fnmatch(target, p): 

86 return True 

87 return False 

88 

89 

90def is_already(prev: set[str], target: str) -> bool: 

91 for p in prev: 91 ↛ 92line 91 didn't jump to line 92 because the loop on line 91 never started

92 if Path(p) in Path(target).parents: 

93 return True 

94 return False 

95 

96 

97def do_kind0( 

98 modified: set[str], 

99 path: str, 

100 link: dict[str, str], 

101 container: docker.models.containers.Container, 

102): # modified 

103 _, stats = container.get_archive(path) 

104 _log.debug("stats %s: %s", path, stats) 

105 special, _ = special_modes(stats["mode"]) 

106 if nonreg & special: 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true

107 _log.debug("skip: %s %s", path, special) 

108 elif "symlink" in special and stats["linkTarget"]: 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true

109 link[path] = stats["linkTarget"] 

110 else: 

111 modified.add(path) 

112 

113 

114def do_kind1( 

115 added: set[str], 

116 path: str, 

117 link: dict[str, str], 

118 container: docker.models.containers.Container, 

119): # added 

120 if is_already(added, path): 120 ↛ 121line 120 didn't jump to line 121 because the condition on line 120 was never true

121 _log.debug("skip(parent-exists): %s", path) 

122 else: 

123 _, stats = container.get_archive(path) 

124 _log.debug("stats %s: %s", path, stats) 

125 special, _ = special_modes(stats["mode"]) 

126 if (nonreg - {"dir"}) & special: 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true

127 _log.debug("skip: %s %s", path, special) 

128 elif "symlink" in special and stats["linkTarget"]: 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true

129 link[path] = stats["linkTarget"] 

130 else: 

131 added.add(path) 

132 

133 

134def do_kind2(deleted: set[str], path: str): # deleted 

135 if is_already(deleted, path): 135 ↛ 136line 135 didn't jump to line 136 because the condition on line 135 was never true

136 _log.debug("skip(parent-exists): %s", path) 

137 else: 

138 deleted.add(path) 

139 

140 

141def get_volumes(container: docker.models.containers.Container) -> set[str]: 

142 return {x["Destination"] for x in container.attrs["Mounts"]} 

143 

144 

145def get_diff( 

146 container: docker.models.containers.Container, ignore: Collection[str] 

147) -> tuple[set[str], set[str], set[str], dict[str, str]]: 

148 deleted: set[str] = set() 

149 added: set[str] = set() 

150 modified: set[str] = set() 

151 link: dict[str, str] = {} 

152 for pathkind in container.diff(): 

153 path = pathkind["Path"] 

154 kind = pathkind["Kind"] 

155 if is_match(ignore, path): 155 ↛ 156line 155 didn't jump to line 156 because the condition on line 155 was never true

156 _log.debug("ignore: %s", path) 

157 continue 

158 if kind == 2: # deleted 

159 do_kind2(deleted, path) 

160 elif kind == 1: # added 

161 do_kind1(added, path, link, container) 

162 elif kind == 0: # modified 162 ↛ 152line 162 didn't jump to line 152 because the condition on line 162 was always true

163 do_kind0(modified, path, link, container) 

164 _log.debug("deleted: %s", deleted) 

165 _log.debug("added: %s", added) 

166 _log.debug("modified: %s", modified) 

167 _log.debug("link: %s", link) 

168 return deleted, added, modified, link