Coverage for ingadhoc-odoo-saas-adhoc / saas_provider_upgrade / models / helpdesk_ticket.py: 49%
282 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-09 18:05 +0000
1import logging
2import math
3from ast import literal_eval
4from datetime import datetime, timedelta
5from typing import Literal
7from markupsafe import Markup
8from odoo import _, api, fields, models
9from odoo.exceptions import RedirectWarning, UserError, ValidationError
10from odoo.tools.safe_eval import safe_eval
11from werkzeug.urls import url_encode
13_logger = logging.getLogger(__name__)
16class HelpdeskTicket(models.Model):
17 _name = "helpdesk.ticket"
18 _inherit = ["helpdesk.ticket", "saas.provider.upgrade.util"]
19 _mailing_enabled = True
21 upgrade_type_id = fields.Many2one("saas.upgrade.type", string="Upgrade Type", tracking=True)
22 upgrade_team = fields.Boolean(related="team_id.upgrade_team", related_sudo=True, store=True)
23 request_line_ids = fields.One2many(
24 "helpdesk.ticket.upgrade.request", "ticket_id", "Requests", domain=[("next_request_id", "=", False)]
25 )
26 all_request_line_ids = fields.One2many("helpdesk.ticket.upgrade.request", "ticket_id", "All Requests")
27 customer_note_ids = fields.One2many("helpdesk.ticket.customer_note", "ticket_id", "Customer Notes")
28 custom_script_ids = fields.Many2many("saas.upgrade.line")
29 environment_variables_dict = fields.Text(
30 default="{}",
31 required=False,
32 help="Dictionary to send environment variables to odoo upgrade service where the key is the variable "
33 "and the value is the value",
34 )
35 consultant_upgrade_id = fields.Many2one(
36 "res.users",
37 string="Upgrade Consultant",
38 compute="_compute_consultant_upgrade_id",
39 store=True,
40 readonly=False,
41 )
42 team_technician_ids = fields.Many2many(
43 "res.users",
44 related="team_id.technician_member_ids",
45 )
46 assigned_technician_id = fields.Many2one(
47 "res.users",
48 string="Técnico asignado",
49 tracking=True,
50 compute="_compute_assigned_technician_id",
51 store=True,
52 readonly=False,
53 help="Responsible technician who will be notified in case of errors during the execution of requests",
54 domain="[('id', 'in', team_technician_ids)]",
55 )
56 customer_note_count = fields.Integer(compute="_compute_customer_note_count", compute_sudo=True)
57 total_customer_note_count = fields.Integer(
58 compute="_compute_customer_note_count",
59 compute_sudo=True,
60 string="Total Customer Notes",
61 )
62 pending_customer_notes = fields.Integer(compute="_compute_customer_note_count", compute_sudo=True)
63 upgrade_pull_request_ids = fields.Many2many(
64 "saas.pull.request",
65 "saas_pull_request_ticket_upgrade_rel",
66 "ticket_id",
67 "pull_id",
68 string="Upgrade PRs",
69 )
70 upgrade_target = fields.Char(
71 related="upgrade_type_id.target",
72 string="Upgrade Major version",
73 )
74 deadline_upgrade = fields.Date(
75 string="Fecha deseada de actualización",
76 help="Planned date for the upgrade (either requested by the customer or set by us).",
77 tracking=True,
78 )
79 date_upgrade_scheduled = fields.Datetime(
80 string="Fecha agendada para actualización",
81 compute="_compute_date_upgrade_scheduled",
82 store=True,
83 help="Scheduled date for the upgrade to production, based on the last production request.",
84 )
85 upgrade_description = fields.Html(related="upgrade_type_id.portal_description")
86 post_upgrade_description = fields.Html(related="upgrade_type_id.post_upgrade_portal_description")
87 upgrade_line_ids = fields.Many2many(
88 "saas.upgrade.line",
89 "saas_upgrade_line_ticket_rel",
90 "ticket_id",
91 "upgrade_line_id",
92 string="Upgrade Lines",
93 help="Upgrade Lines asociadas al ticket.",
94 domain=["|", ("active", "=", True), ("active", "=", False)],
95 context={"active_test": False},
96 )
97 last_test_request_line_id = fields.Many2one(
98 "helpdesk.ticket.upgrade.request",
99 compute="_compute_last_test_request_line",
100 store=True,
101 )
102 last_test_upgraded_database_id = fields.Many2one(
103 "saas.database",
104 related="last_test_request_line_id.upgraded_database_id",
105 string="Last Test Upgraded Database",
106 related_sudo=True,
107 )
108 last_production_request_line_id = fields.Many2one(
109 "helpdesk.ticket.upgrade.request",
110 compute="_compute_date_upgrade_scheduled",
111 store=True,
112 )
113 last_test_request_state = fields.Selection(
114 related="last_test_request_line_id.state",
115 string="Test Request State",
116 related_sudo=True,
117 )
118 last_production_request_state = fields.Selection(
119 related="last_production_request_line_id.state",
120 string="Production Request State",
121 related_sudo=True,
122 )
123 parameters = fields.Properties(
124 definition="upgrade_type_id.parameters_definition",
125 help="Parameters that can be used on scripts",
126 )
127 custom_parameters_dict = fields.Text(compute="_compute_custom_parameters_dict")
128 upgrade_upload_changes_ids = fields.One2many(
129 "saas.upgrade.upload.changes",
130 "ticket_id",
131 )
132 upgrade_client_data_id = fields.Many2one("saas.upgrade.client.data")
133 ticket_upgrade_status = fields.Selection(
134 selection=[
135 ("no_upgraded", "Pending Upgrade"),
136 ("upgraded", "Upgraded Database"),
137 ],
138 compute="_compute_request_state",
139 store=True,
140 )
141 active_production_freeze = fields.Boolean(
142 compute="_compute_active_production_freeze",
143 store=True,
144 readonly=False,
145 )
146 confirm_production_sent = fields.Boolean(
147 default=False,
148 help="Indicates whether the production update has already been committed and the customer has been notified.",
149 )
150 is_last_upgrade_ticket_available = fields.Boolean(compute="_compute_is_last_upgrade_ticket")
151 upgrade_classification = fields.Char(
152 compute="_compute_upgrade_classification",
153 store=True,
154 )
155 gantt_date_stop = fields.Datetime(
156 compute="_compute_gantt_date_stop",
157 store=True,
158 help="Planned end date for the upgrade. Calculated from the start date field specified in context (gantt_start_field) + 7 days. Used in Gantt view.",
159 )
161 _upgrade_uniq = models.Constraint(
162 "unique (project_id, upgrade_type_id)",
163 "Only one upgrade can exist per project and upgrade type.",
164 )
166 @api.depends("team_id", "team_id.upgrade_team")
167 def _compute_assigned_technician_id(self):
168 for ticket in self:
169 if ticket.team_id.upgrade_team and not ticket.assigned_technician_id:
170 technician_dict = ticket.team_id._determine_technician_to_assign()
171 technician = technician_dict.get(ticket.team_id.id, self.env["res.users"])
172 if technician: 172 ↛ 173line 172 didn't jump to line 173 because the condition on line 172 was never true
173 ticket.assigned_technician_id = technician.id
175 @api.depends("upgrade_client_data_id.project_classification")
176 def _compute_upgrade_classification(self):
177 for rec in self:
178 if rec.upgrade_client_data_id and rec.upgrade_client_data_id.last_ticket_id.id == rec.id:
179 rec.upgrade_classification = rec.upgrade_client_data_id.project_classification
180 else:
181 rec.upgrade_classification = False
183 @api.depends("upgrade_type_id")
184 def _compute_is_last_upgrade_ticket(self):
185 last_upgrade_type = self.env["saas.upgrade.type"].search([("stable", "=", True)], limit=1, order="sequence")
186 for rec in self:
187 if rec.upgrade_team and rec.upgrade_type_id:
188 if rec.upgrade_type_id == last_upgrade_type:
189 rec.is_last_upgrade_ticket_available = True
190 else:
191 rec.is_last_upgrade_ticket_available = False
192 else:
193 rec.is_last_upgrade_ticket_available = False
195 @api.depends("last_production_request_line_id.state")
196 def _compute_request_state(self):
197 for rec in self:
198 rec.ticket_upgrade_status = "no_upgraded"
199 if rec.last_production_request_line_id and rec.last_production_request_line_id.state == "done": 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true
200 rec.ticket_upgrade_status = "upgraded"
202 @api.depends("upgrade_type_id.parameters_definition", "parameters", "upgrade_team")
203 def _compute_custom_parameters_dict(self):
204 for rec in self:
205 new_dict = dict()
206 parameters_definition = rec.upgrade_type_id.parameters_definition
207 if not rec.upgrade_team or (not rec.upgrade_type_id.parameters_definition and not rec.parameters): 207 ↛ 210line 207 didn't jump to line 210 because the condition on line 207 was always true
208 rec.custom_parameters_dict = "{}"
209 continue
210 parameters_definition = list(filter(lambda x: x["type"] != "separator", parameters_definition))
211 if not rec.parameters:
212 new_parameters = dict()
213 for definition in parameters_definition:
214 if "default" in definition.keys():
215 new_parameters[definition["name"]] = definition["default"]
216 rec.write({"parameters": new_parameters})
217 rec.parameters = new_parameters
219 for key, value in rec.parameters.items():
220 parameter_definition = list(filter(lambda x: x["name"] == key, parameters_definition))
222 if not parameter_definition:
223 # Elimination case
224 continue
226 parameter_definition = parameter_definition[0]
227 new_key = parameter_definition["string"]
228 new_value = value
229 if not new_value:
230 new_value = parameter_definition["default"]
232 if new_key in new_dict:
233 raise UserError(
234 _(
235 "Duplicate key '%s' in custom_parameters_dict, please use a different name",
236 new_key,
237 )
238 )
239 try:
240 if isinstance(new_value, models.BaseModel):
241 ids = new_value.ids
242 new_dict[new_key] = ids[0] if len(ids) == 1 else ids
243 else:
244 new_dict[new_key] = safe_eval(str(new_value))
245 except Exception:
246 raise UserError(
247 _(
248 "Invalid expression, '%s' value must be a literal python list definition e.g. '[a, b]'",
249 new_key,
250 )
251 )
253 rec.custom_parameters_dict = str(new_dict)
255 @api.depends("project_id.user_id")
256 def _compute_consultant_upgrade_id(self):
257 for rec in self.filtered("upgrade_team"):
258 if not rec.consultant_upgrade_id: 258 ↛ 257line 258 didn't jump to line 257 because the condition on line 258 was always true
259 rec.consultant_upgrade_id = rec.project_id.user_id
261 @api.depends("all_request_line_ids.state", "confirm_production_sent")
262 def _compute_date_upgrade_scheduled(self):
263 for rec in self:
264 requests = rec.all_request_line_ids.filtered(
265 lambda x: x.aim == "production" and x.state not in ["cancel", "error"]
266 ).sorted(key=lambda m: m.planned_start or datetime.min, reverse=True)
267 rec.date_upgrade_scheduled = False
268 rec.last_production_request_line_id = False
269 if requests:
270 request = requests[0]
271 if request.confirm_production_sent: 271 ↛ 272line 271 didn't jump to line 272 because the condition on line 271 was never true
272 rec.date_upgrade_scheduled = request.planned_start
273 rec.last_production_request_line_id = request
275 @api.depends("customer_note_ids")
276 def _compute_customer_note_count(self):
277 for rec in self:
278 pending_customer_notes = rec.customer_note_ids.filtered(lambda cn: cn.status not in ["done"])
279 rec.customer_note_count = len(
280 pending_customer_notes.filtered(lambda cn: not cn.post_production and cn.priority == "1")
281 )
282 rec.total_customer_note_count = len(rec.customer_note_ids)
283 rec.pending_customer_notes = len(pending_customer_notes.ids)
285 @api.depends(
286 "request_line_ids.state",
287 "request_line_ids.with_first_original_database",
288 "request_line_ids.with_upgraded_database",
289 )
290 def _compute_last_test_request_line(self):
291 for rec in self.filtered("request_line_ids"):
292 test_requests = rec.request_line_ids.filtered(
293 lambda x: (
294 x.aim == "test"
295 and x.state != "cancel"
296 and x.with_first_original_database
297 and x.with_upgraded_database
298 )
299 )
300 if test_requests: 300 ↛ 301line 300 didn't jump to line 301 because the condition on line 300 was never true
301 rec.last_test_request_line_id = test_requests[-1]
302 else:
303 rec.last_test_request_line_id = False
305 @api.depends("upgrade_type_id.active_production_freeze")
306 def _compute_active_production_freeze(self):
307 for rec in self:
308 rec.active_production_freeze = rec.upgrade_type_id.active_production_freeze
310 @api.depends("date_upgrade_scheduled", "deadline_upgrade")
311 @api.depends_context("gantt_start_field")
312 def _compute_gantt_date_stop(self):
313 """Compute the Gantt end date based on the start field specified in context."""
314 start_field = self.env.context.get("gantt_start_field", "deadline_upgrade")
315 for rec in self.filtered(start_field): 315 ↛ 316line 315 didn't jump to line 316 because the loop on line 315 never started
316 start_date = getattr(rec, start_field)
317 rec.gantt_date_stop = fields.Datetime.to_datetime(start_date + timedelta(days=7))
319 @api.onchange("upgrade_type_id", "upgrade_team", "project_id")
320 def _suggest_upgrade_ticket_name(self):
321 for rec in self:
322 if rec.upgrade_team and rec.upgrade_type_id:
323 rec.name = "%s - %s" % (
324 rec.upgrade_type_id.with_context(lang="es_AR").name,
325 rec.project_id.name,
326 )
328 def _compute_access_url(self):
329 records_upgrade_team = self.filtered("upgrade_team")
330 super(HelpdeskTicket, self - records_upgrade_team)._compute_access_url()
331 for ticket in records_upgrade_team:
332 ticket.access_url = "/my/upgrade_ticket/%s" % ticket.id
334 @api.constrains("environment_variables_dict")
335 def _check_environment_variables_dict(self):
336 for rec in self:
337 try:
338 dict(safe_eval(rec.environment_variables_dict))
339 except Exception:
340 raise UserError(
341 _(
342 "Invalid expression on Environment Variables Dict, it must be a literal python dictionary "
343 "definition e.g. \"{'field': 'value'}\""
344 )
345 )
347 @api.constrains("project_id", "main_database_id")
348 def _check_upgrade_version_conflict(self):
349 """
350 Verifies that a ticket is not created for a customer who is already on the target version
351 """
352 for rec in self.filtered(lambda x: x.project_id and not x.parent_id and not x.upgrade_team): 352 ↛ 353line 352 didn't jump to line 353 because the loop on line 352 never started
353 upgrade_ticket = self.search(
354 [
355 ("project_id", "=", rec.project_id.id),
356 ("upgrade_team", "=", True),
357 ("ticket_upgrade_status", "=", "no_upgraded"),
358 ],
359 limit=1,
360 order="upgrade_target desc",
361 )
362 if (
363 upgrade_ticket == rec
364 and rec.main_database_id
365 and upgrade_ticket.upgrade_target == rec.main_database_id.major_version
366 ):
367 raise ValidationError(_("The client is already on the target version."))
369 def _get_estimated_production_upgrade_duration(
370 self, measure: Literal["hours", "minutes", "seconds"] = "hours"
371 ) -> int:
372 """
373 Returns the estimated duration for the production upgrade based on the last test request.
374 If no test request exists, returns the default appointment duration.
376 :param measure: The unit of time to return the duration in. Can be 'hours', 'minutes' or 'seconds'.
377 :return: Estimated duration for the production upgrade in the specified unit of time.
378 """
379 self.ensure_one()
380 if self.last_test_request_line_id:
381 time = math.ceil(self.last_test_request_line_id.cum_total_time)
382 else:
383 time = math.ceil(self.upgrade_type_id.appointment_type_id.appointment_duration)
385 match measure:
386 case "hours":
387 return time
388 case "minutes":
389 return time * 60
390 case "seconds":
391 return time * 3600
392 case _:
393 return time
395 # ---------------------------------------------------
396 # Portal Upgrade Methods
397 # ---------------------------------------------------
399 @api.model
400 def get_portal_upgrade_info(self, partner):
401 """
402 Central method to get all necessary information for the upgrade portal.
403 Returns a dictionary with structured information about upgrade status and available options.
405 :param partner: Partner recordset to get upgrade information for
406 :return: Dictionary with keys: stable_upgrade_type, stable_ticket, is_on_stable_version,
407 show_upgrade_button, show_beta_tester, show_beta_tester_message,
408 next_upgrade_version, next_upgrade_type
409 """
410 stable_upgrade_type = (
411 self.env["saas.upgrade.type"]
412 .sudo()
413 .search([("stable", "=", True), ("active", "=", True)], limit=1, order="sequence")
414 )
415 stable_ticket = self.sudo().search(
416 [
417 ("partner_id", "=", partner.id),
418 ("upgrade_team", "=", True),
419 ("upgrade_type_id", "=", stable_upgrade_type.id),
420 ],
421 limit=1,
422 order="id desc",
423 )
424 is_on_stable_version = bool(stable_ticket and stable_ticket.ticket_upgrade_status == "upgraded")
425 result = {
426 "stable_upgrade_type": stable_upgrade_type,
427 "stable_ticket": stable_ticket,
428 "is_on_stable_version": is_on_stable_version,
429 "show_upgrade_button": not is_on_stable_version,
430 "show_beta_tester": False,
431 "show_beta_tester_message": False,
432 "next_upgrade_version": False,
433 "next_upgrade_type": False,
434 }
435 if is_on_stable_version:
436 next_upgrade_type = (
437 self.env["saas.upgrade.type"]
438 .sudo()
439 .search(
440 [
441 ("active", "=", True),
442 ("sequence", "<", stable_upgrade_type.sequence),
443 ],
444 limit=1,
445 order="sequence",
446 )
447 )
448 if next_upgrade_type:
449 beta_ticket = self.sudo().search(
450 [
451 ("partner_id", "=", partner.id),
452 ("upgrade_team", "=", True),
453 ("upgrade_type_id", "=", next_upgrade_type.id),
454 ("ticket_upgrade_status", "=", "no_upgraded"),
455 ],
456 limit=1,
457 order="id desc",
458 )
459 result["next_upgrade_type"] = next_upgrade_type
460 result["next_upgrade_version"] = next_upgrade_type.target
461 result["show_beta_tester_message"] = bool(beta_ticket)
462 result["show_beta_tester"] = not bool(beta_ticket)
463 return result
465 # ---------------------------------------------------
466 # upgrade ticket sharing
467 # ---------------------------------------------------
469 def action_open_customer_notes(self) -> dict:
470 self.ensure_one()
471 action = self.env["ir.actions.act_window"]._for_xml_id(
472 "saas_provider_upgrade.action_helpdesk_ticket_customer_note"
473 )
474 new_context = literal_eval(action.get("context", "{}"))
475 new_context.update({"default_ticket_id": self.id})
476 action["context"] = new_context
477 action["domain"] = [["ticket_id", "=", self.id]]
478 return action
480 def action_open_requests_logs(self) -> dict:
481 self.ensure_one()
482 action = self.env["ir.actions.act_window"]._for_xml_id(
483 "saas_provider_upgrade.action_saas_upgrade_line_request_log_entry"
484 )
485 new_context = literal_eval(action.get("context", "{}"))
486 new_context.update({"default_upgrade_ticket_id": self.id})
487 action["context"] = new_context
488 action["domain"] = [["upgrade_ticket_id", "=", self.id]]
489 return action
491 def action_confirm_production(self) -> dict:
492 self.ensure_one()
493 upgrade_type = self.upgrade_type_id
494 if not upgrade_type.confirm_stage_id or not upgrade_type.confirm_mail_template_id: 494 ↛ 495line 494 didn't jump to line 495 because the condition on line 494 was never true
495 action = self.env["ir.actions.act_window"]._for_xml_id("saas_provider_upgrade.action_saas_upgrade_type")
496 action["res_id"] = upgrade_type.id
497 action["views"] = [
498 (
499 self.env.ref("saas_provider_upgrade.view_saas_upgrade_type_form").id,
500 "form",
501 )
502 ]
503 raise RedirectWarning(
504 message=_(
505 "The confirmation stage and/or mail template are not defined in the Upgrade Type. Please complete them before confirming."
506 ),
507 action=action,
508 button_text=_("Go to Upgrade Type"),
509 )
511 self.stage_id = upgrade_type.confirm_stage_id.id
512 self.message_post_with_source(
513 upgrade_type.confirm_mail_template_id,
514 message_type="comment",
515 subtype_xmlid="mail.mt_comment",
516 )
517 self.confirm_production_sent = True
518 msg = _("Production upgrade was confirmed and customer was notified. Stage changed.")
519 action_to_return = self.build_display_notification_action(
520 title=_("Confirmation done"),
521 message=msg,
522 type="success",
523 )
524 return action_to_return
526 def action_customer_portal(self) -> dict:
527 self.ensure_one()
528 return {
529 "type": "ir.actions.act_url",
530 "url": self.access_url,
531 "target": "new",
532 }
534 def action_schedule_upgrade_production(self):
535 """
536 Action to program the upgrade to production from portal
537 """
538 self.ensure_one()
539 request_production = self.last_production_request_line_id
541 # If a production request exists, redirect to its calendar event
542 # All production requests now have an associated event (via portal, wizard, or migration)
543 if request_production:
544 request_production = request_production.sudo()
545 event_upgrade = request_production.event_upgrade_id
546 event_token = event_upgrade.access_token
547 url = f"/calendar/view/{event_token}"
548 params = {
549 "ticket_id": self.id,
550 }
551 if event_upgrade.partner_ids:
552 params["partner_id"] = event_upgrade.partner_ids.id
553 return {
554 "type": "ir.actions.act_url",
555 "url": f"{url}?{url_encode(params)}",
556 "target": "new",
557 }
559 # No production request exists - run validations before allowing scheduling
560 action_to_return = {
561 "type": "ir.actions.act_url",
562 "url": f"/appointment/{self.sudo().upgrade_type_id.appointment_type_id.id}?ticket_id={self.id}",
563 "target": "new",
564 }
565 # Create a dummy request with aim 'production' to trigger the ULs on_create
566 # Raises an exception if any of the conditions are not met
567 try:
568 request = (
569 self.env["helpdesk.ticket.upgrade.request"].sudo().new({"aim": "production", "ticket_id": self.id})
570 )
571 request.run_on_create_scripts()
572 except UserError as e:
573 title = self.env._("Check conditions!")
574 message = e.args[0] # e.args is a tuple
575 action_to_return = self.build_display_notification_action(title, message, "warning")
576 except Exception as e:
577 body = "Error in programming the production upgrade by the user, error: \n <blockquote>%s</blockquote>" % e
578 self.message_post(body=Markup(body))
579 title = self.env._("Unexpected failure!")
580 message = self.env._(
581 "An error occurred while processing the request, please try again. "
582 "If the problem persists contact support."
583 )
584 action_to_return = self.build_display_notification_action(title, message, "warning")
585 return action_to_return
587 def generate_appointment_link(self) -> str:
588 """
589 Genera el enlace de actualización para el ticket.
590 """
591 self.ensure_one()
592 if not self.upgrade_type_id.appointment_type_id:
593 raise ValueError(_("There is not appointment type associated with this upgrade type."))
595 # Create an appointment.invite associated with the ticket
596 invite = self.env["appointment.invite"].create(
597 {
598 "appointment_type_ids": [(6, 0, [self.upgrade_type_id.appointment_type_id.id])],
599 "ticket_id": self.id,
600 }
601 )
602 url = invite.redirect_url
603 return url
605 def get_major_version_from_to_upg(self) -> tuple[str, str]:
606 """
607 Retrieve the major version for main database and target version for an upgrade ticket
609 return: (old_major_version, new_major_version)
610 """
611 self.ensure_one()
612 self = self.sudo()
613 return (
614 self.main_database_id.odoo_version_group_id.major_version_id.name,
615 self.upgrade_target,
616 )
619class HelpdeskStage(models.Model):
620 _inherit = "helpdesk.stage"
622 show_in_upgrade_portal = fields.Boolean()