From 693821eb536e8f93b2a5ca42537ba3252b6084d1 Mon Sep 17 00:00:00 2001 From: git_admin Date: Mon, 27 Apr 2026 08:43:31 +0000 Subject: [PATCH] Tower: upload cetmix_tower_server 16.0.2.2.9 (via marketplace) --- .../tests/test_variable.py | 1130 +++++++++++++++++ 1 file changed, 1130 insertions(+) create mode 100644 addons/cetmix_tower_server/tests/test_variable.py diff --git a/addons/cetmix_tower_server/tests/test_variable.py b/addons/cetmix_tower_server/tests/test_variable.py new file mode 100644 index 0000000..29838fd --- /dev/null +++ b/addons/cetmix_tower_server/tests/test_variable.py @@ -0,0 +1,1130 @@ +# Copyright (C) 2022 Cetmix OÜ +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). + +from unittest.mock import patch + +from psycopg2 import IntegrityError + +from odoo import _, fields +from odoo.exceptions import AccessError, ValidationError +from odoo.tests.common import Form +from odoo.tools.misc import mute_logger + +from .common import TestTowerCommon + + +class TestTowerVariable(TestTowerCommon): + """Testing variables and variable values.""" + + def check_variable_values(self, vals, server_ids=None): + """Check if variable values are correctly stored in db + + Args: + vals (List of tuples): format ("variable_id", "value") + server_id (cx.tower.server()): Servers those variables belong to. + """ + if server_ids: + variable_records = server_ids.variable_value_ids + else: + variable_records = self.VariableValue.search([("is_global", "=", True)]) + len_vals = len(vals) + + # Ensure correct number of records + self.assertEqual( + len(variable_records), len_vals, msg="Must be %s records" % str(len_vals) + ) + + # Check variable values + for val in vals: + variable_line = variable_records.filtered( + lambda v, val=val: v.variable_id.id == val[0] + ) + self.assertEqual( + len(variable_line), 1, msg="Must be a single variable line" + ) + expected_value = val[1] or False + self.assertEqual( + variable_line.value_char, + expected_value, + msg="Variable value does not match provided one", + ) + + def test_variable_values(self): + """Test common variable operations""" + + # -- 1 -- + # Server specific variables + + # Add two variables + with Form(self.server_test_1) as f: + with f.variable_value_ids.new() as line: + line.variable_id = self.variable_dir + line.value_char = "/opt/odoo" + with f.variable_value_ids.new() as line: + line.variable_id = self.variable_url + line.value_char = "example.com" + f.save() + + vals = [ + (self.variable_url.id, "example.com"), + (self.variable_dir.id, "/opt/odoo"), + ] + self.check_variable_values(vals=vals, server_ids=self.server_test_1) + + # Add another variable and edit the existing one + with Form(self.server_test_1) as f: + with f.variable_value_ids.edit(1) as line: + line.value_char = "meme.example.com" + with f.variable_value_ids.new() as line: + line.variable_id = self.variable_version + line.value_char = "10.0" + f.save() + + vals = [ + (self.variable_url.id, "meme.example.com"), + (self.variable_dir.id, "/opt/odoo"), + (self.variable_version.id, "10.0"), + ] + self.check_variable_values(vals=vals, server_ids=self.server_test_1) + + # Delete two variables, add a new one + with Form(self.server_test_1) as f: + f.variable_value_ids.remove(index=0) + f.variable_value_ids.remove(index=0) + with f.variable_value_ids.new() as line: + line.variable_id = self.variable_os + line.value_char = "Debian" + + # Add an empty variable value + with f.variable_value_ids.new() as line: + line.variable_id = self.variable_url + f.save() + + vals = [ + (self.variable_os.id, "Debian"), + (self.variable_version.id, "10.0"), + (self.variable_url.id, False), + ] + self.check_variable_values(vals=vals, server_ids=self.server_test_1) + + # Test 'get_variable_values' function + res = self.server_test_1.get_variable_values( + ["test_dir", "test_os", "test_url", "test_version"] + ) + self.assertEqual(len(res), 1, "Must be a single record key in the result") + + res_vars = res.get(self.server_test_1.id) + var_dir = res_vars["test_dir"] + var_os = res_vars["test_os"] + var_url = res_vars["test_url"] + var_version = res_vars["test_version"] + + self.assertIsNone(var_dir, msg="Variable 'dir' must be None") + self.assertFalse(var_url, msg="Variable 'url' must be False") + self.assertEqual(var_os, "Debian", msg="Variable 'os' must be 'Debian'") + self.assertEqual(var_version, "10.0", msg="Variable 'version' must be '10.0'") + + # -- 2 -- + # Test global variable values + + # Create a global value for the 'dir' variable + self.VariableValue.create( + {"variable_id": self.variable_dir.id, "value_char": "/global/dir"} + ) + res = self.server_test_1.get_variable_values( + ["test_dir", "test_os", "test_url", "test_version"] + ) + self.assertEqual(len(res), 1, "Must be a single record key in the result") + + res_vars = res.get(self.server_test_1.id) + var_dir = res_vars["test_dir"] + var_os = res_vars["test_os"] + var_url = res_vars["test_url"] + var_version = res_vars["test_version"] + + self.assertEqual( + var_dir, "/global/dir", msg="Variable 'dir' must be equal to '/global/dir'" + ) + self.assertFalse(var_url, msg="Variable 'url' must be False") + self.assertEqual(var_os, "Debian", msg="Variable 'os' must be 'Debian'") + self.assertEqual(var_version, "10.0", msg="Variable 'version' must be '10.0'") + + # Now save a local value for the variable + with Form(self.server_test_1) as f: + with f.variable_value_ids.new() as line: + line.variable_id = self.variable_dir + line.value_char = "/opt/odoo" + f.save() + + # Check + res = self.server_test_1.get_variable_values( + ["test_dir", "test_os", "test_url", "test_version"] + ) + self.assertEqual(len(res), 1, "Must be a single record key in the result") + + res_vars = res.get(self.server_test_1.id) + var_dir = res_vars["test_dir"] + var_os = res_vars["test_os"] + var_url = res_vars["test_url"] + var_version = res_vars["test_version"] + + self.assertEqual( + var_dir, "/opt/odoo", msg="Variable 'dir' must be equal to '/opt/odoo'" + ) + self.assertFalse(var_url, msg="Variable 'url' must be False") + self.assertEqual(var_os, "Debian", msg="Variable 'os' must be 'Debian'") + self.assertEqual(var_version, "10.0", msg="Variable 'version' must be '10.0'") + + def test_variables_in_variable_values(self): + """Test variables in variable values + eg + home: /home + user: bob + home_dir: {{ home }}/{{ user }} --> /home/bob + """ + + # Add local variables + with Form(self.server_test_1) as f: + with f.variable_value_ids.new() as line: + line.variable_id = self.variable_dir + line.value_char = "/web" + with f.variable_value_ids.new() as line: + line.variable_id = self.variable_path + line.value_char = "{{ test_dir }}/{{ test_version }}" + with f.variable_value_ids.new() as line: + line.variable_id = self.variable_url + line.value_char = "{{ test_path_ }}/example.com" + f.save() + + # Create a global value for the 'Version' variable + self.VariableValue.create( + {"variable_id": self.variable_version.id, "value_char": "10.0"} + ) + + # Check values + res = self.server_test_1.get_variable_values( + ["test_dir", "test_url", "test_version"] + ) + self.assertEqual(len(res), 1, "Must be a single record key in the result") + + res_vars = res.get(self.server_test_1.id) + var_dir = res_vars["test_dir"] + var_url = res_vars["test_url"] + var_version = res_vars["test_version"] + + self.assertEqual(var_dir, "/web", msg="Variable 'dir' must be '/web'") + self.assertEqual( + var_url, + "/web/10.0/example.com", + msg="Variable 'url' must be '/web/10.0/example.com'", + ) + self.assertEqual(var_version, "10.0", msg="Variable 'version' must be '10.0'") + + def test_variable_values_unlink(self): + """Ensure variable values are deleted properly + - Create a new server + - Add 2 variable values + - Delete server + - Ensure variable values are deleted + """ + + def get_value_count(variable): + """helper function to count variable value records + Arg: (cx.tower.variable) variable rec + Returns: (int) record count + """ + return self.VariableValue.search_count([("variable_id", "=", variable.id)]) + + # Get variable values count before adding variables to server + count_dir_before = get_value_count(self.variable_dir) + count_url_before = get_value_count(self.variable_url) + + # Create new server + server_test_var = self.Server.create( + { + "name": "Test Var", + "os_id": self.os_debian_10.id, + "ip_v4_address": "localhost", + "ssh_username": "bob", + "ssh_password": "pass", + } + ) + + # Add two variables to server + with Form(server_test_var) as f: + with f.variable_value_ids.new() as line: + line.variable_id = self.variable_dir + line.value_char = "/opt/odoo" + with f.variable_value_ids.new() as line: + line.variable_id = self.variable_url + line.value_char = "example.com" + f.save() + + # Number of values should be incremented + self.assertEqual( + get_value_count(self.variable_dir), + count_dir_before + 1, + msg="Value count must be incremented!", + ) + self.assertEqual( + get_value_count(self.variable_url), + count_url_before + 1, + msg="Value count must be incremented!", + ) + + # Delete the server + server_test_var.unlink() + self.assertEqual( + get_value_count(self.variable_dir), + count_dir_before, + msg="Value count must be same as before server creation!", + ) + self.assertEqual( + get_value_count(self.variable_url), + count_url_before, + msg="Value count must be same as before server creation!", + ) + + def test_variable_value_toggle_global(self): + """Test what happens when variable value 'global' setting is togged""" + + variable_meme = self.Variable.create({"name": "meme"}) + variable_value_pepe = self.VariableValue.create( + {"variable_id": variable_meme.id, "value_char": "Pepe"} + ) + + self.assertEqual( + variable_value_pepe.is_global, True, msg="Value 'Pepe' must be global" + ) + + # Test `_check_is_global` function + self.assertEqual( + variable_value_pepe._check_is_global(), + True, + msg="Value 'Pepe' must be global", + ) + + # Try to create another global value for the same variable + with self.assertRaises(ValidationError) as err: + self.VariableValue.create( + {"variable_id": variable_meme.id, "value_char": "Doge"} + ) + + # We check the message in order to ensure that + # exception was raised by the correct event. + self.assertEqual( + err.exception.args[0], + _("Only one global value can be defined for variable 'meme'"), + msg="Error message doesn't match. Check if you have modified it in code:" + "models/cx_tower_server.py", + ) + + # Try to disable 'global' for a global variable explicitly + with self.assertRaises(ValidationError) as err: + variable_value_pepe.is_global = False + + # We check the message in order to ensure that + # exception was raised by the correct event. + self.assertEqual( + err.exception.args[0], + _( + "Cannot change 'global' status for " + "'meme' with value 'Pepe'." + "\nTry to assigns it to a record instead." + ), + msg="Error message doesn't match. Check if you have modified it in code:" + "models/cx_tower_server.py", + ) + + def test_system_variable_server_type_values(self): + """Test system variables of `server` type""" + + # Modify server record for testing + self.server_test_1.ip_v6_address = "suchmuchipv6" + self.server_test_1.url = "meme.example.com" + self.server_test_1.partner_id = ( + self.env["res.partner"].create({"name": "Pepe Frog"}).id + ) + + # Create new command with system variables + command = self.Command.create( + { + "name": "Super System Command", + "code": "echo {{ tower.server.name }} " + "{{ tower.server.username}} " + "{{ tower.server.partner_name }} " + "{{ tower.server.ipv4 }} " + "{{ tower.server.ipv6 }} " + "{{ tower.server.url }} ", + } + ) + + # Get variables + variables = command.get_variables().get(str(command.id)) + # Get variable values + variable_values = self.server_test_1.get_variable_values(variables).get( + self.server_test_1.id + ) + + # Check values + self.assertEqual( + variable_values["tower"]["server"]["name"], + self.server_test_1.name, + "System variable doesn't match server property", + ) + self.assertEqual( + variable_values["tower"]["server"]["reference"], + self.server_test_1.reference, + "System variable doesn't match server property", + ) + self.assertEqual( + variable_values["tower"]["server"]["username"], + self.server_test_1.ssh_username, + "System variable doesn't match server property", + ) + self.assertEqual( + variable_values["tower"]["server"]["username"], + self.server_test_1.ssh_username, + "System variable doesn't match server property", + ) + self.assertEqual( + variable_values["tower"]["server"]["partner_name"], + self.server_test_1.partner_id.name, + "System variable doesn't match server property", + ) + self.assertEqual( + variable_values["tower"]["server"]["ipv4"], + self.server_test_1.ip_v4_address, + "System variable doesn't match server property", + ) + self.assertEqual( + variable_values["tower"]["server"]["ipv6"], + self.server_test_1.ip_v6_address, + "System variable doesn't match server property", + ) + self.assertEqual( + variable_values["tower"]["server"]["url"], + self.server_test_1.url, + "System variable doesn't match server property", + ) + + @patch( + "odoo.addons.cetmix_tower_server.models.cx_tower_variable_mixin.fields.Datetime.now", + return_value=fields.Datetime.now(), + ) + @patch( + "odoo.addons.cetmix_tower_server.models.cx_tower_variable_mixin.fields.Date.today", + return_value=fields.Date.today(), + ) + @patch( + "odoo.addons.cetmix_tower_server.models.cx_tower_variable_mixin.uuid.uuid4", + return_value="suchmuchuuid4", + ) + def test_system_variable_tools_type_values(self, mock_uuid4, mock_today, mock_now): + """Test system variables of `tools` type""" + + # Create new command with system variables + command = self.Command.create( + {"name": "Super System Command", "code": "echo {{ tower.tools.uuid}}"} + ) + + # Get variables + variables = command.get_variables().get(str(command.id)) + # Get variable values + variable_values = self.server_test_1.get_variable_values(variables).get( + self.server_test_1.id + ) + + # Check values + self.assertEqual( + variable_values["tower"]["tools"]["uuid"], + mock_uuid4.return_value, + "System variable doesn't match result provided by tools", + ) + self.assertEqual( + variable_values["tower"]["tools"]["today"], + str(mock_today.return_value), + "System variable doesn't match result provided by tools", + ) + self.assertEqual( + variable_values["tower"]["tools"]["now"], + str(mock_now.return_value), + "System variable doesn't match result provided by tools", + ) + self.assertEqual( + variable_values["tower"]["tools"]["today_underscore"], + str(mock_today.return_value) + .replace("-", "_") + .replace(" ", "_") + .replace(":", "_") + .replace(".", "_") + .replace("/", "_"), + "System variable doesn't match result provided by tools", + ) + self.assertEqual( + variable_values["tower"]["tools"]["now_underscore"], + str(mock_now.return_value) + .replace("-", "_") + .replace(":", "_") + .replace(" ", "_") + .replace(".", "_") + .replace("/", "_"), + "System variable doesn't match result provided by tools", + ) + + def test_make_value_pythonic(self): + """Test making variable values 'pythonic`""" + + # Number + value = 12.34 + expected_value = '"12.34"' + result_value = self.Command._make_value_pythonic(value) + + self.assertEqual( + expected_value, result_value, "Result value doesn't match expected" + ) + + # Text + value = "Doge much like" + expected_value = '"Doge much like"' + result_value = self.Command._make_value_pythonic(value) + + self.assertEqual( + expected_value, result_value, "Result value doesn't match expected" + ) + + # Boolean + value = True + expected_value = True + result_value = self.Command._make_value_pythonic(value) + + self.assertEqual( + expected_value, result_value, "Result value doesn't match expected" + ) + + # None + value = None + expected_value = None + result_value = self.Command._make_value_pythonic(value) + + self.assertEqual( + expected_value, result_value, "Result value doesn't match expected" + ) + + # Dict + value = {"doge": {"likes": "memes", "much": 200}} + expected_value = {"doge": {"likes": '"memes"', "much": '"200"'}} + result_value = self.Command._make_value_pythonic(value) + + self.assertEqual( + expected_value, result_value, "Result value doesn't match expected" + ) + + def test_get_by_variable_reference(self): + """Test getting variable values by variable reference""" + + 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 + server_result = self.VariableValue.get_by_variable_reference( + variable_meme.reference, server_id=self.server_test_1.id + ) + self.assertIsNone(server_result.get("server")) + self.assertIsNone(server_result.get("server_template")) + self.assertEqual(server_result.get("global"), 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, + } + ) + server_result = self.VariableValue.get_by_variable_reference( + variable_meme.reference, server_id=self.server_test_1.id + ) + self.assertEqual(server_result.get("server"), server_value.value_char) + self.assertEqual(server_result.get("global"), global_value.value_char) + self.assertIsNone(server_result.get("server_template")) + + # -- 3 -- Do not fetch global value now + server_result = self.VariableValue.get_by_variable_reference( + variable_meme.reference, server_id=self.server_test_1.id, check_global=False + ) + self.assertIsNone(server_result.get("global")) + self.assertEqual(server_result.get("server"), server_value.value_char) + self.assertIsNone(server_result.get("server_template")) + + # -- 4 -- Check server template value + server_template_value = self.VariableValue.create( + { + "variable_id": variable_meme.id, + "value_char": "Memes Servervs Templatvs", + "server_template_id": self.server_template_sample.id, + } + ) + server_result = self.VariableValue.get_by_variable_reference( + variable_meme.reference, server_template_id=self.server_template_sample.id + ) + self.assertEqual(server_result.get("global"), global_value.value_char) + self.assertIsNone(server_result.get("server")) + self.assertEqual( + server_result.get("server_template"), server_template_value.value_char + ) + + def test_single_assignment(self): + """Test that a variable can only be assigned to one model at a time.""" + # Create a variable value assigned to the server + variable_value = self.env["cx.tower.variable.value"].create( + { + "variable_id": self.variable_os.id, + "value_char": "Branch = Main", + "server_id": self.server_test_1.id, + } + ) + + # Try to assign the same variable value to + # server template and expect a ValidationError + with self.assertRaises(ValidationError): + variable_value.write({"server_template_id": self.server_template_sample.id}) + + # Try to assign the same variable value to + # plan line action and expect a ValidationError + with self.assertRaises(ValidationError): + variable_value.write({"plan_line_action_id": self.plan_line_1_action_1.id}) + + def test_unique_assignment(self): + """Test that the same variable value cannot be + assigned multiple times to the same record. + """ + + # Create a variable + variable = self.env["cx.tower.variable"].create( + {"name": "Environment Type", "note": "The environment type for the server."} + ) + + # Create a server + server = self.env["cx.tower.server"].create( + { + "name": "Test Server", + "ip_v4_address": "127.0.0.1", + "ssh_username": "testuser", + "ssh_password": "testpassword", + "ssh_auth_mode": "p", + } + ) + + # Create a variable value for the server + self.env["cx.tower.variable.value"].create( + { + "variable_id": variable.id, + "value_char": "Production", + "server_id": server.id, + } + ) + + # Try to create a second variable value with the same variable and server + with mute_logger("odoo.sql_db"), self.assertRaises( + IntegrityError, + msg="A variable value cannot be assigned multiple times to the same server", + ): + self.env["cx.tower.variable.value"].create( + { + "variable_id": variable.id, + "value_char": "Production", + "server_id": server.id, + } + ) + + def test_value_access_level_consistency(self): + """Test that variable value access level cannot be lower + than variable access level.""" + + # Create test servers + server_2 = self.Server.create( + { + "name": "Test Server 2", + "ip_v4_address": "localhost", + "ssh_username": "admin", + "ssh_password": "password", + "os_id": self.os_debian_10.id, + } + ) + + server_3 = self.Server.create( + { + "name": "Test Server 3", + "ip_v4_address": "localhost", + "ssh_username": "admin", + "ssh_password": "password", + "os_id": self.os_debian_10.id, + } + ) + + # Create a variable with access level "2" + variable_restricted = self.Variable.create( + { + "name": "restricted_variable", + "access_level": "2", + } + ) + + # Should succeed: value with same access level as variable + try: + self.VariableValue.create( + { + "variable_id": variable_restricted.id, + "value_char": "test_value1", + "access_level": "2", + "is_global": True, + } + ) + except ValidationError: + self.fail("Should allow creating value with same access level as variable") + + # Should succeed: value with higher access level than variable + try: + self.VariableValue.create( + { + "variable_id": variable_restricted.id, + "value_char": "test_value2", + "access_level": "3", + "server_id": server_2.id, + } + ) + except ValidationError: + self.fail( + "Should allow creating value with higher access level than variable" + ) + + # Should fail: value with lower access level than variable + with self.assertRaises( + ValidationError, + msg="Should not allow creating value with lower access level than variable", + ): + self.VariableValue.create( + { + "variable_id": variable_restricted.id, + "value_char": "test_value3", + "access_level": "1", + "server_id": server_3.id, + } + ) + + # Test updating existing value's access level + value = self.VariableValue.create( + { + "variable_id": self.variable_dir.id, # Using a different variable + "value_char": "test_value4", + "access_level": "2", + "server_id": server_3.id, + } + ) + + # Should fail: updating to lower access level than variable + with self.assertRaises( + ValidationError, + msg="Should not allow updating value to lower access level than variable", + ): + value.write({"access_level": "1"}) + + # Should succeed: updating to higher access level than variable + try: + value.write({"access_level": "3"}) + except ValidationError: + self.fail( + "Should allow updating value to higher access level than variable" + ) + + def test_variable_access_rights(self): + """Test access rights for variables based on access levels and user roles.""" + + # Create variables with different access levels + variable_level_1 = self.Variable.create( + { + "name": "Level 1 Variable", + "access_level": "1", + } + ) + + variable_level_2 = self.Variable.create( + { + "name": "Level 2 Variable", + "access_level": "2", + } + ) + + variable_level_3 = self.Variable.create( + { + "name": "Level 3 Variable", + "access_level": "3", + } + ) + manager2 = self.Users.create( + { + "name": "Manager 2", + "login": "manager2@example.com", + "groups_id": [(4, self.group_manager.id)], + } + ) + + # Test User Access + # --------------- + # Should see level 1 variables + records = self.Variable.with_user(self.user).search( + [ + ( + "id", + "in", + [variable_level_1.id, variable_level_2.id, variable_level_3.id], + ) + ] + ) + self.assertEqual(len(records), 1, "User should only see level 1 variables") + self.assertEqual( + records.id, variable_level_1.id, "User should only see level 1 variables" + ) + + # Test Manager Access + # ----------------- + # Should see level 1 and 2 variables + records = self.Variable.with_user(self.manager).search( + [ + ( + "id", + "in", + [variable_level_1.id, variable_level_2.id, variable_level_3.id], + ) + ] + ) + self.assertEqual(len(records), 2, "Manager should see level 1 and 2 variables") + self.assertIn( + variable_level_1.id, records.ids, "Manager should see level 1 variables" + ) + self.assertIn( + variable_level_2.id, records.ids, "Manager should see level 2 variables" + ) + + # Test Manager Write Access + # ----------------------- + # Create a variable as manager + manager_variable = self.Variable.with_user(self.manager).create( + { + "name": "Manager Created Variable", + "access_level": "2", + } + ) + + # Manager should be able to modify their own variable + try: + manager_variable.with_user(self.manager).write({"name": "Updated Name"}) + except AccessError: + self.fail("Manager should be able to modify their own variables") + + # Manager should not be able to modify another manager's variable + manager2_variable = self.Variable.with_user(manager2).create( + { + "name": "Other Manager Variable", + "access_level": "2", + } + ) + + with self.assertRaises(AccessError): + manager2_variable.with_user(self.manager).write({"name": "Try Update"}) + + # Manager should not be able to create level 3 variable + with self.assertRaises(AccessError): + self.Variable.with_user(self.manager).create( + { + "name": "Try Level 3", + "access_level": "3", + } + ) + + # Test Root Access + # -------------- + # Root should see all variables + records = self.Variable.with_user(self.root).search( + [ + ( + "id", + "in", + [variable_level_1.id, variable_level_2.id, variable_level_3.id], + ) + ] + ) + self.assertEqual(len(records), 3, "Root should see all variables") + + # Root should be able to create any level variable + try: + self.Variable.with_user(self.root).create( + { + "name": "Root Level 3", + "access_level": "3", + } + ) + except AccessError: + self.fail("Root should be able to create any level variable") + + # Root should be able to modify any variable + try: + variable_level_3.with_user(self.root).write({"name": "Updated by Root"}) + except AccessError: + self.fail("Root should be able to modify any variable") + + def test_validate_value(self): + """Test variable value validation""" + # Create variable with validation pattern + variable_with_pattern = self.Variable.create( + { + "name": "Test Pattern", + "validation_pattern": "^[a-z0-9]+$", + "validation_message": "Only lowercase letters and numbers allowed", + } + ) + + # Test valid values + valid_value = "abc123" + is_valid, message = variable_with_pattern._validate_value(valid_value) + self.assertTrue(is_valid, "Value should be valid") + self.assertIsNone(message, "No message should be returned for valid value") + + # Test invalid values + invalid_value = "ABC123!" + is_valid, message = variable_with_pattern._validate_value(invalid_value) + self.assertFalse(is_valid, "Value should be invalid") + self.assertEqual( + message, + f"Variable: {variable_with_pattern.name}, Value: {invalid_value}\n" + "Only lowercase letters and numbers allowed", + "Invalid value message doesn't match", + ) + + # Test empty value + is_valid, message = variable_with_pattern._validate_value(None) + self.assertTrue(is_valid, "Empty value should be valid") + self.assertIsNone(message, "No message should be returned for empty value") + + # Test variable without pattern + variable_no_pattern = self.Variable.create( + { + "name": "No Pattern", + } + ) + test_value = "Any Value!" + is_valid, message = variable_no_pattern._validate_value(test_value) + self.assertTrue(is_valid, "Value should be valid when no pattern is set") + self.assertIsNone( + message, "No message should be returned when no pattern is set" + ) + + # Test default validation message + variable_default_message = self.Variable.create( + { + "name": "Default Message", + "validation_pattern": "^[a-z]+$", + } + ) + invalid_value = "123" + is_valid, message = variable_default_message._validate_value(invalid_value) + self.assertFalse(is_valid, "Value should be invalid") + self.assertEqual( + message, + f"Variable: {variable_default_message.name}, Value: {invalid_value}\n" + f"{variable_default_message.DEFAULT_VALIDATION_MESSAGE}", + "Default validation message doesn't match", + ) + + +class TestVariableReferenceRename(TestTowerCommon): + """Ensure variable rename updates all Jinja references using shared fixtures.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + + cls.ref_old = cls.variable_version.reference + cls.ref_new = "software_version" + + cls.command = cls.Command.create( + { + "name": "Show version (test)", + "code": f"echo {{ {{ {cls.ref_old} }} }}", + "variable_ids": [(6, 0, [cls.variable_version.id])], + } + ) + + cls.file = cls.File.create( + { + "name": "test_version.txt", + "server_dir": "/tmp", + "code": f"{{ {{ {cls.ref_old} }} }}", + "variable_ids": [(6, 0, [cls.variable_version.id])], + } + ) + + def _rename(self): + """Rename variable and invalidate caches for records under test.""" + self.variable_version.write({"reference": self.ref_new}) + self.command.invalidate_recordset() + self.file.invalidate_recordset() + + def test_false_references_are_ignored(self): + """Ignore malformed or non-Jinja references.""" + cmd_plain = self.Command.create( + { + "name": "Plain", + "code": "print(test_version)", + "variable_ids": [(6, 0, [self.variable_version.id])], + } + ) + cmd_bad = self.Command.create( + { + "name": "BadBrackets", + "code": "{test_version}", + "variable_ids": [(6, 0, [self.variable_version.id])], + } + ) + + self._rename() + cmd_plain.invalidate_recordset() + cmd_bad.invalidate_recordset() + + self.assertEqual(cmd_plain.code, "print(test_version)") + self.assertEqual(cmd_bad.code, "{test_version}") + + def test_multiple_occurrences_replace_all(self): + """Replace all valid Jinja references in one field.""" + code = "A: {{ test_version }}, B: {{ test_version }}, C-end" + cmd_multi = self.Command.create( + { + "name": "Multi", + "code": code, + "variable_ids": [(6, 0, [self.variable_version.id])], + } + ) + + self._rename() + cmd_multi.invalidate_recordset() + actual_ref = self.variable_version.reference + expected = f"A: {{{{ {actual_ref} }}}}, " f"B: {{{{ {actual_ref} }}}}, C-end" + self.assertEqual(cmd_multi.code, expected) + + def test_template_files_updated(self): + """Propagate rename in template and generated file.""" + tpl = self.env["cx.tower.file.template"].create( + { + "name": "TmpTpl", + "file_name": "tpl.txt", + "server_dir": "/tmp", + "code": "{{ test_version }}", + "variable_ids": [(6, 0, [self.variable_version.id])], + } + ) + tpl_file = self.File.create( + { + "name": "from_tpl.txt", + "server_dir": "/tmp", + "template_id": tpl.id, + "code": "{{ test_version }}", + } + ) + + self._rename() + tpl.invalidate_recordset() + tpl_file.invalidate_recordset() + + actual_ref = self.variable_version.reference + expected = f"{{{{ {actual_ref} }}}}" + self.assertEqual(tpl.code, expected) + self.assertEqual(tpl_file.code, expected) + + def test_value_and_plan_line_update(self): + """Update value_char and plan line condition.""" + + def patched_mapping(_): + return { + "cx.tower.command": ["code", "path"], + "cx.tower.file": ["code", "server_dir", "name"], + "cx.tower.file.template": ["code", "server_dir", "file_name"], + "cx.tower.variable.value": ["value_char"], + "cx.tower.plan.line": ["condition"], + } + + with patch.object( + type(self.variable_version), + "_get_propagation_field_mapping", + patched_mapping, + ): + val = self.env["cx.tower.variable.value"].create( + { + "variable_id": self.variable_version.id, + "value_char": "hello {{ test_version }} world", + } + ) + + pl = self.plan_line_1 + pl.write( + { + "variable_ids": [(6, 0, [self.variable_version.id])], + "condition": "if {{ test_version }} then", + } + ) + + self.assertIn(self.variable_version.id, pl.variable_ids.ids) + + self._rename() + val.invalidate_recordset() + pl.invalidate_recordset() + + actual_ref = self.variable_version.reference + expected_val = f"hello {{{{ {actual_ref} }}}} world" + self.assertEqual(val.value_char, expected_val) + expected_cond = f"if {{{{ {actual_ref} }}}} then" + self.assertEqual(pl.condition, expected_cond) + + def test_variable_reference_update(self): + """Test variable reference update cascades to dependent models""" + # 1. Add a variable value to variable_os + variable_value = self.VariableValue.create( + { + "variable_id": self.variable_os.id, + "value_char": "Ubuntu 20.04", + "server_id": self.server_test_1.id, + } + ) + + # Store original references for comparison + original_variable_reference = self.variable_os.reference + original_variable_value_reference = variable_value.reference + + # 2. Change the reference for variable_os to "awesome_variable" + self.variable_os.write({"reference": "awesome_variable"}) + + # 3. Verify that references are updated for dependent models + # Invalidate models to refresh all references + self.env["cx.tower.variable"].invalidate_model(["reference"]) + self.env["cx.tower.variable.value"].invalidate_model(["reference"]) + + # Check that variable reference was updated + self.assertEqual(self.variable_os.reference, "awesome_variable") + self.assertNotEqual(self.variable_os.reference, original_variable_reference) + + # Check that variable value reference was updated + # to include the new variable reference + self.assertIn("awesome_variable", variable_value.reference) + self.assertNotEqual(variable_value.reference, original_variable_value_reference) + + # Verify the reference pattern for variable value follows the expected format: + # ___ # noqa: E501 + expected_variable_pattern = ( + f"{self.variable_os.reference}_variable_value_server_" + f"{self.server_test_1.reference}" + ) + self.assertEqual(variable_value.reference, expected_variable_pattern)