starlingx/test
Code Issues Proposed changes

system cluster keywords

Browse Source 
Change-Id: I59c9f2f3dda1674fff4ad89c8b109402fa9731b1
This commit is contained in:
Rity Menon
2025-01-13 09:26:15 -08:00
parent 7da7e728af
commit d48819f002
4 changed files with 284 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
keywords/cloud_platform/system/cluster/objects/system_cluster_object.py Normal file
View File

@@ -0,0 +1,99 @@
class SystemClusterObject:
"""
This class represents a cluster as an object.
This is typically a line in the system cluster output table.
"""
def __init__(self):
self.uuid:str = None
self.cluster_uuid:str = None
self.type:str = None
self.name:str = None
self.deployment_model:str = None
self.replication_groups:list[str] = []
self.storage_tiers:list[str] = []
def set_uuid(self, uuid: str):
"""
Setter for the uuid
"""
self.uuid = uuid
def get_uuid(self) -> str:
"""
Getter for this uuid
"""
return self.uuid
def set_cluster_uuid(self, cluster_uuid: str):
"""
Setter for the cluster_uuid
"""
self.cluster_uuid = cluster_uuid
def get_cluster_uuid(self) -> str:
"""
Getter for cluster_uuid
"""
return self.cluster_uuid
def set_type(self, c_type: str):
"""
Setter for the type
"""
self.type = c_type
def get_type(self) -> str:
"""
Getter for the type
"""
return self.type
def set_name(self, name: str):
"""
Setter for the name
"""
self.name = name
def get_name(self) -> str:
"""
Getter for the name
"""
return self.name
def set_deployment_model(self, deployment_model: str):
"""
Setter for the deployment_model
"""
self.deployment_model = deployment_model
def get_deployment_model(self) -> str:
"""
Getter for the deployment_model
"""
return self.deployment_model
def set_replication_groups(self, replication_groups: []):
"""
Setter for replication_groups
"""
self.replication_groups = replication_groups
def get_replication_groups(self) -> []:
"""
Getter for replication_groups
"""
return self.replication_groups
def set_storage_tiers(self, storage_tiers: []):
"""
Setter for storage_tiers
"""
self.storage_tiers = storage_tiers
def get_storage_tiers(self) -> []:
"""
Getter for storage_tiers
"""
return self.storage_tiers

67
keywords/cloud_platform/system/cluster/objects/system_cluster_output.py Normal file
View File

@@ -0,0 +1,67 @@
from framework.exceptions.keyword_exception import KeywordException
from framework.logging.automation_logger import get_logger
from keywords.cloud_platform.system.cluster.objects.system_cluster_object import SystemClusterObject
from keywords.cloud_platform.system.system_table_parser import SystemTableParser
class SystemClusterOutput:
"""
This class parses the output of 'system cluster-list' command into an object of type SystemClusterObject.
"""
def __init__(self, system_output):
"""
Constructor
Args:
system_output (str): Output of the 'system cluster-list' command.
Raises:
KeywordException: If the output is not valid.
"""
self.system_cluster : [SystemClusterObject] = []
system_table_parser = SystemTableParser(system_output)
output_values = system_table_parser.get_output_values_list()
for value in output_values:
if self.is_valid_output(value):
system_cluster = SystemClusterObject()
system_cluster.set_uuid(value['uuid'])
system_cluster.set_cluster_uuid(value['cluster_uuid'])
system_cluster.set_type(value['type'])
system_cluster.set_name(value['name'])
system_cluster.set_deployment_model(value['deployment_model'])
self.system_cluster.append(system_cluster)
else:
raise KeywordException(f"The output line {value} was not valid")
def get_system_cluster_list(self):
"""
Returns the parsed system cluster_list object.
Returns:
SystemClusterMonObject: The parsed system cluster object.
"""
return self.system_cluster
@staticmethod
def is_valid_output(value):
"""
Checks if the output contains all the expected fields.
Args:
value (dict): The dictionary of output values.
Returns:
bool: True if the output contains all required fields, False otherwise.
"""
required_fields = ["uuid", "cluster_uuid", "type", "name", "deployment_model"]
valid = True
for field in required_fields:
if field not in value:
get_logger().log_error(f'{field} is not in the output value')
valid = False
break
return valid

67
keywords/cloud_platform/system/cluster/objects/system_cluster_show_output.py Normal file
View File

@@ -0,0 +1,67 @@
from framework.exceptions.keyword_exception import KeywordException
from framework.logging.automation_logger import get_logger
from keywords.cloud_platform.system.cluster.objects.system_cluster_object import SystemClusterObject
from keywords.cloud_platform.system.system_vertical_table_parser import SystemVerticalTableParser
class SystemClusterShowOutput:
"""
This class parses the output of 'system cluster-show' command into an object of type SystemClusterObject.
"""
def __init__(self, system_output):
"""
Constructor
Args:
system_output (str): Output of the 'system cluster-show' command.
Raises:
KeywordException: If the output is not valid.
"""
system_vertical_table_parser = SystemVerticalTableParser(system_output)
output_values = system_vertical_table_parser.get_output_values_dict()
if self.is_valid_output(output_values):
self.system_cluster = SystemClusterObject()
self.system_cluster.set_uuid(output_values['uuid'])
self.system_cluster.set_cluster_uuid(output_values['cluster_uuid'])
self.system_cluster.set_type(output_values['type'])
self.system_cluster.set_name(output_values['name'])
self.system_cluster.set_replication_groups(output_values['replication_groups'])
self.system_cluster.set_storage_tiers(output_values['storage_tiers'])
self.system_cluster.set_deployment_model(output_values['deployment_model'])
else:
raise KeywordException(f"The output line {output_values} was not valid")
def get_system_cluster_show(self):
"""
Returns the parsed system cluster object.
Returns:
SystemClusterObject: The parsed system cluster object.
"""
return self.system_cluster
@staticmethod
def is_valid_output(value):
"""
Checks if the output contains all the expected fields.
Args:
value (dict): The dictionary of output values.
Returns:
bool: True if the output contains all required fields, False otherwise.
"""
required_fields = ["uuid", "cluster_uuid", "type", "name", "replication_groups", "storage_tiers",
"deployment_model"]
valid = True
for field in required_fields:
if field not in value:
get_logger().log_error(f'{field} is not in the output value')
valid = False
break
return valid

51
keywords/cloud_platform/system/cluster/system_cluster_keywords.py Normal file
View File

@@ -0,0 +1,51 @@
from keywords.base_keyword import BaseKeyword
from keywords.cloud_platform.command_wrappers import source_openrc
from keywords.cloud_platform.system.cluster.objects.system_cluster_output import SystemClusterOutput
from keywords.cloud_platform.system.cluster.objects.system_cluster_show_output import SystemClusterShowOutput
class SystemClusterKeywords(BaseKeyword):
"""
This class contains all the keywords related to the 'system cluster' commands.
"""
def __init__(self, ssh_connection):
"""
Constructor
Args:
ssh_connection:
"""
self.ssh_connection = ssh_connection
def get_system_cluster_list(self) -> SystemClusterOutput:
"""
Gets the system cluster-list
Args:
Returns:
SystemClusterOutput object with the list of cluster.
"""
command = source_openrc(f'system cluster-list')
output = self.ssh_connection.send(command)
self.validate_success_return_code(self.ssh_connection)
system_cluster_output = SystemClusterOutput(output)
return system_cluster_output
def get_system_cluster_show(self, uuid) -> SystemClusterShowOutput:
"""
Gets the system cluster-show
Args:
uuid: uuid of the cluster
Returns:
SystemClusterShowOutput object.
"""
command = source_openrc(f'system cluster-show {uuid}')
output = self.ssh_connection.send(command)
self.validate_success_return_code(self.ssh_connection)
system_cluster_show_output = SystemClusterShowOutput(output)
return system_cluster_show_output