starlingx/test
Code Issues Proposed changes

dcmanager subcloud-group keywords

Browse Source 
Change-Id: I2efb75fefe30abd5f0671d238594b07bc5d95428
This commit is contained in:
Rity Menon 2025-02-20 13:27:48 -08:00
parent 6165897ebb
commit 17f499ac1b
5 changed files with 274 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.pre-commit-config.yaml
View File

@ -17,7 +17,7 @@
# - `black` uses `pyproject.toml` for consistency across formatting tools.
# - `pydocstyle` and `pydoclint` ensure compliance with Google-style docstrings.
# -----------------------------------------------------------------------------
default_stages: [commit]
default_stages: [pre-commit]
default_language_version:
python: python3.11

53
keywords/cloud_platform/dcmanager/dcmanager_subcloud_group_keywords.py Normal file
View File

@ -0,0 +1,53 @@
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_group_output import (
DcmanagerSubcloudGroupOutput,
)
from keywords.cloud_platform.dcmanager.objects.dcmanager_subcloud_group_show_output import (
DcmanagerSubcloudGroupShowOutput,
)
class DcmanagerSubcloudGroupKeywords(BaseKeyword):
"""
This class contains all the keywords related to the 'dcmanager subcloud-group' commands.
"""
def __init__(self, ssh_connection: SSHConnection) -> None:
"""
Initializes DcmanagerSubcloudGroupKeywords.
Args:
ssh_connection (SSHConnection): The SSH connection object used for executing commands.
"""
self.ssh_connection = ssh_connection
def get_dcmanager_subcloud_group_list(self) -> DcmanagerSubcloudGroupOutput:
"""
Gets the dcmanager subcloud-group list.
Returns:
DcmanagerSubcloudGroupOutput: An object containing the list of subcloud groups.
"""
command = source_openrc('dcmanager subcloud-group list')
output = self.ssh_connection.send(command)
self.validate_success_return_code(self.ssh_connection)
return DcmanagerSubcloudGroupOutput(output)
def get_dcmanager_subcloud_group_show(
self, group_id: str
) -> DcmanagerSubcloudGroupShowOutput:
"""
Gets the dcmanager subcloud-group details for a specific group.
Args:
group_id (str): The identifier of the subcloud group.
Returns:
DcmanagerSubcloudGroupShowOutput: An object containing details of the subcloud group.
"""
command = source_openrc(f'dcmanager subcloud-group show {group_id}')
output = self.ssh_connection.send(command)
self.validate_success_return_code(self.ssh_connection)
return DcmanagerSubcloudGroupShowOutput(output)

73
keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_group_object.py Normal file
View File

@ -0,0 +1,73 @@
from typing import Optional
class DcmanagerSubcloudGroupObject:
"""
This class represents a dcmanager subcloud-group as an object.
"""
def __init__(self) -> None:
"""Initializes a DcmanagerSubcloudGroupObject with default values."""
self.id: int = -1
self.name: Optional[str] = None
self.description: Optional[str] = None
self.update_apply_type: Optional[str] = None
self.max_parallel_subclouds: int = -1
self.created_at: Optional[str] = None
self.updated_at: Optional[str] = None
def set_id(self, group_id: int) -> None:
"""Sets the ID of the subcloud group."""
self.id = group_id
def get_id(self) -> int:
"""Gets the ID of the subcloud group."""
return self.id
def set_name(self, name: str) -> None:
"""Sets the name of the subcloud group."""
self.name = name
def get_name(self) -> Optional[str]:
"""Gets the name of the subcloud group."""
return self.name
def set_description(self, description: str) -> None:
"""Sets the description of the subcloud group."""
self.description = description
def get_description(self) -> Optional[str]:
"""Gets the description of the subcloud group."""
return self.description
def set_update_apply_type(self, update_apply_type: str) -> None:
"""Sets the update apply type of the subcloud group."""
self.update_apply_type = update_apply_type
def get_update_apply_type(self) -> Optional[str]:
"""Gets the update apply type of the subcloud group."""
return self.update_apply_type
def set_max_parallel_subclouds(self, max_parallel_subclouds: int) -> None:
"""Sets the maximum number of parallel subclouds."""
self.max_parallel_subclouds = max_parallel_subclouds
def get_max_parallel_subclouds(self) -> int:
"""Gets the maximum number of parallel subclouds."""
return self.max_parallel_subclouds
def set_created_at(self, created_at: str) -> None:
"""Sets the creation timestamp of the subcloud group."""
self.created_at = created_at
def get_created_at(self) -> Optional[str]:
"""Gets the creation timestamp of the subcloud group."""
return self.created_at
def set_updated_at(self, updated_at: str) -> None:
"""Sets the last updated timestamp of the subcloud group."""
self.updated_at = updated_at
def get_updated_at(self) -> Optional[str]:
"""Gets the last updated timestamp of the subcloud group."""
return self.updated_at

67
keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_group_output.py Normal file
View File

@ -0,0 +1,67 @@
from typing import Dict, List
from framework.exceptions.keyword_exception import KeywordException
from framework.logging.automation_logger import get_logger
from keywords.cloud_platform.dcmanager.dcmanager_table_parser import (
DcManagerTableParser,
)
from keywords.cloud_platform.dcmanager.objects.dcmanager_subcloud_group_object import (
DcmanagerSubcloudGroupObject,
)
class DcmanagerSubcloudGroupOutput:
"""
Parses the output of the 'dcmanager subcloud-group list' command into a list of DcmanagerSubcloudGroupObject instances.
"""
def __init__(self, dcmanager_output: str) -> None:
"""
Initializes DcmanagerSubcloudGroupOutput.
Args:
dcmanager_output (str): Output of the 'dcmanager subcloud-group list' command.
Raises:
KeywordException: If the output format is invalid.
"""
self.dcmanager_subcloud_group: List[DcmanagerSubcloudGroupObject] = []
dc_table_parser = DcManagerTableParser(dcmanager_output)
output_values = dc_table_parser.get_output_values_list()
for value in output_values:
if self.is_valid_output(value):
dcmanager_subcloud_group = DcmanagerSubcloudGroupObject()
dcmanager_subcloud_group.set_id(value["id"])
dcmanager_subcloud_group.set_name(value["name"])
dcmanager_subcloud_group.set_description(value["description"])
self.dcmanager_subcloud_group.append(dcmanager_subcloud_group)
else:
raise KeywordException(f"The output line {value} was not valid")
def get_dcmanager_subcloud_group_list(self) -> List[DcmanagerSubcloudGroupObject]:
"""
Retrieves the parsed dcmanager subcloud-group list.
Returns:
List[DcmanagerSubcloudGroupObject]: A list of parsed DcmanagerSubcloudGroupObject instances.
"""
return self.dcmanager_subcloud_group
@staticmethod
def is_valid_output(value: Dict[str, str]) -> bool:
"""
Checks if the output dictionary contains all required fields.
Args:
value (Dict[str, str]): The dictionary of output values.
Returns:
bool: True if the output contains all required fields, False otherwise.
"""
required_fields = ["id", "name", "description"]
for field in required_fields:
if field not in value:
get_logger().log_error(f"{field} is not in the output value")
return False
return True

80
keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_group_show_output.py Normal file
View File

@ -0,0 +1,80 @@
from typing import Dict
from framework.exceptions.keyword_exception import KeywordException
from framework.logging.automation_logger import get_logger
from keywords.cloud_platform.dcmanager.dcmanager_vertical_table_parser import (
DcManagerVerticalTableParser,
)
from keywords.cloud_platform.dcmanager.objects.dcmanager_subcloud_group_object import (
DcmanagerSubcloudGroupObject,
)
class DcmanagerSubcloudGroupShowOutput:
"""
Parses the output of the 'dcmanager subcloud-group show' command into a DcmanagerSubcloudGroupObject instance.
"""
def __init__(self, dcmanager_output: str) -> None:
"""
Initializes DcmanagerSubcloudGroupShowOutput.
Args:
dcmanager_output (str): Output of the 'dcmanager subcloud-group show' command.
Raises:
KeywordException: If the output format is invalid.
"""
dc_vertical_table_parser = DcManagerVerticalTableParser(dcmanager_output)
output_values = dc_vertical_table_parser.get_output_values_dict()
if self.is_valid_output(output_values):
self.dcmanager_subcloud_group = DcmanagerSubcloudGroupObject()
self.dcmanager_subcloud_group.set_id(output_values["id"])
self.dcmanager_subcloud_group.set_name(output_values["name"])
self.dcmanager_subcloud_group.set_description(output_values["description"])
self.dcmanager_subcloud_group.set_update_apply_type(
output_values["update apply type"]
)
self.dcmanager_subcloud_group.set_max_parallel_subclouds(
output_values["max parallel subclouds"]
)
self.dcmanager_subcloud_group.set_created_at(output_values["created_at"])
self.dcmanager_subcloud_group.set_updated_at(output_values["updated_at"])
else:
raise KeywordException(f"The output line {output_values} was not valid")
def get_dcmanager_subcloud_group_show(self) -> DcmanagerSubcloudGroupObject:
"""
Retrieves the parsed dcmanager subcloud-group show object.
Returns:
DcmanagerSubcloudGroupObject: The parsed subcloud-group show object.
"""
return self.dcmanager_subcloud_group
@staticmethod
def is_valid_output(value: Dict[str, str]) -> bool:
"""
Checks if the output contains all the required fields.
Args:
value (Dict[str, str]): The dictionary of output values.
Returns:
bool: True if all required fields are present, False otherwise.
"""
required_fields = [
"id",
"name",
"description",
"update apply type",
"max parallel subclouds",
"created_at",
"updated_at",
]
for field in required_fields:
if field not in value:
get_logger().log_error(f"{field} is not in the output value")
return False
return True