system load keywords
Change-Id: Ib5dc76d801b77043bb0d9078ebef1992cbafb51b
This commit is contained in:
@@ -0,0 +1,73 @@
|
|||||||
|
class SystemLoadObject:
|
||||||
|
"""
|
||||||
|
This class represents a service as an object.
|
||||||
|
This is typically a line in the system load output table.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self):
|
||||||
|
self.id:int = -1
|
||||||
|
self.state:str = None
|
||||||
|
self.software_version:str = None
|
||||||
|
self.compatible_version:str = None
|
||||||
|
self.required_patches: str = None
|
||||||
|
|
||||||
|
|
||||||
|
def set_id(self, load_id: int):
|
||||||
|
"""
|
||||||
|
Setter for the id
|
||||||
|
"""
|
||||||
|
self.id = load_id
|
||||||
|
|
||||||
|
def get_id(self) -> int:
|
||||||
|
"""
|
||||||
|
Getter for the id
|
||||||
|
"""
|
||||||
|
return self.id
|
||||||
|
|
||||||
|
def set_state(self, state: str):
|
||||||
|
"""
|
||||||
|
Setter for the state
|
||||||
|
"""
|
||||||
|
self.state = state
|
||||||
|
|
||||||
|
def get_state(self) -> str:
|
||||||
|
"""
|
||||||
|
Getter for state
|
||||||
|
"""
|
||||||
|
return self.state
|
||||||
|
|
||||||
|
def set_software_version(self, software_version: str):
|
||||||
|
"""
|
||||||
|
Setter for the software_version
|
||||||
|
"""
|
||||||
|
self.software_version = software_version
|
||||||
|
|
||||||
|
def get_software_version(self) -> str:
|
||||||
|
"""
|
||||||
|
Getter for the software_version
|
||||||
|
"""
|
||||||
|
return self.software_version
|
||||||
|
|
||||||
|
def set_compatible_version(self, compatible_version: str):
|
||||||
|
"""
|
||||||
|
Setter for the compatible_version
|
||||||
|
"""
|
||||||
|
self.compatible_version = compatible_version
|
||||||
|
|
||||||
|
def get_compatible_version(self) -> str:
|
||||||
|
"""
|
||||||
|
Getter for the compatible_version
|
||||||
|
"""
|
||||||
|
return self.compatible_version
|
||||||
|
|
||||||
|
def set_required_patches(self, required_patches: str):
|
||||||
|
"""
|
||||||
|
Setter for required_patches
|
||||||
|
"""
|
||||||
|
self.required_patches = required_patches
|
||||||
|
|
||||||
|
def get_required_patches(self) -> str:
|
||||||
|
"""
|
||||||
|
Getter for required_patches
|
||||||
|
"""
|
||||||
|
return self.required_patches
|
@@ -0,0 +1,64 @@
|
|||||||
|
from framework.exceptions.keyword_exception import KeywordException
|
||||||
|
from framework.logging.automation_logger import get_logger
|
||||||
|
from keywords.cloud_platform.system.load.objects.system_load_object import SystemLoadObject
|
||||||
|
from keywords.cloud_platform.system.system_vertical_table_parser import SystemVerticalTableParser
|
||||||
|
|
||||||
|
class SystemLoadShowOutput:
|
||||||
|
"""
|
||||||
|
This class parses the output of 'system load-show' command into an object of type SystemLoadShowObject.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, system_output):
|
||||||
|
"""
|
||||||
|
Constructor
|
||||||
|
|
||||||
|
Args:
|
||||||
|
system_output (str): Output of the 'system load-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_load = SystemLoadObject()
|
||||||
|
self.system_load.set_id(output_values['id'])
|
||||||
|
self.system_load.set_state(output_values['state'])
|
||||||
|
self.system_load.set_software_version(output_values['software_version'])
|
||||||
|
self.system_load.set_compatible_version(output_values['compatible_version'])
|
||||||
|
self.system_load.set_required_patches(output_values['required_patches'])
|
||||||
|
else:
|
||||||
|
raise KeywordException(f"The output line {output_values} was not valid")
|
||||||
|
|
||||||
|
def get_system_load_show(self) -> SystemLoadObject:
|
||||||
|
"""
|
||||||
|
Returns the parsed system load-show object.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
SystemLoadObject: The parsed system load-show object.
|
||||||
|
"""
|
||||||
|
|
||||||
|
return self.system_load
|
||||||
|
|
||||||
|
@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", "state", "software_version", "compatible_version", "required_patches"]
|
||||||
|
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
|
34
keywords/cloud_platform/system/load/system_load_keywords.py
Normal file
34
keywords/cloud_platform/system/load/system_load_keywords.py
Normal file
@@ -0,0 +1,34 @@
|
|||||||
|
from keywords.base_keyword import BaseKeyword
|
||||||
|
from keywords.cloud_platform.command_wrappers import source_openrc
|
||||||
|
from keywords.cloud_platform.system.load.objects.system_load_show_output import SystemLoadShowOutput
|
||||||
|
|
||||||
|
|
||||||
|
class SystemLoadKeywords(BaseKeyword):
|
||||||
|
"""
|
||||||
|
This class contains all the keywords related to the 'system load' commands.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, ssh_connection):
|
||||||
|
"""
|
||||||
|
Constructor
|
||||||
|
Args:
|
||||||
|
ssh_connection:
|
||||||
|
"""
|
||||||
|
self.ssh_connection = ssh_connection
|
||||||
|
|
||||||
|
def get_system_load_show(self, load_id:int) -> SystemLoadShowOutput:
|
||||||
|
"""
|
||||||
|
Gets the system load-show
|
||||||
|
|
||||||
|
Args:
|
||||||
|
load_id (int): load id.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
SystemLoadShowOutput object.
|
||||||
|
|
||||||
|
"""
|
||||||
|
command = source_openrc(f'system load-show {load_id}')
|
||||||
|
output = self.ssh_connection.send(command)
|
||||||
|
self.validate_success_return_code(self.ssh_connection)
|
||||||
|
system_load_show_output = SystemLoadShowOutput(output)
|
||||||
|
return system_load_show_output
|
Reference in New Issue
Block a user