Tower: upload cetmix_tower_server 16.0.3.0.1 (via marketplace)

addons/cetmix_tower_server/tests/test_cetmix_tower.py

@@ -0,0 +1,244 @@
+# Copyright (C) 2024 Cetmix OÜ
+# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
+from unittest.mock import patch
+
+from odoo.tools import mute_logger
+
+from ..models.constants import GENERAL_ERROR, NOT_FOUND, SSH_CONNECTION_ERROR
+from .common import TestTowerCommon
+
+
+class TestCetmixTower(TestTowerCommon):
+    """
+    Tests for the 'cetmix.tower' helper model
+    """
+
+    @mute_logger("odoo.addons.cetmix_tower_server.models.cetmix_tower")
+    def test_server_set_variable_value(self):
+        """Test plan line action naming"""
+
+        # -- 1--
+        # Create new variable
+        variable_meme = self.Variable.create(
+            {"name": "Meme Variable", "reference": "meme_variable"}
+        )
+
+        # Set variable for Server 1
+        result = self.CetmixTower.server_set_variable_value(
+            server_reference=self.server_test_1.reference,
+            variable_reference=variable_meme.reference,
+            value="Doge",
+        )
+
+        # Check exit code
+        self.assertEqual(result["exit_code"], 0, "Exit code must be equal to 0")
+
+        # Check variable value
+        variable_value = self.VariableValue.search(
+            [("variable_id", "=", variable_meme.id)]
+        )
+
+        self.assertEqual(len(variable_value), 1, "Must be 1 result")
+        self.assertEqual(variable_value.value_char, "Doge", "Must be Doge!")
+
+        # -- 2 --
+        # Update existing variable value
+
+        # Set variable for Server 1
+        result = self.CetmixTower.server_set_variable_value(
+            server_reference=self.server_test_1.reference,
+            variable_reference=variable_meme.reference,
+            value="Pepe",
+        )
+
+        # Check exit code
+        self.assertEqual(result["exit_code"], 0, "Exit code must be equal to 0")
+
+        # Check variable value
+        variable_value = self.VariableValue.search(
+            [("variable_id", "=", variable_meme.id)]
+        )
+
+        self.assertEqual(len(variable_value), 1, "Must be 1 result")
+        self.assertEqual(variable_value.value_char, "Pepe", "Must be Pepe!")
+
+    @mute_logger("odoo.addons.cetmix_tower_server.models.cetmix_tower")
+    def test_server_get_variable_value(self):
+        """Test getting value for server"""
+        variable_meme = self.Variable.create(
+            {"name": "Meme Variable", "reference": "meme_variable"}
+        )
+        global_value = self.VariableValue.create(
+            {"variable_id": variable_meme.id, "value_char": "Memes Globalvs"}
+        )
+
+        # -- 1 -- Get value for Server with no server value defined
+        value = self.CetmixTower.server_get_variable_value(
+            self.server_test_1.reference, variable_meme.reference
+        )
+        self.assertEqual(value, global_value.value_char)
+
+        # -- 2 -- Add server value and try again
+        server_value = self.VariableValue.create(
+            {
+                "variable_id": variable_meme.id,
+                "value_char": "Memes Servervs",
+                "server_id": self.server_test_1.id,
+            }
+        )
+        value = self.CetmixTower.server_get_variable_value(
+            self.server_test_1.reference, variable_meme.reference
+        )
+        self.assertEqual(value, server_value.value_char)
+
+    @mute_logger("odoo.addons.cetmix_tower_server.models.cetmix_tower")
+    def test_server_check_ssh_connection(self):
+        """
+        Test SSH connection check with a mocked function that
+        either returns a dictionary or raises an exception.
+        """
+
+        # Test successful connection
+        result = self.env["cetmix.tower"].server_check_ssh_connection(
+            self.server_test_1.reference,
+        )
+        self.assertEqual(result["exit_code"], 0, "SSH connection should be successful.")
+
+        def test_ssh_connection(this, *args, **kwargs):
+            return {"status": GENERAL_ERROR}
+
+        with patch.object(
+            self.registry["cx.tower.server"], "test_ssh_connection", test_ssh_connection
+        ):
+            # Test connection timeout after max attempts
+            result = self.env["cetmix.tower"].server_check_ssh_connection(
+                self.server_test_1.reference,
+                attempts=2,
+                wait_time=1,
+            )
+            self.assertEqual(
+                result["exit_code"],
+                SSH_CONNECTION_ERROR,
+                "SSH connection should timeout after maximum attempts.",
+            )
+
+    @mute_logger("odoo.addons.cetmix_tower_server.models.cetmix_tower")
+    def test_server_run_command(self):
+        """Test running command on server"""
+        # Create test command
+        command = self.Command.create(
+            {
+                "name": "Test Command",
+                "reference": "test_command",
+                "code": "echo 'Hello World'",
+                "action": "ssh_command",
+            }
+        )
+
+        # -- 1 -- Test with non-existent server
+        result = self.CetmixTower.server_run_command(
+            server_reference="non_existent",
+            command_reference=command.reference,
+        )
+        self.assertEqual(result["exit_code"], NOT_FOUND)
+        self.assertEqual(result["message"], "Server not found")
+
+        # -- 2 -- Test with non-existent command
+        result = self.CetmixTower.server_run_command(
+            server_reference=self.server_test_1.reference,
+            command_reference="non_existent",
+        )
+        self.assertEqual(result["exit_code"], NOT_FOUND)
+        self.assertEqual(result["message"], "Command not found")
+
+        # -- 3 -- Test successful command execution
+        result = self.CetmixTower.server_run_command(
+            server_reference=self.server_test_1.reference,
+            command_reference=command.reference,
+        )
+        self.assertEqual(result["exit_code"], 0)
+
+    @mute_logger("odoo.addons.cetmix_tower_server.models.cetmix_tower")
+    def test_server_run_flight_plan(self):
+        """Test running flight plan on server"""
+        # Create test flight plan
+        flight_plan = self.Plan.create(
+            {
+                "name": "Test Flight Plan",
+                "reference": "test_flight_plan",
+            }
+        )
+
+        # -- 1 -- Test with non-existent server
+        result = self.CetmixTower.server_run_flight_plan(
+            server_reference="non_existent",
+            flight_plan_reference=flight_plan.reference,
+        )
+        self.assertFalse(result, "Should return False for non-existent server")
+
+        # -- 2 -- Test with non-existent flight plan
+        result = self.CetmixTower.server_run_flight_plan(
+            server_reference=self.server_test_1.reference,
+            flight_plan_reference="non_existent",
+        )
+        self.assertFalse(result, "Should return False for non-existent flight plan")
+
+        # -- 3 -- Test successful flight plan execution
+        with patch.object(self.server_test_1.__class__, "run_flight_plan") as mock_run:
+            # Setup mock to return a plan log record
+            plan_log = self.PlanLog.create(
+                {
+                    "name": "Test Log",
+                    "server_id": self.server_test_1.id,
+                    "plan_id": flight_plan.id,
+                }
+            )
+            mock_run.return_value = plan_log
+
+            # Run flight plan
+            result = self.CetmixTower.server_run_flight_plan(
+                server_reference=self.server_test_1.reference,
+                flight_plan_reference=flight_plan.reference,
+            )
+
+            # Verify result
+            self.assertEqual(result, plan_log, "Should return plan log record")
+            mock_run.assert_called_once_with(flight_plan)
+
+    @mute_logger("odoo.addons.cetmix_tower_server.models.cetmix_tower")
+    def test_server_run_command_with_variable_values(self):
+        """Test running command with variable values"""
+        # Create test command
+        command = self.Command.create(
+            {
+                "name": "Test Command",
+                "reference": "test_command",
+                "code": "result = {'exit_code': 0, 'message': {{ test_version }}}",
+                "action": "python_code",
+            }
+        )
+        # Set variable value for the server
+        self.CetmixTower.server_set_variable_value(
+            server_reference=self.server_test_1.reference,
+            variable_reference=self.variable_version.reference,
+            value="prod",
+        )
+
+        # -- 1 --
+        # Run command without modifying variable values
+        result = self.CetmixTower.server_run_command(
+            server_reference=self.server_test_1.reference,
+            command_reference=command.reference,
+        )
+        self.assertEqual(result["exit_code"], 0)
+        self.assertEqual(result["message"], "prod")
+
+        # -- 2 --
+        # Run command with modified variable values
+        result = self.CetmixTower.server_run_command(
+            server_reference=self.server_test_1.reference,
+            command_reference=command.reference,
+            **{"test_version": "dev"},
+        )
+        self.assertEqual(result["exit_code"], 0)
+        self.assertEqual(result["message"], "dev")