Added Test to Deploy In service patch
Change-Id: I777de3a2140d3c5b4bc2e647687bf7d20755f8c9 Signed-off-by: Abhishek jaiswal <abhishek.jaiswal@windriver.com>
This commit is contained in:
Abhishek jaiswal
@@ -1,6 +1,11 @@
|
||||
import time
|
||||
|
||||
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.objects.dcmanager_prestage_strategy_object import DcmanagerPrestageStrategyObject
|
||||
from keywords.cloud_platform.dcmanager.objects.dcmanager_prestage_strategy_show_output import DcmanagerPrestageStrategyShowOutput
|
||||
|
||||
|
||||
@@ -18,28 +23,37 @@ class DcmanagerPrestageStrategyKeywords(BaseKeyword):
|
||||
"""
|
||||
self.ssh_connection = ssh_connection
|
||||
|
||||
def get_dcmanager_prestage_strategy_abort(self) -> None:
|
||||
def get_dcmanager_prestage_strategy_abort(self) -> str:
|
||||
"""Gets the prestage-strategy abort."""
|
||||
command = source_openrc("dcmanager prestage-strategy abort")
|
||||
output = self.ssh_connection.send(command)
|
||||
self.validate_success_return_code(self.ssh_connection)
|
||||
return output
|
||||
|
||||
def get_dcmanager_prestage_strategy_apply(self) -> None:
|
||||
"""Gets the prestage-strategy apply."""
|
||||
command = source_openrc("dcmanager prestage-strategy apply")
|
||||
output = self.ssh_connection.send(command)
|
||||
self.validate_success_return_code(self.ssh_connection)
|
||||
return output
|
||||
def get_dcmanager_prestage_strategy_apply(self) -> DcmanagerPrestageStrategyObject:
|
||||
"""Gets the prestage-strategy apply.
|
||||
|
||||
def get_dcmanager_prestage_strategy_create(self) -> DcmanagerPrestageStrategyShowOutput:
|
||||
Returns:
|
||||
DcmanagerPrestageStrategyObject: An object containing details of the prestage strategy.
|
||||
"""
|
||||
Gets the prestage-strategy create.
|
||||
command = source_openrc("dcmanager prestage-strategy apply")
|
||||
self.ssh_connection.send(command)
|
||||
self.validate_success_return_code(self.ssh_connection)
|
||||
# wait for apply to complete
|
||||
return self.wait_for_state(["complete", "apply-failed"])
|
||||
|
||||
def get_dcmanager_prestage_strategy_create(self, sysadmin_password: str, sw_deploy: bool = True) -> 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.
|
||||
|
||||
Returns:
|
||||
DcmanagerPrestageStrategyShowOutput: The output of the prestage strategy.
|
||||
"""
|
||||
command = source_openrc("dcmanager prestage-strategy create")
|
||||
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}")
|
||||
output = self.ssh_connection.send(command)
|
||||
self.validate_success_return_code(self.ssh_connection)
|
||||
return DcmanagerPrestageStrategyShowOutput(output)
|
||||
@@ -56,14 +70,37 @@ class DcmanagerPrestageStrategyKeywords(BaseKeyword):
|
||||
self.validate_success_return_code(self.ssh_connection)
|
||||
return DcmanagerPrestageStrategyShowOutput(output)
|
||||
|
||||
def get_dcmanager_prestage_strategy_delete(self) -> DcmanagerPrestageStrategyShowOutput:
|
||||
"""
|
||||
Gets the prestage-strategy delete.
|
||||
|
||||
Returns:
|
||||
DcmanagerPrestageStrategyShowOutput: An object containing details of the prestage strategy .
|
||||
"""
|
||||
def check_dcmanager_prestage_strategy_delete(self):
|
||||
"""Gets the prestage-strategy delete."""
|
||||
command = source_openrc("dcmanager prestage-strategy delete")
|
||||
output = self.ssh_connection.send(command)
|
||||
self.validate_success_return_code(self.ssh_connection)
|
||||
return DcmanagerPrestageStrategyShowOutput(output)
|
||||
if "Unable to delete strategy of type" in "".join(output):
|
||||
get_logger().log_info("No strategy to be deleted, moving on...")
|
||||
return True
|
||||
return False
|
||||
|
||||
def get_dcmanager_prestage_strategy_delete(self):
|
||||
"""
|
||||
Starts sw-deploy-strategy deletion process if there is a strategy in progress and waits for its deletion.
|
||||
"""
|
||||
validate_equals_with_retry(function_to_execute=self.check_dcmanager_prestage_strategy_delete, expected_value=True, validation_description="Waits for strategy deletion", timeout=600, polling_sleep_time=10)
|
||||
|
||||
def wait_for_state(self, state: list, timeout: int = 1800) -> DcmanagerPrestageStrategyObject:
|
||||
"""Waits for the prestage-strategy to reach a specific state.
|
||||
|
||||
Args:
|
||||
state (list): The desired state to wait for.
|
||||
timeout (int): The maximum time to wait in seconds.
|
||||
|
||||
Returns:
|
||||
DcmanagerPrestageStrategyObject: The output of the prestage-strategy show command when the desired state is reached.
|
||||
"""
|
||||
# possibles states : initial, deleting, deleting, complete, aborting, applying
|
||||
interval = 10
|
||||
start_time = time.time()
|
||||
while time.time() - start_time < timeout:
|
||||
output = self.get_dcmanager_prestage_strategy_show().get_dcmanager_prestage_strategy()
|
||||
if output.get_state() in state:
|
||||
return output
|
||||
time.sleep(interval)
|
||||
raise TimeoutError(f"Timed out waiting for prestage-strategy to reach state '{state}' after {timeout} seconds.")
|
||||
|
||||
@@ -65,7 +65,7 @@ class DcmanagerSwDeployStrategy(BaseKeyword):
|
||||
"""
|
||||
Starts sw-deploy-strategy deletion process if there is a strategy in progress and waits for its deletion.
|
||||
"""
|
||||
validate_equals_with_retry(function_to_execute=self.check_sw_deploy_strategy_delete_output, expected_value=True, validation_description="Waits for strategy deletion", timeout=60, polling_sleep_time=10)
|
||||
validate_equals_with_retry(function_to_execute=self.check_sw_deploy_strategy_delete_output, expected_value=True, validation_description="Waits for strategy deletion", timeout=240, polling_sleep_time=10)
|
||||
|
||||
def wait_sw_deployment(
|
||||
self,
|
||||
|
||||
@@ -27,19 +27,19 @@ class SwManagerSwDeployStrategyShowOutput:
|
||||
if self.is_valid_output(output_values):
|
||||
self.swmanager_sw_deploy_strategy = SwManagerSwDeployStrategyObject()
|
||||
sw_deploy_strat = output_values["Strategy Software Deploy Strategy"]
|
||||
self.swmanager_sw_deploy_strategy.set_strategy_uuid(sw_deploy_strat["strategy-uuid"])
|
||||
self.swmanager_sw_deploy_strategy.set_release_id(sw_deploy_strat["release-id"])
|
||||
self.swmanager_sw_deploy_strategy.set_controller_apply_type(sw_deploy_strat["controller-apply-type"])
|
||||
self.swmanager_sw_deploy_strategy.set_storage_apply_type(sw_deploy_strat["storage-apply-type"])
|
||||
self.swmanager_sw_deploy_strategy.set_worker_apply_type(sw_deploy_strat["worker-apply-type"])
|
||||
self.swmanager_sw_deploy_strategy.set_default_instance_action(sw_deploy_strat["default-instance-action"])
|
||||
self.swmanager_sw_deploy_strategy.set_alarm_restrictions(sw_deploy_strat["alarm-restrictions"])
|
||||
self.swmanager_sw_deploy_strategy.set_current_phase(sw_deploy_strat["current-phase"])
|
||||
self.swmanager_sw_deploy_strategy.set_current_stage(sw_deploy_strat["current-stage"])
|
||||
self.swmanager_sw_deploy_strategy.set_current_step(sw_deploy_strat["current-step"])
|
||||
self.swmanager_sw_deploy_strategy.set_current_phase_completion(sw_deploy_strat["current-phase-completion"])
|
||||
self.swmanager_sw_deploy_strategy.set_state(sw_deploy_strat["state"])
|
||||
self.swmanager_sw_deploy_strategy.set_inprogress(sw_deploy_strat["inprogress"])
|
||||
self.swmanager_sw_deploy_strategy.set_strategy_uuid(sw_deploy_strat.get("strategy-uuid"))
|
||||
self.swmanager_sw_deploy_strategy.set_release_id(sw_deploy_strat.get("release-id"))
|
||||
self.swmanager_sw_deploy_strategy.set_controller_apply_type(sw_deploy_strat.get("controller-apply-type"))
|
||||
self.swmanager_sw_deploy_strategy.set_storage_apply_type(sw_deploy_strat.get("storage-apply-type"))
|
||||
self.swmanager_sw_deploy_strategy.set_worker_apply_type(sw_deploy_strat.get("worker-apply-type"))
|
||||
self.swmanager_sw_deploy_strategy.set_default_instance_action(sw_deploy_strat.get("default-instance-action"))
|
||||
self.swmanager_sw_deploy_strategy.set_alarm_restrictions(sw_deploy_strat.get("alarm-restrictions"))
|
||||
self.swmanager_sw_deploy_strategy.set_current_phase(sw_deploy_strat.get("current-phase"))
|
||||
self.swmanager_sw_deploy_strategy.set_current_stage(sw_deploy_strat.get("current-stage"))
|
||||
self.swmanager_sw_deploy_strategy.set_current_step(sw_deploy_strat.get("current-step"))
|
||||
self.swmanager_sw_deploy_strategy.set_current_phase_completion(sw_deploy_strat.get("current-phase-completion"))
|
||||
self.swmanager_sw_deploy_strategy.set_state(sw_deploy_strat.get("state"))
|
||||
self.swmanager_sw_deploy_strategy.set_inprogress(sw_deploy_strat.get("inprogress"))
|
||||
else:
|
||||
raise KeywordException(f"The output line {output_values} was not valid")
|
||||
|
||||
@@ -63,7 +63,7 @@ class SwManagerSwDeployStrategyShowOutput:
|
||||
Returns:
|
||||
bool: True if all required fields are present, False otherwise.
|
||||
"""
|
||||
required_fields = ["strategy-uuid", "release-id", "controller-apply-type", "storage-apply-type", "worker-apply-type", "default-instance-action", "alarm-restrictions", "current-phase", "current-stage", "current-step", "current-phase-completion", "state", "inprogress"]
|
||||
required_fields = ["strategy-uuid", "release-id", "controller-apply-type", "storage-apply-type", "worker-apply-type", "default-instance-action", "alarm-restrictions", "current-phase", "current-phase-completion", "state"]
|
||||
if "Strategy Software Deploy Strategy" not in value:
|
||||
get_logger().log_error("Strategy Software Deploy Strategy is not in the output value")
|
||||
return False
|
||||
|
||||
@@ -1,16 +1,22 @@
|
||||
import time
|
||||
|
||||
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.objects.dcmanager_prestage_strategy_object import DcmanagerPrestageStrategyObject
|
||||
from keywords.cloud_platform.swmanager.objects.swmanager_sw_deploy_strategy_object import SwManagerSwDeployStrategyObject
|
||||
from keywords.cloud_platform.swmanager.objects.swmanager_sw_deploy_strategy_show_output import SwManagerSwDeployStrategyShowOutput
|
||||
|
||||
|
||||
class SwmanagerSwDeployStrategyKeywords(BaseKeyword):
|
||||
class SwManagerSwDeployStrategyKeywords(BaseKeyword):
|
||||
"""
|
||||
This class contains all the keywords related to the 'sw-manager sw-deploy-strategy' commands.
|
||||
"""
|
||||
|
||||
def __init__(self, ssh_connection: SSHConnection) -> None:
|
||||
"""Initializes SwmanagerSwDeployStrategyKeywords.
|
||||
"""Initializes SwManagerSwDeployStrategyKeywords.
|
||||
|
||||
Args:
|
||||
ssh_connection (SSHConnection): The SSH connection object used for executing commands.
|
||||
@@ -23,31 +29,35 @@ class SwmanagerSwDeployStrategyKeywords(BaseKeyword):
|
||||
self.ssh_connection.send(command)
|
||||
self.validate_success_return_code(self.ssh_connection)
|
||||
|
||||
def get_sw_deploy_strategy_apply(self):
|
||||
"""Gets the sw-deploy-strategy apply."""
|
||||
def get_sw_deploy_strategy_apply(self) -> DcmanagerPrestageStrategyObject:
|
||||
"""Gets the sw-deploy-strategy apply.
|
||||
|
||||
Returns:
|
||||
DcmanagerPrestageStrategyObject: An object containing details of the sw-deploy strategy.
|
||||
"""
|
||||
command = source_openrc("sw-manager sw-deploy-strategy apply")
|
||||
self.ssh_connection.send(command)
|
||||
self.validate_success_return_code(self.ssh_connection)
|
||||
# wait for apply to complete
|
||||
return self.wait_for_state(["applied", "apply-failed"])
|
||||
|
||||
def get_sw_deploy_strategy_create(self, release: str, delete: bool) -> SwManagerSwDeployStrategyShowOutput:
|
||||
def get_sw_deploy_strategy_create(self, release: str, delete: bool) -> DcmanagerPrestageStrategyObject:
|
||||
"""Gets the sw-deploy-strategy create.
|
||||
|
||||
Args:
|
||||
release (str): The release to be used for the strategy.
|
||||
delete (bool): If True, pass --delete as an argument to the command.
|
||||
release (str): The release to be deployed.
|
||||
delete (bool): If True, include the --delete argument in the command.
|
||||
|
||||
Returns:
|
||||
SwManagerSwDeployStrategyShowOutput: The output of the prestage strategy.
|
||||
DcmanagerPrestageStrategyObject: The output of the sw-deploy strategy.
|
||||
"""
|
||||
if delete:
|
||||
delete_arg = "--delete"
|
||||
else:
|
||||
delete_arg = ""
|
||||
delete_arg = "--delete" if delete else ""
|
||||
|
||||
command = source_openrc(f"sw-manager sw-deploy-strategy create {delete_arg} {release}")
|
||||
output = self.ssh_connection.send(command)
|
||||
self.ssh_connection.send(command)
|
||||
self.validate_success_return_code(self.ssh_connection)
|
||||
return SwManagerSwDeployStrategyShowOutput(output)
|
||||
# wait for build to complete
|
||||
return self.wait_for_state(["ready-to-apply", "build-failed"])
|
||||
|
||||
def get_sw_deploy_strategy_show(self) -> SwManagerSwDeployStrategyShowOutput:
|
||||
"""Gets the sw-deploy-strategy show.
|
||||
@@ -57,16 +67,40 @@ class SwmanagerSwDeployStrategyKeywords(BaseKeyword):
|
||||
"""
|
||||
command = source_openrc("sw-manager sw-deploy-strategy show")
|
||||
output = self.ssh_connection.send(command)
|
||||
self.validate_success_return_code(self.ssh_connection)
|
||||
return SwManagerSwDeployStrategyShowOutput(output)
|
||||
|
||||
def get_sw_deploy_strategy_delete(self) -> None:
|
||||
"""Gets the sw-deploy-strategy delete.
|
||||
def check_sw_deploy_strategy_delete(self) -> bool:
|
||||
"""Checks the output of the sw-deploy-strategy delete command.
|
||||
|
||||
Returns:
|
||||
None: None.
|
||||
bool: True if no strategy exists to be deleted, False if a strategy is in the process of being deleted.
|
||||
"""
|
||||
command = source_openrc("sw-manager sw-deploy-strategy delete")
|
||||
output = self.ssh_connection.send(command)
|
||||
self.validate_success_return_code(self.ssh_connection)
|
||||
return output
|
||||
if "Nothing to delete" in "".join(output):
|
||||
get_logger().log_info("No sw-deploy-strategy to be deleted, moving on...")
|
||||
return True
|
||||
return False
|
||||
|
||||
def get_sw_deploy_strategy_delete(self) -> None:
|
||||
"""Starts sw-deploy-strategy deletion process if there is a strategy in progress and waits for its deletion."""
|
||||
validate_equals_with_retry(function_to_execute=self.check_sw_deploy_strategy_delete, expected_value=True, validation_description="Waits for strategy deletion", timeout=600, polling_sleep_time=10)
|
||||
|
||||
def wait_for_state(self, state: list, timeout: int = 1800) -> SwManagerSwDeployStrategyObject:
|
||||
"""Waits for the sw-deploy-strategy to reach a specific state.
|
||||
|
||||
Args:
|
||||
state (list): The desired state to wait for.
|
||||
timeout (int): The maximum time to wait in seconds.
|
||||
|
||||
Returns:
|
||||
SwManagerSwDeployStrategyObject: The output of the sw-deploy-strategy show command when the desired state is reached.
|
||||
"""
|
||||
interval = 10
|
||||
start_time = time.time()
|
||||
while time.time() - start_time < timeout:
|
||||
output = self.get_sw_deploy_strategy_show().get_swmanager_sw_deploy_strategy_show()
|
||||
if output.get_state() in state:
|
||||
return output
|
||||
time.sleep(interval)
|
||||
raise TimeoutError(f"Timed out waiting for sw-deploy-strategy to reach state '{state}' after {timeout} seconds.")
|
||||
|
||||
@@ -2,13 +2,17 @@ from pytest import fail, mark
|
||||
|
||||
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
|
||||
from keywords.cloud_platform.dcmanager.dcmanager_prestage_strategy_keywords import DcmanagerPrestageStrategyKeywords
|
||||
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_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
|
||||
from keywords.cloud_platform.upgrade.software_list_keywords import SoftwareListKeywords
|
||||
from keywords.cloud_platform.upgrade.usm_keywords import USMKeywords
|
||||
|
||||
|
||||
@mark.p2
|
||||
@@ -60,3 +64,150 @@ def test_patch_apply(request):
|
||||
|
||||
# Verify that the strategy was applied correctly
|
||||
validate_equals(strategy_status, "complete", f"Software deploy completed successfully for subcloud {subcloud_name}.")
|
||||
|
||||
|
||||
def swman_sw_deploy_strategy_create_apply(release: str):
|
||||
"""Helper function to create and apply a software deploy strategy.
|
||||
|
||||
Args:
|
||||
release (str): The software release to be used for the strategy.
|
||||
"""
|
||||
central_ssh = LabConnectionKeywords().get_active_controller_ssh()
|
||||
swman_deploy_kw = SwManagerSwDeployStrategyKeywords(central_ssh)
|
||||
|
||||
get_logger().log_test_case_step("Through the VIM orchestration deploy the patch in the system controller")
|
||||
swman_strat_out = swman_deploy_kw.get_sw_deploy_strategy_create(release=release, delete=True)
|
||||
swman_obj = swman_strat_out.get_swmanager_sw_deploy_strategy_show()
|
||||
get_logger().log_info(f"Created sw-deploy strategy: {swman_obj.get_strategy_uuid()} for release {swman_obj.get_release_id()}")
|
||||
get_logger().log_info(f"release = {release} get_release_id = {swman_obj.get_release_id()}")
|
||||
get_logger().log_test_case_step("Apply the strategy")
|
||||
swman_obj = swman_deploy_kw.get_sw_deploy_strategy_apply()
|
||||
# Verify that the strategy applied
|
||||
validate_equals(swman_obj.get_state(), "applied", "deploy strategy applied successfully.")
|
||||
|
||||
# Verify that the current phase completed successfully
|
||||
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)
|
||||
|
||||
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:
|
||||
"""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.
|
||||
|
||||
Returns:
|
||||
list: A list of software releases.
|
||||
"""
|
||||
get_logger().log_test_case_step(log_messgae)
|
||||
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_in_service_patch(request):
|
||||
"""
|
||||
Verify insvc patch application on subcloud + rehomed
|
||||
|
||||
Test Steps:
|
||||
- perform software list
|
||||
- Upload the IN-service patch in the system controller through command software upload
|
||||
- create and apply the sw-deploy strategy for the IN-service patch using below commands:
|
||||
- sw-manager sw-deploy-strategy create --delete
|
||||
- sw-manager sw-deploy-strategy apply
|
||||
- perform software list on system controller
|
||||
- create and apply the dcmanager prestage strategy using below commands:
|
||||
- dcmanager prestage-strategy create --for-sw-deploy
|
||||
- dcmanager prestage-strategy apply
|
||||
- perform software list on subcloud
|
||||
- delete the dcmanager prestage strategy using below command:
|
||||
- dcmanager prestage-strategy delete
|
||||
- create and apply the dcmanager sw-deploy strategy using below commands:
|
||||
- dcmanager sw-deploy-strategy create <subcloud_name> <release>
|
||||
- dcmanager sw-deploy-strategy apply <subcloud_name>
|
||||
- perform software list on subcloud
|
||||
- delete the dcmanager sw-deploy strategy using below command:
|
||||
- dcmanager sw-deploy-strategy delete
|
||||
"""
|
||||
central_ssh = LabConnectionKeywords().get_active_controller_ssh()
|
||||
# Gets the lowest subcloud (the subcloud with the lowest id).
|
||||
dcman_sc_list_kw = DcManagerSubcloudListKeywords(central_ssh)
|
||||
lowest_subcloud = dcman_sc_list_kw.get_dcmanager_subcloud_list().get_healthy_subcloud_with_lowest_id()
|
||||
subcloud_name = lowest_subcloud.get_name()
|
||||
|
||||
# get subcloud ssh
|
||||
subcloud_ssh = LabConnectionKeywords().get_subcloud_ssh(subcloud_name)
|
||||
msg = "Fetch software list before patching on "
|
||||
fetch_sw_list(central_ssh, f"{msg} Systemcontroller")
|
||||
fetch_sw_list(subcloud_ssh, f"{msg} subcloud ==> {subcloud_name}")
|
||||
|
||||
# Upload the IN-service patch in the system controller
|
||||
usm_keywords = USMKeywords(central_ssh)
|
||||
usm_config = ConfigurationManager.get_usm_config()
|
||||
patch_file_path = usm_config.get_patch_path()
|
||||
get_logger().log_info(f"Uploading IN-service patch file: {patch_file_path}")
|
||||
|
||||
upload_patch_out = usm_keywords.upload_patch_file(patch_file_path)
|
||||
upload_patch_obj = upload_patch_out.get_software_uploaded()
|
||||
if not upload_patch_obj:
|
||||
raise Exception(f"Failed to upload patch file: {patch_file_path}")
|
||||
uploaded_file = upload_patch_obj[0].get_uploaded_file()
|
||||
release = upload_patch_obj[0].get_release()
|
||||
get_logger().log_info(f"Uploaded patch file: {uploaded_file} with release ID: {release}")
|
||||
|
||||
# sw-manager sw-deploy-strategy create / apply / complete
|
||||
swman_sw_deploy_strategy_create_apply(release=release)
|
||||
msg = "Fetch software list after sw-deploy-strategy on "
|
||||
fetch_sw_list(central_ssh, f"{msg} Systemcontroller")
|
||||
fetch_sw_list(subcloud_ssh, f"{msg} subcloud ==> {subcloud_name}")
|
||||
|
||||
# dcmanager prestage-strategy create / apply / delete
|
||||
dcman_prestage_strategy_create_apply()
|
||||
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)
|
||||
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.")
|
||||
|
||||
Reference in New Issue
Block a user