From a3f387f59d1f45c3449f5d8e0810ff122768d939 Mon Sep 17 00:00:00 2001 From: git_admin Date: Mon, 27 Apr 2026 08:18:17 +0000 Subject: [PATCH] Tower: upload cetmix_tower_server 16.0.3.0.1 (via marketplace) --- .../tests/test_command_wizard.py | 572 ++++++++++++++++++ 1 file changed, 572 insertions(+) create mode 100644 addons/cetmix_tower_server/tests/test_command_wizard.py diff --git a/addons/cetmix_tower_server/tests/test_command_wizard.py b/addons/cetmix_tower_server/tests/test_command_wizard.py new file mode 100644 index 0000000..ca44338 --- /dev/null +++ b/addons/cetmix_tower_server/tests/test_command_wizard.py @@ -0,0 +1,572 @@ +# Copyright (C) 2022 Cetmix OÜ +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). + +from odoo.exceptions import AccessError, ValidationError + +from .common import TestTowerCommon + + +class TestTowerCommandWizard(TestTowerCommon): + """Test Tower Command Run Wizard""" + + def test_user_access_rules(self): + """Test user access rules""" + + # Add Bob to `root` group in order to create a wizard + self.add_to_group(self.user_bob, "cetmix_tower_server.group_root") + + # Create new wizard + test_wizard = ( + self.env["cx.tower.command.run.wizard"] + .with_user(self.user_bob) + .create( + { + "server_ids": [self.server_test_1.id], + "command_id": self.command_create_dir.id, + } + ) + ).with_user(self.user_bob) + + # Force rendered code computation + test_wizard._compute_rendered_code() + + # Remove bob from all cxtower_server groups + self.remove_from_group( + self.user_bob, + [ + "cetmix_tower_server.group_user", + "cetmix_tower_server.group_manager", + "cetmix_tower_server.group_root", + ], + ) + # Ensure that regular user cannot execute command in wizard + with self.assertRaises(AccessError): + test_wizard.run_command_in_wizard() + + # Add bob back to `user` group and try again + self.add_to_group(self.user_bob, "cetmix_tower_server.group_user") + with self.assertRaises(AccessError): + test_wizard.run_command_in_wizard() + + # Now promote bob to `manager` group and try again + self.add_to_group(self.user_bob, "cetmix_tower_server.group_manager") + test_wizard.run_command_in_wizard() + + def test_execute_code_without_a_command(self): + """Run command code without a command selected""" + + # Add Bob to `root` group in order to create a wizard + self.add_to_group(self.user_bob, "cetmix_tower_server.group_root") + + # Create new wizard + test_wizard = ( + self.env["cx.tower.command.run.wizard"] + .with_user(self.user_bob) + .create( + { + "server_ids": [self.server_test_1.id], + } + ) + ).with_user(self.user_bob) + + # Should not allow to run command on server if no command is selected + with self.assertRaises(ValidationError): + test_wizard.run_command_on_server() + + def test_run_command_on_server_access_rights(self): + """Test access rights for executing command on server""" + + # Add Bob to `root` group + self.add_to_group(self.user_bob, "cetmix_tower_server.group_root") + + # Create new wizard with Bob as a root user + test_wizard = ( + self.env["cx.tower.command.run.wizard"] + .with_user(self.user_bob) + .create( + { + "server_ids": [self.server_test_1.id], + "command_id": self.command_create_dir.id, + } + ) + ).with_user(self.user_bob) + + # Ensure command can be executed by root + test_wizard.run_command_on_server() + + # Remove Bob from all tower server groups + self.remove_from_group( + self.user_bob, + [ + "cetmix_tower_server.group_user", + "cetmix_tower_server.group_manager", + "cetmix_tower_server.group_root", + ], + ) + + # Ensure that regular user cannot execute command on server + with self.assertRaises(AccessError): + test_wizard.run_command_on_server() + + # Add Bob to `user` group and ensure he can execute commands + self.add_to_group(self.user_bob, "cetmix_tower_server.group_user") + test_wizard.run_command_on_server() + # Ensure that Bob has access to path field but can't read its value + allowed_path = ( + self.user_bob.has_group("cetmix_tower_server.group_manager") + and test_wizard.path + ) + + self.assertEqual(allowed_path, False) + # Ensure that Bob can write to the path field as a member of `group_user` + # the result will be None + test_wizard.write({"path": "/new/invalid/path"}) + allowed_path = ( + test_wizard.path + if self.user_bob.has_group("cetmix_tower_server.group_manager") + and test_wizard.path + else None + ) + self.assertEqual(allowed_path, None) + + # Add Bob to `manager` group and ensure access to execute commands + self.add_to_group(self.user_bob, "cetmix_tower_server.group_manager") + test_wizard.run_command_on_server() + # Check that path access is valid for the manager + test_wizard.read(["path"]) + + def test_run_command_with_sensitive_vars_on_server_access_rights(self): + """Test access rights for executing command on server""" + # create new command + command = self.Command.create( + { + "name": "Create new command", + "action": "python_code", + "code": """ + properties = { + "Server Name": {{ tower.server.name }}, + "Server Reference": {{ tower.server.reference }}, + "SSH Username": {{ tower.server.username }}, + "IPv4 Address": {{ tower.server.ipv4 }}, + "IPv6 Address": {{ tower.server.ipv6 }}, + "Partner Name": {{ tower.server.partner_name }} + } + result = {"exit_code": 0, "message": properties} + """, + "access_level": "1", + } + ) + + # Add Bob to `root` group in order to create a wizard + self.add_to_group(self.user_bob, "cetmix_tower_server.group_root") + + server = self.Server.with_user(self.user_bob).create( + { + "name": "Test 2", + "ip_v4_address": "localhost", + "ssh_username": "root", + "ssh_password": "password", + "ssh_auth_mode": "p", + "os_id": self.os_debian_10.id, + } + ) + + self.remove_from_group( + self.user_bob, + [ + "cetmix_tower_server.group_user", + "cetmix_tower_server.group_manager", + "cetmix_tower_server.group_root", + ], + ) + + # Add user bob to group user + self.add_to_group(self.user_bob, "cetmix_tower_server.group_user") + + # Create new wizard with Bob + test_wizard = ( + self.env["cx.tower.command.run.wizard"] + .with_user(self.user_bob) + .create( + { + "server_ids": [server.id], + "command_id": command.id, + } + ) + ).with_user(self.user_bob) + + # Add Bob as a user to the command + command.write({"user_ids": [(4, self.user_bob.id)]}) + + # Ensure command can be executed by user + test_wizard.run_command_on_server() + + def test_run_command_in_wizard_multiple_servers(self): + """ + Test that raises an error when multiple servers are selected + """ + + # Add Bob to `root` group in order to create a wizard + + server_test_2 = self.Server.create( + { + "name": "Test 2", + "ip_v4_address": "localhost", + "ssh_username": "root", + "ssh_password": "password", + "ssh_auth_mode": "p", + "os_id": self.os_debian_10.id, + } + ) + + self.add_to_group(self.user_bob, "cetmix_tower_server.group_root") + + # Create new wizard with multiple servers selected + test_wizard = ( + self.env["cx.tower.command.run.wizard"] + .with_user(self.user_bob) + .create( + { + "server_ids": [self.server_test_1.id, server_test_2.id], + "command_id": self.command_create_dir.id, + } + ) + ).with_user(self.user_bob) + + # Force rendered code computation + test_wizard._compute_rendered_code() + + # Ensure that executing command with multiple servers + # selected raises a ValidationError + with self.assertRaises( + ValidationError, + msg="You cannot run custom code on multiple servers at once.", + ): + test_wizard.run_command_in_wizard() + + # Now, test with a single server selected + test_wizard.server_ids = [self.server_test_1.id] + + # Ensure that executing command works with a single server selected + test_wizard.run_command_in_wizard() + self.assertTrue( + test_wizard.result, + msg="Command execution should succeed with a single server selected", + ) + + def test_custom_variable_values_creation(self): + """ + Test that custom variable values are created properly + when command has variables + """ + # Add manager as server user + self.server_test_1.write({"user_ids": [(4, self.manager.id)]}) + + # Create variables that will be used in command + variable = self.Variable.create( + { + "name": "Test Variable", + "reference": "test_var", + "variable_type": "s", # string type + } + ) + option_variable = self.Variable.create( + { + "name": "Option Variable", + "reference": "opt_var", + "variable_type": "o", # option type + } + ) + option = self.VariableOption.create( + { + "name": "Test Option", + "value_char": "option_value", + "variable_id": option_variable.id, + } + ) + + # Add variable values to server + self.VariableValue.create( + [ + { + "variable_id": variable.id, + "server_id": self.server_test_1.id, + "value_char": "server value", + }, + { + "variable_id": option_variable.id, + "server_id": self.server_test_1.id, + "value_char": "option_value", + }, + ] + ) + + # Create command that uses these variables in its code + command = self.Command.create( + { + "name": "Test Command with Variables", + "action": "ssh_command", + "code": "echo {{ test_var }} && echo {{ opt_var }}", + } + ) + + # Create wizard + wizard = ( + self.env["cx.tower.command.run.wizard"] + .with_user(self.manager) + .create( + { + "server_ids": [self.server_test_1.id], + "command_id": command.id, + "action": "ssh_command", + } + ) + ) + + # Trigger onchange to generate custom_variable_values + wizard._onchange_command_variable_ids() + + # Check that custom variable values were created + self.assertEqual(len(wizard.custom_variable_value_ids), 2) + + # Check char variable value + char_value = wizard.custom_variable_value_ids.filtered( + lambda v: v.variable_id == variable + ) + self.assertTrue(char_value) + self.assertEqual(char_value.value_char, "server value") + + # Check option variable value + option_value = wizard.custom_variable_value_ids.filtered( + lambda v: v.variable_id == option_variable + ) + self.assertTrue(option_value) + self.assertEqual(option_value.value_char, "option_value") + self.assertEqual(option_value.option_id, option) + + # Try to change variable value when user doesn't have write access + char_value.value_char = "custom value" + + # Run command + wizard.run_command_on_server() + + # Get latest command log + command_log = self.env["cx.tower.command.log"].search( + [ + ("server_id", "=", self.server_test_1.id), + ("command_id", "=", command.id), + ], + order="create_date desc", + limit=1, + ) + + # Verify that original server values were used + self.assertEqual(command_log.code, "echo server value && echo option_value") + + def test_custom_variable_values_with_manager_access(self): + """ + Test that custom variable values are applied + when manager has write access + """ + # Add manager as server manager + self.server_test_1.write({"manager_ids": [(4, self.manager.id)]}) + + # Create variables that will be used in command + variable = self.Variable.create( + { + "name": "Test Variable", + "reference": "test_var", + "variable_type": "s", # string type + } + ) + + # Add variable value to server + self.VariableValue.create( + { + "variable_id": variable.id, + "server_id": self.server_test_1.id, + "value_char": "server value", + } + ) + + # Create command that uses the variable + command = self.Command.create( + { + "name": "Test Command with Variables", + "action": "ssh_command", + "code": "echo {{ test_var }}", + } + ) + + # Create wizard + wizard = ( + self.env["cx.tower.command.run.wizard"] + .with_user(self.manager) + .create( + { + "server_ids": [self.server_test_1.id], + "command_id": command.id, + "action": "ssh_command", + } + ) + ) + + # Trigger onchange to generate custom_variable_value_ids + wizard._onchange_command_variable_ids() + + # Modify variable value + wizard.custom_variable_value_ids.filtered( + lambda v: v.variable_id == variable + ).value_char = "manager value" + + # Run command + wizard.run_command_on_server() + + # Get latest command log + command_log = self.env["cx.tower.command.log"].search( + [ + ("server_id", "=", self.server_test_1.id), + ("command_id", "=", command.id), + ], + order="create_date desc", + limit=1, + ) + + # Verify that custom value was used + self.assertEqual(command_log.code, "echo manager value") + + def test_default_applicability_for_regular_and_manager(self): + """sets applicability='this' for regular users, keeps default for managers.""" + # Regular user (no special groups) + default_usr = ( + self.env["cx.tower.command.run.wizard"] + .with_user(self.user_bob) + .default_get(["applicability"]) + ) + self.assertEqual(default_usr.get("applicability"), "this") + + # Manager user should receive the original default ("shared") + self.add_to_group(self.user_bob, "cetmix_tower_server.group_manager") + default_mgr = ( + self.env["cx.tower.command.run.wizard"] + .with_user(self.user_bob) + .default_get(["applicability"]) + ) + self.assertEqual(default_mgr.get("applicability"), "shared") + + def test_compute_show_servers_behavior(self): + """Should enforce 'this' for regular users but preserve manager choice.""" + # Grant Bob the basic 'user' group so he can read servers and create the wizard + self.add_to_group(self.user_bob, "cetmix_tower_server.group_user") + + # Ensure Bob has read access to the first server + self.server_test_1.write({"user_ids": [(4, self.user_bob.id)]}) + # Create a second server and grant Bob read access to it + srv2 = self.Server.create( + { + "name": "Server 2", + "ip_v4_address": "127.0.0.2", + "ssh_username": "root", + "ssh_password": "pwd", + "ssh_auth_mode": "p", + "os_id": self.os_debian_10.id, + } + ) + srv2.write({"user_ids": [(4, self.user_bob.id)]}) + + # --- Regular user scenario --- + wiz_usr = ( + self.env["cx.tower.command.run.wizard"] + .with_user(self.user_bob) + .create({"server_ids": [self.server_test_1.id, srv2.id]}) + ) + # Compute show_servers under Bob; he should see both servers + wiz_usr._compute_show_servers() + self.assertTrue(wiz_usr.show_servers) + # Enforcement should set applicability to 'this' + self.assertEqual(wiz_usr.applicability, "this") + + # --- Manager user scenario --- + self.add_to_group(self.user_bob, "cetmix_tower_server.group_manager") + # Grant Bob manager access to both servers + self.server_test_1.write({"manager_ids": [(4, self.user_bob.id)]}) + srv2.write({"manager_ids": [(4, self.user_bob.id)]}) + + wiz_mgr = ( + self.env["cx.tower.command.run.wizard"] + .with_user(self.user_bob) + .create({"server_ids": [self.server_test_1.id, srv2.id]}) + ) + # Compute show_servers under Bob as manager + wiz_mgr._compute_show_servers() + # Manager should also see both servers + self.assertTrue(wiz_mgr.show_servers) + # Enforcement should not override manager's choice of 'shared' + self.assertEqual(wiz_mgr.applicability, "shared") + + def test_required_variable_validation(self): + """ + Wizard must block execution when a required variable is empty + and allow it after the value is provided. + """ + # Create a required variable + var = self.Variable.create( + { + "name": "Req Var", + "reference": "req_var", + "variable_type": "s", + } + ) + self.VariableValue.create( + { + "variable_id": var.id, + "server_id": self.server_test_1.id, + "required": True, + "value_char": "", + } + ) + + # Create command that uses this variable + cmd = self.Command.create( + { + "name": "Echo Req Var", + "action": "ssh_command", + "code": "echo {{ req_var }}", + "variable_ids": [(4, var.id)], + } + ) + + self.server_test_1.write({"user_ids": [(4, self.manager.id)]}) + + # Create wizard as manager user + wiz = ( + self.env["cx.tower.command.run.wizard"] + .with_user(self.manager) + .create( + { + "server_ids": [self.server_test_1.id], + "command_id": cmd.id, + } + ) + ) + + # Create lines of configuration + wiz._onchange_command_variable_ids() + wiz._compute_has_missing_required_values() + + # Test blocking behavior + self.assertTrue(wiz.has_missing_required_values) + with self.assertRaises(ValidationError): + wiz.run_command_on_server() + + # Fill the value directly in the wizard line + wiz.custom_variable_value_ids.filtered( + lambda line: line.variable_id == var + ).value_char = "filled" + + # Recompute the flag + wiz._compute_has_missing_required_values() + self.assertFalse(wiz.has_missing_required_values) + + # Now the execution should pass + wiz.run_command_on_server()