Coverage for core/forms/model_forms.py: 39%

1import copy 

2import json 

3 

4from django import forms 

5from django.conf import settings 

6from django.forms.fields import JSONField as _JSONField 

7from django.utils.translation import gettext_lazy as _ 

8 

9from core.forms.mixins import SyncedDataMixin 

10from core.models import * 

11from netbox.config import PARAMS, get_config 

12from netbox.forms import NetBoxModelForm, PrimaryModelForm 

13from netbox.registry import registry 

14from netbox.utils import get_data_backend_choices 

15from utilities.forms import get_field_value 

16from utilities.forms.fields import ChoiceField, JSONField 

17from utilities.forms.rendering import FieldSet 

18from utilities.forms.widgets import HTMXSelect 

19 

20__all__ = ( 

21 'ConfigRevisionForm', 

22 'DataSourceForm', 

23 'ManagedFileForm', 

24) 

25 

26EMPTY_VALUES = ('', None, [], ()) 

27 

28 

29class DataSourceForm(PrimaryModelForm): 

30 type = ChoiceField( 

31 choices=get_data_backend_choices, 

32 # No hx_target_id: changing type adds/removes the Backend Parameters fieldset entirely. 

33 widget=HTMXSelect() 

34 ) 

35 

36 class Meta: 

37 model = DataSource 

38 fields = [ 

39 'name', 'type', 'source_url', 'enabled', 'description', 'sync_interval', 'ignore_rules', 'owner', 

40 'comments', 'tags', 

41 ] 

42 widgets = { 

43 'ignore_rules': forms.Textarea( 

44 attrs={ 

45 'rows': 5, 

46 'class': 'font-monospace', 

47 'placeholder': '.cache\n*.txt\nsubdir/*' 

48 } 

49 ), 

50 } 

51 

52 @property 

53 def fieldsets(self): 

54 fieldsets = [ 

55 FieldSet( 

56 'name', 'type', 'source_url', 'description', 'tags', 'ignore_rules', name=_('Source') 

57 ), 

58 FieldSet('enabled', 'sync_interval', name=_('Sync')), 

59 ] 

60 if self.backend_fields: 

61 fieldsets.append( 

62 FieldSet(*self.backend_fields, name=_('Backend Parameters')) 

63 ) 

64 

65 return fieldsets 

66 

67 def __init__(self, *args, **kwargs): 

68 super().__init__(*args, **kwargs) 

69 # Determine the selected backend type 

70 backend_type = get_field_value(self, 'type') 

71 backend = registry['data_backends'].get(backend_type) 

72 

73 # Add backend-specific form fields 

74 self.backend_fields = [] 

75 if backend: 

76 for name, form_field in backend.parameters.items(): 

77 field_name = f'backend_{name}' 

78 self.backend_fields.append(field_name) 

79 self.fields[field_name] = copy.copy(form_field) 

80 if self.instance and self.instance.parameters: 

81 self.fields[field_name].initial = self.instance.parameters.get(name) 

82 

83 def save(self, *args, **kwargs): 

84 

85 parameters = {} 

86 for name in self.fields: 

87 if name.startswith('backend_'): 

88 parameters[name[8:]] = self.cleaned_data[name] 

89 self.instance.parameters = parameters 

90 

91 return super().save(*args, **kwargs) 

92 

93 

94class ManagedFileForm(SyncedDataMixin, NetBoxModelForm): 

95 upload_file = forms.FileField( 

96 required=False 

97 ) 

98 

99 fieldsets = ( 

100 FieldSet('upload_file', name=_('File Upload')), 

101 FieldSet('data_source', 'data_file', 'auto_sync_enabled', name=_('Data Source')), 

102 ) 

103 

104 class Meta: 

105 model = ManagedFile 

106 fields = ('data_source', 'data_file', 'auto_sync_enabled') 

107 

108 def clean(self): 

109 super().clean() 

110 

111 if self.cleaned_data.get('upload_file') and self.cleaned_data.get('data_file'): 

112 raise forms.ValidationError(_("Cannot upload a file and sync from an existing file")) 

113 if not self.cleaned_data.get('upload_file') and not self.cleaned_data.get('data_file'): 

114 raise forms.ValidationError(_("Must upload a file or select a data file to sync")) 

115 

116 return self.cleaned_data 

117 

118 

119class ConfigFormMetaclass(forms.models.ModelFormMetaclass): 

120 

121 def __new__(mcs, name, bases, attrs): 

122 

123 # Emulate a declared field for each supported configuration parameter 

124 param_fields = {} 

125 for param in PARAMS: 

126 field_kwargs = { 

127 'required': False, 

128 'label': param.label, 

129 'help_text': param.description, 

130 } 

131 field_kwargs.update(**param.field_kwargs) 

132 if param.field is _JSONField: 

133 # Replace with our own JSONField to get pretty JSON in config editor 

134 param.field = JSONField 

135 param_fields[param.name] = param.field(**field_kwargs) 

136 attrs.update(param_fields) 

137 

138 return super().__new__(mcs, name, bases, attrs) 

139 

140 

141class ConfigRevisionForm(forms.ModelForm, metaclass=ConfigFormMetaclass): 

142 """ 

143 Form for creating a new ConfigRevision. 

144 """ 

145 

146 fieldsets = ( 

147 FieldSet( 

148 'RACK_ELEVATION_DEFAULT_UNIT_HEIGHT', 'RACK_ELEVATION_DEFAULT_UNIT_WIDTH', name=_('Rack Elevations') 

149 ), 

150 FieldSet( 

151 'POWERFEED_DEFAULT_VOLTAGE', 'POWERFEED_DEFAULT_AMPERAGE', 'POWERFEED_DEFAULT_MAX_UTILIZATION', 

152 name=_('Power') 

153 ), 

154 FieldSet('ENFORCE_GLOBAL_UNIQUE', 'PREFER_IPV4', name=_('IPAM')), 

155 FieldSet('ALLOWED_URL_SCHEMES', name=_('Security')), 

156 FieldSet('BANNER_LOGIN', 'BANNER_MAINTENANCE', 'BANNER_TOP', 'BANNER_BOTTOM', name=_('Banners')), 

157 FieldSet('PAGINATE_COUNT', 'MAX_PAGE_SIZE', name=_('Pagination')), 

158 FieldSet('CUSTOM_VALIDATORS', 'PROTECTION_RULES', name=_('Validation')), 

159 FieldSet('DEFAULT_USER_PREFERENCES', name=_('User Preferences')), 

160 FieldSet('CHANGELOG_RETENTION', 'CHANGELOG_RETAIN_CREATE_LAST_UPDATE', name=_('Change Log')), 

161 FieldSet( 

162 'MAINTENANCE_MODE', 'COPILOT_ENABLED', 'GRAPHQL_ENABLED', 'JOB_RETENTION', 'MAPS_URL', 

163 name=_('Miscellaneous'), 

164 ), 

165 FieldSet('comment', name=_('Config Revision')) 

166 ) 

167 

168 class Meta: 

169 model = ConfigRevision 

170 fields = '__all__' 

171 widgets = { 

172 'comment': forms.Textarea(), 

173 } 

174 

175 def __init__(self, *args, **kwargs): 

176 super().__init__(*args, **kwargs) 

177 

178 # Append current parameter values to form field help texts and check for static configurations 

179 config = get_config() 

180 for param in PARAMS: 

181 value = getattr(config, param.name) 

182 

183 # Set the field's initial value, if it can be serialized. (This may not be the case e.g. for 

184 # CUSTOM_VALIDATORS, which may reference Python objects.) 

185 try: 

186 json.dumps(value) 

187 if type(value) in (tuple, list): 

188 self.fields[param.name].initial = ', '.join(value) 

189 else: 

190 self.fields[param.name].initial = value 

191 except TypeError: 

192 pass 

193 

194 # Check whether this parameter is statically configured (e.g. in configuration.py) 

195 if hasattr(settings, param.name): 

196 self.fields[param.name].disabled = True 

197 self.fields[param.name].help_text = _( 

198 'This parameter has been defined statically and cannot be modified.' 

199 ) 

200 continue 

201 

202 # Set the field's help text 

203 help_text = self.fields[param.name].help_text 

204 if help_text: 

205 help_text += '<br />' # Line break 

206 help_text += _('Current value: <strong>{value}</strong>').format(value=value or '&mdash;') 

207 if value == param.default: 

208 help_text += _(' (default)') 

209 self.fields[param.name].help_text = help_text 

210 

211 def save(self, commit=True): 

212 instance = super().save(commit=False) 

213 

214 # Populate JSON data on the instance 

215 instance.data = self.render_json() 

216 

217 if commit: 

218 instance.save() 

219 

220 return instance 

221 

222 def render_json(self): 

223 json = {} 

224 

225 # Iterate through each field and populate non-empty values 

226 for field_name in self.declared_fields: 

227 if field_name in self.cleaned_data and self.cleaned_data[field_name] not in EMPTY_VALUES: 

228 json[field_name] = self.cleaned_data[field_name] 

229 

230 return json