diff --git a/keywords/cloud_platform/dcmanager/dcmanager_subcloud_backup_keywords.py b/keywords/cloud_platform/dcmanager/dcmanager_subcloud_backup_keywords.py new file mode 100644 index 00000000..1d629665 --- /dev/null +++ b/keywords/cloud_platform/dcmanager/dcmanager_subcloud_backup_keywords.py @@ -0,0 +1,190 @@ +from typing import Optional + +from framework.ssh.ssh_connection import SSHConnection +from framework.validation.validation import validate_equals_with_retry +from keywords.base_keyword import BaseKeyword +from keywords.cloud_platform.command_wrappers import source_openrc +from keywords.files.file_keywords import FileKeywords + + +class DcManagerSubcloudBackupKeywords(BaseKeyword): + """ + This class contains all the keywords related to the 'dcmanager subcloud-backup ' command. + """ + + def __init__(self, ssh_connection: SSHConnection): + """ + Constructor. + + Args: + ssh_connection (SSHConnection): SSH connection to the target system. + """ + self.ssh_connection = ssh_connection + + def create_subcloud_backup( + self, + sysadmin_password: str, + con_ssh: SSHConnection, + path: str, + subcloud: Optional[str] = None, + local_only: bool = False, + backup_yaml: Optional[str] = None, + group: Optional[str] = None, + registry: bool = False, + ) -> None: + """ + Creates a backup of the specified subcloud. + + Args: + sysadmin_password (str): Subcloud sysadmin password needed for backup creation. + con_ssh (SSHConnection): SSH connection to execute the command (central_ssh or subcloud_ssh). + path (str): The directory path where the backup file will be checked. + subcloud (Optional[str]): The name of the subcloud to backup. Defaults to None. + local_only (bool): If True, backup will be stored only in the subcloud. Defaults to False. + backup_yaml (Optional[str]): path to use the yaml file. Defaults to None. + group (Optional[str]): Subcloud group name to create backup. Defaults to None. + registry (bool): Option to add the registry backup in the same task. Defaults to False. + + Returns: + None: + """ + # Command construction + cmd = ( + f"dcmanager subcloud-backup create --sysadmin-password {sysadmin_password}" + ) + if subcloud: + cmd += f" --subcloud {subcloud}" + if local_only: + cmd += " --local-only" + if backup_yaml: + cmd += f" --backup-values {backup_yaml}" + if group: + cmd += f" --group {group}" + if registry: + cmd += " --registry-images" + + self.ssh_connection.send(source_openrc(cmd)) + self.validate_success_return_code(self.ssh_connection) + + # Use wait_for_backup_creation to ensure the file is created + self.wait_for_backup_creation(con_ssh, path, subcloud) + + def wait_for_backup_creation( + self, + con_ssh: SSHConnection, + path: str, + subcloud: Optional[str], + check_interval: int = 30, + timeout: int = 600, + ) -> None: + """ + Waits for the backup file to be created in the specified path. + + Args: + con_ssh (SSHConnection): SSH connection to execute the command (central_ssh or subcloud_ssh). + path (str): The path where the backup file is expected. + subcloud (Optional[str]): The name of the subcloud to check. + check_interval (int): Time interval (in seconds) to check for file creation. Defaults to 30. + timeout (int): Maximum time (in seconds) to wait for file creation. Defaults to 600. + + Returns: + None: + """ + + def check_backup_created() -> str: + """ + Checks if the backup has been created. + + Returns: + str: A message indicating whether the backup has been successfully created or not. + """ + check_file = FileKeywords(con_ssh).validate_file_exists_with_sudo(path) + if check_file: + return f"Backup should be created at {path}" + else: + return "Backup not created yet." + + validate_equals_with_retry( + function_to_execute=check_backup_created, + expected_value=f"Backup should be created at {path}", + validation_description=f"Backup creation for subcloud {subcloud} completed.", + timeout=timeout, + polling_sleep_time=check_interval, + ) + + def delete_subcloud_backup( + self, + con_ssh: SSHConnection, + path: str, + release: str, + subcloud: Optional[str] = None, + local_only: bool = False, + group: Optional[str] = None, + sysadmin_password: str = None, + ) -> None: + """ + Sends the command to delete the backup of the specified subcloud and waits for confirmation of its deletion. + + Args: + con_ssh (SSHConnection): SSH connection to execute the command (central_ssh or subcloud_ssh). + path (str): The path where the backup file is located. + release (str): Required to delete a release backup. + subcloud (Optional[str]): The name of the subcloud to delete the backup. Defaults to None. + local_only (bool): If True, only deletes the local backup in the subcloud. Defaults to False. + group (Optional[str]): Subcloud group name to delete backup. Defaults to None. + sysadmin_password (str): Subcloud sysadmin password needed for deletion on local_path. Defaults to None. + + Returns: + None: + """ + # Command construction for backup deletion + cmd = f"dcmanager subcloud-backup delete {release}" + if subcloud: + cmd += f" --subcloud {subcloud}" + if local_only: + cmd += " --local-only" + if group: + cmd += f" --group {group}" + if sysadmin_password: + cmd += f" --sysadmin-password {sysadmin_password}" + + self.ssh_connection.send(source_openrc(cmd)) + self.validate_success_return_code(self.ssh_connection) + + # Call wait_for_backup_deletion method to wait and verify the backup deletion. + self.wait_for_backup_deletion(con_ssh, path, subcloud) + + def wait_for_backup_deletion( + self, con_ssh: SSHConnection, path: str, subcloud: str + ) -> None: + """ + Waits for the backup to be deleted by checking for the absence of the backup file. + + Args: + con_ssh (SSHConnection): SSH connection object to execute the command. + path (str): The path where the backup file was located. + subcloud (str): The name of the subcloud to delete the backup. + + Returns: + None: + """ + + def check_backup_deleted() -> str: + """ + Checks if the backup has been deleted. + + Returns: + str: Confirmation message if the backup is deleted, otherwise an error message. + """ + check_file = FileKeywords(con_ssh).validate_file_exists_with_sudo(path) + if not check_file: + return f"Backup successfully deleted from {path} for {subcloud}" + else: + return f"Backup still exists at {path}." + + # Using validate_equals_with_retry to ensure the backup is deleted. + validate_equals_with_retry( + function_to_execute=check_backup_deleted, + expected_value=f"Backup successfully deleted from {path} for {subcloud}", + validation_description=f"Backup deletion for subcloud {subcloud} completed.", + ) diff --git a/keywords/files/file_keywords.py b/keywords/files/file_keywords.py index 326361d9..d859b529 100644 --- a/keywords/files/file_keywords.py +++ b/keywords/files/file_keywords.py @@ -33,10 +33,10 @@ class FileKeywords(BaseKeyword): sftp_client.get(remote_file_path, local_file_path) except Exception as e: get_logger().log_error( - f'Exception while downloading remote file [{remote_file_path}] to [{local_file_path}]. {e}' + f"Exception while downloading remote file [{remote_file_path}] to [{local_file_path}]. {e}" ) raise KeywordException( - f'Exception while downloading remote file [{remote_file_path}] to [{local_file_path}]. {e}' + f"Exception while downloading remote file [{remote_file_path}] to [{local_file_path}]. {e}" ) return True @@ -66,10 +66,10 @@ class FileKeywords(BaseKeyword): sftp_client.put(local_file_path, remote_file_path) except Exception as e: get_logger().log_error( - f'Exception while uploading local file [{local_file_path}] to [{remote_file_path}]. {e}' + f"Exception while uploading local file [{local_file_path}] to [{remote_file_path}]. {e}" ) raise KeywordException( - f'Exception while uploading local file [{local_file_path}] to [{remote_file_path}]. {e}' + f"Exception while uploading local file [{local_file_path}] to [{remote_file_path}]. {e}" ) return True @@ -102,7 +102,7 @@ class FileKeywords(BaseKeyword): Returns: bool: True if delete successful, False otherwise. """ - self.ssh_connection.send_as_sudo(f'rm {file_name}') + self.ssh_connection.send_as_sudo(f"rm {file_name}") return self.file_exists(file_name) def get_files_in_dir(self, file_dir: str) -> list[str]: @@ -137,7 +137,7 @@ class FileKeywords(BaseKeyword): end_line = 10000 # we can handle 10000 lines without issue end_time = time.time() + 300 - grep_arg = '' + grep_arg = "" if grep_pattern: grep_arg = f"| grep {grep_pattern}" @@ -152,3 +152,31 @@ class FileKeywords(BaseKeyword): end_line = end_line + 10000 return total_output + + def validate_file_exists_with_sudo(self, path: str) -> bool: + """ + Validates whether a file or directory exists at the specified path using sudo. + + Args: + path (str): The path to the file or directory. + + Returns: + bool: True if the file/directory exists, False otherwise. + + Raises: + KeywordException: If there is an error executing the SSH command. + """ + try: + cmd = f"find {path} -mtime 0" + output = self.ssh_connection.send_as_sudo(cmd) + + # Handle encoding issues + output = "".join( + [line.replace("‘", "").replace("’", "") for line in output] + ) + + return "No such file or directory" not in output + + except Exception as e: + get_logger().log_error(f"Failed to check file existence at {path}: {e}") + raise KeywordException(f"Failed to check file existence at {path}: {e}") diff --git a/testcases/cloud_platform/regression/dc/backup_restore/__init__.py b/testcases/cloud_platform/regression/dc/backup_restore/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/testcases/cloud_platform/regression/dc/backup_restore/test_delete_subcloud_backup.py b/testcases/cloud_platform/regression/dc/backup_restore/test_delete_subcloud_backup.py new file mode 100644 index 00000000..f77e65a4 --- /dev/null +++ b/testcases/cloud_platform/regression/dc/backup_restore/test_delete_subcloud_backup.py @@ -0,0 +1,150 @@ +from pytest import mark + +from config.configuration_manager import ConfigurationManager +from framework.logging.automation_logger import get_logger +from keywords.cloud_platform.dcmanager.dcmanager_subcloud_backup_keywords import ( + DcManagerSubcloudBackupKeywords, +) +from keywords.cloud_platform.dcmanager.dcmanager_subcloud_list_keywords import ( + DcManagerSubcloudListKeywords, +) +from keywords.cloud_platform.rest.bare_metal.hosts.get_hosts_keywords import ( + GetHostsKeywords, +) +from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords +from keywords.cloud_platform.system.host.system_host_list_keywords import ( + SystemHostListKeywords, +) + + +@mark.p2 +@mark.lab_has_subcloud +def test_delete_backup_central(request): + """ + Verify delete centralized subcloud backup + + Test Steps: + - Create a Subcloud backup and check it on central path + - Delete the backup created and the backup is deleted + Teardown: + - Remove files created while the Tc was running. + + """ + central_ssh = LabConnectionKeywords().get_active_controller_ssh() + host = SystemHostListKeywords(central_ssh).get_active_controller().get_host_name() + host_show_output = GetHostsKeywords().get_hosts().get_system_host_show_object(host) + + # Gets the lowest subcloud (the subcloud with the lowest id). + dcmanager_subcloud_list_keywords = DcManagerSubcloudListKeywords(central_ssh) + lowest_subcloud = ( + dcmanager_subcloud_list_keywords.get_dcmanager_subcloud_list().get_healthy_subcloud_with_lowest_id() + ) + subcloud_name = lowest_subcloud.get_name() + subcloud_ssh = LabConnectionKeywords().get_subcloud_ssh(subcloud_name) + + # Gets the lowest subcloud sysadmin password needed for backup creation. + lab_config = ConfigurationManager.get_lab_config().get_subcloud(subcloud_name) + subcloud_password = lab_config.get_admin_credentials().get_password() + + # Get the sw_version if available (used in vbox environments). + release = host_show_output.get_sw_version() + # If sw_version is not available, fall back to software_load (used in physical labs). + if not release: + release = host_show_output.get_software_load() + + dc_manager_backup = DcManagerSubcloudBackupKeywords(central_ssh) + + # Path to where the backup file will store. + central_path = f"/opt/dc-vault/backups/{subcloud_name}/{release}" + + def teardown(): + get_logger().log_info("Removing test files during teardown") + central_ssh.send_as_sudo("rm -r -f /opt/dc-vault/backups/") + subcloud_ssh.send_as_sudo("rm -r -f /opt/platform-backup/backups/") + + request.addfinalizer(teardown) + + # Create a sbcloud backup + get_logger().log_info(f"Create {subcloud_name} backup on Central Cloud") + dc_manager_backup.create_subcloud_backup( + subcloud_password, central_ssh, central_path, subcloud=subcloud_name + ) + + # Delete the backup created + get_logger().log_info(f"Delete {subcloud_name} backup on Central Cloud") + dc_manager_backup.delete_subcloud_backup( + central_ssh, central_path, release, subcloud=subcloud_name + ) + + +@mark.p2 +@mark.lab_has_subcloud +def test_delete_backup_local(request): + """ + Verify delete subcloud backup on local path + + Test Steps: + - Create a Subcloud backup and check it on local path + - Delete the backup created and verify the backup is deleted + Teardown: + - Remove files created while the Tc was running. + + """ + central_ssh = LabConnectionKeywords().get_active_controller_ssh() + host = SystemHostListKeywords(central_ssh).get_active_controller().get_host_name() + host_show_output = GetHostsKeywords().get_hosts().get_system_host_show_object(host) + + # Gets the lowest subcloud (the subcloud with the lowest id). + dcmanager_subcloud_list_keywords = DcManagerSubcloudListKeywords(central_ssh) + lowest_subcloud = ( + dcmanager_subcloud_list_keywords.get_dcmanager_subcloud_list().get_healthy_subcloud_with_lowest_id() + ) + subcloud_name = lowest_subcloud.get_name() + subcloud_ssh = LabConnectionKeywords().get_subcloud_ssh(subcloud_name) + + # Gets the lowest subcloud sysadmin password needed for backup backup creation and deletion on local_path. + lab_config = ConfigurationManager.get_lab_config().get_subcloud(subcloud_name) + subcloud_password = lab_config.get_admin_credentials().get_password() + + # Get the sw_version if available (used in vbox environments). + release = host_show_output.get_sw_version() + # If sw_version is not available, fall back to software_load (used in physical labs). + if not release: + release = host_show_output.get_software_load() + + dc_manager_backup = DcManagerSubcloudBackupKeywords(central_ssh) + + # Path to where the backup file will store. + local_path = ( + f"/opt/platform-backup/backups/{release}/{subcloud_name}_platform_backup_*.tgz" + ) + + def teardown(): + get_logger().log_info("Removing test files during teardown") + subcloud_ssh.send_as_sudo("rm -r -f /opt/platform-backup/backups/") + + request.addfinalizer(teardown) + + # Create a subcloud backup on local + get_logger().log_info(f"Create {subcloud_name} backup on local") + dc_manager_backup.create_subcloud_backup( + subcloud_password, + subcloud_ssh, + local_path, + subcloud=subcloud_name, + local_only=True, + ) + + # path where the backup directory should be checked for deletion. + path = f"/opt/platform-backup/backups/{release}/" + + # Delete the backup created on subcloud + get_logger().log_info(f"Delete {subcloud_name} backup on Central Cloud") + dc_manager_backup.delete_subcloud_backup( + subcloud_ssh, + path, + release, + subcloud=subcloud_name, + local_only=True, + sysadmin_password=subcloud_password, + )