Merge "Add dcmanager subcloud peer group keywords"
This commit is contained in:
Zuul
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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__()
|
||||
@@ -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__()
|
||||
@@ -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})"
|
||||
@@ -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)})"
|
||||
@@ -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__()
|
||||
@@ -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__()
|
||||
@@ -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__()
|
||||
Reference in New Issue
Block a user