Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pip/_internal/utils/filesystem.py: 27%

1from __future__ import annotations 

2 

3import fnmatch 

4import os 

5import os.path 

6import random 

7import sys 

8from collections.abc import Generator 

9from contextlib import contextmanager 

10from pathlib import Path 

11from tempfile import NamedTemporaryFile 

12from typing import Any, BinaryIO, Callable, cast 

13 

14from pip._internal.utils.compat import get_path_uid 

15from pip._internal.utils.misc import format_size 

16from pip._internal.utils.retry import retry 

17 

18 

19def check_path_owner(path: str) -> bool: 

20 # If we don't have a way to check the effective uid of this process, then 

21 # we'll just assume that we own the directory. 

22 if sys.platform == "win32" or not hasattr(os, "geteuid"): 

23 return True 

24 

25 assert os.path.isabs(path) 

26 

27 previous = None 

28 while path != previous: 

29 if os.path.lexists(path): 

30 # Check if path is writable by current user. 

31 if os.geteuid() == 0: 

32 # Special handling for root user in order to handle properly 

33 # cases where users use sudo without -H flag. 

34 try: 

35 path_uid = get_path_uid(path) 

36 except OSError: 

37 return False 

38 return path_uid == 0 

39 else: 

40 return os.access(path, os.W_OK) 

41 else: 

42 previous, path = path, os.path.dirname(path) 

43 return False # assume we don't own the path 

44 

45 

46@contextmanager 

47def adjacent_tmp_file(path: str, **kwargs: Any) -> Generator[BinaryIO, None, None]: 

48 """Return a file-like object pointing to a tmp file next to path. 

49 

50 The file is created securely and is ensured to be written to disk 

51 after the context reaches its end. 

52 

53 kwargs will be passed to tempfile.NamedTemporaryFile to control 

54 the way the temporary file will be opened. 

55 """ 

56 with NamedTemporaryFile( 

57 delete=False, 

58 dir=os.path.dirname(path), 

59 prefix=os.path.basename(path), 

60 suffix=".tmp", 

61 **kwargs, 

62 ) as f: 

63 result = cast(BinaryIO, f) 

64 try: 

65 yield result 

66 finally: 

67 result.flush() 

68 os.fsync(result.fileno()) 

69 

70 

71replace = retry(stop_after_delay=1, wait=0.25)(os.replace) 

72 

73 

74# test_writable_dir and _test_writable_dir_win are copied from Flit, 

75# with the author's agreement to also place them under pip's license. 

76def test_writable_dir(path: str) -> bool: 

77 """Check if a directory is writable. 

78 

79 Uses os.access() on POSIX, tries creating files on Windows. 

80 """ 

81 # If the directory doesn't exist, find the closest parent that does. 

82 while not os.path.isdir(path): 

83 parent = os.path.dirname(path) 

84 if parent == path: 

85 break # Should never get here, but infinite loops are bad 

86 path = parent 

87 

88 if os.name == "posix": 

89 return os.access(path, os.W_OK) 

90 

91 return _test_writable_dir_win(path) 

92 

93 

94def _test_writable_dir_win(path: str) -> bool: 

95 # os.access doesn't work on Windows: http://bugs.python.org/issue2528 

96 # and we can't use tempfile: http://bugs.python.org/issue22107 

97 basename = "accesstest_deleteme_fishfingers_custard_" 

98 alphabet = "abcdefghijklmnopqrstuvwxyz0123456789" 

99 for _ in range(10): 

100 name = basename + "".join(random.choice(alphabet) for _ in range(6)) 

101 file = os.path.join(path, name) 

102 try: 

103 fd = os.open(file, os.O_RDWR | os.O_CREAT | os.O_EXCL) 

104 except FileExistsError: 

105 pass 

106 except PermissionError: 

107 # This could be because there's a directory with the same name. 

108 # But it's highly unlikely there's a directory called that, 

109 # so we'll assume it's because the parent dir is not writable. 

110 # This could as well be because the parent dir is not readable, 

111 # due to non-privileged user access. 

112 return False 

113 else: 

114 os.close(fd) 

115 os.unlink(file) 

116 return True 

117 

118 # This should never be reached 

119 raise OSError("Unexpected condition testing for writable directory") 

120 

121 

122def find_files(path: str, pattern: str) -> list[str]: 

123 """Returns a list of absolute paths of files beneath path, recursively, 

124 with filenames which match the UNIX-style shell glob pattern.""" 

125 result: list[str] = [] 

126 for root, _, files in os.walk(path): 

127 matches = fnmatch.filter(files, pattern) 

128 result.extend(os.path.join(root, f) for f in matches) 

129 return result 

130 

131 

132def file_size(path: str) -> int | float: 

133 # If it's a symlink, return 0. 

134 if os.path.islink(path): 

135 return 0 

136 return os.path.getsize(path) 

137 

138 

139def format_file_size(path: str) -> str: 

140 return format_size(file_size(path)) 

141 

142 

143def directory_size(path: str) -> int | float: 

144 size = 0.0 

145 for root, _dirs, files in os.walk(path): 

146 for filename in files: 

147 file_path = os.path.join(root, filename) 

148 size += file_size(file_path) 

149 return size 

150 

151 

152def format_directory_size(path: str) -> str: 

153 return format_size(directory_size(path)) 

154 

155 

156def copy_directory_permissions(directory: str, target_file: BinaryIO) -> None: 

157 mode = ( 

158 os.stat(directory).st_mode & 0o666 # select read/write permissions of directory 

159 | 0o600 # set owner read/write permissions 

160 ) 

161 # Change permissions only if there is no risk of following a symlink. 

162 if os.chmod in os.supports_fd: 

163 os.chmod(target_file.fileno(), mode) 

164 elif os.chmod in os.supports_follow_symlinks: 

165 os.chmod(target_file.name, mode, follow_symlinks=False) 

166 

167 

168def _subdirs_without_generic( 

169 path: str, predicate: Callable[[str, list[str]], bool] 

170) -> Generator[Path]: 

171 """Yields every subdirectory of +path+ that has no files matching the 

172 predicate under it.""" 

173 

174 directories = [] 

175 excluded = set() 

176 

177 for root_str, _, filenames in os.walk(Path(path).resolve()): 

178 root = Path(root_str) 

179 if predicate(root_str, filenames): 

180 # This directory should be excluded, so exclude it and all of its 

181 # parent directories. 

182 # The last item in root.parents is ".", so we ignore it. 

183 # 

184 # Wrapping this in `list()` is only needed for Python 3.9. 

185 excluded.update(list(root.parents)[:-1]) 

186 excluded.add(root) 

187 directories.append(root) 

188 

189 for d in sorted(directories, reverse=True): 

190 if d not in excluded: 

191 yield d 

192 

193 

194def subdirs_without_files(path: str) -> Generator[Path]: 

195 """Yields every subdirectory of +path+ that has no files under it.""" 

196 return _subdirs_without_generic(path, lambda root, filenames: len(filenames) > 0) 

197 

198 

199def subdirs_without_wheels(path: str) -> Generator[Path]: 

200 """Yields every subdirectory of +path+ that has no .whl files under it.""" 

201 return _subdirs_without_generic( 

202 path, lambda root, filenames: any(x.endswith(".whl") for x in filenames) 

203 )