Merge "system servicenode keywords"
This commit is contained in:
Zuul
@@ -0,0 +1,124 @@
|
||||
class SystemServicenodeObject:
|
||||
"""
|
||||
This class represents a servicenode as an object.
|
||||
This is typically a line in the system servicenode output table.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.id:str = None
|
||||
self.name:str = None
|
||||
self.administrative:str = None
|
||||
self.operational:str = None
|
||||
self.availability:str = None
|
||||
self.ready_state: str = None
|
||||
self.administrative_state: str = None
|
||||
self.availability_status: str = None
|
||||
self.operational_state: str = None
|
||||
|
||||
def set_id(self, host_id: str):
|
||||
"""
|
||||
Setter for the host_id
|
||||
"""
|
||||
self.id = host_id
|
||||
|
||||
def get_id(self) -> str:
|
||||
"""
|
||||
Getter for this id
|
||||
"""
|
||||
return self.id
|
||||
|
||||
def set_name(self, name: str):
|
||||
"""
|
||||
Setter for the name
|
||||
"""
|
||||
self.name = name
|
||||
|
||||
def get_name(self) -> str:
|
||||
"""
|
||||
Getter for name
|
||||
"""
|
||||
return self.name
|
||||
|
||||
def set_administrative(self, administrative: str):
|
||||
"""
|
||||
Setter for the administrative
|
||||
"""
|
||||
self.administrative = administrative
|
||||
|
||||
def get_administrative(self) -> str:
|
||||
"""
|
||||
Getter for the administrative
|
||||
"""
|
||||
return self.administrative
|
||||
|
||||
def set_operational(self, operational: str):
|
||||
"""
|
||||
Setter for the operational
|
||||
"""
|
||||
self.operational = operational
|
||||
|
||||
def get_operational(self) -> str:
|
||||
"""
|
||||
Getter for the operational
|
||||
"""
|
||||
return self.operational
|
||||
|
||||
def set_availability(self, availability: str):
|
||||
"""
|
||||
Setter for the availability
|
||||
"""
|
||||
self.availability = availability
|
||||
|
||||
def get_availability(self) -> str:
|
||||
"""
|
||||
Getter for the availability
|
||||
"""
|
||||
return self.availability
|
||||
|
||||
def set_ready_state(self, ready_state: str):
|
||||
"""
|
||||
Setter for the ready_state
|
||||
"""
|
||||
self.ready_state = ready_state
|
||||
|
||||
def get_ready_state(self) -> str:
|
||||
"""
|
||||
Getter for the ready_state
|
||||
"""
|
||||
return self.ready_state
|
||||
|
||||
def set_administrative_state(self, administrative_state: str):
|
||||
"""
|
||||
Setter for the administrative_state
|
||||
"""
|
||||
self.administrative_state = administrative_state
|
||||
|
||||
def get_administrative_state(self) -> str:
|
||||
"""
|
||||
Getter for the administrative_state
|
||||
"""
|
||||
return self.administrative_state
|
||||
|
||||
def set_availability_status(self, availability_status):
|
||||
"""
|
||||
Setter for availability_status
|
||||
"""
|
||||
self.availability_status = availability_status
|
||||
|
||||
def get_availability_status(self) -> str:
|
||||
"""
|
||||
Getter for availability_status
|
||||
"""
|
||||
return self.availability_status
|
||||
|
||||
def set_operational_state(self, operational_state: str):
|
||||
"""
|
||||
Setter for the operational_state
|
||||
"""
|
||||
self.operational_state = operational_state
|
||||
|
||||
def get_operational_state(self) -> str:
|
||||
"""
|
||||
Getter for the operational_state
|
||||
"""
|
||||
return self.operational_state
|
||||
@@ -0,0 +1,68 @@
|
||||
from framework.exceptions.keyword_exception import KeywordException
|
||||
from framework.logging.automation_logger import get_logger
|
||||
from keywords.cloud_platform.system.servicenode.objects.system_servicenode_object import SystemServicenodeObject
|
||||
from keywords.cloud_platform.system.system_table_parser import SystemTableParser
|
||||
|
||||
class SystemServicenodeOutput:
|
||||
"""
|
||||
This class parses the output of 'system servicenode-list' command into an object of type SystemServicenodeObject.
|
||||
"""
|
||||
|
||||
def __init__(self, system_output):
|
||||
"""
|
||||
Constructor
|
||||
|
||||
Args:
|
||||
system_output (str): Output of the 'system servicenode-list' command.
|
||||
|
||||
Raises:
|
||||
KeywordException: If the output is not valid.
|
||||
"""
|
||||
|
||||
self.system_servicenodes : [SystemServicenodeObject] = []
|
||||
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_servicenode = SystemServicenodeObject()
|
||||
system_servicenode.set_id(value['id'])
|
||||
system_servicenode.set_name(value['name'])
|
||||
system_servicenode.set_administrative(value['administrative'])
|
||||
system_servicenode.set_operational(value['operational'])
|
||||
system_servicenode.set_availability(value['availability'])
|
||||
system_servicenode.set_ready_state(value['ready_state'])
|
||||
self.system_servicenodes.append(system_servicenode)
|
||||
else:
|
||||
raise KeywordException(f"The output line {value} was not valid")
|
||||
|
||||
def get_system_servicenode_list(self) -> list[SystemServicenodeObject]:
|
||||
"""
|
||||
Returns the parsed system servicenode object.
|
||||
|
||||
Returns:
|
||||
list[SystemServicenodeObject]: list of parsed system servicenode objects.
|
||||
"""
|
||||
|
||||
return self.system_servicenodes
|
||||
|
||||
@staticmethod
|
||||
def is_valid_output(value) -> bool:
|
||||
"""
|
||||
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 = ["id", "name", "administrative", "operational", "availability", "ready_state"]
|
||||
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
|
||||
@@ -0,0 +1,65 @@
|
||||
from framework.exceptions.keyword_exception import KeywordException
|
||||
from framework.logging.automation_logger import get_logger
|
||||
from keywords.cloud_platform.system.servicenode.objects.system_servicenode_object import SystemServicenodeObject
|
||||
from keywords.cloud_platform.system.system_vertical_table_parser import SystemVerticalTableParser
|
||||
|
||||
class SystemServicenodeShowOutput:
|
||||
"""
|
||||
This class parses the output of 'system servicenode-show' command into an object of type SystemServicenodeObject.
|
||||
"""
|
||||
|
||||
def __init__(self, system_output):
|
||||
"""
|
||||
Constructor
|
||||
|
||||
Args:
|
||||
system_output (str): Output of the 'system servicenode-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_servicenode = SystemServicenodeObject()
|
||||
self.system_servicenode.set_administrative_state(output_values['administrative_state'])
|
||||
self.system_servicenode.set_availability_status(output_values['availability_status'])
|
||||
self.system_servicenode.set_id(output_values['id'])
|
||||
self.system_servicenode.set_name(output_values['name'])
|
||||
self.system_servicenode.set_operational_state(output_values['operational_state'])
|
||||
self.system_servicenode.set_ready_state(output_values['ready_state'])
|
||||
else:
|
||||
raise KeywordException(f"The output line {output_values} was not valid")
|
||||
|
||||
def get_system_servicenode_show(self) -> SystemServicenodeObject:
|
||||
"""
|
||||
Returns the parsed system servicenode-show object.
|
||||
|
||||
Returns:
|
||||
SystemServicenodeObject: The parsed system servicenode-show object.
|
||||
"""
|
||||
|
||||
return self.system_servicenode
|
||||
|
||||
@staticmethod
|
||||
def is_valid_output(value) -> bool:
|
||||
"""
|
||||
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 = ["administrative_state", "availability_status", "id", "name", "operational_state", "ready_state"]
|
||||
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
|
||||
@@ -0,0 +1,51 @@
|
||||
from keywords.base_keyword import BaseKeyword
|
||||
from keywords.cloud_platform.command_wrappers import source_openrc
|
||||
from keywords.cloud_platform.system.servicenode.objects.system_servicenode_output import SystemServicenodeOutput
|
||||
from keywords.cloud_platform.system.servicenode.objects.system_servicenode_show_output import SystemServicenodeShowOutput
|
||||
|
||||
|
||||
class SystemServicenodeKeywords(BaseKeyword):
|
||||
"""
|
||||
This class contains all the keywords related to the 'system servicenode' commands.
|
||||
"""
|
||||
|
||||
def __init__(self, ssh_connection):
|
||||
"""
|
||||
Constructor
|
||||
Args:
|
||||
ssh_connection:
|
||||
"""
|
||||
self.ssh_connection = ssh_connection
|
||||
|
||||
def get_system_servicenode_list(self) -> SystemServicenodeOutput:
|
||||
"""
|
||||
Gets the system servicenode-list
|
||||
|
||||
Args:
|
||||
|
||||
Returns:
|
||||
SystemServicenodeOutput object with the list of servicenode.
|
||||
|
||||
"""
|
||||
command = source_openrc(f'system servicenode-list')
|
||||
output = self.ssh_connection.send(command)
|
||||
self.validate_success_return_code(self.ssh_connection)
|
||||
system_servicenode_output = SystemServicenodeOutput(output)
|
||||
return system_servicenode_output
|
||||
|
||||
def get_system_servicenode_show(self, host_id) -> SystemServicenodeShowOutput:
|
||||
"""
|
||||
Gets the system servicenode-show
|
||||
|
||||
Args:
|
||||
host_id: host id
|
||||
|
||||
Returns:
|
||||
SystemServicenodeShowOutput object.
|
||||
|
||||
"""
|
||||
command = source_openrc(f'system servicenode-show {host_id}')
|
||||
output = self.ssh_connection.send(command)
|
||||
self.validate_success_return_code(self.ssh_connection)
|
||||
system_servicenode_show_output = SystemServicenodeShowOutput(output)
|
||||
return system_servicenode_show_output
|
||||
Reference in New Issue
Block a user