Add tests to verify cert-manager and nginx ingress controller are active and completed
- Enhanced application validation to verify that both 'cert-manager' and 'nginx-ingress-controller' apps are in 'applied' state, marked as 'active', and their
progress status is 'completed'.
- This ensures proper deployment and readiness of platform components on the active controller.
Change-Id: If632facd267c455d2b7524e800c0406f16c9e0a6
Signed-off-by: Thomas Sunil <sunil.thomas@windriver.com>
This commit is contained in:
@@ -0,0 +1,115 @@
|
||||
class SystemApplicationShowObject:
|
||||
"""
|
||||
Represents a system application returned from the 'system application-show' command
|
||||
|
||||
Provides access to key attributes such as status, version, manifest name, etc.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
"""
|
||||
Initialize with an empty property dictionary
|
||||
"""
|
||||
self._properties = {}
|
||||
|
||||
def set_property(self, key: str, value: str):
|
||||
"""
|
||||
Sets a property in the application's properties dictionary
|
||||
|
||||
Args:
|
||||
key (str): The name of the property
|
||||
value (str): The value of the property
|
||||
"""
|
||||
self._properties[key] = value
|
||||
|
||||
def get_property(self, key: str) -> str:
|
||||
"""
|
||||
Retrieves a property value from the application's properties dictionary
|
||||
|
||||
Args:
|
||||
key (str): The name of the property
|
||||
|
||||
Returns:
|
||||
str: The value of the property
|
||||
"""
|
||||
return self._properties.get(key)
|
||||
|
||||
def get_status(self) -> str:
|
||||
"""
|
||||
Retrieves the status of the application
|
||||
|
||||
Returns:
|
||||
str: The status of the application
|
||||
"""
|
||||
return self.get_property("status")
|
||||
|
||||
def get_progress(self) -> str:
|
||||
"""
|
||||
Retrieves the progress of the application
|
||||
|
||||
Returns:
|
||||
str: The progress of the application
|
||||
"""
|
||||
return self.get_property("progress")
|
||||
|
||||
def get_active(self) -> str:
|
||||
"""
|
||||
Retrieves the active status of the application
|
||||
|
||||
Returns:
|
||||
str: 'True' if the application is active, 'False' otherwise
|
||||
"""
|
||||
return self.get_property("active")
|
||||
|
||||
def get_name(self) -> str:
|
||||
"""
|
||||
Retrieves the name of the application
|
||||
|
||||
Returns:
|
||||
str: The name of the application
|
||||
"""
|
||||
return self.get_property("name")
|
||||
|
||||
def get_version(self) -> str:
|
||||
"""
|
||||
Retrieves the version of the application
|
||||
|
||||
Returns:
|
||||
str: The version of the application
|
||||
"""
|
||||
return self.get_property("app_version")
|
||||
|
||||
def get_manifest_name(self) -> str:
|
||||
"""
|
||||
Retrieves the manifest name used by the application
|
||||
|
||||
Returns:
|
||||
str: The manifest name
|
||||
"""
|
||||
return self.get_property("manifest_name")
|
||||
|
||||
def get_manifest_file(self) -> str:
|
||||
"""
|
||||
Retrieves the manifest file used by the application
|
||||
|
||||
Returns:
|
||||
str: The manifest file
|
||||
"""
|
||||
return self.get_property("manifest_file")
|
||||
|
||||
def get_created_at(self) -> str:
|
||||
"""
|
||||
Retrieves the creation timestamp of the application
|
||||
|
||||
Returns:
|
||||
str: The creation timestamp
|
||||
"""
|
||||
return self.get_property("created_at")
|
||||
|
||||
def get_updated_at(self) -> str:
|
||||
"""
|
||||
Retrieves the last updated timestamp of the application
|
||||
|
||||
Returns:
|
||||
str: The last updated timestamp
|
||||
"""
|
||||
return self.get_property("updated_at")
|
||||
@@ -0,0 +1,52 @@
|
||||
from keywords.cloud_platform.system.application.object.system_application_show_object import SystemApplicationShowObject
|
||||
|
||||
|
||||
class SystemApplicationShowOutput:
|
||||
"""
|
||||
Parses the output from the 'system application-show <app_name>' command
|
||||
|
||||
Methods:
|
||||
get_system_application_object(): Returns the parsed system application object
|
||||
"""
|
||||
|
||||
def __init__(self, raw_output: str | list):
|
||||
"""
|
||||
Initialize with raw output and parse it.
|
||||
|
||||
Args:
|
||||
raw_output (str | list): The raw command output from the system
|
||||
"""
|
||||
if isinstance(raw_output, list):
|
||||
raw_output = "".join(raw_output)
|
||||
self.raw_output = raw_output
|
||||
self.application_object = self._parse_output(self.raw_output)
|
||||
|
||||
def _parse_output(self, output: str) -> SystemApplicationShowObject:
|
||||
"""
|
||||
Parses the CLI output into a SystemApplicationShowObject
|
||||
|
||||
Args:
|
||||
output (str): Raw command output
|
||||
|
||||
Returns:
|
||||
SystemApplicationShowObject: Parsed application data
|
||||
"""
|
||||
app_obj = SystemApplicationShowObject()
|
||||
for line in output.strip().splitlines():
|
||||
if "|" not in line or "Property" in line:
|
||||
continue
|
||||
parts = [p.strip() for p in line.split("|") if p.strip()]
|
||||
if len(parts) == 2:
|
||||
key = parts[0].lower().replace(" ", "_")
|
||||
value = parts[1]
|
||||
app_obj.set_property(key, value)
|
||||
return app_obj
|
||||
|
||||
def get_system_application_object(self) -> SystemApplicationShowObject:
|
||||
"""
|
||||
Returns the parsed system application object
|
||||
|
||||
Returns:
|
||||
SystemApplicationShowObject: The application object containing parsed data
|
||||
"""
|
||||
return self.application_object
|
||||
@@ -0,0 +1,72 @@
|
||||
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.system.application.object.system_application_show_output import SystemApplicationShowOutput
|
||||
|
||||
|
||||
class SystemApplicationShowKeywords(BaseKeyword):
|
||||
"""
|
||||
Provides methods to interact with the 'system application-show' command
|
||||
|
||||
Methods:
|
||||
get_system_application_show(app_name): Run the command and parse output
|
||||
"""
|
||||
|
||||
def __init__(self, ssh_connection: SSHConnection) -> None:
|
||||
"""
|
||||
Constructor
|
||||
|
||||
Args:
|
||||
ssh_connection (SSHConnection): The SSH connection object
|
||||
"""
|
||||
self.ssh_connection = ssh_connection
|
||||
|
||||
def get_system_application_show(self, app_name: str) -> SystemApplicationShowOutput:
|
||||
"""
|
||||
Executes the 'system application-show <app_name>' command and returns parsed output
|
||||
|
||||
Args:
|
||||
app_name (str): The application name to query
|
||||
|
||||
Returns:
|
||||
SystemApplicationShowOutput: The parsed output from the command
|
||||
|
||||
Raises:
|
||||
Exception: If the command execution or parsing fails
|
||||
"""
|
||||
try:
|
||||
cmd = f"system application-show {app_name}"
|
||||
output = self.ssh_connection.send(source_openrc(cmd))
|
||||
self.validate_success_return_code(self.ssh_connection)
|
||||
return SystemApplicationShowOutput(output)
|
||||
except Exception as ex:
|
||||
get_logger().log_exception(f"Failed to run 'system application-show {app_name}'")
|
||||
raise ex
|
||||
|
||||
def validate_app_status(self, app_name: str, expected_status: str, expected_active: str, expected_progress: str) -> None:
|
||||
"""
|
||||
Validates that the application reaches the expected status, active state, and progress.
|
||||
|
||||
Args:
|
||||
app_name (str): Name of the application
|
||||
expected_status (str): Expected status (e.g., 'applied')
|
||||
expected_active (str): Expected active state (e.g., 'True')
|
||||
expected_progress (str): Expected progress (e.g., 'completed')
|
||||
"""
|
||||
|
||||
def get_status():
|
||||
return self.get_system_application_show(app_name).get_system_application_object().get_status()
|
||||
|
||||
def get_active():
|
||||
return self.get_system_application_show(app_name).get_system_application_object().get_active()
|
||||
|
||||
def get_progress():
|
||||
return self.get_system_application_show(app_name).get_system_application_object().get_progress()
|
||||
|
||||
validate_equals_with_retry(get_status, expected_status, f"Waiting for application '{app_name}' to reach status '{expected_status}'", timeout=300)
|
||||
|
||||
validate_equals_with_retry(get_active, expected_active, f"Waiting for application '{app_name}' to become active '{expected_active}'", timeout=300)
|
||||
|
||||
validate_equals_with_retry(get_progress, expected_progress, f"Waiting for application '{app_name}' to complete progress '{expected_progress}'", timeout=300)
|
||||
@@ -31,6 +31,7 @@ from keywords.cloud_platform.system.application.system_application_apply_keyword
|
||||
from keywords.cloud_platform.system.application.system_application_delete_keywords import SystemApplicationDeleteKeywords
|
||||
from keywords.cloud_platform.system.application.system_application_list_keywords import SystemApplicationListKeywords
|
||||
from keywords.cloud_platform.system.application.system_application_remove_keywords import SystemApplicationRemoveKeywords
|
||||
from keywords.cloud_platform.system.application.system_application_show_keywords import SystemApplicationShowKeywords
|
||||
from keywords.cloud_platform.system.application.system_application_upload_keywords import SystemApplicationUploadKeywords
|
||||
from keywords.cloud_platform.system.host.system_host_list_keywords import SystemHostListKeywords
|
||||
from keywords.cloud_platform.system.host.system_host_lock_keywords import SystemHostLockKeywords
|
||||
@@ -124,14 +125,12 @@ def test_cert_manager_applied():
|
||||
|
||||
Test Steps:
|
||||
- connect to active controller
|
||||
- run system cmd - system application-list
|
||||
- validate that cert manager is in applied state
|
||||
- run system cmd - system application-show nginx-ingress-controller
|
||||
- validate cert manager state is 'applied', 'active' and progress 'completed'
|
||||
|
||||
"""
|
||||
ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
|
||||
system_applications = SystemApplicationListKeywords(ssh_connection).get_system_application_list()
|
||||
application_status = system_applications.get_application("cert-manager").get_status()
|
||||
assert application_status == "applied", f"cert-manager was not applied. Status was {application_status}"
|
||||
SystemApplicationShowKeywords(ssh_connection).validate_app_status(app_name="cert-manager", expected_status="applied", expected_active="True", expected_progress="completed")
|
||||
|
||||
|
||||
@mark.p0
|
||||
@@ -141,13 +140,12 @@ def test_nginx_ingress_controller_applied():
|
||||
|
||||
Test Steps:
|
||||
- connect to active controller
|
||||
- run system cmd - system application-list
|
||||
- validate that nginx ingress controller is in applied state
|
||||
- run system cmd - system application-show nginx-ingress-controller
|
||||
- validate nginx ingress controller state is 'applied', 'active' and progress 'completed'
|
||||
|
||||
"""
|
||||
ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
|
||||
system_applications = SystemApplicationListKeywords(ssh_connection)
|
||||
system_applications.validate_app_status("nginx-ingress-controller", "applied")
|
||||
SystemApplicationShowKeywords(ssh_connection).validate_app_status(app_name="nginx-ingress-controller", expected_status="applied", expected_active="True", expected_progress="completed")
|
||||
|
||||
|
||||
@mark.p0
|
||||
|
||||
Reference in New Issue
Block a user