Coverage for src/backend/InvenTree/InvenTree/backends.py: 46%

1"""Custom backend implementation for maintenance-mode.""" 

2 

3import datetime 

4import time 

5 

6from django.conf import settings 

7from django.core.exceptions import ValidationError 

8from django.core.mail.backends.base import BaseEmailBackend 

9from django.core.mail.backends.locmem import EmailBackend as LocMemEmailBackend 

10from django.core.mail.message import EmailMessage, EmailMultiAlternatives 

11from django.utils.module_loading import import_string 

12 

13import structlog 

14from maintenance_mode.backends import AbstractStateBackend 

15 

16import common.models 

17from common.settings import get_global_setting, set_global_setting 

18 

19logger = structlog.get_logger('inventree') 

20 

21 

22class InvenTreeMaintenanceModeBackend(AbstractStateBackend): 

23 """Custom backend for managing state of maintenance mode. 

24 

25 Stores a timestamp in the database to determine when maintenance mode will elapse. 

26 """ 

27 

28 SETTING_KEY = '_MAINTENANCE_MODE' 

29 

30 def get_value(self) -> bool: 

31 """Get the current state of the maintenance mode. 

32 

33 Returns: 

34 bool: True if maintenance mode is active, False otherwise. 

35 """ 

36 try: 

37 value = get_global_setting(self.SETTING_KEY) 

38 except Exception: 

39 # Database is inaccessible - assume we are not in maintenance mode 

40 logger.debug('Failed to read maintenance mode state - assuming False') 

41 return False 

42 

43 # Extract timestamp from string 

44 try: 

45 # If the timestamp is in the past, we are now *out* of maintenance mode 

46 timestamp = datetime.datetime.fromisoformat(value) 

47 return timestamp > datetime.datetime.now() 

48 except ValueError: 

49 # If the value is not a valid timestamp, assume maintenance mode is not active 

50 return False 

51 

52 def set_value(self, value: bool, retries: int = 5, minutes: int = 5): 

53 """Set the state of the maintenance mode. 

54 

55 Instead of simply writing "true" or "false" to the setting, 

56 we write a timestamp to the setting, which is used to determine 

57 when maintenance mode will elapse. 

58 This ensures that we will always *exit* maintenance mode after a certain time period. 

59 """ 

60 logger.debug('Setting maintenance mode state: %s', value) 

61 

62 if value: 

63 # Save as isoformat 

64 timestamp = datetime.datetime.now() + datetime.timedelta(minutes=minutes) 

65 timestamp = timestamp.isoformat() 

66 else: 

67 # Blank timestamp means maintenance mode is not active 

68 timestamp = '' 

69 

70 r = retries 

71 

72 while r > 0: 

73 try: 

74 set_global_setting(self.SETTING_KEY, timestamp) 

75 # Read the value back to confirm 

76 if self.get_value() == value: 

77 break 

78 except Exception: 

79 # In the database is locked, then 

80 logger.debug( 

81 'Failed to set maintenance mode state (%s retries left)', r 

82 ) 

83 time.sleep(0.1) 

84 

85 r -= 1 

86 

87 # Disable this warning message (for now) as it is confusing users with no upside 

88 # if r == 0: 

89 # logger.warning( 

90 # 'Failed to set maintenance mode state after %s retries', retries 

91 # ) 

92 

93 

94class InvenTreeMailLoggingBackend(BaseEmailBackend): 

95 """Backend that logs send mails to the database.""" 

96 

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

98 """Initialize the email backend.""" 

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

100 klass = import_string(settings.INTERNAL_EMAIL_BACKEND) 

101 self.backend: BaseEmailBackend = klass(*args, **kwargs) 

102 

103 def open(self): 

104 """Open the email backend connection.""" 

105 return self.backend.open() 

106 

107 def close(self): 

108 """Close the email backend connection.""" 

109 return self.backend.open() 

110 

111 def send_messages( 

112 self, email_messages: list[EmailMessage | EmailMultiAlternatives] 

113 ) -> int: 

114 """Send email messages and log them to the database. 

115 

116 Args: 

117 email_messages (list): List of EmailMessage objects to send. 

118 """ 

119 from plugin.base.mail.mail import process_mail_out 

120 

121 # Issue mails to plugins 

122 process_mail_out(email_messages) 

123 

124 # Process 

125 msg_ids: list[common.models.EmailMessage] = [] 

126 try: 

127 msg_ids = common.models.log_email_messages(email_messages) 

128 except Exception: # pragma: no cover 

129 logger.exception('INVE-W10: Problem logging recipients, ignoring') 

130 

131 # Anymail: pre-processing 

132 if settings.INTERNAL_EMAIL_BACKEND.startswith('anymail.backends.'): 

133 for a in email_messages: 

134 if a.extra_headers and common.models.HEADER_MSG_ID in a.extra_headers: 

135 # Remove the Message-ID header from the email 

136 # This is because some ESPs do not like it being set 

137 # in the headers, and will ignore the email 

138 a.extra_headers.pop(common.models.HEADER_MSG_ID) 

139 # Add tracking if requested: TODO 

140 # a.track_opens = True 

141 

142 # Send 

143 try: 

144 ret_val = self.backend.send_messages(email_messages) 

145 except Exception as e: 

146 logger.exception('INVE-W10: Problem sending email: %s', e) 

147 # If we fail to send the email, we need to set the status to ERROR 

148 for msg in msg_ids: 

149 msg.status = common.models.EmailMessage.EmailStatus.FAILED 

150 msg.error_message = str(e) 

151 msg.save() 

152 raise ValidationError(f'INVE-W10: Failed to send email: {e}') from e 

153 

154 # Log 

155 if ret_val == 0: # pragma: no cover 

156 logger.info('INVE-W10: No emails sent') 

157 else: 

158 logger.info('INVE-W10: %s emails sent', ret_val) 

159 if settings.INTERNAL_EMAIL_BACKEND.startswith('anymail.backends.'): 

160 # Anymail: ESP does return the message ID for us so we need to set it in the database 

161 for k, v in { 

162 a.extra_headers['X-InvenTree-MsgId-1']: a.anymail_status.message_id 

163 for a in email_messages 

164 }.items(): 

165 current = common.models.EmailMessage.objects.get(global_id=k) 

166 current.message_id_key = v 

167 current.status = common.models.EmailMessage.EmailStatus.SENT 

168 current.save() 

169 else: 

170 common.models.EmailMessage.objects.filter( 

171 pk__in=[msg.pk for msg in msg_ids] 

172 ).update(status=common.models.EmailMessage.EmailStatus.SENT) 

173 return ret_val 

174 

175 

176class InvenTreeErrorMailBackend(LocMemEmailBackend): 

177 """Backend that generates an error when sending - for testing.""" 

178 

179 def send_messages(self, email_messages): 

180 """Issues an error when sending email messages.""" 

181 super().send_messages(email_messages) 

182 # Simulate an error 

183 raise ValueError('Test error sending email')