Coverage for ingadhoc-odoo-saas-adhoc / saas_provider_upgrade / models / saas_provider_upgrade_util.py: 54%

1############################################################################## 

2# For copyright and license notices, see __manifest__.py file in module root 

3# directory 

4############################################################################## 

5import difflib 

6import json 

7import re 

8from typing import Any, Literal 

9 

10import requests 

11from markupsafe import Markup 

12from odoo import api, models 

13from odoo.models import BaseModel 

14from odoo.tools import safe_eval 

15 

16from ..constants import UL_TYPES_FREEZE_IGNORES 

17 

18 

19class SaasProviderUpgradeUtil(models.AbstractModel): 

20 _name = "saas.provider.upgrade.util" 

21 _description = "saas.provider.upgrade.util" 

22 

23 @api.model 

24 def build_display_notification_action( 

25 self, title: str, message: str, type: Literal["success", "warning", "danger", "info"] = "success" 

26 ) -> dict[str, Any]: 

27 """ 

28 Used to build "display_notification" action. Default type is "success" 

29 

30 :param title: Notification title 

31 :param message: Notification message 

32 :param type: Notification type (success, warning, danger, info) 

33 :return: A dictionary representing the notification action 

34 """ 

35 return { 

36 "type": "ir.actions.client", 

37 "tag": "display_notification", 

38 "params": { 

39 "title": title, 

40 "type": type, 

41 "message": message, 

42 "next": {"type": "ir.actions.act_window_close"}, 

43 }, 

44 } 

45 

46 @api.model 

47 def create_message(self, title: str, body: str) -> Markup: 

48 """ 

49 Creates a message with a title and a body. It uses 'type' to determine the message type. 

50 By default is an error message. 

51 

52 :param title: The title of the message 

53 :param body: The body of the message (can contain HTML) 

54 """ 

55 message = (Markup("<p><b>") + "ERROR: " + Markup("</b>") + "%s") % (title,) 

56 message += Markup("<blockquote>%s</blockquote>") % (Markup(body)) 

57 return message 

58 

59 @api.model 

60 def urgent_communication(self, msg: str): 

61 """ 

62 Method used to send a message to the Telegram channel configured in the saas_provider_upgrade module. 

63 Is used in the after_done UL scripts, and in the after_error scripts in the Upgrade Types. 

64 

65 :param msg: The message to send to the Telegram channel 

66 """ 

67 chat_id = self.env["ir.config_parameter"].sudo().get_param("saas_provider_upgrade.telegram_chat_id", False) 

68 bot_token = self.env["ir.config_parameter"].sudo().get_param("saas_provider_upgrade.telegram_bot_token", False) 

69 channel = self._get_channel_upgrade() 

70 url = f"https://api.telegram.org/bot{bot_token}/sendMessage" 

71 if bot_token and chat_id: 

72 response = requests.post( 

73 url, 

74 json={"chat_id": chat_id, "text": msg}, 

75 headers={"Content-Type": "application/json"}, 

76 timeout=30, 

77 ) 

78 if response.status_code != 200: 

79 msg += f" (Fail posting on telegram with: error {response.status_code}" 

80 channel.message_post( 

81 body=msg, 

82 subtype_id=self.env.ref("mail.mt_comment").id, 

83 message_type="comment", 

84 body_is_html=True, 

85 ) 

86 

87 if not (bot_token and chat_id): 

88 channel.message_post( 

89 body=msg, 

90 subtype_id=self.env.ref("mail.mt_comment").id, 

91 message_type="comment", 

92 body_is_html=True, 

93 ) 

94 

95 @api.model 

96 def similar_content(self, content_1: str, content_2: str, threshold: float = 0.85): 

97 """ 

98 Check if two contents have similar content based on a threshold. 

99 

100 :param content_1: First content 

101 :param content_2: Second content 

102 :param threshold: Similarity threshold (default is 0.85) 

103 :return: True if contents are similar, False otherwise 

104 """ 

105 if content_1 and content_2: 

106 ratio = difflib.SequenceMatcher(None, content_1, content_2).ratio() 

107 return ratio >= threshold 

108 

109 return False 

110 

111 @api.model 

112 def is_serializable(self, value: Any) -> bool: 

113 """ 

114 Method to check if a value is JSON serializable. 

115 

116 :param value: Value to check 

117 :return: True if the value is serializable, False otherwise 

118 """ 

119 try: 

120 json.dumps(value) 

121 return True 

122 except Exception: 

123 return False 

124 

125 def action_reload(self) -> dict: 

126 """ 

127 Reload the client. 

128 

129 :return: Action to reload the client 

130 """ 

131 return {"type": "ir.actions.client", "tag": "soft_reload"} 

132 

133 @api.model 

134 def _production_freeze_ignore(self, ul_type: str) -> bool: 

135 """ 

136 Check if the upgrade line type should be ignored during production freeze. 

137 

138 :param ul_type: Upgrade line type 

139 :return: True if the upgrade line type should be ignored, False otherwise 

140 """ 

141 return ul_type in UL_TYPES_FREEZE_IGNORES 

142 

143 @api.model 

144 def get_portal_requests_domain(self, ticket_id: int) -> list: 

145 return [ 

146 ("ticket_id.id", "=", ticket_id), 

147 ("hide_from_portal", "=", False), 

148 "&", 

149 ("aim", "!=", "production"), 

150 "|", 

151 "&", 

152 ("original_database_id", "=", False), 

153 ("upgraded_database_id", "=", False), 

154 ("state", "!=", "cancel"), 

155 "|", 

156 ("original_database_state", "!=", "deleted"), 

157 ("upgraded_database_state", "!=", "deleted"), 

158 ("next_request_id", "=", False), 

159 ] 

160 

161 @api.model 

162 def _get_filters_eval_context(self) -> dict: 

163 return { 

164 "datetime": safe_eval.datetime, 

165 "dateutil": safe_eval.dateutil, 

166 "today": safe_eval.datetime.datetime.today(), 

167 "time": safe_eval.time, 

168 "user": self.env.user, 

169 } 

170 

171 @api.model 

172 def extract_qweb_variables(self, message: str) -> set[str]: 

173 """ 

174 Extracts variable names used in a QWeb message template. 

175 

176 Matches attributes like `t-out="var"`, `t-esc='var'`, `t-if="cond"`, 

177 `t-elif`, `t-foreach`, `t-set`, etc., returning the variable names found. 

178 

179 :param message: The QWeb template message to analyze 

180 :return: Set of variable names used in the template 

181 """ 

182 # Pattern to match t-* attributes with simple variable values 

183 pattern = r't-(?:out|esc|if|elif|foreach|set)\s*=\s*["\']([a-zA-Z_][a-zA-Z0-9_]*)["\']' 

184 return set(re.findall(pattern, message)) 

185 

186 @api.model 

187 def json_to_display(self, val: dict | list) -> str: 

188 try: 

189 return json.dumps(val, ensure_ascii=False, separators=(",", ": ")) 

190 except Exception: 

191 return str(val) 

192 

193 @api.model 

194 def _get_channel_upgrade(self) -> BaseModel: 

195 """Retrieve the id of the upgrade discuss channel 

196 

197 Returns: 

198 Recordset: Record of the upgrade discuss channel 

199 """ 

200 channel_id = self.env["ir.config_parameter"].sudo().get_param("saas_provider_upgrade.discuss_channel_id", "0") 

201 return self.env["discuss.channel"].sudo().browse(int(channel_id))