Coverage for netbox/plugins/__init__.py: 32%

1import collections 

2from importlib import import_module 

3 

4from django.apps import AppConfig, apps 

5from django.core.exceptions import ImproperlyConfigured 

6from django.utils.module_loading import import_string 

7from packaging import version 

8 

9from core.exceptions import IncompatiblePluginError 

10from netbox.event_rules import register_event_rule_action 

11from netbox.registry import registry 

12from netbox.search import register_search 

13from netbox.utils import register_data_backend 

14 

15from .navigation import * 

16from .registration import * 

17from .templates import * 

18from .utils import * 

19 

20# Initialize plugin registry 

21registry['plugins'].update({ 

22 'installed': [], 

23 'graphql_schemas': [], 

24 'jinja_filters': {}, 

25 'graphql_type_extensions': collections.defaultdict(list), 

26 'graphql_filter_extensions': collections.defaultdict(list), 

27 # Assembled (store key, model label) pairs. Registering an extension for an assembled target raises. 

28 'graphql_extensions_assembled': set(), 

29 'menus': [], 

30 'menu_items': {}, 

31 'preferences': {}, 

32 'template_extensions': collections.defaultdict(list), 

33}) 

34 

35DEFAULT_RESOURCE_PATHS = { 

36 'search_indexes': 'search.indexes', 

37 'data_backends': 'data_backends.backends', 

38 'event_rule_actions': 'event_rules.event_rule_actions', 

39 'graphql_schema': 'graphql.schema', 

40 'graphql_type_extensions': 'graphql_extensions.type_extensions', 

41 'graphql_filter_extensions': 'graphql_extensions.filter_extensions', 

42 'jinja_filters': 'jinja_env.filters', 

43 'menu': 'navigation.menu', 

44 'menu_items': 'navigation.menu_items', 

45 'template_extensions': 'template_content.template_extensions', 

46 'user_preferences': 'preferences.preferences', 

47} 

48 

49 

50# 

51# Plugin AppConfig class 

52# 

53 

54class PluginConfig(AppConfig): 

55 """ 

56 Subclass of Django's built-in AppConfig class, to be used for NetBox plugins. 

57 """ 

58 # Plugin metadata 

59 author = '' 

60 author_email = '' 

61 description = '' 

62 version = '' 

63 release_track = '' 

64 

65 # Root URL path under /plugins. If not set, the plugin's label will be used. 

66 base_url = None 

67 

68 # Minimum/maximum compatible versions of NetBox 

69 min_version = None 

70 max_version = None 

71 

72 # Default configuration parameters 

73 default_settings = {} 

74 

75 # Mandatory configuration parameters 

76 required_settings = [] 

77 

78 # Middleware classes provided by the plugin 

79 middleware = [] 

80 

81 # Django-rq queues dedicated to the plugin 

82 queues = [] 

83 

84 # Django apps to append to INSTALLED_APPS when plugin requires them. 

85 django_apps = [] 

86 

87 # Optional plugin resources 

88 search_indexes = None 

89 data_backends = None 

90 event_rule_actions = None 

91 graphql_schema = None 

92 jinja_filters = None 

93 # Extension resources load from ready() and must not import core GraphQL modules. Schemas load at assembly. 

94 graphql_type_extensions = None 

95 graphql_filter_extensions = None 

96 menu = None 

97 menu_items = None 

98 serializer_resolver = None 

99 template_extensions = None 

100 user_preferences = None 

101 events_pipeline = [] 

102 

103 def get_jinja_context(self): 

104 """ 

105 Return a dict of additional variables to inject into the Jinja template context 

106 when rendering ConfigTemplates. Override this in a PluginConfig subclass to expose 

107 plugin-managed data to config templates without requiring template authors to know 

108 internal model names. 

109 

110 The returned dict is merged into the template context after the standard 

111 ObjectType-based model population, so keys here can shadow the auto-populated 

112 entries if needed. 

113 """ 

114 return {} 

115 

116 def _load_resource(self, name): 

117 # Import from the configured path, if defined. 

118 if path := getattr(self, name, None): 

119 return import_string(f"{self.__module__}.{path}") 

120 

121 # Fall back to the default path. Only the module's own absence returns None, nested errors propagate. 

122 default_path = f'{self.__module__}.{DEFAULT_RESOURCE_PATHS[name]}' 

123 default_module, resource_name = default_path.rsplit('.', 1) 

124 try: 

125 module = import_module(default_module) 

126 except ModuleNotFoundError as exc: 

127 if exc.name and (default_module == exc.name or default_module.startswith(f'{exc.name}.')): 

128 return None 

129 raise 

130 return getattr(module, resource_name, None) 

131 

132 def ready(self): 

133 from netbox.models.features import register_models 

134 

135 # Register models 

136 register_models(*self.get_models()) 

137 

138 plugin_name = self.name.rsplit('.', 1)[-1] 

139 

140 # Register search extensions (if defined) 

141 search_indexes = self._load_resource('search_indexes') or [] 

142 for idx in search_indexes: 

143 register_search(idx) 

144 

145 # Register data backends (if defined) 

146 data_backends = self._load_resource('data_backends') or [] 

147 for backend in data_backends: 

148 register_data_backend()(backend) 

149 

150 # Register event rule actions (if defined) 

151 event_rule_actions = self._load_resource('event_rule_actions') or [] 

152 for action in event_rule_actions: 

153 register_event_rule_action(action) 

154 

155 # Register Jinja filters (if defined) 

156 if jinja_filters := self._load_resource('jinja_filters'): 

157 register_jinja_filters(jinja_filters) 

158 

159 # Register template content (if defined) 

160 if template_extensions := self._load_resource('template_extensions'): 

161 register_template_extensions(template_extensions) 

162 

163 # Register navigation menu and/or menu items (if defined) 

164 if menu := self._load_resource('menu'): 

165 register_menu(menu) 

166 if menu_items := self._load_resource('menu_items'): 

167 register_menu_items(self.verbose_name, menu_items) 

168 

169 # Register GraphQL type & filter extensions (if defined) 

170 if graphql_type_extensions := self._load_resource('graphql_type_extensions'): 

171 register_graphql_type_extensions(graphql_type_extensions) 

172 if graphql_filter_extensions := self._load_resource('graphql_filter_extensions'): 

173 register_graphql_filter_extensions(graphql_filter_extensions) 

174 

175 # Register user preferences (if defined) 

176 if user_preferences := self._load_resource('user_preferences'): 

177 register_user_preferences(plugin_name, user_preferences) 

178 

179 # Register serializer resolver (if defined) 

180 if self.serializer_resolver: 

181 resolver_path = f"{self.__module__}.{self.serializer_resolver}" 

182 try: 

183 resolver = import_string(resolver_path) 

184 except ImportError as e: 

185 raise ImproperlyConfigured( 

186 f"Invalid serializer resolver path for plugin {self.__module__}: {resolver_path}" 

187 ) from e 

188 register_serializer_resolver(self.label, resolver) 

189 

190 @classmethod 

191 def validate(cls, user_config, netbox_version): 

192 

193 # Enforce version constraints 

194 current_version = version.parse(netbox_version) 

195 if cls.min_version is not None: 

196 min_version = version.parse(cls.min_version) 

197 if current_version < min_version: 

198 raise IncompatiblePluginError( 

199 f"Plugin {cls.__module__} requires NetBox minimum version {cls.min_version} (current: " 

200 f"{netbox_version})." 

201 ) 

202 if cls.max_version is not None: 

203 max_version = version.parse(cls.max_version) 

204 if current_version > max_version: 

205 raise IncompatiblePluginError( 

206 f"Plugin {cls.__module__} requires NetBox maximum version {cls.max_version} (current: " 

207 f"{netbox_version})." 

208 ) 

209 

210 # Verify required configuration settings 

211 for setting in cls.required_settings: 

212 if setting not in user_config: 

213 raise ImproperlyConfigured( 

214 f"Plugin {cls.__module__} requires '{setting}' to be present in the PLUGINS_CONFIG section of " 

215 f"configuration.py." 

216 ) 

217 

218 # Apply default configuration values 

219 for setting, value in cls.default_settings.items(): 

220 if setting not in user_config: 

221 user_config[setting] = value 

222 

223 

224def _load_plugin_graphql_schemas(): 

225 """ 

226 Load and register every installed plugin's GraphQL schema resource. Runs during root schema assembly, after 

227 all plugins have initialized, so plugin schema modules may import core GraphQL types freely. 

228 """ 

229 configs = {config.name: config for config in apps.get_app_configs()} 

230 for plugin_name in registry['plugins']['installed']: 230 ↛ 231line 230 didn't jump to line 231 because the loop on line 230 never started

231 if (config := configs.get(plugin_name)) is None: 

232 raise ImproperlyConfigured( 

233 f"Plugin '{plugin_name}' has no AppConfig named after its PLUGINS entry. PluginConfig.name " 

234 f"must match the configured plugin name." 

235 ) 

236 if graphql_schema := config._load_resource('graphql_schema'): 

237 # Avoid duplicate registration if the loader is invoked more than once. 

238 registered = registry['plugins']['graphql_schemas'] 

239 register_graphql_schema([cls for cls in graphql_schema if cls not in registered])