diff --git a/keywords/cloud_platform/dcmanager/dcmanager_subcloud_prestage.py b/keywords/cloud_platform/dcmanager/dcmanager_subcloud_prestage.py index 649e8777..1da18a8b 100644 --- a/keywords/cloud_platform/dcmanager/dcmanager_subcloud_prestage.py +++ b/keywords/cloud_platform/dcmanager/dcmanager_subcloud_prestage.py @@ -1,4 +1,3 @@ -from config.configuration_manager import ConfigurationManager from framework.ssh.ssh_connection import SSHConnection from framework.validation.validation import validate_equals_with_retry from keywords.base_keyword import BaseKeyword @@ -20,19 +19,18 @@ class DcmanagerSubcloudPrestage(BaseKeyword): """ self.ssh_connection = ssh_connection - def dcmanager_subcloud_prestage(self, subcloud_name: str, syspass: str) -> None: + def dcmanager_subcloud_prestage(self, subcloud_name: str, syspass: str, for_sw_deploy: bool = False) -> None: """ Runs dcmanager subcloud prestage command. Args: subcloud_name (str): The name of the subcloud to check. syspass (str): The sysadmin password to be passed to the command. + for_sw_deploy (bool): whether to enable --for-sw-deploy flag. """ - lab_config = ConfigurationManager.get_lab_config() - pasw = lab_config.get_admin_credentials().get_password() - - command = source_openrc(f"dcmanager subcloud prestage --for-sw-deploy {subcloud_name}" f" --sysadmin-password {pasw}") + cmd_options = "--for-sw-deploy" if for_sw_deploy else "" + command = source_openrc(f"dcmanager subcloud prestage {cmd_options} {subcloud_name}" f" --sysadmin-password {syspass}") self.ssh_connection.send(command) self.validate_success_return_code(self.ssh_connection) diff --git a/keywords/cloud_platform/health/health_keywords.py b/keywords/cloud_platform/health/health_keywords.py new file mode 100644 index 00000000..34b54d99 --- /dev/null +++ b/keywords/cloud_platform/health/health_keywords.py @@ -0,0 +1,47 @@ +from framework.ssh.ssh_connection import SSHConnection +from keywords.base_keyword import BaseKeyword +from keywords.cloud_platform.fault_management.alarms.alarm_list_keywords import AlarmListKeywords +from keywords.cloud_platform.system.application.system_application_list_keywords import SystemApplicationListKeywords +from keywords.k8s.pods.kubectl_get_pods_keywords import KubectlGetPodsKeywords + + +class HealthKeywords(BaseKeyword): + """Class for health Keywords for Cloud Platform""" + + def __init__(self, ssh_connection: SSHConnection): + """Constructor + + Args: + ssh_connection (SSHConnection): ssh object + + """ + self.ssh_connection = ssh_connection + + def validate_healty_cluster(self): + """Function to validate the health of the cluster + + This function checks the health of the cluster which consists of + 1. validate all pods are healty + 2. validate no alarms are present + 3. validate all apps are healty and applied + """ + # Validate all pods are healthy + self.validate_pods_health() + # Validate no alarms are present + self.validate_no_alarms() + # Validate all apps are healthy and applied + self.validate_apps_health_and_applied() + + def validate_pods_health(self): + """Function to validate the health of all pods in the cluster""" + healthy_status = ["Running", "Succeeded", "Completed"] + KubectlGetPodsKeywords(self.ssh_connection).wait_for_all_pods_status(healthy_status, timeout=300) + + def validate_no_alarms(self): + """Function to validate no alarms are present in the cluster""" + AlarmListKeywords(self.ssh_connection).wait_for_all_alarms_cleared() + + def validate_apps_health_and_applied(self): + """Function to validate all apps are healthy and applied""" + healthy_status = ["applied", "uploaded"] + SystemApplicationListKeywords(self.ssh_connection).validate_all_apps_status(healthy_status) diff --git a/keywords/cloud_platform/system/application/system_application_list_keywords.py b/keywords/cloud_platform/system/application/system_application_list_keywords.py index 661b255d..6c35ba1d 100644 --- a/keywords/cloud_platform/system/application/system_application_list_keywords.py +++ b/keywords/cloud_platform/system/application/system_application_list_keywords.py @@ -1,13 +1,12 @@ -import time from typing import List from framework.exceptions.keyword_exception import KeywordException from framework.logging.automation_logger import get_logger +from framework.ssh.ssh_connection import SSHConnection from framework.validation.validation import validate_equals_with_retry, validate_list_contains_with_retry from keywords.base_keyword import BaseKeyword from keywords.cloud_platform.command_wrappers import source_openrc from keywords.cloud_platform.system.application.object.system_application_list_output import SystemApplicationListOutput -from keywords.cloud_platform.system.application.object.system_application_list_status_tracking_input import SystemApplicationListStatusTrackingInput class SystemApplicationListKeywords(BaseKeyword): @@ -15,11 +14,12 @@ class SystemApplicationListKeywords(BaseKeyword): Class for System application list keywords. """ - def __init__(self, ssh_connection): - """ - Constructor + def __init__(self, ssh_connection: SSHConnection): + """Constructor + Args: - ssh_connection: + ssh_connection (SSHConnection): ssh object + """ self.ssh_connection = ssh_connection @@ -33,20 +33,20 @@ class SystemApplicationListKeywords(BaseKeyword): SystemApplicationListOutput: an instance of the SystemApplicationOutput object representing the applications on the host, as a result of the execution of the 'system application-list' command. """ - output = self.ssh_connection.send(source_openrc('system application-list')) + output = self.ssh_connection.send(source_openrc("system application-list")) self.validate_success_return_code(self.ssh_connection) system_application_list_output = SystemApplicationListOutput(output) return system_application_list_output - def validate_app_status(self, application_name: str, status: str, timeout=300, polling_sleep_time=5): - """ - This function will validate that the application specified reaches the desired status. + def validate_app_status(self, application_name: str, status: str, timeout: int = 300, polling_sleep_time: int = 5): + """This function will validate that the application specified reaches the desired status. + Args: - application_name: Name of the application that we are waiting for. - status: Status in which we want to wait for the application to reach. - timeout: Timeout in seconds - polling_sleep_time: wait time in seconds before the next attempt when unsuccessful validation + application_name (str): Name of the application that we are waiting for. + status (str): Status in which we want to wait for the application to reach. + timeout (int): Timeout in seconds + polling_sleep_time (int): wait time in seconds before the next attempt when unsuccessful validation Returns: None @@ -60,16 +60,17 @@ class SystemApplicationListKeywords(BaseKeyword): message = f"Application {application_name}'s status is {status}" validate_equals_with_retry(get_app_status, status, message, timeout, polling_sleep_time) - def validate_app_status_in_list(self, application_name: str, status: List[str], timeout=3600, polling_sleep_time=30) -> str: - """ - This function will validate that the application specified reaches the desired status. - Args: - application_name: Name of the application that we are waiting for. - status: Status in which we want to wait for the application to reach. - timeout: Timeout in seconds - polling_sleep_time: wait time in seconds before the next attempt when unsuccessful validation + def validate_app_status_in_list(self, application_name: str, status: List[str], timeout: int = 3600, polling_sleep_time: int = 30) -> str: + """This function will validate that the application specified reaches the desired status. - Returns: observed_status of the application + Args: + application_name (str): Name of the application that we are waiting for. + status (List[str]): Status in which we want to wait for the application to reach. + timeout (int): Timeout in seconds + polling_sleep_time (int): wait time in seconds before the next attempt when unsuccessful validation + + Returns: + str: observed_status of the application """ @@ -79,26 +80,42 @@ class SystemApplicationListKeywords(BaseKeyword): return application_status message = f"Application {application_name}'s status is in {status}" - observed_status=validate_list_contains_with_retry(get_app_status, status, message, timeout, polling_sleep_time) + observed_status = validate_list_contains_with_retry(get_app_status, status, message, timeout, polling_sleep_time) return observed_status - def is_app_present(self, application_name: str) -> bool: - """ - This function will validate that the application is present in application list. + """This function will validate that the application is present in application list. + Args: - application_name: Name of the application that we want to check. - - Returns: boolean value True if application is found in list else False + application_name (str): Name of the application that we want to check. + Returns: + bool: True if application is found in list else False """ - try: system_applications = self.get_system_application_list() application_status = system_applications.get_application(application_name).get_status() - get_logger().log_info(f'{application_name} is present. Status is {application_status}') + get_logger().log_info(f"{application_name} is present. Status is {application_status}") return True except KeywordException: - get_logger().log_info(f'{application_name} is not found.') - return False \ No newline at end of file + get_logger().log_info(f"{application_name} is not found.") + return False + + def validate_all_apps_status(self, expected_statuses: [str]) -> bool: + """Validates That all apps are in the expected status. + + Args: + expected_statuses ([str]): list of expected statuses ex. ['applied' , 'uploaded'] + + Returns: + bool: True if all apps are in the expected status, False otherwise. + """ + apps = self.get_system_application_list().get_applications() + not_applied_apps = list(filter(lambda app: app.get_status() not in expected_statuses, apps)) + if len(not_applied_apps) == 0: + return True + else: + for app in not_applied_apps: + get_logger().log_info(f"Application {app.get_application()} is in status {app.get_status()}") + raise KeywordException("All applications are not in the expected status.") diff --git a/keywords/k8s/pods/kubectl_get_pods_keywords.py b/keywords/k8s/pods/kubectl_get_pods_keywords.py index b06fdab0..546bf52b 100644 --- a/keywords/k8s/pods/kubectl_get_pods_keywords.py +++ b/keywords/k8s/pods/kubectl_get_pods_keywords.py @@ -21,7 +21,7 @@ class KubectlGetPodsKeywords(BaseKeyword): ssh_connection (SSHConnection): An SSH connection object to the target system. """ self.ssh_connection = ssh_connection - + def get_pods(self, namespace: str = None, label: str = None) -> KubectlGetPodsOutput: """ Gets the k8s pods that are available using '-o wide'. @@ -49,7 +49,7 @@ class KubectlGetPodsKeywords(BaseKeyword): pods_list_output = KubectlGetPodsOutput(kubectl_get_pods_output) return pods_list_output - + def get_pods_no_validation(self, namespace: str = None) -> KubectlGetPodsOutput: """ Gets the k8s pods that are available using '-o wide'. @@ -161,7 +161,7 @@ class KubectlGetPodsKeywords(BaseKeyword): return True time.sleep(5) - return False + raise KeywordException("All pods are not in the expected status") def wait_for_pods_to_reach_status(self, expected_status: str, pod_names: list = None, namespace: str = None, poll_interval: int = 5, timeout: int = 180) -> bool: """ diff --git a/testcases/cloud_platform/patching/dc/orchestrator/test_sw_patch_release_orchestration.py b/testcases/cloud_platform/patching/dc/orchestrator/test_sw_patch_release_orchestration.py index 4f05784a..c99c1079 100644 --- a/testcases/cloud_platform/patching/dc/orchestrator/test_sw_patch_release_orchestration.py +++ b/testcases/cloud_platform/patching/dc/orchestrator/test_sw_patch_release_orchestration.py @@ -46,7 +46,7 @@ def test_patch_apply(request): # Prestage the subcloud with the latest software deployed in the controller get_logger().log_info(f"Prestage {subcloud_name} with {sw_release}.") - DcmanagerSubcloudPrestage(central_ssh).dcmanager_subcloud_prestage(subcloud_name=subcloud_name, syspass=subcloud_password) + DcmanagerSubcloudPrestage(central_ssh).dcmanager_subcloud_prestage(subcloud_name=subcloud_name, syspass=subcloud_password, for_sw_deploy=True) # Create software deploy strategy get_logger().log_info(f"Create sw-deploy strategy for {subcloud_name}.") diff --git a/testcases/cloud_platform/regression/dc/prestage/test_prestage.py b/testcases/cloud_platform/regression/dc/prestage/test_prestage.py new file mode 100644 index 00000000..334aecf1 --- /dev/null +++ b/testcases/cloud_platform/regression/dc/prestage/test_prestage.py @@ -0,0 +1,35 @@ +from pytest import mark + +from config.configuration_manager import ConfigurationManager +from framework.logging.automation_logger import get_logger +from framework.validation.validation import validate_equals +from keywords.cloud_platform.dcmanager.dcmanager_subcloud_list_keywords import DcManagerSubcloudListKeywords +from keywords.cloud_platform.dcmanager.dcmanager_subcloud_prestage import DcmanagerSubcloudPrestage +from keywords.cloud_platform.health.health_keywords import HealthKeywords +from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords + + +@mark.p0 +@mark.lab_has_subcloud +def test_subcloud_prestage(): + """Test the prestage of a subcloud.""" + # Gets the SSH connection to the active controller of the central cloud. + ssh_connection = LabConnectionKeywords().get_active_controller_ssh() + # Gets the lowest subcloud (the subcloud with the lowest id). + dcm_sc_list_kw = DcManagerSubcloudListKeywords(ssh_connection) + lowest_subcloud = dcm_sc_list_kw.get_dcmanager_subcloud_list().get_healthy_subcloud_with_lowest_id() + sc_name = lowest_subcloud.get_name() + # Gets the lowest subcloud sysadmin password + lab_config = ConfigurationManager.get_lab_config().get_subcloud(sc_name) + syspass = lab_config.get_admin_credentials().get_password() + + get_logger().log_info(f"Subcloud selected for prestage: {sc_name}") + DcmanagerSubcloudPrestage(ssh_connection).dcmanager_subcloud_prestage(sc_name, syspass) + # validate prestage status + obj_subcloud = dcm_sc_list_kw.get_dcmanager_subcloud_list().get_subcloud_by_name(sc_name) + # Verify that the prestage is completed successfully + validate_equals(obj_subcloud.get_prestage_status(), "complete", f"subcloud {sc_name} successfully.") + + # validate Healthy status + subcloud_ssh = LabConnectionKeywords().get_subcloud_ssh(sc_name) + HealthKeywords(subcloud_ssh).validate_healty_cluster()