# 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()