diff --git a/keywords/cloud_platform/system/application/object/system_application_show_object.py b/keywords/cloud_platform/system/application/object/system_application_show_object.py new file mode 100644 index 00000000..19cb76ca --- /dev/null +++ b/keywords/cloud_platform/system/application/object/system_application_show_object.py @@ -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") diff --git a/keywords/cloud_platform/system/application/object/system_application_show_output.py b/keywords/cloud_platform/system/application/object/system_application_show_output.py new file mode 100644 index 00000000..8606e0b8 --- /dev/null +++ b/keywords/cloud_platform/system/application/object/system_application_show_output.py @@ -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 ' 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 diff --git a/keywords/cloud_platform/system/application/system_application_show_keywords.py b/keywords/cloud_platform/system/application/system_application_show_keywords.py new file mode 100644 index 00000000..a8241149 --- /dev/null +++ b/keywords/cloud_platform/system/application/system_application_show_keywords.py @@ -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 ' 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) diff --git a/testcases/cloud_platform/sanity/test_sanity.py b/testcases/cloud_platform/sanity/test_sanity.py index 9288890b..f94f15cc 100644 --- a/testcases/cloud_platform/sanity/test_sanity.py +++ b/testcases/cloud_platform/sanity/test_sanity.py @@ -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