diff --git a/keywords/cloud_platform/system/service/objects/system_service_parameter_object.py b/keywords/cloud_platform/system/service/objects/system_service_parameter_object.py new file mode 100644 index 00000000..3f6376b1 --- /dev/null +++ b/keywords/cloud_platform/system/service/objects/system_service_parameter_object.py @@ -0,0 +1,45 @@ +class SystemServiceParameterObject: + """ + Class to represent a single service parameter from the service parameter list output. + """ + + def __init__(self): + """ + Constructor. + """ + self.service = None + self.section = None + self.name = None + self.value = None + + def set_service(self, service: str): + """Setter for service""" + self.service = service + + def get_service(self) -> str: + """Getter for service""" + return self.service + + def set_section(self, section: str): + """Setter for section""" + self.section = section + + def get_section(self) -> str: + """Getter for section""" + return self.section + + def set_name(self, name: str): + """Setter for name""" + self.name = name + + def get_name(self) -> str: + """Getter for name""" + return self.name + + def set_value(self, value: str): + """Setter for value""" + self.value = value + + def get_value(self) -> str: + """Getter for value""" + return self.value diff --git a/keywords/cloud_platform/system/service/objects/system_service_parameter_output.py b/keywords/cloud_platform/system/service/objects/system_service_parameter_output.py new file mode 100644 index 00000000..a7af69b8 --- /dev/null +++ b/keywords/cloud_platform/system/service/objects/system_service_parameter_output.py @@ -0,0 +1,82 @@ +from framework.exceptions.keyword_exception import KeywordException +from framework.logging.automation_logger import get_logger +from keywords.cloud_platform.system.service.objects.system_service_parameter_object import SystemServiceParameterObject +from keywords.cloud_platform.system.system_table_parser import SystemTableParser + + +class SystemServiceParameterOutput: + """ + Class to parse and handle the output of 'system service-parameter-list' command. + """ + + def __init__(self, output: str): + """ + Constructor. + + Args: + output (str): Raw output from 'system service-parameter-list' command + + Raises: + KeywordException: If the output is not valid + """ + self.output = output.split("\n") if isinstance(output, str) else output + self.table_parser = SystemTableParser(self.output) + output_values_list = self.table_parser.get_output_values_list() + + if self.is_valid_output(self.output): + # Convert Property/Value pairs to a single parameter object + param = SystemServiceParameterObject() + for item in output_values_list: + property_name = item.get("Property", "") + property_value = item.get("Value", "") + + if property_name == "service": + param.set_service(property_value) + elif property_name == "section": + param.set_section(property_value) + elif property_name == "name": + param.set_name(property_value) + elif property_name == "value": + param.set_value(property_value) + + self.parameters = [param] + else: + raise KeywordException(f"The output {self.output} was not valid") + + def get_parameters(self) -> list: + """ + Get list of service parameter objects. + + Returns: + list: List of SystemServiceParameterObject instances + """ + return self.parameters + + def get_raw_output(self) -> str: + """ + Get the raw output from the command. + + Returns: + str: Raw command output + """ + return self.output + + @staticmethod + def is_valid_output(output: list) -> bool: + """ + Checks if the output contains valid service parameter data. + + Args: + output (list): The command output to validate + + Returns: + bool: True if the output is valid, False otherwise + """ + output_str = "\n".join(output) if isinstance(output, list) else output + + # Check for Property/Value table structure + if "Property" not in output_str or "Value" not in output_str: + get_logger().log_error("Required Property/Value table structure not found in output") + return False + + return True diff --git a/keywords/cloud_platform/system/service/system_service_keywords.py b/keywords/cloud_platform/system/service/system_service_keywords.py index 4fdaf601..5fd730fb 100644 --- a/keywords/cloud_platform/system/service/system_service_keywords.py +++ b/keywords/cloud_platform/system/service/system_service_keywords.py @@ -1,9 +1,10 @@ +from framework.ssh.ssh_connection import SSHConnection from keywords.base_keyword import BaseKeyword from keywords.cloud_platform.command_wrappers import source_openrc from keywords.cloud_platform.system.service.objects.system_service_output import SystemServiceOutput +from keywords.cloud_platform.system.service.objects.system_service_parameter_output import SystemServiceParameterOutput from keywords.cloud_platform.system.service.objects.system_service_show_output import SystemServiceShowOutput from keywords.k8s.pods.kubectl_get_pods_keywords import KubectlGetPodsKeywords -from time import sleep class SystemServiceKeywords(BaseKeyword): @@ -11,66 +12,100 @@ class SystemServiceKeywords(BaseKeyword): This class contains all the keywords related to the 'system service' commands. """ - def __init__(self, ssh_connection): + def __init__(self, ssh_connection: SSHConnection): """ - Constructor + Constructor. + Args: - ssh_connection: + ssh_connection (SSHConnection): SSH connection instance """ self.ssh_connection = ssh_connection def get_system_service_list(self) -> SystemServiceOutput: """ - Gets the system service-list - - Args: + Gets the system service-list. Returns: - SystemServiceOutput object with the list of service. - + SystemServiceOutput: Object with the list of service. """ - command = source_openrc(f'system service-list') + command = source_openrc("system service-list") output = self.ssh_connection.send(command) self.validate_success_return_code(self.ssh_connection) system_service_output = SystemServiceOutput(output) return system_service_output - def get_system_service_show(self, service_id) -> SystemServiceShowOutput: + def get_system_service_show(self, service_id: str) -> SystemServiceShowOutput: """ - Gets the system service-show + Gets the system service-show. Args: - service_id: service_id + service_id (str): Service ID Returns: - SystemServiceShowOutput object. - + SystemServiceShowOutput: Service show output object. """ - command = source_openrc(f'system service-show {service_id}') + command = source_openrc(f"system service-show {service_id}") output = self.ssh_connection.send(command) self.validate_success_return_code(self.ssh_connection) system_service_show_output = SystemServiceShowOutput(output) return system_service_show_output - - def add_service_parameter(self, service: str, parameter: str, value: str): + def add_service_parameter(self, service: str, parameter: str, value: str) -> SystemServiceParameterOutput: """ Adds a service parameter. Args: - service (str): The service name. - parameter (str): The parameter to add. - value (str): The value of the parameter. + service (str): The service name + parameter (str): The parameter to add + value (str): The value of the parameter + + Returns: + SystemServiceParameterOutput: Output object """ - command = source_openrc(f'system service-parameter-add {service} {parameter}={value}') + command = source_openrc(f"system service-parameter-add {service} {parameter}={value}") + output = self.ssh_connection.send(command) + self.validate_success_return_code(self.ssh_connection) + system_service_parameter_add_output = SystemServiceParameterOutput(output) + return system_service_parameter_add_output + + def modify_service_parameter(self, service: str, section: str, parameter: str, value: str) -> SystemServiceParameterOutput: + """ + Modifies a service parameter. + + Args: + service (str): The service name (e.g., 'platform') + section (str): The section name (e.g., 'client') + parameter (str): The parameter to modify (e.g., 'cli_confirmations') + value (str): The value of the parameter (e.g., 'enabled') + + Returns: + SystemServiceParameterOutput: Output object + """ + command = source_openrc(f'system service-parameter-modify {service} {section} {parameter}="{value}"') + output = self.ssh_connection.send(command) + self.validate_success_return_code(self.ssh_connection) + system_service_parameter_modify_output = SystemServiceParameterOutput(output) + return system_service_parameter_modify_output + + def apply_service_parameters(self, service: str) -> None: + """ + Applies service parameters. + + Args: + service (str): The service name (e.g., 'platform', 'kubernetes') + """ + command = source_openrc(f"system service-parameter-apply {service}") self.ssh_connection.send(command) self.validate_success_return_code(self.ssh_connection) - def apply_kubernetes_service_parameters(self): + def apply_kubernetes_service_parameters(self) -> None: """ - Applies kubernetes service parameters and waits for it to restart. + Applies kubernetes service parameters and waits for Kubernetes to restart. + + This method includes validation to ensure Kubernetes stability after parameter changes. """ - command = source_openrc(f'system service-parameter-apply kubernetes') + command = source_openrc("system service-parameter-apply kubernetes") self.ssh_connection.send(command) self.validate_success_return_code(self.ssh_connection) + # Wait for Kubernetes to restart and stabilize after parameter changes KubectlGetPodsKeywords(self.ssh_connection).wait_for_kubernetes_to_restart() diff --git a/testcases/cloud_platform/regression/security/test_cli_confirmation.py b/testcases/cloud_platform/regression/security/test_cli_confirmation.py new file mode 100755 index 00000000..586f2d69 --- /dev/null +++ b/testcases/cloud_platform/regression/security/test_cli_confirmation.py @@ -0,0 +1,73 @@ +from pytest import FixtureRequest, mark + +from framework.logging.automation_logger import get_logger +from framework.ssh.prompt_response import PromptResponse +from framework.validation.validation import validate_str_contains +from keywords.cloud_platform.command_wrappers import source_openrc +from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords +from keywords.cloud_platform.system.service.system_service_keywords import SystemServiceKeywords + + +def cli_confirmations_enabled(request: FixtureRequest): + """ + Function to enable CLI confirmations in setup and disable in teardown. + """ + ssh_connection = LabConnectionKeywords().get_active_controller_ssh() + service_keywords = SystemServiceKeywords(ssh_connection) + + get_logger().log_info("Enabling CLI confirmations in setup") + service_keywords.modify_service_parameter("platform", "client", "cli_confirmations", "enabled") + service_keywords.apply_service_parameters("platform") + + get_logger().log_info("Waiting for CLI confirmation settings to take effect...") + # Exit from lab session to force fresh login + get_logger().log_info("Exiting from lab session to force fresh login") + ssh_connection.send("exit") + ssh_connection.close() + + # Completely destroy the connection object + ssh_connection = None + get_logger().log_info("Setup complete - CLI confirmations enabled") + + def teardown(): + teardown_ssh_connection = LabConnectionKeywords().get_active_controller_ssh() + teardown_service_keywords = SystemServiceKeywords(teardown_ssh_connection) + get_logger().log_info("Disabling CLI confirmations in teardown") + teardown_service_keywords.modify_service_parameter("platform", "client", "cli_confirmations", "disabled") + teardown_service_keywords.apply_service_parameters("platform") + + request.addfinalizer(teardown) + + +@mark.p1 +def test_cli_with_confirmations_enabled(request: FixtureRequest): + """ + Test system host-lock command with CLI confirmations enabled. + Verifies confirmation prompts appear for destructive commands. + """ + # Call the setup function directly + cli_confirmations_enabled(request) + + ssh_connection = LabConnectionKeywords().get_active_controller_ssh() + + get_logger().log_info("Testing system host-lock command with confirmations enabled") + cmd = "system host-lock controller-1" + command = source_openrc(cmd) + + # Set up prompts to expect confirmation and respond with 'No' + confirmation_prompt = PromptResponse("Do you want to continue?", "no") + command_completed = PromptResponse("keystone_admin", None) # Wait for command prompt to return + prompts = [confirmation_prompt, command_completed] + + output = ssh_connection.send_expect_prompts(command, prompts) + + # Verify confirmation prompt appeared and was handled correctly + output_str = str(output) + get_logger().log_info(f"Full output: {output_str}") + + # Validate that CLI confirmation behavior occurred + validate_str_contains(output_str, "Operation cancelled by the user", "Verify CLI confirmation prompt appeared and operation was cancelled") + + get_logger().log_info("CLI confirmation prompt appeared and was handled correctly - operation cancelled") + + get_logger().log_info("system host-lock confirmation test completed")