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

1# SPDX-FileCopyrightText: Copyright (c) 2025 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 (empty)

12 

13import glob (empty)

14import os (empty)

15from collections.abc import Sequence (empty)

16from dataclasses import dataclass (empty)

17from typing import Protocol, cast (empty)

18 

19from cuda.pathfinder._dynamic_libs.lib_descriptor import LibDescriptor (empty)

20from cuda.pathfinder._dynamic_libs.supported_nvidia_libs import is_suppressed_dll_file (empty)

21from cuda.pathfinder._utils.find_sub_dirs import find_sub_dirs_all_sitepackages (empty)

22from cuda.pathfinder._utils.platform_aware import IS_WINDOWS (empty)

23 

24 

25def _no_such_file_in_sub_dirs( (empty)

26 sub_dirs: Sequence[str], file_wild: str, error_messages: list[str], attachments: list[str] 

27) -> None: 

28 error_messages.append(f"No such file: {file_wild}") 5 ctx1abecd

29 for sub_dir in find_sub_dirs_all_sitepackages(sub_dirs): 5 ctx1abecd

30 attachments.append(f' listdir("{sub_dir}"):') 1 ctx1d

31 for node in sorted(os.listdir(sub_dir)): 1 ctx1d

32 attachments.append(f" {node}") 

33 

34 

35def _find_so_in_rel_dirs( (empty)

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

37 so_basename: str, 

38 error_messages: list[str], 

39 attachments: list[str], 

40) -> str | None: 

41 sub_dirs_searched: list[tuple[str, ...]] = [] 5 ctx1abeld

42 file_wild = so_basename + "*" 5 ctx1abeld

43 for rel_dir in rel_dirs: 5 ctx1abeld

44 sub_dir = tuple(rel_dir.split(os.path.sep)) 5 ctx1abeld

45 for abs_dir in find_sub_dirs_all_sitepackages(sub_dir): 5 ctx1abeld

46 so_name = os.path.join(abs_dir, so_basename) 2 ctx1ld

47 if os.path.isfile(so_name): 2 ctx1ld

48 return so_name 1 ctx1l

49 for so_name in sorted(glob.glob(os.path.join(abs_dir, file_wild))): 1 ctx1d

50 if os.path.isfile(so_name): 

51 return so_name 

52 sub_dirs_searched.append(sub_dir) 4 ctx1abed

53 for sub_dir in sub_dirs_searched: 4 ctx1abed

54 _no_such_file_in_sub_dirs(sub_dir, file_wild, error_messages, attachments) 4 ctx1abed

55 return None 4 ctx1abed

56 

57 

58def _find_dll_under_dir(dirpath: str, file_wild: str) -> str | None: (empty)

59 for path in sorted(glob.glob(os.path.join(dirpath, file_wild))): 10 ctx1afghijcmnk

60 if not os.path.isfile(path): 10 ctx1afghijcmnk

61 continue 

62 if not is_suppressed_dll_file(os.path.basename(path)): 10 ctx1afghijcmnk

63 return path 10 ctx1afghijcmnk

64 return None 

65 

66 

67def _find_dll_in_rel_dirs( (empty)

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

69 lib_searched_for: str, 

70 error_messages: list[str], 

71 attachments: list[str], 

72) -> str | None: 

73 sub_dirs_searched: list[tuple[str, ...]] = [] 4 ctx1abck

74 for rel_dir in rel_dirs: 4 ctx1abck

75 sub_dir = tuple(rel_dir.split(os.path.sep)) 4 ctx1abck

76 for abs_dir in find_sub_dirs_all_sitepackages(sub_dir): 4 ctx1abck

77 dll_name = _find_dll_under_dir(abs_dir, lib_searched_for) 1 ctx1k

78 if dll_name is not None: 1 ctx1k

79 return dll_name 1 ctx1k

80 sub_dirs_searched.append(sub_dir) 3 ctx1abc

81 for sub_dir in sub_dirs_searched: 3 ctx1abc

82 _no_such_file_in_sub_dirs(sub_dir, lib_searched_for, error_messages, attachments) 3 ctx1abc

83 return None 3 ctx1abc

84 

85 

86class SearchPlatform(Protocol): (empty)

87 def lib_searched_for(self, libname: str) -> str: ... (empty)

88 

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

90 

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

92 

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

94 

95 def find_in_site_packages( (empty)

96 self, 

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

98 lib_searched_for: str, 

99 error_messages: list[str], 

100 attachments: list[str], 

101 ) -> str | None: ... 

102 

103 def find_in_lib_dir( (empty)

104 self, 

105 lib_dir: str, 

106 libname: str, 

107 lib_searched_for: str, 

108 error_messages: list[str], 

109 attachments: list[str], 

110 ) -> str | None: ... 

111 

112 

113@dataclass(frozen=True, slots=True) (empty)

114class LinuxSearchPlatform: (empty)

115 def lib_searched_for(self, libname: str) -> str: (empty)

116 return f"lib{libname}.so" 20 ctx1afghtijbowperqsldxyz

117 

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

119 return cast(tuple[str, ...], desc.site_packages_linux) 6 ctx1abeldA

120 

121 def conda_anchor_point(self, conda_prefix: str) -> str: (empty)

122 return conda_prefix 3 ctx1opq

123 

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

125 return cast(tuple[str, ...], desc.anchor_rel_dirs_linux) 16 ctx1afguhijvopeBCrqs

126 

127 def find_in_site_packages( (empty)

128 self, 

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

130 lib_searched_for: str, 

131 error_messages: list[str], 

132 attachments: list[str], 

133 ) -> str | None: 

134 return _find_so_in_rel_dirs(rel_dirs, lib_searched_for, error_messages, attachments) 5 ctx1abeld

135 

136 def find_in_lib_dir( (empty)

137 self, 

138 lib_dir: str, 

139 _libname: str, 

140 lib_searched_for: str, 

141 error_messages: list[str], 

142 attachments: list[str], 

143 ) -> str | None: 

144 # Most libraries have both unversioned and versioned files/symlinks (exact match first) 

145 so_name = os.path.join(lib_dir, lib_searched_for) 12 ctx1afghijoperqs

146 if os.path.isfile(so_name): 12 ctx1afghijoperqs

147 return so_name 10 ctx1afghijerqs

148 # Some libraries only exist as versioned files (e.g., libcupti.so.13 in conda), 

149 # so the glob fallback is needed 

150 file_wild = lib_searched_for + "*" 2 ctx1op

151 # Only one match is expected, but to ensure deterministic behavior in unexpected 

152 # situations, and to be internally consistent, we sort in reverse order with the 

153 # intent to return the newest version first. 

154 for so_name in sorted(glob.glob(os.path.join(lib_dir, file_wild)), reverse=True): 2 ctx1op

155 if os.path.isfile(so_name): 2 ctx1op

156 return so_name 2 ctx1op

157 error_messages.append(f"No such file: {file_wild}") 

158 attachments.append(f' listdir("{lib_dir}"):') 

159 if not os.path.isdir(lib_dir): 

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

161 else: 

162 for node in sorted(os.listdir(lib_dir)): 

163 attachments.append(f" {node}") 

164 return None 

165 

166 

167@dataclass(frozen=True, slots=True) (empty)

168class WindowsSearchPlatform: (empty)

169 def lib_searched_for(self, libname: str) -> str: (empty)

170 return f"{libname}*.dll" 13 ctx1afghtijbcmnkD

171 

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

173 return cast(tuple[str, ...], desc.site_packages_windows) 4 ctx1abck

174 

175 def conda_anchor_point(self, conda_prefix: str) -> str: (empty)

176 return os.path.join(conda_prefix, "Library") 1 ctx1m

177 

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

179 return cast(tuple[str, ...], desc.anchor_rel_dirs_windows) 12 ctx1afguhijvcEmn

180 

181 def find_in_site_packages( (empty)

182 self, 

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

184 lib_searched_for: str, 

185 error_messages: list[str], 

186 attachments: list[str], 

187 ) -> str | None: 

188 return _find_dll_in_rel_dirs(rel_dirs, lib_searched_for, error_messages, attachments) 4 ctx1abck

189 

190 def find_in_lib_dir( (empty)

191 self, 

192 lib_dir: str, 

193 libname: str, 

194 _lib_searched_for: str, 

195 error_messages: list[str], 

196 attachments: list[str], 

197 ) -> str | None: 

198 file_wild = libname + "*.dll" 9 ctx1afghijcmn

199 dll_name = _find_dll_under_dir(lib_dir, file_wild) 9 ctx1afghijcmn

200 if dll_name is not None: 9 ctx1afghijcmn

201 return dll_name 9 ctx1afghijcmn

202 error_messages.append(f"No such file: {file_wild}") 

203 attachments.append(f' listdir("{lib_dir}"):') 

204 if not os.path.isdir(lib_dir): 

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

206 else: 

207 for node in sorted(os.listdir(lib_dir)): 

208 attachments.append(f" {node}") 

209 return None 

210 

211 

212PLATFORM: SearchPlatform = WindowsSearchPlatform() if IS_WINDOWS else LinuxSearchPlatform() (empty)