starlingx/test
Code Issues Proposed changes

validate confirm message for critical commands

Browse Source 
Change-Id: I5c6ebe89da9686f82451999947e6d33ec00feb4d
Signed-off-by: ssivaman <sudarshanduraisamy.sivamani@windriver.com>
This commit is contained in:
ssivaman
2025-07-22 09:39:11 -04:00
parent 6da8408968
commit 3fc754c14c
4 changed files with 260 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
keywords/cloud_platform/system/service/objects/system_service_parameter_object.py Normal file
View File

@@ -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

82
keywords/cloud_platform/system/service/objects/system_service_parameter_output.py Normal file
View File

@@ -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

85
keywords/cloud_platform/system/service/system_service_keywords.py
View File

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

73
testcases/cloud_platform/regression/security/test_cli_confirmation.py Executable file
View File

@@ -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")