Coverage for ingadhoc-odoo-saas-adhoc / saas_provider_upgrade / controllers / upgrade.py: 13%
121 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-09 18:15 +0000
1##############################################################################
2# For copyright and license notices, see __manifest__.py file in module root
3# directory
4##############################################################################
5import logging
6from typing import TYPE_CHECKING, Literal
8import odoo
9from odoo import _, http
10from odoo.http import request
12from ..constants import UL_TYPE_DB_REQUIREMENTS
13from ..exceptions import UpgradeLineException
15if TYPE_CHECKING:
16 from ..models.helpdesk_ticket_upgrade_request import HelpdeskTicketUpgradeRequest as UpgradeRequest
17 from ..models.saas_upgrade_client_config_line import SaasUpgradeClientConfigLine as UpgradeClientConfigLine
18 from ..models.saas_upgrade_line_request_log import SaasUpgradeLineRequestLog as UpgradeLineRequestLog
19 from ..models.saas_upgrade_line_request_run import SaasUpgradeLineRequestRun as UpgradeLineRequestRun
21_logger = logging.getLogger(__name__)
23RTA_ERROR: dict[str, str | int] = {"state": "error", "description": "Bad Request", "status_code": 400}
24RTA_SUCCESS: dict[str, str | int] = {"state": "ok", "description": "success", "status_code": 200}
27class SaasProviderUpgradeController(http.Controller):
28 @http.route("/saas_provider_upgrade/upgrade_notify", type="jsonrpc", auth="public", methods=["POST"])
29 def upgrade_notify(
30 self, request_id: int, step_type: Literal["submit", "restore", "upgrade", "postupgrade", "run_script"], **kwargs
31 ) -> dict[str, str | int]:
32 token = kwargs.get("token", "")
33 odoo_project_token = request.env["ir.config_parameter"].sudo().get_param("saas_provider.odoo_project_token")
34 if odoo_project_token and odoo_project_token != token:
35 _logger.warning(f"{request_id} with invalid token")
36 return RTA_ERROR
37 try:
38 upgrade_request: UpgradeRequest = (
39 request.env["helpdesk.ticket.upgrade.request"].with_user(odoo.SUPERUSER_ID).browse(request_id)
40 )
41 if not upgrade_request.exists():
42 _logger.warning(f"Request (ID: {request_id}) does not exist")
43 return RTA_ERROR
44 function_name = "step_" + step_type
45 handler = getattr(self, function_name, None)
46 if handler is None:
47 error = f"No handler found for step type: {step_type}"
48 _logger.error(error)
49 self._handle_controller_error(upgrade_request, error, **kwargs)
50 return RTA_SUCCESS
51 rta = handler(upgrade_request, **kwargs)
52 return rta
53 except Exception as e:
54 _logger.exception(f"Error in step handler for {step_type}: {str(e)}")
55 self._handle_controller_error(upgrade_request, str(e), **kwargs)
56 return RTA_SUCCESS
58 def step_submit(self, upgrade_request: "UpgradeRequest", **kwargs) -> dict[str, str | int]:
59 """
60 Odoo Upgrade submit step handler.
62 :param upgrade_request: The upgrade request record
63 :param kwargs: Additional parameters that may include the token and submit URI
64 """
65 request_id = kwargs.get("submit_request_id", "")
66 submit_token = kwargs.get("submit_token", "")
67 submit_uri = kwargs.get("submit_uri", "")
68 if not submit_token or not submit_uri:
69 _logger.warning("Token and URI are required")
70 return RTA_ERROR
72 upgrade_request.write(
73 {
74 "odoo_request_id": request_id,
75 "odoo_request_token": submit_token,
76 "odoo_host_uri": submit_uri,
77 }
78 )
79 upgrade_request._next_state()
80 return RTA_SUCCESS
82 def step_restore(self, upgrade_request: "UpgradeRequest", **kwargs) -> dict[str, str | int]:
83 """
84 Restore step handler for upgrade report.
86 :param upgrade_request: The upgrade request record
87 :param kwargs: Additional parameters that may include the upgrade report
88 """
89 upgrade_report = kwargs.get("upgrade_report", "")
90 if upgrade_report:
91 upgrade_request.write({"upgrade_report_html": upgrade_report})
92 return RTA_SUCCESS
94 def step_upgrade(self, upgrade_request: "UpgradeRequest", **kwargs) -> dict[str, str | int]:
95 """
96 Adhoc upgrade step handler.
98 :param upgrade_request: The upgrade request record
99 :param kwargs: Additional parameters that may include the response type and message
100 """
101 utils = request.env["saas.provider.upgrade.util"].sudo()
102 if "type" not in kwargs or "message" not in kwargs or "log_file" not in kwargs:
103 _logger.warning("Log file or type or message is missing in the request")
104 return RTA_ERROR
106 # Get the type and message from kwargs
107 type, message = kwargs.get("type"), kwargs.get("message")
109 # Save the log file as an attachment
110 encoded_log_file: str = kwargs["log_file"]
111 log_filename = f"{upgrade_request.id}.txt"
112 upgrade_request._create_log_attachment(
113 "upgrade_log_attachment_id",
114 encoded_log_file,
115 log_filename,
116 "Upgrade Log",
117 )
119 match type:
120 case "ok":
121 upgrade_request._next_state()
122 upgrade_request.upgraded_database_id._call_odoo_reload()
123 return RTA_SUCCESS
124 case "error":
125 upgrade_request._error()
126 title = _("Error in upgrade process!")
127 body = utils.create_message(title, message)
128 upgrade_request.message_post(body=body)
129 return RTA_ERROR
130 case _:
131 _logger.error("Invalid response type: %s" % type)
132 upgrade_request._error()
133 return RTA_ERROR
135 def step_postupgrade(self, upgrade_request: "UpgradeRequest", **kwargs) -> dict[str, str | int]:
136 """
137 Post upgrade finalization step handler.
139 :param upgrade_request: The upgrade request record
140 :param kwargs: Additional parameters
141 """
142 error = kwargs.get("error", "")
143 if error:
144 upgrade_request._set_error_and_notify(error)
145 else:
146 upgrade_request._next_state()
147 upgrade_request.upgraded_database_id._call_odoo_reload()
148 return RTA_SUCCESS
150 def step_run_script(self, upgrade_request: "UpgradeRequest", **kwargs) -> dict[str, str | int]:
151 """
152 Process the run script step.
153 If the run info has a next run ID, it will set its state to pending.
154 If the run info has a callback, it will call that method on the upgrade request.
156 :param upgrade_request: The upgrade request object
157 :param kwargs: Additional parameters that may include the run info ID
158 """
159 run_info_id = kwargs.get("run_info_id", False)
160 result = kwargs.get("result", False)
161 breaks = kwargs.get("breaks", False)
162 error = kwargs.get("error", False)
163 placeholders = kwargs.get("placeholders", {})
164 logs: list[dict] = kwargs.get("logs", [])
165 client_configs: list[dict] = kwargs.get("client_configs", [])
167 run_info: UpgradeLineRequestRun = request.env["saas.upgrade.line.request.run"].sudo().browse(run_info_id)
168 context = dict(run_info.bg_job_id.context_json or {})
169 run_info = run_info.with_context(**context).with_user(run_info.create_uid).sudo()
170 env = run_info.env
171 if not run_info.exists():
172 _logger.warning(f"Run info (ID: {run_info_id}) does not exist")
173 return RTA_ERROR
175 upgrade_line = run_info.upgrade_line_id
177 # Generate new logs
178 RequestLog: UpgradeLineRequestLog = env["saas.upgrade.line.request.log"]
179 RequestLog.create_entries_from_json(
180 request=upgrade_request,
181 upgrade_line=upgrade_line,
182 vals_list=logs,
183 )
185 # Process the customer note info
186 upgrade_line._process_customer_note(placeholders, upgrade_request, run_info)
188 # Process new client configs
189 ClientConfig: UpgradeClientConfigLine = env["saas.upgrade.client.config.line"]
190 ClientConfig.create_from_json(
191 upgrade_line,
192 upgrade_request.ticket_id,
193 client_configs,
194 )
196 # Manage error, result and breaks
197 exception = None
198 if error:
199 exception = UpgradeLineException(
200 env,
201 upgrade_request,
202 upgrade_line=upgrade_line,
203 error=error,
204 )
206 if exception:
207 run_info.set_error(exception)
208 else:
209 message = run_info.finish(result=breaks or result)
210 if message:
211 run_info.bg_job_id._notify_user(message)
213 # Lookup the correct database to call reload
214 db_requirement = UL_TYPE_DB_REQUIREMENTS.get(run_info.upgrade_line_id.type, None)
215 database = (
216 upgrade_request.original_database_id
217 if db_requirement == "with_original_db"
218 else upgrade_request.upgraded_database_id
219 )
220 database._call_odoo_reload()
221 return RTA_SUCCESS
223 def _handle_controller_error(self, upgrade_request: "UpgradeRequest", error: str, **kwargs):
224 """
225 Handle errors in the controller by logging the error, updating the upgrade request state, and returning an error response.
227 :param upgrade_request: The upgrade request record
228 :param error: The error message to log and post on the upgrade request
229 :return: A dictionary representing an error response
230 """
231 run_info_id = kwargs.get("run_info_id", False)
232 run_info: UpgradeLineRequestRun = request.env["saas.upgrade.line.request.run"].browse(run_info_id)
233 if not run_info.exists():
234 upgrade_request._set_error_and_notify(error)
235 else:
236 run_info.set_error(error)