Coverage for extras/forms/scripts.py: 28%

1from django import forms 

2from django.core.files.storage import storages 

3from django.utils.translation import gettext_lazy as _ 

4 

5from core.choices import JobIntervalChoices, JobNotificationChoices 

6from core.forms import ManagedFileForm 

7from extras.utils import validate_script_content 

8from utilities.datetime import local_now 

9from utilities.forms.widgets import DateTimePicker, NumberWithOptions 

10 

11__all__ = ( 

12 'ScriptFileForm', 

13 'ScriptForm', 

14) 

15 

16 

17class ScriptForm(forms.Form): 

18 _commit = forms.BooleanField( 

19 required=False, 

20 initial=True, 

21 label=_("Commit changes"), 

22 help_text=_("Commit changes to the database (uncheck for a dry-run)") 

23 ) 

24 _schedule_at = forms.DateTimeField( 

25 required=False, 

26 widget=DateTimePicker(), 

27 label=_("Schedule at"), 

28 help_text=_("Schedule execution of script to a set time"), 

29 ) 

30 _interval = forms.IntegerField( 

31 required=False, 

32 min_value=1, 

33 label=_("Recurs every"), 

34 widget=NumberWithOptions( 

35 options=JobIntervalChoices 

36 ), 

37 help_text=_("Interval at which this script is re-run (in minutes)") 

38 ) 

39 _notifications = forms.ChoiceField( 

40 required=False, 

41 choices=JobNotificationChoices, 

42 initial=JobNotificationChoices.NOTIFICATION_ALWAYS, 

43 label=_("Notifications"), 

44 help_text=_("When to notify the user of job completion") 

45 ) 

46 

47 def __init__(self, *args, scheduling_enabled=True, **kwargs): 

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

49 

50 # Annotate the current system time for reference 

51 now = local_now().strftime('%Y-%m-%d %H:%M:%S %Z') 

52 self.fields['_schedule_at'].help_text += _(' (current time: <strong>{now}</strong>)').format(now=now) 

53 

54 # Remove scheduling fields if scheduling is disabled 

55 if not scheduling_enabled: 

56 self.fields.pop('_schedule_at') 

57 self.fields.pop('_interval') 

58 

59 def clean(self): 

60 scheduled_time = self.cleaned_data.get('_schedule_at') 

61 if scheduled_time and scheduled_time < local_now(): 

62 raise forms.ValidationError(_('Scheduled time must be in the future.')) 

63 

64 # When interval is used without schedule at, schedule for the current time 

65 if self.cleaned_data.get('_interval') and not scheduled_time: 

66 self.cleaned_data['_schedule_at'] = local_now() 

67 

68 # Fall back to the field's initial value if no notification preference was submitted 

69 # (e.g. when running a script via the "Run Script" button on the scripts list view) 

70 if not self.cleaned_data.get('_notifications'): 

71 self.cleaned_data['_notifications'] = self.fields['_notifications'].initial 

72 

73 return self.cleaned_data 

74 

75 

76class ScriptFileForm(ManagedFileForm): 

77 """ 

78 ManagedFileForm with a custom save method to use django-storages. 

79 """ 

80 def clean(self): 

81 super().clean() 

82 

83 if upload_file := self.cleaned_data.get('upload_file'): 

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

85 content = upload_file.read() 

86 upload_file.seek(0) 

87 try: 

88 validate_script_content(content, upload_file.name) 

89 except Exception as e: 

90 raise forms.ValidationError( 

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

92 ) 

93 elif data_file := self.cleaned_data.get('data_file'): 

94 # Validate scripts synced from a data source as well, to avoid creating a broken 

95 # script module that cannot be loaded or corrected 

96 try: 

97 validate_script_content(data_file.data, data_file.path) 

98 except Exception as e: 

99 raise forms.ValidationError( 

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

101 ) 

102 

103 return self.cleaned_data 

104 

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

106 # If a file was uploaded, save it to disk 

107 if self.cleaned_data['upload_file']: 

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

109 

110 filename = self.cleaned_data['upload_file'].name 

111 self.instance.file_path = filename 

112 data = self.cleaned_data['upload_file'] 

113 storage.save(filename, data) 

114 

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