Coverage for extras/api/serializers_/scripts.py: 43%

1import logging 

2 

3from django.core.files.storage import storages 

4from django.db import IntegrityError, router, transaction 

5from django.utils.translation import gettext_lazy as _ 

6from drf_spectacular.utils import extend_schema_field 

7from rest_framework import serializers 

8 

9from core.api.serializers_.jobs import JobSerializer 

10from core.choices import JobNotificationChoices, ManagedFileRootPathChoices 

11from extras.models import Script, ScriptModule 

12from extras.utils import validate_script_content 

13from netbox.api.serializers import ValidatedModelSerializer 

14from utilities.datetime import local_now 

15 

16logger = logging.getLogger(__name__) 

17 

18__all__ = ( 

19 'ScriptDetailSerializer', 

20 'ScriptInputSerializer', 

21 'ScriptModuleSerializer', 

22 'ScriptSerializer', 

23) 

24 

25 

26class ScriptModuleSerializer(ValidatedModelSerializer): 

27 file = serializers.FileField(write_only=True) 

28 file_path = serializers.CharField(read_only=True) 

29 

30 class Meta: 

31 model = ScriptModule 

32 fields = ['id', 'display', 'file_path', 'file', 'created', 'last_updated'] 

33 brief_fields = ('id', 'display') 

34 

35 def validate(self, data): 

36 # ScriptModule.save() sets file_root; inject it here so full_clean() succeeds. 

37 # Pop 'file' before model instantiation — ScriptModule has no such field. 

38 file = data.pop('file', None) 

39 data['file_root'] = ManagedFileRootPathChoices.SCRIPTS 

40 

41 if self.instance is None: 41 ↛ 47line 41 didn't jump to line 47 because the condition on line 41 was always true

42 # Reject duplicates before writing to storage so a failed upload can't touch the existing file 

43 if file is not None and ScriptModule.objects.filter( 43 ↛ 46line 43 didn't jump to line 46 because the condition on line 43 was never true

44 file_root=ManagedFileRootPathChoices.SCRIPTS, file_path=file.name 

45 ).exists(): 

46 raise serializers.ValidationError(_("A script module with this file name already exists.")) 

47 elif file is None: 

48 # Replacing a module's content requires a file upload, even for a partial update 

49 raise serializers.ValidationError({'file': _("This field is required.")}) 

50 elif file.name != self.instance.file_path: 

51 raise serializers.ValidationError({ 

52 'file': _( 

53 "The uploaded file name must match the existing file path ({path})." 

54 ).format(path=self.instance.file_path) 

55 }) 

56 

57 data = super().validate(data) 

58 data.pop('file_root', None) 

59 if file is not None: 59 ↛ 70line 59 didn't jump to line 70 because the condition on line 59 was always true

60 # Validate that the uploaded script can be loaded as a Python module 

61 content = file.read() 

62 file.seek(0) 

63 try: 

64 validate_script_content(content, file.name) 

65 except Exception as e: 

66 raise serializers.ValidationError( 

67 _("Error loading script: {error}").format(error=e) 

68 ) 

69 data['file'] = file 

70 return data 

71 

72 def create(self, validated_data): 

73 file = validated_data.pop('file') 

74 storage = storages.create_storage(storages.backends["scripts"]) 

75 validated_data['file_path'] = storage.save(file.name, file) 

76 created = False 

77 try: 

78 instance = super().create(validated_data) 

79 created = True 

80 return instance 

81 except IntegrityError as e: 

82 if 'file_path' in str(e): 

83 raise serializers.ValidationError( 

84 _("A script module with this file name already exists.") 

85 ) 

86 raise 

87 finally: 

88 # Don't delete a path another ScriptModule still references (e.g. a concurrent upload won the race) 

89 file_path = validated_data.get('file_path') 

90 if not created and file_path and not ScriptModule.objects.filter( 

91 file_root=ManagedFileRootPathChoices.SCRIPTS, file_path=file_path 

92 ).exists(): 

93 try: 

94 storage.delete(file_path) 

95 except Exception: 

96 logger.warning(f"Failed to delete orphaned script file '{file_path}' from storage.") 

97 

98 def update(self, instance, validated_data): 

99 file = validated_data.pop('file') 

100 storage = storages.create_storage(storages.backends["scripts"]) 

101 

102 # Overwrite the existing file in place, keeping file_path stable 

103 file.seek(0) 

104 saved_path = storage.save(instance.file_path, file) 

105 if saved_path != instance.file_path: 

106 # The backend saved under an alternate name instead of overwriting; drop the orphan and reject 

107 try: 

108 storage.delete(saved_path) 

109 except Exception: 

110 logger.warning(f"Failed to delete orphaned script file '{saved_path}' from storage.") 

111 raise serializers.ValidationError({ 

112 'file': _( 

113 "The scripts storage backend did not overwrite the existing file. Ensure the " 

114 "backend is configured to allow overwrites." 

115 ) 

116 }) 

117 

118 # Discard any cached class discovery so save() re-syncs from the new content 

119 instance.__dict__.pop('module_scripts', None) 

120 instance.last_updated = local_now() 

121 # Keep Script row sync all-or-nothing; the storage write above cannot join the transaction 

122 with transaction.atomic(using=router.db_for_write(ScriptModule)): 

123 instance.save() 

124 

125 return instance 

126 

127 

128class ScriptSerializer(ValidatedModelSerializer): 

129 description = serializers.SerializerMethodField(read_only=True) 

130 vars = serializers.SerializerMethodField(read_only=True) 

131 result = JobSerializer(nested=True, read_only=True) 

132 

133 class Meta: 

134 model = Script 

135 fields = [ 

136 'id', 'url', 'display_url', 'module', 'name', 'description', 'vars', 'result', 'display', 'is_executable', 

137 ] 

138 brief_fields = ('id', 'url', 'display', 'name', 'description') 

139 

140 @extend_schema_field(serializers.JSONField(allow_null=True)) 

141 def get_vars(self, obj): 

142 if obj.python_class: 

143 return { 

144 k: v.__class__.__name__ for k, v in obj.python_class()._get_vars().items() 

145 } 

146 return {} 

147 

148 @extend_schema_field(serializers.CharField()) 

149 def get_display(self, obj): 

150 return f'{obj.name} ({obj.module})' 

151 

152 @extend_schema_field(serializers.CharField(allow_null=True)) 

153 def get_description(self, obj): 

154 if obj.python_class: 

155 return obj.python_class().description 

156 return None 

157 

158 

159class ScriptDetailSerializer(ScriptSerializer): 

160 result = serializers.SerializerMethodField(read_only=True) 

161 

162 @extend_schema_field(JobSerializer()) 

163 def get_result(self, obj): 

164 job = obj.jobs.all().order_by('-created').first() 

165 context = { 

166 'request': self.context['request'] 

167 } 

168 data = JobSerializer(job, context=context).data 

169 return data 

170 

171 

172class ScriptInputSerializer(serializers.Serializer): 

173 data = serializers.JSONField() 

174 commit = serializers.BooleanField() 

175 schedule_at = serializers.DateTimeField(required=False, allow_null=True) 

176 interval = serializers.IntegerField(required=False, allow_null=True) 

177 notifications = serializers.ChoiceField( 

178 choices=JobNotificationChoices, 

179 required=False, 

180 default=JobNotificationChoices.NOTIFICATION_ALWAYS, 

181 ) 

182 

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

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

185 

186 # Default to script's Meta.notifications_default if set 

187 script = self.context.get('script') 

188 if script and script.python_class: 188 ↛ 189line 188 didn't jump to line 189 because the condition on line 188 was never true

189 self.fields['notifications'].default = script.python_class.notifications_default 

190 

191 def validate_data(self, value): 

192 """ 

193 Validates that the script input is an object mapping variable names to values. 

194 """ 

195 if not isinstance(value, dict): 

196 raise serializers.ValidationError( 

197 _('Invalid data payload; expected an object mapping variable names to values.') 

198 ) 

199 return value 

200 

201 def validate_schedule_at(self, value): 

202 """ 

203 Validates the specified schedule time for a script execution. 

204 """ 

205 if value: 

206 if not self.context['script'].python_class.scheduling_enabled: 

207 raise serializers.ValidationError(_('Scheduling is not enabled for this script.')) 

208 if value < local_now(): 

209 raise serializers.ValidationError(_('Scheduled time must be in the future.')) 

210 return value 

211 

212 def validate_interval(self, value): 

213 """ 

214 Validates the provided interval based on the script's scheduling configuration. 

215 """ 

216 if value and not self.context['script'].python_class.scheduling_enabled: 

217 raise serializers.ValidationError(_('Scheduling is not enabled for this script.')) 

218 return value 

219 

220 def validate(self, data): 

221 """ 

222 Validates the given data and ensures the necessary fields are populated. 

223 """ 

224 # Set the schedule_at time to now if only an interval is provided 

225 # while handling the case where schedule_at is null. 

226 if data.get('interval') and not data.get('schedule_at'): 

227 data['schedule_at'] = local_now() 

228 

229 return super().validate(data)