243 lines
8.7 KiB
Python
243 lines
8.7 KiB
Python
# Copyright (C) 2024 Cetmix OÜ
|
|
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
|
|
from unittest.mock import patch
|
|
|
|
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
|
|
"""
|
|
|
|
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!")
|
|
|
|
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 -- Do not fetch global value now
|
|
value = self.CetmixTower.server_get_variable_value(
|
|
self.server_test_1.reference, variable_meme.reference, check_global=False
|
|
)
|
|
self.assertIsNone(value)
|
|
|
|
# -- 3 -- 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)
|
|
|
|
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.",
|
|
)
|
|
|
|
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)
|
|
|
|
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)
|
|
|
|
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")
|