From ac9207b0c6200475b4e273b7a4c55b572ac5e3e2 Mon Sep 17 00:00:00 2001 From: git_admin Date: Mon, 27 Apr 2026 08:18:14 +0000 Subject: [PATCH] Tower: upload cetmix_tower_server 16.0.3.0.1 (via marketplace) --- addons/cetmix_tower_server/tests/common.py | 509 +++++++++++++++++++++ 1 file changed, 509 insertions(+) create mode 100644 addons/cetmix_tower_server/tests/common.py diff --git a/addons/cetmix_tower_server/tests/common.py b/addons/cetmix_tower_server/tests/common.py new file mode 100644 index 0000000..7cb0061 --- /dev/null +++ b/addons/cetmix_tower_server/tests/common.py @@ -0,0 +1,509 @@ +# Copyright (C) 2022 Cetmix OÜ +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). +import os +from unittest.mock import MagicMock, patch + +from odoo import _ +from odoo.exceptions import ValidationError + +from odoo.addons.base.tests.common import BaseCommon + +from ..models.constants import GENERAL_ERROR +from ..ssh.ssh import SftpService, SSHConnection + + +class TestTowerCommon(BaseCommon): + """ + Common test class for Cetmix Tower. + """ + + @classmethod + def setUpClass(cls): + super().setUpClass() + # Disable transaction commit to avoid race conditions + cls.env = cls.env["base"].with_context(cetmix_tower_no_commit=True).env + + # ---------------------------------------------- + # -- Create core elements invoked in the tests + # ---------------------------------------------- + # Group XML records + cls.group_user = cls.env.ref("cetmix_tower_server.group_user") + cls.group_manager = cls.env.ref("cetmix_tower_server.group_manager") + cls.group_root = cls.env.ref("cetmix_tower_server.group_root") + + # Cetmix Tower helper model + cls.CetmixTower = cls.env["cetmix.tower"] + + # Tags + cls.Tag = cls.env["cx.tower.tag"] + cls.tag_test_staging = cls.Tag.create({"name": "Test Staging"}) + cls.tag_test_production = cls.Tag.create({"name": "Test Production"}) + + # Users + cls.Users = cls.env["res.users"] + cls.user_bob = cls.Users.create( + { + "name": "Bob", + "login": "bob", + "groups_id": [(4, cls.env.ref("base.group_user").id)], + } + ) + cls.user = cls.Users.create( + { + "name": "Test User", + "login": "test_user", + "email": "test_user@example.com", + "groups_id": [(6, 0, [cls.group_user.id])], + } + ) + cls.manager = cls.Users.create( + { + "name": "Test Manager", + "login": "test_manager", + "email": "test_manager@example.com", + "groups_id": [(6, 0, [cls.group_manager.id])], + } + ) + cls.root = cls.Users.create( + { + "name": "Test Root", + "login": "test_root", + "email": "test_root@example.com", + "groups_id": [(6, 0, [cls.group_root.id])], + } + ) + + # OS + cls.os_debian_10 = cls.env["cx.tower.os"].create({"name": "Test Debian 10"}) + + # Server + cls.Server = cls.env["cx.tower.server"] + cls.server_test_1 = cls.Server.create( + { + "name": "Test 1", + "ip_v4_address": "localhost", + "ssh_username": "admin", + "ssh_password": "password", + "ssh_auth_mode": "p", + "host_key": "test_key", + "os_id": cls.os_debian_10.id, + } + ) + + # Server Template + cls.ServerTemplate = cls.env["cx.tower.server.template"] + cls.server_template_sample = cls.ServerTemplate.create( + { + "name": "Sample Template", + "ssh_port": 22, + "ssh_username": "admin", + "ssh_password": "password", + "ssh_auth_mode": "p", + "os_id": cls.os_debian_10.id, + } + ) + + # Server log + cls.ServerLog = cls.env["cx.tower.server.log"] + + # Variable + cls.Variable = cls.env["cx.tower.variable"] + cls.VariableValue = cls.env["cx.tower.variable.value"] + cls.VariableOption = cls.env["cx.tower.variable.option"] + + cls.variable_path = cls.Variable.create({"name": "test_path_"}) + cls.variable_dir = cls.Variable.create({"name": "test_dir"}) + cls.variable_os = cls.Variable.create({"name": "test_os"}) + cls.variable_url = cls.Variable.create({"name": "test_url"}) + cls.variable_version = cls.Variable.create({"name": "test_version"}) + + # Key + cls.Key = cls.env["cx.tower.key"] + cls.KeyValue = cls.env["cx.tower.key.value"] + + cls.key_1 = cls.Key.create( + {"name": "Test Key 1", "key_type": "k", "secret_value": "much key"} + ) + cls.secret_2 = cls.Key.create( + {"name": "Test Key 2", "key_type": "s", "secret_value": "secret top"} + ) + + # Command + cls.sudo_prefix = "sudo -S -p ''" + cls.Command = cls.env["cx.tower.command"] + cls.command_create_dir = cls.Command.create( + { + "name": "Test create directory", + "path": "/home/{{ tower.server.username }}", + "code": "cd {{ test_path_ }} && mkdir {{ test_dir }}", + } + ) + cls.command_list_dir = cls.Command.create( + { + "name": "Test create directory", + "path": "/home/{{ tower.server.username }}", + "code": "cd {{ test_path_ }} && ls -l", + } + ) + + cls.template_file_tower = cls.env["cx.tower.file.template"].create( + { + "name": "Test file template", + "file_name": "test_os.txt", + "source": "tower", + "server_dir": "/home/{{ tower.server.username }}", + "code": "Hello, world!", + } + ) + + cls.template_file_server = cls.env["cx.tower.file.template"].create( + { + "name": "Test file template", + "file_name": "test_os.txt", + "source": "server", + "server_dir": "/home/{{ tower.server.username }}", + } + ) + + cls.command_create_file_with_template_tower_source = cls.Command.create( + { + "name": "Test create file with template with tower source", + "path": "/home/{{ tower.server.username }}", + "action": "file_using_template", + "file_template_id": cls.template_file_tower.id, + "if_file_exists": "raise", + } + ) + + cls.command_create_file_with_template_server_source = cls.Command.create( + { + "name": "Test create file with template with server source", + "path": "/home/{{ tower.server.username }}", + "action": "file_using_template", + "file_template_id": cls.template_file_server.id, + "if_file_exists": "raise", + } + ) + + # Command log + cls.CommandLog = cls.env["cx.tower.command.log"] + + # File template + cls.FileTemplate = cls.env["cx.tower.file.template"] + + # File + cls.File = cls.env["cx.tower.file"] + + # Flight Plans + cls.Plan = cls.env["cx.tower.plan"] + cls.plan_line = cls.env["cx.tower.plan.line"] + cls.plan_line_action = cls.env["cx.tower.plan.line.action"] + + cls.plan_1 = cls.Plan.create( + { + "name": "Test plan 1", + "note": "Create directory and list its content", + "tag_ids": [(6, 0, [cls.tag_test_staging.id])], + } + ) + cls.plan_line_1 = cls.plan_line.create( + { + "sequence": 5, + "plan_id": cls.plan_1.id, + "command_id": cls.command_create_dir.id, + "path": "/such/much/path", + } + ) + cls.plan_line_2 = cls.plan_line.create( + { + "sequence": 20, + "plan_id": cls.plan_1.id, + "command_id": cls.command_list_dir.id, + } + ) + cls.plan_line_1_action_1 = cls.plan_line_action.create( + { + "line_id": cls.plan_line_1.id, + "sequence": 1, + "condition": "==", + "value_char": "0", + } + ) + cls.plan_line_1_action_2 = cls.plan_line_action.create( + { + "line_id": cls.plan_line_1.id, + "sequence": 2, + "condition": ">", + "value_char": "0", + "action": "ec", + "custom_exit_code": 255, + } + ) + cls.plan_line_2_action_1 = cls.plan_line_action.create( + { + "line_id": cls.plan_line_2.id, + "sequence": 1, + "condition": "==", + "value_char": "-1", + "action": "ec", + "custom_exit_code": 100, + } + ) + cls.plan_line_2_action_2 = cls.plan_line_action.create( + { + "line_id": cls.plan_line_2.id, + "sequence": 2, + "condition": ">=", + "value_char": "3", + "action": "n", + } + ) + + # Flight plan log + cls.PlanLog = cls.env["cx.tower.plan.log"] + + # Shortcut + cls.Shortcut = cls.env["cx.tower.shortcut"] + + # Model references + cls.OS = cls.env["cx.tower.os"] + cls.PlanLineAction = cls.env["cx.tower.plan.line.action"] + + # Scheduled task + cls.ScheduledTask = cls.env["cx.tower.scheduled.task"] + cls.ScheduledTaskCv = cls.env["cx.tower.scheduled.task.cv"] + # Jet State + cls.JetState = cls.env["cx.tower.jet.state"] + + # Jet Action + cls.JetAction = cls.env["cx.tower.jet.action"] + + # Jet Template Install + cls.JetTemplateInstall = cls.env["cx.tower.jet.template.install"] + + # Jet Template Install Line + cls.JetTemplateInstallLine = cls.env["cx.tower.jet.template.install.line"] + + # Jet Template Dependency + cls.JetTemplateDependency = cls.env["cx.tower.jet.template.dependency"] + + # Jet Template + cls.JetTemplate = cls.env["cx.tower.jet.template"] + cls.jet_template_sample = cls.JetTemplate.create( + { + "name": "Sample Jet Template", + "server_ids": [(4, cls.server_test_1.id)], + "variable_value_ids": [ + ( + 0, + 0, + { + "variable_id": cls.variable_path.id, + "value_char": "/jets/templates/template1", + }, + ), + ( + 0, + 0, + {"variable_id": cls.variable_os.id, "value_char": "Debian 10"}, + ), + ( + 0, + 0, + { + "variable_id": cls.variable_url.id, + "value_char": "https://jets.example.com", + }, + ), + ( + 0, + 0, + { + "variable_id": cls.variable_dir.id, + "value_char": "jet_templates", + }, + ), + ], + } + ) + + # Jets + cls.Jet = cls.env["cx.tower.jet"] + cls.jet_sample = cls.Jet.create( + { + "name": "Sample Jet", + "jet_template_id": cls.jet_template_sample.id, + "server_id": cls.server_test_1.id, + "variable_value_ids": [ + ( + 0, + 0, + { + "variable_id": cls.variable_path.id, + "value_char": "/jets/jet1", + }, + ) + ], + } + ) + + # apply ssh connection patches + cls.apply_patches() + + @classmethod + def apply_patches(cls): + """ + Apply mock patches for SSH-related methods to simulate various + scenarios during testing. + + Patches: + 1. SSHConnection.connect: + - Returns a mock connection with a fake exec_command method, + which returns a successful or unsuccessful result depending on the + command content. + 2. SftpService.download_file: + - Returns b"ok\x00" for files with the .zip extension and + b"ok" for the rest. + 3. SftpService.upload_file: + - Returns MagicMock, simulating file upload. + 4. SftpService.delete_file: + - Returns MagicMock, simulating file deletion. + """ + + # Patch connection SSH method + def ssh_connect(self): + connection_mock = MagicMock() + + # set up stdin with a condition for error simulation + def exec_command_side_effect(command, *args, **kwargs): + # Create mocks for stdin, stdout, and stderr + stdin_mock = MagicMock() + stdout_mock = MagicMock() + stderr_mock = MagicMock() + + if "fail" in command: + # Simulate failure + stdout_mock.channel.recv_exit_status.return_value = GENERAL_ERROR + stdout_mock.readlines.return_value = [] + stderr_mock.readlines.return_value = ["error"] + return stdin_mock, stdout_mock, stderr_mock + elif "raise" in command: + # Simulate an exception + raise Exception("error") # pylint: disable=broad-exception-raised + else: + # Simulate success + stdout_mock.channel.recv_exit_status.return_value = 0 + stdout_mock.readlines.return_value = ["ok"] + stderr_mock.readlines.return_value = [] + return stdin_mock, stdout_mock, stderr_mock + + # Apply side effect to exec_command + connection_mock.exec_command.side_effect = exec_command_side_effect + + return connection_mock + + connect_patch = patch.object(SSHConnection, "connect", new=ssh_connect) + connect_patch.start() + cls.addClassCleanup(connect_patch.stop) + + # Patch file manipulation methods for testing + def ssh_download_file(self, remote_path): + if hasattr(self, "env"): + error = self.env.context.get("raise_download_error") + if error: + raise ValidationError(error) + + _, extension = os.path.splitext(remote_path) + if extension == ".zip": + return b"ok\x00" + return b"ok" + + download_patch = patch.object( + SftpService, "download_file", new=ssh_download_file + ) + download_patch.start() + cls.addClassCleanup(download_patch.stop) + + def ssh_upload_file(self, file, remote_path): + if hasattr(self, "env"): + error = self.env.context.get("raise_upload_error") + if error: + raise ValidationError(error) + return MagicMock() + + upload_patch = patch.object(SftpService, "upload_file", new=ssh_upload_file) + upload_patch.start() + cls.addClassCleanup(upload_patch.stop) + + def ssh_delete_file(self, remote_path): + return MagicMock() + + delete_patch = patch.object(SftpService, "delete_file", new=ssh_delete_file) + delete_patch.start() + cls.addClassCleanup(delete_patch.stop) + + @classmethod + def add_to_group(cls, user, group_refs): + """Add user to groups + + Args: + user (res.users): User record + group_refs (list): Group ref OR List of group references + eg ['base.group_user', 'some_module.some_group'...] + """ + if isinstance(group_refs, str): + group = cls.env.ref(group_refs, raise_if_not_found=False) + if not group: + raise ValidationError(_("Group reference %s not found!") % group_refs) + action = [(4, group.id)] + elif isinstance(group_refs, list): + action = [] + for group_ref in group_refs: + group = cls.env.ref(group_ref, raise_if_not_found=False) + if not group: + raise ValidationError( + _("Group reference %s not found!") % group_ref + ) + action.append((4, group.id)) + else: + raise ValidationError(_("groups_ref must be string or list of strings!")) + user.write({"groups_id": action}) + + @classmethod + def remove_from_group(cls, user, group_refs): + """Remove user from groups + + Args: + user (res.users): User record + group_refs (list): List of group references + eg ['base.group_user', 'some_module.some_group'...] + """ + if isinstance(group_refs, str): + group = cls.env.ref(group_refs, raise_if_not_found=False) + if not group: + raise ValidationError(_("Group reference %s not found!") % group_refs) + action = [(3, group.id)] + elif isinstance(group_refs, list): + action = [] + for group_ref in group_refs: + group = cls.env.ref(group_ref, raise_if_not_found=False) + if not group: + raise ValidationError( + _("Group reference %s not found!") % group_ref + ) + action.append((3, group.id)) + else: + raise ValidationError(_("groups_ref must be string or list of strings!")) + user.write({"groups_id": action}) + + @classmethod + def write_and_invalidate(cls, records, **values): + """Write values and invalidate cache + + Args: + records (recordset): recordset to save values + **values (dict): values to set + """ + if values: + records.write(values) + records.invalidate_recordset(values.keys())