starlingx/test
Code Issues Proposed changes

BNR SX Automation-Part1

Browse Source 
Added common keywords required for backup

Jira: CGTS-73831

Change-Id: I81bc0c903464329c2452c4649774c5df649fcee8
Signed-off-by: satapathy Durga <durgaprasad.satapathy@windriver.com>
This commit is contained in:
satapathy Durga
2025-03-27 12:02:07 -04:00
parent 6f75a949a3
commit 7d3eca2960
5 changed files with 181 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
keywords/cloud_platform/ansible_playbook/ansible_playbook_keywords.py Normal file
View File

@@ -0,0 +1,43 @@
from config.configuration_manager import ConfigurationManager
from framework.logging.automation_logger import get_logger
from keywords.base_keyword import BaseKeyword
from keywords.cloud_platform.ansible_playbook.object.ansible_playbook_backup_output import AnsiblePlaybookBackUpOutput
class AnsiblePlaybookKeywords(BaseKeyword):
"""Provides keyword functions for ansible playbook commands."""
def __init__(self, ssh_connection: str):
"""Initializes AnsiblePlaybookKeywords with an SSH connection.
Args:
ssh_connection (str): SSH connection to the target system.
"""
self.ssh_connection = ssh_connection
def ansible_playbook_backup(self, backup_dir: str, backup_registry: bool = False) -> bool:
"""
Executes the `ansible-playbook` backup command and returns the parsed output.
Args:
backup_dir (str): backuo playbook path
backup_registry (bool): backup registry
Returns:
bool: Parsed output to verify successful backup
"""
backup_playbook_path = "/usr/share/ansible/stx-ansible/playbooks/backup.yml"
backup_registry_argument = ""
if backup_registry:
backup_registry_argument = '-e "backup_registry_filesystem=true"'
admin_password = ConfigurationManager.get_lab_config().get_admin_credentials().get_password()
command = f'ansible-playbook {backup_playbook_path} -e "ansible_become_pass={admin_password}" -e "admin_password={admin_password}" -e "backup_dir={backup_dir}" {backup_registry_argument}'
cmd_out = self.ssh_connection.send(command, reconnect_timeout=3600)
self.validate_success_return_code(self.ssh_connection)
get_logger().log_info("get ansible playbook backup output")
backup_output = AnsiblePlaybookBackUpOutput(cmd_out)
backup_output.validate_ansible_playbook_backup_result()
return backup_output

47
keywords/cloud_platform/ansible_playbook/backup_files_upload_keywords.py Normal file
View File

@@ -0,0 +1,47 @@
import os
from framework.logging.automation_logger import get_logger
from keywords.base_keyword import BaseKeyword
from keywords.files.file_keywords import FileKeywords
class BackUpFilesUploadKeywords(BaseKeyword):
"""Provides keyword functions for ansible playbook commands."""
def __init__(self, ssh_connection: str):
"""Initializes AnsiblePlaybookKeywords with an SSH connection.
Args:
ssh_connection (str): SSH connection to the target system.
"""
self.ssh_connection = ssh_connection
def backup_file(self, backup_file_path: str, local_backup_folder_path: str) -> bool:
"""
Check backup files present in backup playbook path scp backup files to local machine.
Args:
backup_file_path (str): backup file path
local_backup_folder_path (str): Local backup folder path
Returns:
bool: Parsed output to verify successful backup
"""
get_logger().log_info("create a temporary directory in local server to copy the backup files")
os.makedirs(local_backup_folder_path, exist_ok=True)
get_logger().log_info("Check backup files present in backup playbook path")
cmd = f"ls -tr {backup_file_path} | grep --color=never backup"
backup_files = self.ssh_connection.send(cmd)
self.validate_success_return_code(self.ssh_connection)
file_download_status = False
for backup_file in backup_files:
if "backup" in backup_file:
backup_src_path = (os.path.join(backup_file_path, backup_file)).strip()
backup_dest_path = os.path.join(local_backup_folder_path, backup_file).strip()
get_logger().log_info(f"Upload backup file from {backup_src_path} to local {backup_dest_path} path")
file_download_status = FileKeywords(self.ssh_connection).download_file(backup_src_path, backup_dest_path)
return file_download_status

51
keywords/cloud_platform/ansible_playbook/object/ansible_playbook_backup_output.py Normal file
View File

@@ -0,0 +1,51 @@
from framework.logging.automation_logger import get_logger
class AnsiblePlaybookBackUpOutput:
"""
This class parses the output of 'Ansible-playbook backup' command to verify the output.
"""
def __init__(self, cmd_output: str):
"""
Constructor
Args:
cmd_output (str): Output of the 'ansible-playbook backup <cmd>' command.
"""
self.cmd_output = cmd_output
def validate_ansible_playbook_backup_result(self) -> bool:
"""
Checks if the output contains all the expected values.
Returns:
bool: True if the output contains all required fields, False otherwise.
sample output format:
PLAY RECAP ******************************************************************************************************************************************
localhost : ok=274 changed=151 unreachable=0 failed=0 skipped=270 rescued=0 ignored=0
Thursday 10 April 2025 18:08:50 +0000 (0:00:00.020) 0:25:50.203 ********
===============================================================================
common/push-docker-images : Download images and push to local registry --------------------------------------------------------------------- 450.63s
optimized-restore/apply-manifest : Applying puppet restore manifest ------------------------------------------------------------------------ 136.43s
optimized-restore/apply-manifest : Create puppet hieradata runtime configuration ------------------------------------------------------------ 80.92s
"""
successful_backup = False
output = "".join(self.cmd_output)
if output and len(output) > 0:
lastlines = output[output.rfind("PLAY RECAP") :].splitlines()
result_line = [line for line in lastlines if "PLAY RECAP" not in line]
result_line = result_line[0] if len(result_line) > 0 else None
get_logger().log_info(f"Ansible result line = {result_line}")
if result_line and ":" in result_line:
result = result_line.split(":")[1].split()
failed = [i for i in result if "failed" in i]
if failed:
if int(failed[0].split("=")[1]) == 0:
successful_backup = True
get_logger().log_info(f"successful backup : {successful_backup}")
return successful_backup

40
testcases/cloud_platform/backup_and_restore/test_system_backup.py Normal file
View File

@@ -0,0 +1,40 @@
from pytest import mark
from framework.logging.automation_logger import get_logger
from framework.validation.validation import validate_equals
from keywords.cloud_platform.ansible_playbook.ansible_playbook_keywords import AnsiblePlaybookKeywords
from keywords.cloud_platform.ansible_playbook.backup_files_upload_keywords import BackUpFilesUploadKeywords
from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords
from keywords.files.file_keywords import FileKeywords
@mark.p0
def test_backup():
"""
Test system backup using ansible playbook
Test Steps:
- Take a system backup
- Verify backup is successful
- copy to local test server
"""
backup_dir = "/opt/backups"
local_backup_folder_path = "/tmp/bnr"
ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
get_logger().log_info("Delete old backup files if present in back up directory")
backup_files = FileKeywords(ssh_connection).get_files_in_dir(backup_dir)
for backup_file in backup_files:
if "backup" in backup_file:
get_logger().log_info(f"Deleting old backup file {backup_file}")
file_exists_post_deletion = FileKeywords(ssh_connection).delete_file(f"{backup_dir}/{backup_file}")
validate_equals(file_exists_post_deletion, False, "Old Back up file deletion")
get_logger().log_info("Run backup ansible playbook")
ansible_playbook_backup_output = AnsiblePlaybookKeywords(ssh_connection).ansible_playbook_backup(backup_dir)
validate_equals(ansible_playbook_backup_output, True, "Ansible backup command execution")
backup_file_upload_status = BackUpFilesUploadKeywords(ssh_connection).backup_file(backup_dir, local_backup_folder_path)
validate_equals(backup_file_upload_status, True, "Backup file upload to local directory")

0
testcases/cloud_platform/backup_and_restore/test_system_restore.py Normal file
View File