Coverage for cuda/pathfinder/_dynamic_libs/search_platform.py: 95.62%

137 statements  

« prev     ^ index     » next       coverage.py v7.16.0, created at 2026-09-03 02:41 +0000

1# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. 

2# SPDX-License-Identifier: Apache-2.0 

3 

4"""Platform abstraction for filesystem search steps. 

5 

6The goal is to keep :mod:`search_steps` platform-agnostic: it should not branch 

7on OS flags like ``IS_WINDOWS``. Instead, it calls through the single 

8``PLATFORM`` instance exported here. 

9""" 

10 

11from __future__ import annotations 

12 

13import os 

14from collections.abc import Sequence 

15from dataclasses import dataclass 

16from pathlib import PurePath 

17from typing import Protocol, cast 

18 

19from cuda.pathfinder._dynamic_libs.lib_descriptor import LibDescriptor 

20from cuda.pathfinder._dynamic_libs.supported_nvidia_libs import is_suppressed_dll_file 

21from cuda.pathfinder._utils.find_sub_dirs import find_sub_dirs_all_sitepackages 

22from cuda.pathfinder._utils.platform_aware import IS_WINDOWS 

23from cuda.pathfinder._utils.windows_arch import windows_pe_matches_arch, windows_python_arch 

24 

25 

26def _no_such_file_in_sub_dirs( 

27 sub_dirs: Sequence[str], file_description: str, error_messages: list[str], attachments: list[str] 

28) -> None: 

29 error_messages.append(f"No such file: {file_description}") 1agFijklref

30 for sub_dir in find_sub_dirs_all_sitepackages(sub_dirs): 1agFijklref

31 attachments.append(f' listdir("{sub_dir}"):') 1ijklref

32 for node in sorted(os.listdir(sub_dir)): 1ijklref

33 attachments.append(f" {node}") 1ijklef

34 

35 

36def _find_descriptor_so_under_dir(dirpath: str, desc: LibDescriptor) -> str | None: 

37 for soname in desc.linux_sonames: 1ambcnoST2UVBCDE3GHIijklrJWXYZ01

38 path = os.path.join(dirpath, soname) 1ambcnoST2UVBCDE3GHIijklrJWXYZ01

39 if os.path.isfile(path): 1ambcnoST2UVBCDE3GHIijklrJWXYZ01

40 return path 1ambcnoST2UV3GHIJWXYZ01

41 return None 1BCDEijklr

42 

43 

44def _find_so_in_rel_dirs( 

45 rel_dirs: tuple[str, ...], 

46 desc: LibDescriptor, 

47 file_description: str, 

48 error_messages: list[str], 

49 attachments: list[str], 

50) -> str | None: 

51 sub_dirs_searched: list[tuple[str, ...]] = [] 1agGHIijklrJ

52 for rel_dir in rel_dirs: 1agGHIijklrJ

53 sub_dir = PurePath(rel_dir).parts 1agGHIijklrJ

54 for abs_dir in find_sub_dirs_all_sitepackages(sub_dir): 1agGHIijklrJ

55 so_path = _find_descriptor_so_under_dir(abs_dir, desc) 1GHIijklrJ

56 if so_path is not None: 1GHIijklrJ

57 return so_path 1GHIJ

58 sub_dirs_searched.append(sub_dir) 1agijklr

59 for sub_dir in sub_dirs_searched: 1agijklr

60 _no_such_file_in_sub_dirs(sub_dir, file_description, error_messages, attachments) 1agijklr

61 return None 1agijklr

62 

63 

64def _find_descriptor_dll_under_dir( 

65 dirpath: str, 

66 desc: LibDescriptor, 

67 target_arch: str | None = None, 

68) -> str | None: 

69 def candidate_is_usable(path: str) -> bool: 1ambcnoOwKLPQpMNqtusve456789fhyzAdx

70 if not os.path.isfile(path): 1ambcnoOwKLPQpMNqtusve456789fhyzAdx

71 return False 1ambcnopMNqtuse56789fhd

72 if is_suppressed_dll_file(os.path.basename(path)): 1ambcnoOwKLPQpMNqtusv4yzAdx

73 return False 

74 return target_arch is None or windows_pe_matches_arch(path, target_arch) 1ambcnoOwKLPQpMNqtusv4yzAdx

75 

76 for dll_basename in desc.windows_dlls: 1ambcnoOwKLPQpMNqtusve456789fhyzAdx

77 path = os.path.join(dirpath, dll_basename) 1ambcnoOwKLPQpMNqtusve456789fhyzAdx

78 if candidate_is_usable(path): 1ambcnoOwKLPQpMNqtusve456789fhyzAdx

79 return path 1ambcnoOwKLPQMNtusv4yzAdx

80 return None 1pqe56789fhd

81 

82 

83def _find_dll_in_rel_dirs( 

84 rel_dirs: tuple[str, ...], 

85 desc: LibDescriptor, 

86 target_arch: str, 

87 lib_searched_for: str, 

88 error_messages: list[str], 

89 attachments: list[str], 

90) -> str | None: 

91 sub_dirs_searched: list[tuple[str, ...]] = [] 1agtFusvef

92 checked_arch = target_arch if desc.requires_windows_binary_arch_check else None 1agtFusvef

93 for rel_dir in rel_dirs: 1agtFusvef

94 sub_dir = PurePath(rel_dir).parts 1agtFusvef

95 for abs_dir in find_sub_dirs_all_sitepackages(sub_dir): 1agtFusvef

96 dll_name = _find_descriptor_dll_under_dir(abs_dir, desc, checked_arch) 1tusvef

97 if dll_name is not None: 1tusvef

98 return dll_name 1tusv

99 sub_dirs_searched.append(sub_dir) 1agFsef

100 for sub_dir in sub_dirs_searched: 1agFef

101 _no_such_file_in_sub_dirs(sub_dir, lib_searched_for, error_messages, attachments) 1agFef

102 return None 1agFef

103 

104 

105class SearchPlatform(Protocol): 

106 def lib_searched_for(self, desc: LibDescriptor) -> str: ... 

107 

108 def site_packages_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]: ... 

109 

110 def conda_anchor_point(self, conda_prefix: str) -> str: ... 

111 

112 def anchor_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]: ... 

113 

114 def install_root_env_vars(self, desc: LibDescriptor) -> tuple[str, ...]: ... 

115 

116 def install_root_env_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]: ... 

117 

118 def program_files_root_globs(self, desc: LibDescriptor) -> tuple[str, ...]: ... 

119 

120 def find_in_site_packages( 

121 self, 

122 rel_dirs: tuple[str, ...], 

123 desc: LibDescriptor, 

124 error_messages: list[str], 

125 attachments: list[str], 

126 ) -> str | None: ... 

127 

128 def find_in_lib_dir( 

129 self, 

130 lib_dir: str, 

131 desc: LibDescriptor, 

132 error_messages: list[str], 

133 attachments: list[str], 

134 ) -> str | None: ... 

135 

136 

137@dataclass(frozen=True, slots=True) 

138class LinuxSearchPlatform: 

139 def lib_searched_for(self, desc: LibDescriptor) -> str: 

140 return " or ".join(desc.linux_sonames) 1aRg)BCDEGHIijklrJ*+,

141 

142 def site_packages_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]: 

143 return cast(tuple[str, ...], desc.site_packages_linux) 1agGHIijklrJ-

144 

145 def conda_anchor_point(self, conda_prefix: str) -> str: 

146 return conda_prefix 1STUVBCDE

147 

148 def anchor_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]: 

149 return cast(tuple[str, ...], desc.anchor_rel_dirs_linux) 1amb%cno'ST./2UVBCDE3

150 

151 def install_root_env_vars(self, desc: LibDescriptor) -> tuple[str, ...]: 

152 return cast(tuple[str, ...], desc.install_root_env_vars_linux) 1abcRWXYZ01

153 

154 def install_root_env_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]: 

155 return cast(tuple[str, ...], desc.install_root_env_rel_dirs_linux) 1abcRWXYZ01

156 

157 def program_files_root_globs(self, _desc: LibDescriptor) -> tuple[str, ...]: 

158 return () 1bR

159 

160 def find_in_site_packages( 

161 self, 

162 rel_dirs: tuple[str, ...], 

163 desc: LibDescriptor, 

164 error_messages: list[str], 

165 attachments: list[str], 

166 ) -> str | None: 

167 return _find_so_in_rel_dirs(rel_dirs, desc, self.lib_searched_for(desc), error_messages, attachments) 1agGHIijklrJ

168 

169 def find_in_lib_dir( 

170 self, 

171 lib_dir: str, 

172 desc: LibDescriptor, 

173 error_messages: list[str], 

174 attachments: list[str], 

175 ) -> str | None: 

176 so_path = _find_descriptor_so_under_dir(lib_dir, desc) 1ambcnoST2UVBCDE3WXYZ01

177 if so_path is not None: 1ambcnoST2UVBCDE3WXYZ01

178 return so_path 1ambcnoST2UV3WXYZ01

179 error_messages.append(f"No such file: {self.lib_searched_for(desc)}") 1BCDE

180 attachments.append(f' listdir("{lib_dir}"):') 1BCDE

181 if not os.path.isdir(lib_dir): 1BCDE

182 attachments.append(" DIRECTORY DOES NOT EXIST") 

183 else: 

184 for node in sorted(os.listdir(lib_dir)): 1BCDE

185 attachments.append(f" {node}") 1BCDE

186 return None 1BCDE

187 

188 

189@dataclass(frozen=True, slots=True) 

190class WindowsSearchPlatform: 

191 target_arch: str 

192 

193 def lib_searched_for(self, desc: LibDescriptor) -> str: 

194 return f"known {desc.name} DLL" 1aRgpqtFusvef:hd

195 

196 def site_packages_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]: 

197 return cast(tuple[str, ...], desc.site_packages_windows.for_arch(self.target_arch)) 1agwtFusvef!

198 

199 def conda_anchor_point(self, conda_prefix: str) -> str: 

200 return os.path.join(conda_prefix, "Library") 1KL!

201 

202 def anchor_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]: 

203 return cast(tuple[str, ...], desc.anchor_rel_dirs_windows.for_arch(self.target_arch)) 1amb%cno'O;=?@KLPQpMNq!dx

204 

205 def install_root_env_vars(self, desc: LibDescriptor) -> tuple[str, ...]: 

206 if self.target_arch not in desc.supported_windows_arch: 1abcRw#$hyzA

207 return () 

208 return cast(tuple[str, ...], desc.install_root_env_vars_windows) 1abcRw#$hyzA

209 

210 def install_root_env_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]: 

211 if self.target_arch not in desc.supported_windows_arch: 1abcRw#$hyzA

212 return () 

213 return cast(tuple[str, ...], desc.install_root_env_rel_dirs_windows.for_arch(self.target_arch)) 1abcRw#$hyzA

214 

215 def program_files_root_globs(self, desc: LibDescriptor) -> tuple[str, ...]: 

216 program_files = os.environ.get("PROGRAMW6432") or os.environ.get("PROGRAMFILES") 1bR!dx

217 if not program_files: 1bR!dx

218 return () 

219 rel_globs = desc.program_files_root_globs_windows.for_arch(self.target_arch) 1bR!dx

220 return tuple(os.path.join(program_files, rel_glob) for rel_glob in rel_globs) 1bR!dx

221 

222 def find_in_site_packages( 

223 self, 

224 rel_dirs: tuple[str, ...], 

225 desc: LibDescriptor, 

226 error_messages: list[str], 

227 attachments: list[str], 

228 ) -> str | None: 

229 return _find_dll_in_rel_dirs( 1agtFusvef

230 rel_dirs, 

231 desc, 

232 self.target_arch, 

233 self.lib_searched_for(desc), 

234 error_messages, 

235 attachments, 

236 ) 

237 

238 def find_in_lib_dir( 

239 self, 

240 lib_dir: str, 

241 desc: LibDescriptor, 

242 error_messages: list[str], 

243 attachments: list[str], 

244 ) -> str | None: 

245 target_arch = self.target_arch if desc.requires_windows_binary_arch_check else None 1ambcnoOwKLPQpMNqhyzAdx

246 dll_name = _find_descriptor_dll_under_dir(lib_dir, desc, target_arch) 1ambcnoOwKLPQpMNqhyzAdx

247 if dll_name is not None: 1ambcnoOwKLPQpMNqhyzAdx

248 return dll_name 1ambcnoOwKLPQMNyzAdx

249 lib_searched_for = self.lib_searched_for(desc) 1pqhd

250 if target_arch is None: 1pqhd

251 error_messages.append(f"No such file: {lib_searched_for}") 1hd

252 else: 

253 error_messages.append(f"No {target_arch}-compatible PE file: {lib_searched_for}") 1pq

254 attachments.append(f' listdir("{lib_dir}"):') 1pqhd

255 if not os.path.isdir(lib_dir): 1pqhd

256 attachments.append(" DIRECTORY DOES NOT EXIST") 

257 else: 

258 for node in sorted(os.listdir(lib_dir)): 1pqhd

259 attachments.append(f" {node}") 1pqhd

260 return None 1pqhd

261 

262 

263def _platform_for_current_system() -> SearchPlatform: 

264 if IS_WINDOWS: 1a(

265 return WindowsSearchPlatform(target_arch=windows_python_arch()) 

266 return LinuxSearchPlatform() 1a(

267 

268 

269PLATFORM = _platform_for_current_system()