From 5e4cf9723bf79f9567c0a9dfffe8cf4845cb2bfb Mon Sep 17 00:00:00 2001 From: git_admin Date: Mon, 27 Apr 2026 08:15:48 +0000 Subject: [PATCH] Tower: upload cetmix_tower_server 16.0.3.0.1 (via marketplace) --- .../models/cetmix_tower.py | 313 ++++++++++++++++++ 1 file changed, 313 insertions(+) create mode 100644 addons/cetmix_tower_server/models/cetmix_tower.py diff --git a/addons/cetmix_tower_server/models/cetmix_tower.py b/addons/cetmix_tower_server/models/cetmix_tower.py new file mode 100644 index 0000000..ef9786c --- /dev/null +++ b/addons/cetmix_tower_server/models/cetmix_tower.py @@ -0,0 +1,313 @@ +# Copyright (C) 2024 Cetmix OÜ +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). +import logging +import time +import warnings + +from odoo import _, api, models +from odoo.exceptions import ValidationError + +from . import tools +from .constants import NOT_FOUND, SSH_CONNECTION_ERROR + +_logger = logging.getLogger(__name__) + + +class CetmixTower(models.AbstractModel): + """Generic model used to simplify Odoo automation. + Used to keep main integration function in a single place. + """ + + _name = "cetmix.tower" + _description = "Cetmix Tower Odoo Automation" + + @api.model + def server_create_from_template(self, template_reference, server_name, **kwargs): + """ + THIS METHOD IS DEPRECATED. USE THE 'cx.tower.server.template' MODEL DIRECTLY. + """ + _logger.warning( + "server_create_from_template: This method is deprecated " + "and will be removed in the future. " + "Use the 'cx.tower.server.template' model directly instead." + ) + return self.env["cx.tower.server.template"].create_server_from_template( + template_reference=template_reference, server_name=server_name, **kwargs + ) + + @api.model + def server_run_command( + self, server_reference, command_reference, get_result=True, **variable_values + ): + """ + THIS METHOD IS DEPRECATED. USE THE 'cx.tower.server' MODEL DIRECTLY. + """ + + _logger.warning( + "server_run_command: This method is deprecated and " + "will be removed in the future. " + "Use the 'cx.tower.server' model directly instead." + ) + server = self.env["cx.tower.server"].get_by_reference(server_reference) + if not server: + return {"exit_code": NOT_FOUND, "message": _("Server not found")} + command = self.env["cx.tower.command"].get_by_reference(command_reference) + if not command: + return {"exit_code": NOT_FOUND, "message": _("Command not found")} + + # Will return command result if get_result is True + # Otherwise will save to log and return None + command_result = server.with_context(no_command_log=get_result).run_command( + command, **{"variable_values": variable_values} if variable_values else {} + ) + + # Return command result if get_result is True + if command_result: + status = command_result.get("status") + response = command_result.get("response", "") + error = command_result.get("error", "") + return { + "exit_code": status, + "message": response or error, + } + + def server_run_flight_plan( + self, server_reference, flight_plan_reference, **variable_values + ): + """THIS METHOD IS DEPRECATED. USE THE 'cx.tower.server' MODEL DIRECTLY.""" + _logger.warning( + "server_run_flight_plan: This method is deprecated and " + "will be removed in the future. " + "Use the 'cx.tower.server' model directly instead." + ) + server = self.env["cx.tower.server"].get_by_reference(server_reference) + if not server: + # This is not the best way to handle this, but it's the only way to + # avoid complex response handling + return False + flight_plan = self.env["cx.tower.plan"].get_by_reference(flight_plan_reference) + if not flight_plan: + # This is not the best way to handle this, but it's the only way to + # avoid complex response handling + return False + return server.run_flight_plan( + flight_plan, + **{"variable_values": variable_values} if variable_values else {}, + ) + + @api.model + def server_set_variable_value(self, server_reference, variable_reference, value): + """THIS METHOD IS DEPRECATED. USE THE 'cx.tower.server' MODEL DIRECTLY.""" + _logger.warning( + "server_set_variable_value: This method is deprecated and " + "will be removed in the future. " + "Use the 'cx.tower.server' model directly instead." + ) + server = self.env["cx.tower.server"].get_by_reference(server_reference) + if not server: + return {"exit_code": NOT_FOUND, "message": _("Server not found")} + variable = self.env["cx.tower.variable"].get_by_reference(variable_reference) + if not variable: + return {"exit_code": NOT_FOUND, "message": _("Variable not found")} + + # Check if variable is already defined for the server + variable_value_record = variable.value_ids.filtered( + lambda v: v.server_id == server + ) + if variable_value_record: + variable_value_record.value_char = value + result = {"exit_code": 0, "message": _("Variable value updated")} + + else: + self.env["cx.tower.variable.value"].create( + { + "variable_id": variable.id, + "server_id": server.id, + "value_char": value, + } + ) + result = {"exit_code": 0, "message": _("Variable value created")} + return result + + @api.model + def server_get_variable_value( + self, server_reference, variable_reference, check_global=True + ): + """THIS METHOD IS DEPRECATED. USE THE 'cx.tower.server' MODEL DIRECTLY.""" + _logger.warning( + "server_get_variable_value: This method is deprecated and " + "will be removed in the future. " + "Use the 'cx.tower.server' model directly instead." + ) + if not check_global: + warnings.warn( + "server_get_variable_value: 'check_global' is deprecated and " + "will be removed in the future. " + "Global values are always checked.", + DeprecationWarning, + stacklevel=2, + ) + + # Get server by reference + server = self.env["cx.tower.server"].get_by_reference(server_reference) + if not server: + _logger.warning( + "server_get_variable_value: Server not found for reference '%s'", + server_reference, + ) + return None + return ( + self.env["cx.tower.variable"] + ._get_variable_values_by_references( + variable_references=[variable_reference], server=server + ) + .get(variable_reference) + ) + + @api.model + def server_check_ssh_connection( + self, + server_reference, + attempts=5, + wait_time=10, + try_command=True, + try_file=True, + ): + """ + Check if SSH connection to the server is available. + This method uses the `test_ssh_connection` method + of the 'cx.tower.server' model. + It tries to connect to the server multiple times + and is designed to be used in the Python commands or + Odoo automated actions. + + Args: + server_reference (Char): Server reference. + attempts (int): Number of attempts to try the connection. + Default is 5. + wait_time (int): Wait time in seconds between connection attempts. + Default is 10 seconds. + try_command (bool): Try to execute a command. + Default is True. + try_file (bool): Try file operations. + Default is True. + Raises: + ValidationError: + If the provided server reference is invalid or + the server cannot be found. + Returns: + dict: { + "exit_code": int, + 0 for success, + error code for failure + "message": str # Description of the result + } + """ + server = self.env["cx.tower.server"].get_by_reference(server_reference) + if not server: + raise ValidationError(_("No server found for the provided reference.")) + + # Try connecting multiple times + for attempt in range(1, attempts + 1): + try: + _logger.info( + "Attempt %s of %s to connect to server %s", + attempt, + attempts, + server_reference, + ) + result = server.test_ssh_connection( + raise_on_error=True, + return_notification=False, + try_command=try_command, + try_file=try_file, + ) + if result.get("status") == 0: + return { + "exit_code": 0, + "message": _("Connection successful."), + } + if attempt == attempts: + return { + "exit_code": SSH_CONNECTION_ERROR, + "message": _( + "Failed to connect after %(attempts)s attempts. " + "Error: %(err)s", + attempts=attempts, + err=result.get("error", ""), + ), + } + except Exception as e: # pylint: disable=broad-except + if attempt == attempts: + return { + "exit_code": SSH_CONNECTION_ERROR, + "message": _("Failed to connect. Error: %(err)s", err=e), + } + time.sleep(wait_time) + + @api.model + def server_validate_secret( + self, secret_value, secret_reference, server_reference=None + ): + """ + Validates the provided secret value against the actual secret. + + Accepts either a full inline reference (e.g. #!cxtower.secret.!#) + or just a . + + Args: + secret_value (Char): Value to validate + secret_reference (Char): Reference code or inline reference + server_reference (Char, optional): Reference code of the server + Returns: + Bool: True if the value matches the secret, False otherwise + """ + server = self.env["cx.tower.server"] + if server_reference: + server = server.get_by_reference(server_reference) + + # Try to extract reference from inline format using _extract_key_parts + key_parts = self.env["cx.tower.key"]._extract_key_parts(secret_reference) + if key_parts: + # _extract_key_parts returns a tuple: (key_type, reference). + # We only need the reference part here. + secret_reference = key_parts[1] + + value = self.env["cx.tower.key"]._resolve_key_type_secret( + secret_reference, server_id=server.id + ) + return value == secret_value + + @api.model + def generate_random_id(self, sections=1, population=4, separator="-"): + """ + Helper method that allows to generate a random id + with customizable sections and population. + Such ids are more human readable and less likely to collide. + + + Args: + sections (int): Number of sections to generate. + population (int): Population of the sections. + separator (str): Separator between sections. + Returns: + str: Random id + """ + return tools.generate_random_id( + sections=sections, population=population, separator=separator + ) + + @api.model + def is_valid_url(self, url, no_scheme_check=False): + """ + Check if the provided URL is a valid URL. + The `urlparse` function from the `urllib.parse` module is used. + + Args: + url (str): URL to check + no_scheme_check (bool): If True, the scheme check will be skipped. + Defaults to False. + Returns: + bool: True if the URL is valid, False otherwise + """ + return tools.is_valid_url(url=url, no_scheme_check=no_scheme_check)