diff --git a/keywords/cloud_platform/dcmanager/dcmanager_subcloud_peer_group_add_keywords.py b/keywords/cloud_platform/dcmanager/dcmanager_subcloud_peer_group_add_keywords.py new file mode 100644 index 00000000..afa93d3b --- /dev/null +++ b/keywords/cloud_platform/dcmanager/dcmanager_subcloud_peer_group_add_keywords.py @@ -0,0 +1,49 @@ +from typing import Optional + +from framework.ssh.ssh_connection import SSHConnection +from keywords.base_keyword import BaseKeyword +from keywords.cloud_platform.command_wrappers import source_openrc + + +class DcManagerSubcloudPeerGroupAddKeywords(BaseKeyword): + """Keywords for 'dcmanager subcloud-peer-group add' commands.""" + + def __init__(self, ssh_connection: SSHConnection): + """Initialize the keywords class. + + Args: + ssh_connection (SSHConnection): SSH connection to the active controller. + """ + self.ssh_connection = ssh_connection + + def dcmanager_subcloud_peer_group_add( + self, + peer_group_name: str, + group_priority: Optional[int] = None, + group_description: Optional[str] = None, + max_subclouds: Optional[int] = None, + ) -> None: + """Add a subcloud peer group using 'dcmanager subcloud-peer-group add' command. + + Args: + peer_group_name (str): Name of the peer group. + group_priority (Optional[int]): Priority level of the peer group. + group_description (Optional[str]): Description of the peer group. + max_subclouds (Optional[int]): Maximum number of subclouds per group. + """ + cmd_parts = [ + "dcmanager subcloud-peer-group add", + f"--peer-group-name {peer_group_name}", + ] + if group_description: + cmd_parts.append(f"--group-priority '{group_priority}'") + + if group_description: + cmd_parts.append(f"--group-description '{group_description}'") + + if max_subclouds is not None: + cmd_parts.append(f"--max-subclouds {max_subclouds}") + + cmd = " ".join(cmd_parts) + self.ssh_connection.send(source_openrc(cmd)) + self.validate_success_return_code(self.ssh_connection) diff --git a/keywords/cloud_platform/dcmanager/dcmanager_subcloud_peer_group_delete_keywords.py b/keywords/cloud_platform/dcmanager/dcmanager_subcloud_peer_group_delete_keywords.py new file mode 100644 index 00000000..3fc8c3c3 --- /dev/null +++ b/keywords/cloud_platform/dcmanager/dcmanager_subcloud_peer_group_delete_keywords.py @@ -0,0 +1,25 @@ +from framework.ssh.ssh_connection import SSHConnection +from keywords.base_keyword import BaseKeyword +from keywords.cloud_platform.command_wrappers import source_openrc + + +class DcManagerSubcloudPeerGroupDeleteKeywords(BaseKeyword): + """Keywords for 'dcmanager subcloud-peer-group delete' commands.""" + + def __init__(self, ssh_connection: SSHConnection): + """Initialize the keywords class. + + Args: + ssh_connection (SSHConnection): SSH connection to the active controller. + """ + self.ssh_connection = ssh_connection + + def dcmanager_subcloud_peer_group_delete(self, identifier: str) -> None: + """Delete a subcloud peer group using 'dcmanager subcloud-peer-group delete' command. + + Args: + identifier (str): The identifier to delete id / peer_group_name. + """ + cmd = f"dcmanager subcloud-peer-group delete {identifier}" + self.ssh_connection.send(source_openrc(cmd)) + self.validate_success_return_code(self.ssh_connection) diff --git a/keywords/cloud_platform/dcmanager/dcmanager_subcloud_peer_group_list_keywords.py b/keywords/cloud_platform/dcmanager/dcmanager_subcloud_peer_group_list_keywords.py new file mode 100644 index 00000000..2db87fa1 --- /dev/null +++ b/keywords/cloud_platform/dcmanager/dcmanager_subcloud_peer_group_list_keywords.py @@ -0,0 +1,27 @@ +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.dcmanager.objects.dcmanager_subcloud_peer_group_list_output import DcManagerSubcloudPeerGroupListOutput + + +class DcManagerSubcloudPeerGroupListKeywords(BaseKeyword): + """Keywords for 'dcmanager subcloud-peer-group list' commands.""" + + def __init__(self, ssh_connection: SSHConnection): + """Initialize the keywords class. + + Args: + ssh_connection (SSHConnection): SSH connection to the active controller. + """ + self.ssh_connection = ssh_connection + + def get_dcmanager_subcloud_peer_group_list(self) -> DcManagerSubcloudPeerGroupListOutput: + """Get subcloud peer group list using 'dcmanager subcloud-peer-group list' command. + + Returns: + DcManagerSubcloudPeerGroupListOutput: Parsed peer group list output. + """ + cmd = "dcmanager subcloud-peer-group list" + output = self.ssh_connection.send(source_openrc(cmd)) + self.validate_success_return_code(self.ssh_connection) + return DcManagerSubcloudPeerGroupListOutput(output) diff --git a/keywords/cloud_platform/dcmanager/dcmanager_subcloud_peer_group_list_subclouds_keywords.py b/keywords/cloud_platform/dcmanager/dcmanager_subcloud_peer_group_list_subclouds_keywords.py new file mode 100644 index 00000000..e073415f --- /dev/null +++ b/keywords/cloud_platform/dcmanager/dcmanager_subcloud_peer_group_list_subclouds_keywords.py @@ -0,0 +1,30 @@ +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.dcmanager.objects.dcmanager_subcloud_peer_group_list_subclouds_output import DcManagerSubcloudPeerGroupListSubcloudsOutput + + +class DcManagerSubcloudPeerGroupListSubcloudsKeywords(BaseKeyword): + """Keywords for 'dcmanager subcloud-peer-group list-subclouds' commands.""" + + def __init__(self, ssh_connection: SSHConnection): + """Initialize keywords class. + + Args: + ssh_connection (SSHConnection): SSH connection to the active controller. + """ + self.ssh_connection = ssh_connection + + def get_dcmanager_subcloud_peer_group_list_subclouds(self, identifier: str) -> DcManagerSubcloudPeerGroupListSubcloudsOutput: + """Get subclouds in peer group. + + Args: + identifier (str): Peer group identifier (id or name). + + Returns: + DcManagerSubcloudPeerGroupListSubcloudsOutput: Parsed subclouds list output. + """ + cmd = f"dcmanager subcloud-peer-group list-subclouds {identifier}" + output = self.ssh_connection.send(source_openrc(cmd)) + self.validate_success_return_code(self.ssh_connection) + return DcManagerSubcloudPeerGroupListSubcloudsOutput(output) diff --git a/keywords/cloud_platform/dcmanager/dcmanager_subcloud_peer_group_show_keywords.py b/keywords/cloud_platform/dcmanager/dcmanager_subcloud_peer_group_show_keywords.py new file mode 100644 index 00000000..47068453 --- /dev/null +++ b/keywords/cloud_platform/dcmanager/dcmanager_subcloud_peer_group_show_keywords.py @@ -0,0 +1,30 @@ +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.dcmanager.objects.dcmanager_subcloud_peer_group_show_output import DcManagerSubcloudPeerGroupShowOutput + + +class DcManagerSubcloudPeerGroupShowKeywords(BaseKeyword): + """Keywords for 'dcmanager subcloud-peer-group show' commands.""" + + def __init__(self, ssh_connection: SSHConnection): + """Initialize the keywords class. + + Args: + ssh_connection (SSHConnection): SSH connection to the active controller. + """ + self.ssh_connection = ssh_connection + + def get_dcmanager_subcloud_peer_group_show(self, identifier: str) -> DcManagerSubcloudPeerGroupShowOutput: + """Gets the 'cmanager subcloud-peer-group show' output. + + Args: + identifier (str): The identifier to show details for + + Returns: + DcManagerSubcloudPeerGroupShowOutput: Parsed peer group show output. + + """ + output = self.ssh_connection.send(source_openrc(f"dcmanager subcloud-peer-group show {identifier}")) + self.validate_success_return_code(self.ssh_connection) + return DcManagerSubcloudPeerGroupShowOutput(output) diff --git a/keywords/cloud_platform/dcmanager/dcmanager_subcloud_peer_group_status_keywords.py b/keywords/cloud_platform/dcmanager/dcmanager_subcloud_peer_group_status_keywords.py new file mode 100644 index 00000000..7465036a --- /dev/null +++ b/keywords/cloud_platform/dcmanager/dcmanager_subcloud_peer_group_status_keywords.py @@ -0,0 +1,30 @@ +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.dcmanager.objects.dcmanager_subcloud_peer_group_status_output import DcManagerSubcloudPeerGroupStatusOutput + + +class DcManagerSubcloudPeerGroupStatusKeywords(BaseKeyword): + """Keywords for 'dcmanager subcloud-peer-group status' commands.""" + + def __init__(self, ssh_connection: SSHConnection): + """Initialize the keywords class. + + Args: + ssh_connection (SSHConnection): SSH connection to the active controller. + """ + self.ssh_connection = ssh_connection + + def get_dcmanager_subcloud_peer_group_status(self, identifier: str) -> DcManagerSubcloudPeerGroupStatusOutput: + """Get subcloud peer group status using 'dcmanager subcloud-peer-group status' command. + + Args: + identifier (str): The peer group ID or name to get status for. + + Returns: + DcManagerSubcloudPeerGroupStatusOutput: Parsed peer group status output. + """ + cmd = f"dcmanager subcloud-peer-group status {identifier}" + output = self.ssh_connection.send(source_openrc(cmd)) + self.validate_success_return_code(self.ssh_connection) + return DcManagerSubcloudPeerGroupStatusOutput(output) diff --git a/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_list_object.py b/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_list_object.py new file mode 100644 index 00000000..1fa558d6 --- /dev/null +++ b/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_list_object.py @@ -0,0 +1,142 @@ +class DcManagerSubcloudPeerGroupListObject: + """Represents a subcloud peer group from dcmanager subcloud-peer-group list output.""" + + def __init__(self, id: str, peer_group_name: str, group_priority: int, group_state: int, system_leader_id: str, system_leader_name: str, max_subcloud_rehoming: int): + """Initialize the subcloud peer group object. + + Args: + id (str): Peer group ID. + peer_group_name (str): Name of the peer group. + group_priority (int): Priority level of the peer group. + group_state (int): State of the peer group. + system_leader_id (str): ID of the system leader. + system_leader_name (str): Name of the system leader. + max_subcloud_rehoming (int): Maximum subcloud rehoming value. + """ + self.id = id + self.peer_group_name = peer_group_name + self.group_priority = group_priority + self.group_state = group_state + self.system_leader_id = system_leader_id + self.system_leader_name = system_leader_name + self.max_subcloud_rehoming = max_subcloud_rehoming + + def get_id(self) -> str: + """Get the peer group ID. + + Returns: + str: Peer group ID. + """ + return self.id + + def set_id(self, id: str) -> None: + """Set the peer group ID. + + Args: + id (str): Peer group ID. + """ + self.id = id + + def get_peer_group_name(self) -> str: + """Get the peer group name. + + Returns: + str: Peer group name. + """ + return self.peer_group_name + + def set_peer_group_name(self, peer_group_name: str) -> None: + """Set the peer group name. + + Args: + peer_group_name (str): Peer group name. + """ + self.peer_group_name = peer_group_name + + def get_group_priority(self) -> int: + """Get the group priority. + + Returns: + int: Group priority level. + """ + return self.group_priority + + def set_group_priority(self, group_priority: int) -> None: + """Set the group priority. + + Args: + group_priority (int): Group priority level. + """ + self.group_priority = group_priority + + def get_group_state(self) -> int: + """Get the group state. + + Returns: + int: Group state. + """ + return self.group_state + + def set_group_state(self, group_state: int) -> None: + """Set the group state. + + Args: + group_state (int): Group state. + """ + self.group_state = group_state + + def get_system_leader_id(self) -> str: + """Get the system leader ID. + + Returns: + str: System leader ID. + """ + return self.system_leader_id + + def set_system_leader_id(self, system_leader_id: str) -> None: + """Set the system leader ID. + + Args: + system_leader_id (str): System leader ID. + """ + self.system_leader_id = system_leader_id + + def get_system_leader_name(self) -> str: + """Get the system leader name. + + Returns: + str: System leader name. + """ + return self.system_leader_name + + def set_system_leader_name(self, system_leader_name: str) -> None: + """Set the system leader name. + + Args: + system_leader_name (str): System leader name. + """ + self.system_leader_name = system_leader_name + + def get_max_subcloud_rehoming(self) -> int: + """Get the maximum subcloud rehoming value. + + Returns: + int: Maximum subcloud rehoming value. + """ + return self.max_subcloud_rehoming + + def set_max_subcloud_rehoming(self, max_subcloud_rehoming: int) -> None: + """Set the maximum subcloud rehoming value. + + Args: + max_subcloud_rehoming (int): Maximum subcloud rehoming value. + """ + self.max_subcloud_rehoming = max_subcloud_rehoming + + def __str__(self) -> str: + """String representation of the peer group object.""" + return f"{self.__class__.__name__}(Name={self.peer_group_name})" + + def __repr__(self) -> str: + """String representation for debugging.""" + return self.__str__() diff --git a/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_list_output.py b/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_list_output.py new file mode 100644 index 00000000..6b6ac7b4 --- /dev/null +++ b/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_list_output.py @@ -0,0 +1,81 @@ +from typing import List + +from framework.exceptions.keyword_exception import KeywordException +from framework.logging.automation_logger import get_logger +from keywords.cloud_platform.dcmanager.objects.dcmanager_subcloud_peer_group_list_object import DcManagerSubcloudPeerGroupListObject +from keywords.cloud_platform.system.system_table_parser import SystemTableParser + + +class DcManagerSubcloudPeerGroupListOutput: + """Parses the output of 'dcmanager subcloud-peer-group list' command.""" + + def __init__(self, peer_group_list_output: str): + """Initialize and parse the peer group list output. + + Args: + peer_group_list_output (str): Raw output from command. + """ + self.peer_group_list: List[DcManagerSubcloudPeerGroupListObject] = [] + system_table_parser = SystemTableParser(peer_group_list_output) + self.output_values = system_table_parser.get_output_values_list() + for value in self.output_values: + if not self.is_valid_output(value): + raise KeywordException(f"The output line {value} was not valid") + peer_group_object = DcManagerSubcloudPeerGroupListObject( + value.get("id"), + value.get("peer_group_name"), + value.get("group_priority"), + value.get("group_state"), + value.get("system_leader_id"), + value.get("system_leader_name"), + value.get("max_subcloud_rehoming"), + ) + self.peer_group_list.append(peer_group_object) + + def get_peer_group_list(self) -> List[DcManagerSubcloudPeerGroupListObject]: + """Get all peer group objects. + + Returns: + List[DcManagerSubcloudPeerGroupListObject]: List of peer group objects. + """ + return self.peer_group_list + + def get_peer_group_by_name(self, group_name: str) -> DcManagerSubcloudPeerGroupListObject: + """Get peer group by name. + + Args: + group_name (str): Name of the peer group. + + Returns: + DcManagerSubcloudPeerGroupListObject: Peer group object. + """ + for peer_group in self.peer_group_list: + if peer_group.get_group_name() == group_name: + return peer_group + raise ValueError(f"Peer group '{group_name}' not found") + + @staticmethod + def is_valid_output(value: dict) -> bool: + """Checks if the output contains all the required fields. + + Args: + value (dict): The dictionary of output values. + + Returns: + bool: True if all required fields are present, False otherwise. + """ + required_fields = ["peer_group_name", "group_priority", "group_state", "system_leader_id", "system_leader_name", "max_subcloud_rehoming"] + missing_fields = set(required_fields) - set(value.keys()) + + if missing_fields: + get_logger().log_error(f"Missing fields in output: {', '.join(missing_fields)}") + return False + return True + + def __str__(self) -> str: + """String representation of the peer group list.""" + return f"{self.__class__.__name__}(rows={len(self.peer_group_list)})" + + def __repr__(self) -> str: + """String representation for debugging.""" + return self.__str__() diff --git a/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_list_subclouds_object.py b/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_list_subclouds_object.py new file mode 100644 index 00000000..ec3decc3 --- /dev/null +++ b/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_list_subclouds_object.py @@ -0,0 +1,374 @@ +class DcManagerSubcloudPeerGroupListSubcloudsObject: + """Represents a subcloud from 'dcmanager subcloud-peer-group list-subclouds' command output.""" + + def __init__(self, id: str): + """Initialize subcloud object. + + Args: + id (str): The subcloud ID. + """ + self.id: str = id + self.name: str + self.description: str + self.location: str + self.software_version: str + self.management: str + self.availability: str + self.deploy_status: str + self.management_subnet: str + self.management_start_ip: str + self.management_end_ip: str + self.management_gateway_ip: str + self.systemcontroller_gateway_ip: str + self.group_id: str + self.peer_group_id: str + self.created_at: str + self.updated_at: str + self.backup_status: str + self.backup_datetime: str + self.prestage_status: str + self.prestage_versions: str + + def get_id(self) -> str: + """Get subcloud ID. + + Returns: + str: Subcloud ID. + """ + return self.id + + def set_id(self, id: str): + """Set subcloud ID. + + Args: + id (str): Subcloud ID to set. + """ + self.id = id + + def get_name(self) -> str: + """Get subcloud name. + + Returns: + str: Subcloud name. + """ + return self.name + + def set_name(self, name: str): + """Set subcloud name. + + Args: + name (str): Subcloud name to set. + """ + self.name = name + + def get_description(self) -> str: + """Get subcloud description. + + Returns: + str: Subcloud description. + """ + return self.description + + def set_description(self, description: str): + """Set subcloud description. + + Args: + description (str): Subcloud description to set. + """ + self.description = description + + def get_location(self) -> str: + """Get subcloud location. + + Returns: + str: Subcloud location. + """ + return self.location + + def set_location(self, location: str): + """Set subcloud location. + + Args: + location (str): Subcloud location to set. + """ + self.location = location + + def get_software_version(self) -> str: + """Get software version. + + Returns: + str: Software version. + """ + return self.software_version + + def set_software_version(self, software_version: str): + """Set software version. + + Args: + software_version (str): Software version to set. + """ + self.software_version = software_version + + def get_management(self) -> str: + """Get management status. + + Returns: + str: Management status. + """ + return self.management + + def set_management(self, management: str): + """Set management status. + + Args: + management (str): Management status to set. + """ + self.management = management + + def get_availability(self) -> str: + """Get availability status. + + Returns: + str: Availability status. + """ + return self.availability + + def set_availability(self, availability: str): + """Set availability status. + + Args: + availability (str): Availability status to set. + """ + self.availability = availability + + def get_deploy_status(self) -> str: + """Get deploy status. + + Returns: + str: Deploy status. + """ + return self.deploy_status + + def set_deploy_status(self, deploy_status: str): + """Set deploy status. + + Args: + deploy_status (str): Deploy status to set. + """ + self.deploy_status = deploy_status + + def get_management_subnet(self) -> str: + """Get management subnet. + + Returns: + str: Management subnet. + """ + return self.management_subnet + + def set_management_subnet(self, management_subnet: str): + """Set management subnet. + + Args: + management_subnet (str): Management subnet to set. + """ + self.management_subnet = management_subnet + + def get_management_start_ip(self) -> str: + """Get management start IP. + + Returns: + str: Management start IP. + """ + return self.management_start_ip + + def set_management_start_ip(self, management_start_ip: str): + """Set management start IP. + + Args: + management_start_ip (str): Management start IP to set. + """ + self.management_start_ip = management_start_ip + + def get_management_end_ip(self) -> str: + """Get management end IP. + + Returns: + str: Management end IP. + """ + return self.management_end_ip + + def set_management_end_ip(self, management_end_ip: str): + """Set management end IP. + + Args: + management_end_ip (str): Management end IP to set. + """ + self.management_end_ip = management_end_ip + + def get_management_gateway_ip(self) -> str: + """Get management gateway IP. + + Returns: + str: Management gateway IP. + """ + return self.management_gateway_ip + + def set_management_gateway_ip(self, management_gateway_ip: str): + """Set management gateway IP. + + Args: + management_gateway_ip (str): Management gateway IP to set. + """ + self.management_gateway_ip = management_gateway_ip + + def get_systemcontroller_gateway_ip(self) -> str: + """Get system controller gateway IP. + + Returns: + str: System controller gateway IP. + """ + return self.systemcontroller_gateway_ip + + def set_systemcontroller_gateway_ip(self, systemcontroller_gateway_ip: str): + """Set system controller gateway IP. + + Args: + systemcontroller_gateway_ip (str): System controller gateway IP to set. + """ + self.systemcontroller_gateway_ip = systemcontroller_gateway_ip + + def get_group_id(self) -> str: + """Get group ID. + + Returns: + str: Group ID. + """ + return self.group_id + + def set_group_id(self, group_id: str): + """Set group ID. + + Args: + group_id (str): Group ID to set. + """ + self.group_id = group_id + + def get_peer_group_id(self) -> str: + """Get peer group ID. + + Returns: + str: Peer group ID. + """ + return self.peer_group_id + + def set_peer_group_id(self, peer_group_id: str): + """Set peer group ID. + + Args: + peer_group_id (str): Peer group ID to set. + """ + self.peer_group_id = peer_group_id + + def get_created_at(self) -> str: + """Get creation timestamp. + + Returns: + str: Creation timestamp. + """ + return self.created_at + + def set_created_at(self, created_at: str): + """Set creation timestamp. + + Args: + created_at (str): Creation timestamp to set. + """ + self.created_at = created_at + + def get_updated_at(self) -> str: + """Get update timestamp. + + Returns: + str: Update timestamp. + """ + return self.updated_at + + def set_updated_at(self, updated_at: str): + """Set update timestamp. + + Args: + updated_at (str): Update timestamp to set. + """ + self.updated_at = updated_at + + def get_backup_status(self) -> str: + """Get backup status. + + Returns: + str: Backup status. + """ + return self.backup_status + + def set_backup_status(self, backup_status: str): + """Set backup status. + + Args: + backup_status (str): Backup status to set. + """ + self.backup_status = backup_status + + def get_backup_datetime(self) -> str: + """Get backup datetime. + + Returns: + str: Backup datetime. + """ + return self.backup_datetime + + def set_backup_datetime(self, backup_datetime: str): + """Set backup datetime. + + Args: + backup_datetime (str): Backup datetime to set. + """ + self.backup_datetime = backup_datetime + + def get_prestage_status(self) -> str: + """Get prestage status. + + Returns: + str: Prestage status. + """ + return self.prestage_status + + def set_prestage_status(self, prestage_status: str): + """Set prestage status. + + Args: + prestage_status (str): Prestage status to set. + """ + self.prestage_status = prestage_status + + def get_prestage_versions(self) -> str: + """Get prestage versions. + + Returns: + str: Prestage versions. + """ + return self.prestage_versions + + def set_prestage_versions(self, prestage_versions: str): + """Set prestage versions. + + Args: + prestage_versions (str): Prestage versions to set. + """ + self.prestage_versions = prestage_versions + + def __repr__(self) -> str: + """Return string representation. + + Returns: + str: String representation of the object. + """ + return f"{self.__class__.__name__}(Name={self.name})" diff --git a/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_list_subclouds_output.py b/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_list_subclouds_output.py new file mode 100644 index 00000000..a6c0ede4 --- /dev/null +++ b/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_list_subclouds_output.py @@ -0,0 +1,107 @@ +from framework.exceptions.keyword_exception import KeywordException +from keywords.cloud_platform.dcmanager.dcmanager_table_parser import DcManagerTableParser +from keywords.cloud_platform.dcmanager.objects.dcmanager_subcloud_peer_group_list_subclouds_object import DcManagerSubcloudPeerGroupListSubcloudsObject + + +class DcManagerSubcloudPeerGroupListSubcloudsOutput: + """Output parser for 'dcmanager subcloud-peer-group list-subclouds' command.""" + + def __init__(self, dcmanager_output: str): + """Initialize output parser. + + Args: + dcmanager_output (str): Raw command output to parse. + """ + self.dcmanager_output = dcmanager_output + self.subclouds: list[DcManagerSubcloudPeerGroupListSubcloudsObject] = [] + self._parse_output() + + def _parse_output(self): + """Parse dcmanager output and populate subcloud objects.""" + parser = DcManagerTableParser(self.dcmanager_output) + output_values_list = parser.get_output_values_list() + + for output_values in output_values_list: + + if not self.is_valid_output(output_values): + raise KeywordException(f"The output line {output_values} was not valid") + + subcloud = DcManagerSubcloudPeerGroupListSubcloudsObject(output_values.get("id", "")) + subcloud.set_name(output_values.get("name", "")) + subcloud.set_description(output_values.get("description", "")) + subcloud.set_location(output_values.get("location", "")) + subcloud.set_software_version(output_values.get("software_version", "")) + subcloud.set_management(output_values.get("management", "")) + subcloud.set_availability(output_values.get("availability", "")) + subcloud.set_deploy_status(output_values.get("deploy_status", "")) + subcloud.set_management_subnet(output_values.get("management_subnet", "")) + subcloud.set_management_start_ip(output_values.get("management_start_ip", "")) + subcloud.set_management_end_ip(output_values.get("management_end_ip", "")) + subcloud.set_management_gateway_ip(output_values.get("management_gateway_ip", "")) + subcloud.set_systemcontroller_gateway_ip(output_values.get("systemcontroller_gateway_ip", "")) + subcloud.set_group_id(output_values.get("group_id", "")) + subcloud.set_peer_group_id(output_values.get("peer_group_id", "")) + subcloud.set_created_at(output_values.get("created_at", "")) + subcloud.set_updated_at(output_values.get("updated_at", "")) + subcloud.set_backup_status(output_values.get("backup_status", "")) + subcloud.set_backup_datetime(output_values.get("backup_datetime", "")) + subcloud.set_prestage_status(output_values.get("prestage_status", "")) + subcloud.set_prestage_versions(output_values.get("prestage_versions", "")) + self.subclouds.append(subcloud) + + def get_subclouds(self) -> list[DcManagerSubcloudPeerGroupListSubcloudsObject]: + """Get list of subcloud objects. + + Returns: + list[DcManagerSubcloudPeerGroupListSubcloudsObject]: List of subcloud objects. + """ + return self.subclouds + + def get_subcloud_by_name(self, name: str) -> DcManagerSubcloudPeerGroupListSubcloudsObject: + """Get subcloud by name. + + Args: + name (str): Subcloud name to search for. + + Returns: + DcManagerSubcloudPeerGroupListSubcloudsObject: Subcloud object or None if not found. + """ + for subcloud in self.subclouds: + if subcloud.get_name() == name: + return subcloud + return None + + def get_subcloud_by_id(self, id: str) -> DcManagerSubcloudPeerGroupListSubcloudsObject: + """Get subcloud by ID. + + Args: + id (str): Subcloud ID to search for. + + Returns: + DcManagerSubcloudPeerGroupListSubcloudsObject: Subcloud object or None if not found. + """ + for subcloud in self.subclouds: + if subcloud.get_id() == id: + return subcloud + return None + + @staticmethod + def is_valid_output(value: dict) -> bool: + """Check if output contains all required fields. + + Args: + value (dict): Dictionary of output values. + + Returns: + bool: True if all required fields are present, False otherwise. + """ + required_fields = ["id", "name", "description", "location", "software_version", "management", "availability", "deploy_status", "management_subnet", "management_start_ip", "management_end_ip", "management_gateway_ip", "systemcontroller_gateway_ip", "group_id", "peer_group_id", "created_at", "updated_at", "backup_status", "backup_datetime", "prestage_status", "prestage_versions"] + return all(field in value for field in required_fields) + + def __repr__(self) -> str: + """Return string representation. + + Returns: + str: String representation of the object. + """ + return f"{self.__class__.__name__}(SubcloudCount={len(self.subclouds)})" diff --git a/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_show_output.py b/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_show_output.py new file mode 100644 index 00000000..26d16a34 --- /dev/null +++ b/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_show_output.py @@ -0,0 +1,64 @@ +from framework.exceptions.keyword_exception import KeywordException +from keywords.cloud_platform.dcmanager.dcmanager_vertical_table_parser import DcManagerVerticalTableParser +from keywords.cloud_platform.dcmanager.objects.dcmanager_subcloud_peer_group_list_object import DcManagerSubcloudPeerGroupListObject + + +class DcManagerSubcloudPeerGroupShowOutput: + """Parses the output of 'dcmanager subcloud-peer-group show' command.""" + + def __init__(self, peer_group_list_output: str): + """Initialize and parse the peer group list output. + + Args: + peer_group_list_output (str): Raw output from command. + """ + dc_vertical_table_parser = DcManagerVerticalTableParser(peer_group_list_output) + output_values = dc_vertical_table_parser.get_output_values_dict() + if not self.is_valid_output(output_values): + raise KeywordException(f"The output line {output_values} was not valid") + self.peer_group_object = DcManagerSubcloudPeerGroupListObject( + output_values.get("id"), + output_values.get("peer_group_name"), + output_values.get("group_priority"), + output_values.get("group_state"), + output_values.get("system_leader_id"), + output_values.get("system_leader_name"), + output_values.get("max_subcloud_rehoming"), + ) + + def get_dcmanager_subcloud_peer_group_object(self) -> DcManagerSubcloudPeerGroupListObject: + """Get the peer group object. + + Returns: + DcManagerSubcloudPeerGroupListObject: Peer group object. + """ + return self.peer_group_object + + @staticmethod + def is_valid_output(value: dict) -> bool: + """Checks if the output contains all the required fields. + + Args: + value (dict): The dictionary of output values. + + Returns: + bool: True if all required fields are present, False otherwise. + """ + required_fields = ["id", "peer_group_name", "group_priority", "group_state", "system_leader_id", "system_leader_name", "max_subcloud_rehoming"] + return all(field in value for field in required_fields) + + def __str__(self) -> str: + """String representation of the system peer object. + + Returns: + str: String representation. + """ + return f"{self.__class__.__name__}" + + def __repr__(self) -> str: + """String representation for debugging. + + Returns: + str: String representation. + """ + return self.__str__() diff --git a/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_status_object.py b/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_status_object.py new file mode 100644 index 00000000..4a255c02 --- /dev/null +++ b/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_status_object.py @@ -0,0 +1,178 @@ +class DcManagerSubcloudPeerGroupStatusObject: + """Represents a subcloud peer group status from dcmanager subcloud-peer-group status output.""" + + def __init__(self, peer_group_id: str, peer_group_name: str, total_subclouds: int, complete: int, waiting_for_migrate: int, rehoming: int, rehome_failed: int, managed: int, unmanaged: int): + """Initialize the subcloud peer group status object. + + Args: + peer_group_id (str): Peer group ID. + peer_group_name (str): Name of the peer group. + total_subclouds (int): Total number of subclouds. + complete (int): Number of complete subclouds. + waiting_for_migrate (int): Number of subclouds waiting for migration. + rehoming (int): Number of subclouds in rehoming state. + rehome_failed (int): Number of subclouds with failed rehoming. + managed (int): Number of managed subclouds. + unmanaged (int): Number of unmanaged subclouds. + """ + self.peer_group_id = peer_group_id + self.peer_group_name = peer_group_name + self.total_subclouds = total_subclouds + self.complete = complete + self.waiting_for_migrate = waiting_for_migrate + self.rehoming = rehoming + self.rehome_failed = rehome_failed + self.managed = managed + self.unmanaged = unmanaged + + def get_peer_group_id(self) -> str: + """Get the peer group ID. + + Returns: + str: Peer group ID. + """ + return self.peer_group_id + + def set_peer_group_id(self, peer_group_id: str) -> None: + """Set the peer group ID. + + Args: + peer_group_id (str): Peer group ID. + """ + self.peer_group_id = peer_group_id + + def get_peer_group_name(self) -> str: + """Get the peer group name. + + Returns: + str: Peer group name. + """ + return self.peer_group_name + + def set_peer_group_name(self, peer_group_name: str) -> None: + """Set the peer group name. + + Args: + peer_group_name (str): Peer group name. + """ + self.peer_group_name = peer_group_name + + def get_total_subclouds(self) -> int: + """Get the total number of subclouds. + + Returns: + int: Total number of subclouds. + """ + return self.total_subclouds + + def set_total_subclouds(self, total_subclouds: int) -> None: + """Set the total number of subclouds. + + Args: + total_subclouds (int): Total number of subclouds. + """ + self.total_subclouds = total_subclouds + + def get_complete(self) -> int: + """Get the number of complete subclouds. + + Returns: + int: Number of complete subclouds. + """ + return self.complete + + def set_complete(self, complete: int) -> None: + """Set the number of complete subclouds. + + Args: + complete (int): Number of complete subclouds. + """ + self.complete = complete + + def get_waiting_for_migrate(self) -> int: + """Get the number of subclouds waiting for migration. + + Returns: + int: Number of subclouds waiting for migration. + """ + return self.waiting_for_migrate + + def set_waiting_for_migrate(self, waiting_for_migrate: int) -> None: + """Set the number of subclouds waiting for migration. + + Args: + waiting_for_migrate (int): Number of subclouds waiting for migration. + """ + self.waiting_for_migrate = waiting_for_migrate + + def get_rehoming(self) -> int: + """Get the number of subclouds in rehoming state. + + Returns: + int: Number of subclouds in rehoming state. + """ + return self.rehoming + + def set_rehoming(self, rehoming: int) -> None: + """Set the number of subclouds in rehoming state. + + Args: + rehoming (int): Number of subclouds in rehoming state. + """ + self.rehoming = rehoming + + def get_rehome_failed(self) -> int: + """Get the number of subclouds with failed rehoming. + + Returns: + int: Number of subclouds with failed rehoming. + """ + return self.rehome_failed + + def set_rehome_failed(self, rehome_failed: int) -> None: + """Set the number of subclouds with failed rehoming. + + Args: + rehome_failed (int): Number of subclouds with failed rehoming. + """ + self.rehome_failed = rehome_failed + + def get_managed(self) -> int: + """Get the number of managed subclouds. + + Returns: + int: Number of managed subclouds. + """ + return self.managed + + def set_managed(self, managed: int) -> None: + """Set the number of managed subclouds. + + Args: + managed (int): Number of managed subclouds. + """ + self.managed = managed + + def get_unmanaged(self) -> int: + """Get the number of unmanaged subclouds. + + Returns: + int: Number of unmanaged subclouds. + """ + return self.unmanaged + + def set_unmanaged(self, unmanaged: int) -> None: + """Set the number of unmanaged subclouds. + + Args: + unmanaged (int): Number of unmanaged subclouds. + """ + self.unmanaged = unmanaged + + def __str__(self) -> str: + """String representation of the peer group status object.""" + return f"{self.__class__.__name__}(Name={self.peer_group_name})" + + def __repr__(self) -> str: + """String representation for debugging.""" + return self.__str__() diff --git a/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_status_output.py b/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_status_output.py new file mode 100644 index 00000000..be5920f6 --- /dev/null +++ b/keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_peer_group_status_output.py @@ -0,0 +1,66 @@ +from framework.exceptions.keyword_exception import KeywordException +from keywords.cloud_platform.dcmanager.dcmanager_vertical_table_parser import DcManagerVerticalTableParser +from keywords.cloud_platform.dcmanager.objects.dcmanager_subcloud_peer_group_status_object import DcManagerSubcloudPeerGroupStatusObject + + +class DcManagerSubcloudPeerGroupStatusOutput: + """Parses the output of 'dcmanager subcloud-peer-group status' command.""" + + def __init__(self, peer_group_status_output: str): + """Initialize and parse the peer group status output. + + Args: + peer_group_status_output (str): Raw output from command. + """ + dc_vertical_table_parser = DcManagerVerticalTableParser(peer_group_status_output) + output_values = dc_vertical_table_parser.get_output_values_dict() + if not self.is_valid_output(output_values): + raise KeywordException(f"The output line {output_values} was not valid") + self.peer_group_status_object = DcManagerSubcloudPeerGroupStatusObject( + output_values.get("peer_group_id"), + output_values.get("peer_group_name"), + output_values.get("total_subclouds"), + output_values.get("complete"), + output_values.get("waiting_for_migrate"), + output_values.get("rehoming"), + output_values.get("rehome_failed"), + output_values.get("managed"), + output_values.get("unmanaged"), + ) + + def get_dcmanager_subcloud_peer_group_status_object(self) -> DcManagerSubcloudPeerGroupStatusObject: + """Get the peer group status object. + + Returns: + DcManagerSubcloudPeerGroupStatusObject: Peer group status object. + """ + return self.peer_group_status_object + + def __str__(self) -> str: + """String representation of the peer group status output. + + Returns: + str: String representation. + """ + return f"{self.__class__.__name__}" + + @staticmethod + def is_valid_output(value: dict) -> bool: + """Checks if the output contains all the required fields. + + Args: + value (dict): The dictionary of output values. + + Returns: + bool: True if all required fields are present, False otherwise. + """ + required_fields = ["peer_group_id", "peer_group_name", "total_subclouds", "complete", "waiting_for_migrate", "rehoming", "rehome_failed", "managed", "unmanaged"] + return all(field in value for field in required_fields) + + def __repr__(self) -> str: + """String representation for debugging. + + Returns: + str: String representation. + """ + return self.__str__()