Coverage for src/backend/InvenTree/plugin/helpers.py: 55%

1"""Helpers for plugin app.""" 

2 

3import inspect 

4import os 

5import pathlib 

6import pkgutil 

7import sys 

8import sysconfig 

9import traceback 

10from importlib.metadata import entry_points 

11from importlib.util import module_from_spec 

12 

13from django.conf import settings 

14from django.core.exceptions import AppRegistryNotReady 

15from django.db.utils import IntegrityError 

16 

17import structlog 

18 

19logger = structlog.get_logger('inventree') 

20 

21 

22# region logging / errors 

23class IntegrationPluginError(Exception): 

24 """Error that encapsulates another error and adds the path / reference of the raising plugin.""" 

25 

26 def __init__(self, path, message): 

27 """Init a plugin error. 

28 

29 Args: 

30 path: Path on which the error occurred - used to find out which plugin it was 

31 message: The original error message 

32 """ 

33 self.path = path 

34 self.message = message 

35 

36 def __str__(self): 

37 """Returns the error message.""" 

38 return self.message # pragma: no cover 

39 

40 

41class MixinImplementationError(ValueError): 

42 """Error if mixin was implemented wrong in plugin. 

43 

44 Mostly raised if constant is missing 

45 """ 

46 

47 

48class MixinNotImplementedError(NotImplementedError): 

49 """Error if necessary mixin function was not overwritten.""" 

50 

51 

52def log_registry_error(error, reference: str = 'general'): 

53 """Log an plugin error.""" 

54 from plugin.registry import registry 

55 

56 # make sure the registry is set up 

57 if reference not in registry.errors: 

58 registry.errors[reference] = [] 

59 

60 # add error to stack 

61 registry.errors[reference].append(error) 

62 

63 

64def handle_error(error, do_raise: bool = True, do_log: bool = True, log_name: str = ''): 

65 """Handles an error and casts it as an IntegrationPluginError.""" 

66 package_path = traceback.extract_tb(error.__traceback__)[-1].filename 

67 install_path = sysconfig.get_paths()['purelib'] 

68 

69 try: 

70 package_name = pathlib.Path(package_path).relative_to(install_path).parts[0] 

71 except ValueError: 

72 # is file - loaded -> form a name for that 

73 try: 

74 path_obj = pathlib.Path(package_path).relative_to(settings.BASE_DIR) 

75 path_parts = [*path_obj.parts] 

76 path_parts[-1] = path_parts[-1].replace( 

77 path_obj.suffix, '' 

78 ) # remove suffix 

79 

80 # remove path prefixes 

81 if path_parts[0] == 'plugin': 

82 path_parts.remove('plugin') 

83 path_parts.pop(0) 

84 else: 

85 path_parts.remove('plugins') # pragma: no cover 

86 

87 package_name = '.'.join(path_parts) 

88 except Exception: 

89 package_name = package_path 

90 

91 if do_log: 

92 log_kwargs = {} 

93 if log_name: 

94 log_kwargs['reference'] = log_name 

95 log_registry_error({package_name: str(error)}, **log_kwargs) 

96 

97 if do_raise: 

98 # do a straight raise if we are playing with environment variables at execution time, ignore the broken sample 

99 if ( 

100 settings.TESTING_ENV 

101 and package_name != 'integration.broken_sample' 

102 and isinstance(error, IntegrityError) 

103 ): 

104 raise error # pragma: no cover 

105 

106 raise IntegrationPluginError(package_name, str(error)) 

107 

108 

109def get_entrypoints(): 

110 """Returns list for entrypoints for InvenTree plugins.""" 

111 # on python before 3.12, we need to use importlib_metadata 

112 if sys.version_info < (3, 12): 

113 return entry_points().get('inventree_plugins', []) 

114 return entry_points(group='inventree_plugins') 

115 

116 

117# endregion 

118 

119 

120# region git-helpers 

121def get_git_log(path): 

122 """Get dict with info of the last commit to file named in path.""" 

123 import datetime 

124 

125 from dulwich.errors import NotGitRepository 

126 from dulwich.repo import Repo 

127 

128 from InvenTree.ready import isInTestMode 

129 

130 output = None 

131 path = os.path.abspath(path) 

132 

133 if os.path.exists(path) and os.path.isfile(path): 133 ↛ 137line 133 didn't jump to line 137 because the condition on line 133 was always true

134 path = os.path.dirname(path) 

135 

136 # only do this if we are not in test mode 

137 if not isInTestMode(): # pragma: no cover 137 ↛ 156line 137 didn't jump to line 156 because the condition on line 137 was always true

138 try: 

139 repo = Repo(path) 

140 head = repo.head() 

141 commit = repo[head] 

142 

143 output = [ 

144 head.decode(), 

145 commit.author.decode().split('<')[0][:-1], 

146 commit.author.decode().split('<')[1][:-1], 

147 datetime.datetime.fromtimestamp(commit.author_time).isoformat(), 

148 commit.message.decode().split('\n')[0], 

149 ] 

150 repo.close() 

151 except KeyError: 

152 logger.debug('No HEAD tag found in git repo at path %s', path) 

153 except NotGitRepository: 

154 pass 

155 

156 if not output: 156 ↛ 159line 156 didn't jump to line 159 because the condition on line 156 was always true

157 output = 5 * [''] # pragma: no cover 

158 

159 return { 

160 'hash': output[0], 

161 'author': output[1], 

162 'mail': output[2], 

163 'date': output[3], 

164 'message': output[4], 

165 } 

166 

167 

168# endregion 

169 

170 

171# region plugin finders 

172def get_modules(pkg, path=None): 

173 """Get all modules in a package.""" 

174 context = {} 

175 

176 if path is None: 176 ↛ 178line 176 didn't jump to line 178 because the condition on line 176 was always true

177 path = pkg.__path__ 

178 elif type(path) is not list: 

179 path = [path] 

180 

181 packages = pkgutil.walk_packages(path) 

182 

183 while True: 

184 try: 

185 finder, name, _ = next(packages) 

186 except StopIteration: 

187 break 

188 except Exception as error: 

189 log_registry_error({pkg.__name__: str(error)}, 'discovery') 

190 continue 

191 

192 try: 

193 if sys.version_info < (3, 12): 193 ↛ 194line 193 didn't jump to line 194 because the condition on line 193 was never true

194 module = finder.find_module(name).load_module(name) 

195 else: 

196 spec = finder.find_spec(name) 

197 module = module_from_spec(spec) 

198 sys.modules[name] = module 

199 spec.loader.exec_module(module) 

200 pkg_names = getattr(module, '__all__', None) 

201 for k, v in vars(module).items(): 

202 if not k.startswith('_') and (pkg_names is None or k in pkg_names): 

203 context[k] = v 

204 context[name] = module 

205 except AppRegistryNotReady: # pragma: no cover 

206 pass 

207 except Exception as error: 

208 # this 'protects' against malformed plugin modules by more or less silently failing 

209 

210 # log to stack 

211 log_registry_error({name: str(error)}, 'discovery') 

212 

213 return [v for k, v in context.items()] 

214 

215 

216def get_classes(module) -> list: 

217 """Get all classes in a given module.""" 

218 try: 

219 return inspect.getmembers(module, inspect.isclass) 

220 except Exception: 

221 log_registry_error({module.__name__: 'Could not get classes'}, 'discovery') 

222 return [] 

223 

224 

225def get_plugins(pkg, baseclass, path=None): 

226 """Return a list of all modules under a given package. 

227 

228 - Modules must be a subclass of the provided 'baseclass' 

229 - Modules must have a non-empty NAME parameter 

230 """ 

231 plugins = [] 

232 

233 modules = get_modules(pkg, path=path) 

234 

235 # Iterate through each module in the package 

236 for mod in modules: 

237 # Iterate through each class in the module 

238 for item in get_classes(mod): 

239 plugin = item[1] 

240 if issubclass(plugin, baseclass) and plugin.NAME: 

241 plugins.append(plugin) 

242 

243 return plugins 

244 

245 

246# endregion