Coverage for ingadhoc-odoo-saas-adhoc / saas_provider_upgrade / models / saas_upgrade_line_request_log.py: 30%
107 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-09 18:05 +0000
1import datetime
2import logging
3import re
4from typing import TYPE_CHECKING, Literal
6import odoo
7from odoo import _, api, fields, models
9if TYPE_CHECKING:
10 from ..models.helpdesk_ticket_upgrade_request import HelpdeskTicketUpgradeRequest as UpgradeRequest
11 from ..models.saas_upgrade_line import SaasUpgradeLine as UpgradeLine
12 from ..models.saas_upgrade_line_request_log import SaasUpgradeLineRequestLog as UpgradeLog
13 from ..models.saas_upgrade_line_request_log_entry import SaasUpgradeLineRequestLogEntry as UpgradeLogEntry
15_logger = logging.getLogger(__name__)
18class SaasUpgradeLineRequestLog(models.Model):
19 _name = "saas.upgrade.line.request.log"
20 _description = "Upgrade line request log"
21 _inherit = ["mail.thread", "mail.activity.mixin", "saas.provider.upgrade.util"]
22 _order = "id desc"
24 upgrade_line_id = fields.Many2one(
25 "saas.upgrade.line",
26 ondelete="cascade",
27 readonly=True,
28 )
29 type = fields.Selection(
30 [("info", "Info"), ("warning", "Warning"), ("error", "Error")],
31 default="error",
32 readonly=True,
33 )
34 first_content = fields.Text(
35 readonly=True,
36 )
37 entry_ids = fields.One2many(
38 "saas.upgrade.line.request.log.entry",
39 "log_id",
40 readonly=True,
41 help="Entries of the log",
42 )
43 adhoc_product_id = fields.Many2one(
44 related="upgrade_line_id.adhoc_product_id",
45 domain=[("parent_id", "!=", False), ("type", "=", "horizontal")],
46 )
47 ticket_id = fields.Many2one(
48 "helpdesk.ticket",
49 string="Fixing Ticket",
50 readonly=True,
51 help="Ticket where the entry is being or was fixed",
52 )
53 stage_id = fields.Many2one(
54 related="ticket_id.stage_id",
55 store=True,
56 )
57 solved = fields.Boolean(
58 compute="_compute_solved",
59 store=True,
60 readonly=False,
61 )
63 @api.depends("ticket_id.stage_id")
64 def _compute_solved(self):
65 for rec in self:
66 if rec.ticket_id: 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true
67 rec.solved = rec.ticket_id.stage_id.fold
69 @api.depends("upgrade_line_id", "type")
70 def _compute_display_name(self):
71 type_map = {
72 "error": "E",
73 "warning": "W",
74 "info": "I",
75 }
76 for rec in self:
77 ul_name = rec.upgrade_line_id.name or ""
78 msg_type = type_map.get(rec.type, "")
79 rec.display_name = f"{msg_type} - {ul_name}"
81 def write(self, vals):
82 solved = vals.get("solved")
83 if solved:
84 for rec in self.entry_ids.mapped("request_id").filtered(
85 lambda r: r.status == "error" and r.state not in ["done", "cancel"] and not r.automatic
86 ):
87 rec.action_activate()
88 return super().write(vals)
90 def action_run(self):
91 """
92 Action to run the entries of the log. Called from form view.
93 """
94 self.ensure_one()
95 return self.entry_ids.run()
97 @api.model
98 def create_new_entry(
99 self,
100 request: "UpgradeRequest",
101 upgrade_line: "UpgradeLine",
102 type: Literal["info", "warning", "error"],
103 content: str,
104 ) -> tuple["UpgradeLog", "UpgradeLogEntry"]:
105 """
106 Function used to create a new log entry.
107 If an existing and similar log is found, it will add the entry to that log.
108 If no existing and similar log is found, it will create a new log with the entry.
110 :param request: Request record
111 :param upgrade_line: Upgrade Line record
112 :param type: Type of the log entry (info, warning, error)
113 :param content: Content of the log entry
114 :return: Tuple of (log, entry)
115 """
116 # Search for similar UL and type logs
117 self = self.sudo() # Ensure we have access to all logs and entries for the upgrade line
118 logs = self.search(
119 [
120 ("upgrade_line_id", "=", upgrade_line.id),
121 ("type", "=", type),
122 ("solved", "=", False),
123 ],
124 )
126 # If exists, content similar log, it must be one
127 if similar_logs := logs.filtered(lambda log: self.similar_content(log.first_content, content)):
128 log = similar_logs[0]
129 else:
130 log = self.create(
131 {
132 "upgrade_line_id": upgrade_line.id,
133 "type": type,
134 "first_content": content,
135 }
136 )
138 if entries := log.entry_ids.filtered(lambda e: e.content == content and e.request_id == request):
139 entry = entries[0]
140 entry.write({"solved": False})
141 else:
142 entry = log.entry_ids.create(
143 {
144 "log_id": log.id,
145 "request_id": request.id,
146 "content": content,
147 "script_id": upgrade_line.actual_script_id.id,
148 }
149 )
150 return log, entry
152 @api.model
153 def log_message(
154 self,
155 upgrade_line: "UpgradeLine",
156 request: "UpgradeRequest",
157 msg: str,
158 msg_type: Literal["info", "warning"] = "info",
159 ):
160 """
161 Creates a log entry for the upgrade line in the request
163 :param upgrade_line: The upgrade line record
164 :param msg: The message to log
165 :param msg_type: The type of message ('info' or 'warning')
166 """
167 test_dev = self.env.context.get("test_dev", False)
168 if msg_type not in ["info", "warning"]:
169 raise ValueError(_("Invalid message type, it must be 'info' or 'warning'"))
170 if not test_dev:
171 self.env["saas.upgrade.line.request.log"].create_new_entry(
172 request=request,
173 upgrade_line=upgrade_line,
174 type=msg_type,
175 content=msg,
176 )
178 @api.model
179 def create_entries_from_json(self, request: "UpgradeRequest", upgrade_line: "UpgradeLine", vals_list: list[dict]):
180 """
181 Function to create a log from a list of values in JSON format.
183 :param request: The upgrade request record
184 :param upgrade_line: The upgrade line record
185 :param vals_list: List of dictionaries containing log entry values
186 """
187 if self.env.context.get("test_dev", False):
188 return
190 for vals in vals_list:
191 content = vals.get("message", "")
192 if not content:
193 continue
195 self.with_user(odoo.SUPERUSER_ID).create_new_entry(
196 request=request,
197 upgrade_line=upgrade_line,
198 type=vals.get("type", "info"),
199 content=content,
200 )
202 @api.model
203 def _parse(self, content: str):
204 """
205 Method to parse the log and return a formatted version.
207 :param log: The log content to be parsed
208 :return: Formatted log content
209 """
210 SPLIT_WORD = "while evaluating"
211 content = content.split(SPLIT_WORD)[0]
212 content = re.sub(r"line (\d+)", r"\nline \1", content)
214 lines = content.splitlines()
215 new_lines = []
217 for line in lines:
218 flag = True
219 while flag:
220 new_line, rest = self._split_by_max_len(line)
221 new_lines += [new_line]
222 if not rest or len(rest) == 1:
223 flag = False
224 else:
225 line = rest
226 content = "\n".join(new_lines)
227 return content
229 @api.model
230 def _split_by_max_len(self, line: str) -> tuple[str, str | None]:
231 """
232 Function to split a line by the maximum length allowed for logs.
234 :param line: The line to be split
235 :return: tuple of (new_line, rest)
236 """
237 LOG_MAX_LINE_LEN = 200
238 tokens = line.split(" ")
239 new_line = ""
240 rest = None
241 for i in range(len(tokens)):
242 token = tokens[i]
243 if len(new_line) + len(token) > LOG_MAX_LINE_LEN:
244 rest = "\t" + " ".join(tokens[i:])
245 break
246 new_line += token + " "
247 return new_line, rest
249 @api.autovacuum
250 def _gc_childless_logs(self):
251 """
252 Method to delete logs without entries older than 1 day.
253 """
254 cutoff_date = fields.Datetime.now() - datetime.timedelta(days=1)
255 logs_to_delete = self.search(
256 [
257 ("create_date", "<", cutoff_date),
258 ("entry_ids", "=", False),
259 ]
260 )
261 if logs_to_delete:
262 _logger.info("Autovacuum: Deleting %d childless upgrade logs", len(logs_to_delete))
263 logs_to_delete.with_context(bypass_delete=True).unlink()
264 _logger.info("Autovacuum: Successfully deleted childless upgrade logs")