starlingx/test
Code Issues Proposed changes

Add N-1 subcloud upgrade test and updates

Browse Source 
- Add test_upgrade_subcloud.py: New test for upgrading N-1 release
  subclouds to current release
- Enhance dcmanager_prestage_strategy_keywords.py: Add reusable
  methods to reduce code duplication
- Enhance dcmanager_sw_deploy_strategy_keywords.py: Add reusable
  methods to reduce code duplication
- Update dcmanager_subcloud_list_output.py: Add methods to get
  subclouds by release version
- Update test_sw_patch_release_orchestration.py: Refactor to
  use new reusable methods

Change-Id: Iaaece259020c97e4b3762c8e97361aa69f72e78b
Signed-off-by: Abhishek jaiswal <abhishek.jaiswal@windriver.com>
This commit is contained in:
Abhishek jaiswal
2025-06-20 11:34:07 -04:00
parent b4b7aff7fc
commit f5442c2ec1
6 changed files with 183 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
keywords/cloud_platform/dcmanager/dcmanager_prestage_strategy_keywords.py
View File

@@ -1,10 +1,12 @@
import time
from config.configuration_manager import ConfigurationManager
from framework.logging.automation_logger import get_logger
from framework.ssh.ssh_connection import SSHConnection
from framework.validation.validation import validate_equals_with_retry
from framework.validation.validation import validate_equals, validate_equals_with_retry
from keywords.base_keyword import BaseKeyword
from keywords.cloud_platform.command_wrappers import source_openrc
from keywords.cloud_platform.dcmanager.dcmanager_strategy_step_keywords import DcmanagerStrategyStepKeywords
from keywords.cloud_platform.dcmanager.objects.dcmanager_prestage_strategy_object import DcmanagerPrestageStrategyObject
from keywords.cloud_platform.dcmanager.objects.dcmanager_prestage_strategy_show_output import DcmanagerPrestageStrategyShowOutput
@@ -42,18 +44,20 @@ class DcmanagerPrestageStrategyKeywords(BaseKeyword):
# wait for apply to complete
return self.wait_for_state(["complete", "failed"])
def get_dcmanager_prestage_strategy_create(self, sysadmin_password: str, sw_deploy: bool = True) -> DcmanagerPrestageStrategyShowOutput:
def get_dcmanager_prestage_strategy_create(self, sysadmin_password: str, sw_deploy: bool = True, subcloud_name: str = None) -> DcmanagerPrestageStrategyShowOutput:
"""Gets the prestage-strategy create.
Args:
sysadmin_password (str): The password for the sysadmin user.
sw_deploy (bool): If True, include the --for-sw-deploy argument in the command.
subcloud_name (str): The subcloud name.
Returns:
DcmanagerPrestageStrategyShowOutput: The output of the prestage strategy.
"""
sw_deploy_arg = "--for-sw-deploy" if sw_deploy else ""
command = source_openrc(f"dcmanager prestage-strategy create {sw_deploy_arg} --sysadmin-password {sysadmin_password}")
subcloud_name_arg = subcloud_name if subcloud_name else ""
command = source_openrc(f"dcmanager prestage-strategy create {sw_deploy_arg} --sysadmin-password {sysadmin_password} {subcloud_name_arg}")
output = self.ssh_connection.send(command)
self.validate_success_return_code(self.ssh_connection)
return DcmanagerPrestageStrategyShowOutput(output)
@@ -102,5 +106,26 @@ class DcmanagerPrestageStrategyKeywords(BaseKeyword):
output = self.get_dcmanager_prestage_strategy_show().get_dcmanager_prestage_strategy()
if output.get_state() in state:
return output
else:
# execute the dcmanager strategy-step list for monitoring
DcmanagerStrategyStepKeywords(self.ssh_connection).get_dcmanager_strategy_step_list()
time.sleep(interval)
raise TimeoutError(f"Timed out waiting for prestage-strategy to reach state '{state}' after {timeout} seconds.")
def dc_manager_prestage_strategy_create_apply_delete(self, sw_deploy: bool = True, subcloud_name: str = None):
"""Creates, applies, and deletes a prestage strategy.
Args:
sw_deploy (bool): If True, the strategy will be created with the --for-sw-deploy argument.
subcloud_name (str): The subcloud name.
"""
get_logger().log_info("Starting the prestage strategy creation, application, and deletion process.")
sysadmin_password = ConfigurationManager.get_lab_config().get_admin_credentials().get_password()
get_logger().log_test_case_step("Create the prestage strategy")
prestage_strategy_out = self.get_dcmanager_prestage_strategy_create(sysadmin_password=sysadmin_password, sw_deploy=sw_deploy, subcloud_name=subcloud_name)
get_logger().log_info(f"Created prestage strategy state: {prestage_strategy_out.get_dcmanager_prestage_strategy().get_state()}")
get_logger().log_test_case_step("Apply the strategy")
dcman_obj = self.get_dcmanager_prestage_strategy_apply()
validate_equals(dcman_obj.get_state(), "complete", "prestage strategy applied successfully.")
get_logger().log_test_case_step("Delete the strategy")
self.get_dcmanager_prestage_strategy_delete()

34
keywords/cloud_platform/dcmanager/dcmanager_subcloud_list_keywords.py
View File

@@ -1,10 +1,13 @@
import time
from typing import List
from framework.logging.automation_logger import get_logger
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.cloud_platform.dcmanager.dcmanager_subcloud_show_keywords import DcManagerSubcloudShowKeywords
from keywords.cloud_platform.dcmanager.objects.dcmanager_subcloud_list_object import DcManagerSubcloudListObject
from keywords.cloud_platform.dcmanager.objects.dcmanager_subcloud_list_output import DcManagerSubcloudListOutput
@@ -112,3 +115,34 @@ class DcManagerSubcloudListKeywords(BaseKeyword):
return actual_sync_status
validate_equals_with_retry(get_sync, expected_sync_status, f"Sync status of {subcloud_name}", timeout=1200)
def get_all_subcloud_by_release(self, release: str) -> List[DcManagerSubcloudListObject]:
"""Fetch all subclouds by release.
Args:
release (str): The release version to filter subclouds by.
Returns:
List[DcManagerSubcloudListObject]: A list of subcloud objects that match the specified release.
"""
sc_list_with_n_1_release = []
# fetch all the subclouds from the system where the software version matches the release
for sc in self.get_dcmanager_subcloud_list().get_dcmanager_subcloud_list_objects():
subcloud_show_object = DcManagerSubcloudShowKeywords(self.ssh_connection).get_dcmanager_subcloud_show(sc.get_name()).get_dcmanager_subcloud_show_object()
if subcloud_show_object.get_software_version() == release:
sc_list_with_n_1_release.append(sc)
return sc_list_with_n_1_release
def get_one_subcloud_by_release(self, release: str) -> DcManagerSubcloudListObject:
"""Fetch one subcloud by release.
Args:
release (str): The release version to filter subclouds by.
Returns:
DcManagerSubcloudListObject: A subcloud object that matches the specified release.
"""
subclouds = self.get_all_subcloud_by_release(release)
if not subclouds:
raise Exception(f"No subclouds found with release {release}.")
return subclouds[0]

15
keywords/cloud_platform/dcmanager/dcmanager_sw_deploy_strategy_keywords.py
View File

@@ -95,3 +95,18 @@ class DcmanagerSwDeployStrategy(BaseKeyword):
timeout=timeout,
polling_sleep_time=check_interval,
)
def dc_manager_sw_deploy_strategy_create_apply_delete(self, subcloud_name: str, release: str):
"""
Runs dcmanager sw-deploy-strategy create / apply / delete commands.
Args:
subcloud_name (str): The subcloud name.
release (str): The software version to be deployed.
"""
get_logger().log_test_case_step(f"Create the sw-deploy strategy for {subcloud_name} with {release}")
self.dcmanager_sw_deploy_strategy_create(subcloud_name, release)
get_logger().log_test_case_step("Apply the sw-deploy strategy")
self.dcmanager_sw_deploy_strategy_apply(subcloud_name)
get_logger().log_test_case_step("Delete the sw-deploy strategy")
self.dcmanager_sw_deploy_strategy_delete()

12
keywords/cloud_platform/dcmanager/objects/dcmanager_subcloud_list_object.py
View File

@@ -1,5 +1,6 @@
class DcManagerSubcloudListObject:
"""
"""DcManagerSubcloudListObject.
This class represents a subcloud summary as an object.
This is typically a line in the 'dcmanager subcloud list' command output table, as shown below.
@@ -123,3 +124,12 @@ class DcManagerSubcloudListObject:
Setter for the Subcloud Prestage Status
"""
self.prestage_status = prestage_status
def __repr__(self) -> str:
"""
Return the developer-facing representation of the object.
Returns:
str: Class name and row count.
"""
return f"{self.__class__.__name__}(Name={self.name})"

48
testcases/cloud_platform/patching/dc/orchestrator/test_sw_patch_release_orchestration.py
View File

@@ -8,7 +8,6 @@ from keywords.cloud_platform.dcmanager.dcmanager_prestage_strategy_keywords impo
from keywords.cloud_platform.dcmanager.dcmanager_strategy_step_keywords import DcmanagerStrategyStepKeywords
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.dcmanager.dcmanager_subcloud_show_keywords import DcManagerSubcloudShowKeywords
from keywords.cloud_platform.dcmanager.dcmanager_sw_deploy_strategy_keywords import DcmanagerSwDeployStrategy
from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords
from keywords.cloud_platform.swmanager.swmanager_sw_deploy_strategy_keywords import SwManagerSwDeployStrategyKeywords
@@ -89,52 +88,17 @@ def swman_sw_deploy_strategy_create_apply(release: str):
validate_equals(swman_obj.get_current_phase_completion(), "100%", "deploy strategy completed successfully.")
def dcman_prestage_strategy_create_apply():
"""Helper function to create and apply a prestage strategy."""
sysadmin_password = ConfigurationManager.get_lab_config().get_admin_credentials().get_password()
central_ssh = LabConnectionKeywords().get_active_controller_ssh()
dcman_prestage_kw = DcmanagerPrestageStrategyKeywords(central_ssh)
get_logger().log_test_case_step("Create the prestage strategy for the IN-service patch")
prestage_strategy_out = dcman_prestage_kw.get_dcmanager_prestage_strategy_create(sysadmin_password=sysadmin_password, sw_deploy=True)
get_logger().log_info(f"Created prestage strategy state: {prestage_strategy_out.get_dcmanager_prestage_strategy().get_state()}")
get_logger().log_test_case_step("Apply the strategy")
dcman_obj = dcman_prestage_kw.get_dcmanager_prestage_strategy_apply()
validate_equals(dcman_obj.get_state(), "complete", "prestage strategy applied successfully.")
dcman_prestage_kw.get_dcmanager_prestage_strategy_delete()
def dcman_sw_deploy_strategy_create_apply(subcloud_name: str, release: str):
"""Helper function to create and apply a software deploy strategy for a subcloud.
Args:
subcloud_name (str): The name of the subcloud.
release (str): The software release to be used for the strategy.
"""
central_ssh = LabConnectionKeywords().get_active_controller_ssh()
dcman_sw_deploy_kw = DcmanagerSwDeployStrategy(central_ssh)
# before creating the strategy, waith for subcloud software_sync_status to out-of-sync
sc_show_obj = DcManagerSubcloudShowKeywords(central_ssh).wait_for_state(subcloud_name, field="software_sync_status", expected_status="in-sync")
validate_equals(sc_show_obj.get_software_sync_status(), "out-of-sync", "Ready to create the sw-deploy strategy.")
get_logger().log_test_case_step(f"Create the sw-deploy strategy for {subcloud_name} with {release}")
dcman_sw_deploy_kw.dcmanager_sw_deploy_strategy_create(subcloud_name=subcloud_name, sw_version=release)
get_logger().log_test_case_step("Apply the strategy")
dcman_sw_deploy_kw.dcmanager_sw_deploy_strategy_apply(subcloud_name=subcloud_name)
dcman_sw_deploy_kw.dcmanager_sw_deploy_strategy_delete()
def fetch_sw_list(ssh_connection: SSHConnection, log_messgae: str) -> list:
def fetch_sw_list(ssh_connection: SSHConnection, log_message: str) -> list:
"""Fetches the software list from the given SSH connection and logs the information.
Args:
ssh_connection (SSHConnection): The SSH connection to the system controller or subcloud.
log_messgae (str): Optional message to log before fetching the software list.
log_message (str): Optional message to log before fetching the software list.
Returns:
list: A list of software releases.
"""
get_logger().log_test_case_step(log_messgae)
get_logger().log_test_case_step(log_message)
sw_list = SoftwareListKeywords(ssh_connection).get_software_list().get_software_lists()
get_logger().log_info(f"Current software list: {[sw.get_release() for sw in sw_list]}")
return sw_list
@@ -199,13 +163,15 @@ def test_in_service_patch(request):
fetch_sw_list(subcloud_ssh, f"{msg} subcloud ==> {subcloud_name}")
# dcmanager prestage-strategy create / apply / delete
dcman_prestage_strategy_create_apply()
dcman_prestage_kw = DcmanagerPrestageStrategyKeywords(central_ssh)
dcman_prestage_kw.dc_manager_prestage_strategy_create_apply_delete(sw_deploy=True)
msg = "Fetch software list after dcmanager prestage-strategy on "
fetch_sw_list(central_ssh, f"{msg} Systemcontroller")
fetch_sw_list(subcloud_ssh, f"{msg} subcloud ==> {subcloud_name}")
# dcmanager sw-deploy-strategy create / apply / delete
dcman_sw_deploy_strategy_create_apply(subcloud_name=subcloud_name, release=release)
dcman_sw_deploy_kw = DcmanagerSwDeployStrategy(central_ssh)
dcman_sw_deploy_kw.dc_manager_sw_deploy_strategy_create_apply_delete(subcloud_name, release)
msg = "Fetch software list after dcmanager sw-deploy-strategy on "
fetch_sw_list(central_ssh, f"{msg} Systemcontroller")
sw_list = fetch_sw_list(subcloud_ssh, f"{msg} subcloud ==> {subcloud_name}")

88
testcases/cloud_platform/usm/upgrade/test_upgrade_subcloud.py Normal file
View File

@@ -0,0 +1,88 @@
from pytest import mark
from framework.logging.automation_logger import get_logger
from framework.ssh.ssh_connection import SSHConnection
from framework.validation.validation import validate_equals
from keywords.cloud_platform.dcmanager.dcmanager_prestage_strategy_keywords import DcmanagerPrestageStrategyKeywords
from keywords.cloud_platform.dcmanager.dcmanager_subcloud_list_keywords import DcManagerSubcloudListKeywords
from keywords.cloud_platform.dcmanager.dcmanager_sw_deploy_strategy_keywords import DcmanagerSwDeployStrategy
from keywords.cloud_platform.fault_management.alarms.alarm_list_keywords import AlarmListKeywords
from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords
from keywords.cloud_platform.upgrade.software_list_keywords import SoftwareListKeywords
from keywords.cloud_platform.version_info.cloud_platform_version_manager import CloudPlatformVersionManagerClass
def fetch_sw_list(ssh_connection: SSHConnection, log_message: str) -> list:
"""Fetches the software list from the given SSH connection and logs the information.
Args:
ssh_connection (SSHConnection): The SSH connection to the system controller or subcloud.
log_message (str): Optional message to log before fetching the software list.
Returns:
list: A list of software releases.
"""
get_logger().log_test_case_step(log_message)
sw_list = SoftwareListKeywords(ssh_connection).get_software_list().get_software_lists()
get_logger().log_info(f"Current software list: {[sw.get_release() for sw in sw_list]}")
return sw_list
@mark.p2
@mark.lab_has_subcloud
def test_upgrade_n_1_subcloud():
"""
Upgrade the N-1 release subcloud to N release and verify the upgrade was successful.
Assumes that:
- One Subcloud is added with release N-1.
- The subcloud is online and reachable.
"""
get_logger().log_info("Starting N-1 upgrade test for subcloud.")
central_ssh = LabConnectionKeywords().get_active_controller_ssh()
dcm_sc_list_kw = DcManagerSubcloudListKeywords(central_ssh)
cp_version_manager_obj = CloudPlatformVersionManagerClass()
# latest release is the one we are currently running
latest_release = cp_version_manager_obj.get_sw_version().get_name()
# get n-1 release
n_1_release = cp_version_manager_obj.get_last_major_release().get_name()
# get the subcloud with the release
subcloud_obj = dcm_sc_list_kw.get_one_subcloud_by_release(n_1_release)
subcloud_name = subcloud_obj.get_name()
# get subcloud ssh
subcloud_ssh = LabConnectionKeywords().get_subcloud_ssh(subcloud_name)
msg = "Fetch software list before prestage-strategy "
fetch_sw_list(central_ssh, f"{msg} Systemcontroller")
fetch_sw_list(subcloud_ssh, f"{msg} subcloud ==> {subcloud_name}")
AlarmListKeywords(subcloud_ssh).wait_for_all_alarms_cleared()
# dcmanager prestage-strategy create / apply / delete
dcman_prestage_kw = DcmanagerPrestageStrategyKeywords(central_ssh)
dcman_prestage_kw.dc_manager_prestage_strategy_create_apply_delete(sw_deploy=True, subcloud_name=subcloud_name)
msg = "Fetch software list after dcmanager prestage-strategy on "
fetch_sw_list(central_ssh, f"{msg} Systemcontroller")
fetch_sw_list(subcloud_ssh, f"{msg} subcloud ==> {subcloud_name}")
# find out the latest release version from the list of software releases
# latest_release wil have the number part of the release e.g. 25.09
# but we need full name as displayed in software list
sw_list = SoftwareListKeywords(central_ssh).get_software_list().get_software_lists()
full_sw_name = ""
for sw in sw_list:
if latest_release in sw.get_release():
full_sw_name = sw.get_release()
if not full_sw_name:
get_logger().log_error(f"Release {latest_release} not found in the software list.")
raise Exception(f"Release {latest_release} not found in the software list.")
# dcmanager sw-deploy-strategy create / apply / delete
dcman_sw_deploy_kw = DcmanagerSwDeployStrategy(central_ssh)
dcman_sw_deploy_kw.dc_manager_sw_deploy_strategy_create_apply_delete(subcloud_name, full_sw_name)
msg = "Fetch software list after dcmanager sw-deploy-strategy on "
fetch_sw_list(central_ssh, f"{msg} Systemcontroller")
sw_list = fetch_sw_list(subcloud_ssh, f"{msg} subcloud ==> {subcloud_name}")
# verify that the patch is applied on subcloud
for sw in sw_list:
get_logger().log_info(f"Release: {sw.get_release()} is {sw.get_state()}")
validate_equals(sw.get_state(), "deployed", "patch applied successfully.")