Add ceph-rook backend cmd testing for AIO

- Add wait_for_applied argument in system_application_apply()
- Add validate_app_auto_reapplied() in system_application_list_keywords
- Add lock_host_reject() in system_host_lock_keywords
- Add system_storage_backend_delete()
- Add system_storage_backend_delete_with_error()
- Add system_storage_backend_delete_without_force()
- Add TC: test_ceph_rook_backend_operation()
- Add markers: lab_ceph_rook and lab_is_aio
- Automation logs: /folk/cgts_logs/logs/wrcppv-1200/
- Add new logs and run unit test
- Split TC into 5 TCs

Change-Id: I694f3424edbca43e5f8a905e234afbc268ad6bad
Signed-off-by: ppeng <peng.peng@windriver.com>

keywords/cloud_platform/system/application/system_application_apply_keywords.py

@@ -22,16 +22,15 @@ class SystemApplicationApplyKeywords(BaseKeyword):
         """
         self.ssh_connection = ssh_connection
 
-    def system_application_apply(self, app_name: str, timeout: int = 300, polling_sleep_time: int = 5) -> SystemApplicationOutput:
+    def system_application_apply(self, app_name: str, timeout: int = 300, polling_sleep_time: int = 5, wait_for_applied: bool = True) -> SystemApplicationOutput:
         """
-        Executes the applying of an application file by executing the command 'system application-apply'. This method
+        Executes the applying of an application file by executing the command 'system application-apply'.
 
-        returns upon the completion of the 'system application-apply' command, that is, when the 'status' is 'applied'.
-        Executes the installation of an application by executing the command 'system application-apply'.
         Args:
             app_name (str): the name of the application to apply
             timeout (int): Timeout in seconds
             polling_sleep_time (int): wait time in seconds before the next attempt when unsuccessful validation
+            wait_for_applied (bool): whether wait for the app status from applying to applied
 
         Returns:
             SystemApplicationOutput: an object representing status values related to the current installation
@@ -46,12 +45,13 @@ class SystemApplicationApplyKeywords(BaseKeyword):
         self.validate_success_return_code(self.ssh_connection)
         system_application_output = SystemApplicationOutput(output)
 
-        # Tracks the execution of the command 'system application-apply' until its completion or a timeout.
-        system_application_list_keywords = SystemApplicationListKeywords(self.ssh_connection)
-        system_application_list_keywords.validate_app_status(app_name, "applied", timeout, polling_sleep_time)
+        if wait_for_applied:
+            # Tracks the execution of the command 'system application-apply' until its completion or a timeout.
+            system_application_list_keywords = SystemApplicationListKeywords(self.ssh_connection)
+            system_application_list_keywords.validate_app_status(app_name, "applied", timeout, polling_sleep_time)
 
-        # If the execution arrived here the status of the application is 'applied'.
-        system_application_output.get_system_application_object().set_status("applied")
+            # If the execution arrived here the status of the application is 'applied'.
+            system_application_output.get_system_application_object().set_status("applied")
 
         return system_application_output
 

keywords/cloud_platform/system/host/system_host_lock_keywords.py

@@ -46,6 +46,20 @@ class SystemHostLockKeywords(BaseKeyword):
             raise KeywordException("Lock host did not lock in the required time. Host values were: " f"Operational: {host_value.get_operational()} " f"Administrative: {host_value.get_administrative()} " f"Availability: {host_value.get_availability()}")
         return True
 
+    def lock_host_with_error(self, host_name: str) -> str:
+        """
+        Locks the given host name. It's expected that the cmd returns error
+
+        Args:
+            host_name (str): the name of the host
+
+        Returns:
+            str: a str of error message
+
+        """
+        msg = self.ssh_connection.send(source_openrc(f"system host-lock {host_name}"))
+        return msg[0]
+
     def wait_for_host_locked(self, host_name: str) -> bool:
         """
         Waits for the given host name to be locked

keywords/cloud_platform/system/storage/system_storage_backend_keywords.py

@@ -75,7 +75,7 @@ class SystemStorageBackendKeywords(BaseKeyword):
         self.ssh_connection.send(source_openrc(f"system storage-backend-modify {backend}-store {extra_args}"))
         self.validate_success_return_code(self.ssh_connection)
 
-    def system_storage_backend_modify_with_error(self, backend: str, services: str = None, deployment_model: str = None, replication: int = 0, min_replication: int = 0) -> list[str]:
+    def system_storage_backend_modify_with_error(self, backend: str, services: str = None, deployment_model: str = None, replication: int = 0, min_replication: int = 0) -> str:
         """
         Run the "system storage-backend-modify" command with invalid arguments
 
@@ -86,7 +86,7 @@ class SystemStorageBackendKeywords(BaseKeyword):
             replication (int): new replication value
             min_replication (int): min_replication value
         Returns:
-             list[str]: a list of error msg
+             str: a str of error msg
         """
         extra_args = ""
         if deployment_model:
@@ -99,4 +99,39 @@ class SystemStorageBackendKeywords(BaseKeyword):
             extra_args += f"min_replication={min_replication} "
 
         msg = self.ssh_connection.send(source_openrc(f"system storage-backend-modify {backend}-store {extra_args}"))
-        return msg
+        return msg[0]
+
+    def system_storage_backend_delete(self, backend: str):
+        """
+        Delete the storage backend
+
+        Args:
+            backend (str): the backend name ceph or ceph-rook
+
+        """
+        self.ssh_connection.send(source_openrc(f"system storage-backend-delete {backend}-store --force"))
+        self.validate_success_return_code(self.ssh_connection)
+
+    def system_storage_backend_delete_with_error(self, backend: str) -> str:
+        """
+        Run the "system storage-backend-delete" command while rook-ceph app is applied
+
+        Args:
+            backend (str): the backend name ceph or ceph-rook
+        Returns:
+             str: a str of error msg
+        """
+        msg = self.ssh_connection.send(source_openrc(f"system storage-backend-delete {backend}-store --force"))
+        return msg[0]
+
+    def system_storage_backend_delete_without_force(self, backend: str) -> str:
+        """
+        storage-backend-delete cmd negative testing, Delete the storage backend cmd missed argument --force
+
+        Args:
+            backend (str): the backend name ceph or ceph-rook
+        Returns:
+             str: a list of error msg
+        """
+        msg = self.ssh_connection.send(source_openrc(f"system storage-backend-delete {backend}-store"))
+        return msg[0]

testcases/cloud_platform/regression/storage/test_ceph_rook_operations.py

@@ -1,10 +1,13 @@
 from pytest import mark
 
+from config.configuration_manager import ConfigurationManager
 from framework.logging.automation_logger import get_logger
 from framework.validation.validation import validate_equals, validate_equals_with_retry, validate_str_contains
 from keywords.ceph.ceph_osd_pool_ls_detail_keywords import CephOsdPoolLsDetailKeywords
 from keywords.ceph.ceph_status_keywords import CephStatusKeywords
 from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords
+from keywords.cloud_platform.system.application.system_application_apply_keywords import SystemApplicationApplyKeywords
+from keywords.cloud_platform.system.application.system_application_list_keywords import SystemApplicationListKeywords
 from keywords.cloud_platform.system.host.system_host_fs_keywords import SystemHostFSKeywords
 from keywords.cloud_platform.system.host.system_host_list_keywords import SystemHostListKeywords
 from keywords.cloud_platform.system.host.system_host_lock_keywords import SystemHostLockKeywords
@@ -12,7 +15,7 @@ from keywords.cloud_platform.system.storage.system_storage_backend_keywords impo
 
 
 @mark.p2
-@mark.lab_ceph_rook
+@mark.lab_rook_ceph
 def test_ceph_rook_host_fs_operation():
     """
     Test case: [TC_34893] host-fs cmd operations testing. This TC was migrated from cgcs.
@@ -200,7 +203,7 @@ def test_ceph_rook_capabilities_testing_open_model(request):
     get_logger().log_test_case_step("\n\nIt should be rejected if modifying min_replication value great than replication value.\n")
     new_min_replication_value = curr_replication + 1
     msg = system_storage_backend_keywords.system_storage_backend_modify_with_error(backend="ceph-rook", min_replication=new_min_replication_value)
-    validate_str_contains(msg[0], "must be greater than", f"system backend modify should be failed: {msg[0]}")
+    validate_str_contains(msg, "must be greater than", f"system backend modify should be failed: {msg}")
 
     curr_min_replication = system_storage_backend_keywords.get_system_storage_backend_list().get_system_storage_backend("ceph-rook").get_capabilities().get_min_replication()
 
@@ -249,3 +252,289 @@ def test_lock_unlock_multiple_hosts():
 
     get_logger().log_test_case_step("Checking ceph health after Lock/Unlock nodes.")
     ceph_status_keywords.wait_for_ceph_health_status(expect_health_status=True)
+
+
+@mark.p2
+@mark.lab_rook_ceph
+@mark.lab_is_simplex
+def test_ceph_rook_backend_deployment_model_operation_sx():
+    """
+    Test case: ceph-rook backend deployment model testing (SX).
+
+    Test Steps:
+        - Make sure ceph-rook is storage backend
+        - Make sure the deployment model is "open"
+        - Modifying deployment model "open" to "dedicated" should be rejected
+        - change deployment model to "controller"
+        - Modifying deployment model "controller" to "dedicated" should be rejected
+        - Modifying deployment model to open if necessary
+    """
+    ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
+    system_storage_backend_keywords = SystemStorageBackendKeywords(ssh_connection)
+
+    get_logger().log_test_case_step("Check whether rook-ceph is configured as storage backend.")
+
+    if not system_storage_backend_keywords.get_system_storage_backend_list().is_backend_configured("ceph-rook"):
+        get_logger().log_test_case_step("Add rook-ceph as storage backend.")
+        system_storage_backend_keywords.system_storage_backend_add(backend="ceph-rook", confirmed=True)
+
+    backend_name = system_storage_backend_keywords.get_system_storage_backend_list().get_system_storage_backend("ceph-rook").get_backend()
+    get_logger().log_info(f"Backend is: {backend_name}")
+
+    get_logger().log_test_case_step("Get original backend deployment model.")
+    original_deployment_model = system_storage_backend_keywords.get_system_storage_backend_list().get_system_storage_backend("ceph-rook").get_capabilities().get_deployment_model()
+    get_logger().log_info(f"Original Capabilities deployment model is: {original_deployment_model}")
+    if original_deployment_model == "controller":
+        get_logger().log_test_case_step(f"Modify deployment_model from '{original_deployment_model}' to 'open'.")
+        system_storage_backend_keywords.system_storage_backend_modify(backend="ceph-rook", deployment_model="open")
+        curr_deployment_model = system_storage_backend_keywords.get_system_storage_backend_list().get_system_storage_backend("ceph-rook").get_capabilities().get_deployment_model()
+        validate_equals(curr_deployment_model, "open", "Current deployment model should be open.")
+
+    get_logger().log_test_case_step("Modify deployment_model from 'open' to 'dedicated' should be rejected in SX system.")
+    msg = system_storage_backend_keywords.system_storage_backend_modify_with_error(backend="ceph-rook", deployment_model="dedicated")
+    validate_str_contains(msg, "OSDs deployed", f"system backend modify should be failed: {msg}")
+
+    get_logger().log_test_case_step("Modify deployment_model from 'open' to 'controller'.")
+    system_storage_backend_keywords.system_storage_backend_modify(backend="ceph-rook", deployment_model="controller")
+    curr_deployment_model = system_storage_backend_keywords.get_system_storage_backend_list().get_system_storage_backend("ceph-rook").get_capabilities().get_deployment_model()
+    validate_equals(curr_deployment_model, "controller", f"Current deployment model should be {curr_deployment_model}.")
+
+    get_logger().log_test_case_step("Modify deployment_model from 'controller' to 'dedicated' should be rejected in SX system.")
+    msg = system_storage_backend_keywords.system_storage_backend_modify_with_error(backend="ceph-rook", deployment_model="dedicated")
+    validate_str_contains(msg, "not supported", f"system backend modify should be failed: {msg}")
+
+    if original_deployment_model == "open":
+        get_logger().log_test_case_step("Modify deployment_model from 'controller' to 'open'.")
+        system_storage_backend_keywords.system_storage_backend_modify(backend="ceph-rook", deployment_model="open")
+        curr_deployment_model = system_storage_backend_keywords.get_system_storage_backend_list().get_system_storage_backend("ceph-rook").get_capabilities().get_deployment_model()
+        validate_equals(curr_deployment_model, "open", "Current deployment model should be 'open'.")
+
+
+@mark.p2
+@mark.lab_rook_ceph
+@mark.lab_is_duplex
+def test_ceph_rook_backend_deployment_model_operation_dx():
+    """
+    Test case: ceph-rook backend deployment model testing (DX).
+
+    Test Steps:
+        - Make sure ceph-rook is storage backend
+        - Make sure the deployment model is "open"
+        - Modifying deployment model "open" to "dedicated" should be rejected
+        - change deployment model to "controller"
+        - check whether rook-ceph app auto re-applied
+        - Modifying deployment model "controller" to "dedicated" should be rejected
+        - Modifying deployment model to open if necessary
+    """
+    ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
+    system_storage_backend_keywords = SystemStorageBackendKeywords(ssh_connection)
+
+    get_logger().log_test_case_step("Check whether rook-ceph is configured as storage backend.")
+    if not system_storage_backend_keywords.get_system_storage_backend_list().is_backend_configured("ceph-rook"):
+        get_logger().log_test_case_step("Add rook-ceph as storage backend.")
+        system_storage_backend_keywords.system_storage_backend_add(backend="ceph-rook", confirmed=True)
+
+    backend_name = system_storage_backend_keywords.get_system_storage_backend_list().get_system_storage_backend("ceph-rook").get_backend()
+    get_logger().log_info(f"Backend is: {backend_name}")
+
+    get_logger().log_test_case_step("Get original backend deployment model.")
+    original_deployment_model = system_storage_backend_keywords.get_system_storage_backend_list().get_system_storage_backend("ceph-rook").get_capabilities().get_deployment_model()
+    get_logger().log_info(f"Original Capabilities deployment model is: {original_deployment_model}")
+
+    if original_deployment_model == "controller":
+        get_logger().log_test_case_step(f"Modify deployment_model from '{original_deployment_model}' to 'open'.")
+        system_storage_backend_keywords.system_storage_backend_modify(backend="ceph-rook", deployment_model="open")
+        curr_deployment_model = system_storage_backend_keywords.get_system_storage_backend_list().get_system_storage_backend("ceph-rook").get_capabilities().get_deployment_model()
+        validate_equals(curr_deployment_model, "open", "Current deployment model should be open.")
+        get_logger().log_info("Wait for rook-ceph app auto reapplied")
+        app_name = "rook-ceph"
+        app_status_list = ["applying"]
+        SystemApplicationListKeywords(ssh_connection).validate_app_status_in_list(app_name, app_status_list, timeout=360, polling_sleep_time=5)
+        app_status_list = ["applied"]
+        SystemApplicationListKeywords(ssh_connection).validate_app_status_in_list(app_name, app_status_list, timeout=360, polling_sleep_time=10)
+
+    get_logger().log_test_case_step("Modify deployment_model from 'open' to 'dedicated' should be rejected in DX system.")
+    msg = system_storage_backend_keywords.system_storage_backend_modify_with_error(backend="ceph-rook", deployment_model="dedicated")
+    validate_str_contains(msg, "OSDs deployed", f"system backend modify should be failed: {msg}")
+
+    get_logger().log_test_case_step("Modify deployment_model from 'open' to 'controller'.")
+    system_storage_backend_keywords.system_storage_backend_modify(backend="ceph-rook", deployment_model="controller")
+    curr_deployment_model = system_storage_backend_keywords.get_system_storage_backend_list().get_system_storage_backend("ceph-rook").get_capabilities().get_deployment_model()
+    validate_equals(curr_deployment_model, "controller", f"Current deployment model should be {curr_deployment_model}.")
+    get_logger().log_info("Wait for rook-ceph app auto reapplied")
+    app_name = "rook-ceph"
+    app_status_list = ["applying"]
+    SystemApplicationListKeywords(ssh_connection).validate_app_status_in_list(app_name, app_status_list, timeout=360, polling_sleep_time=5)
+    app_status_list = ["applied"]
+    SystemApplicationListKeywords(ssh_connection).validate_app_status_in_list(app_name, app_status_list, timeout=360, polling_sleep_time=10)
+
+    get_logger().log_test_case_step("Modify deployment_model from 'controller' to 'dedicated' should be rejected in DX system.")
+    msg = system_storage_backend_keywords.system_storage_backend_modify_with_error(backend="ceph-rook", deployment_model="dedicated")
+    validate_str_contains(msg, "Unable to change", f"system backend modify should be failed: {msg}")
+
+    if original_deployment_model == "open":
+        get_logger().log_test_case_step("Modify deployment_model from 'controller' to 'open'.")
+        system_storage_backend_keywords.system_storage_backend_modify(backend="ceph-rook", deployment_model="open")
+        curr_deployment_model = system_storage_backend_keywords.get_system_storage_backend_list().get_system_storage_backend("ceph-rook").get_capabilities().get_deployment_model()
+        validate_equals(curr_deployment_model, "open", "Current deployment model should be 'open'.")
+        get_logger().log_info("Wait for rook-ceph app auto reapplied")
+        app_name = "rook-ceph"
+        app_status_list = ["applying"]
+        SystemApplicationListKeywords(ssh_connection).validate_app_status_in_list(app_name, app_status_list, timeout=360, polling_sleep_time=5)
+        app_status_list = ["applied"]
+        SystemApplicationListKeywords(ssh_connection).validate_app_status_in_list(app_name, app_status_list, timeout=360, polling_sleep_time=10)
+
+
+@mark.p2
+@mark.lab_rook_ceph
+def test_ceph_rook_backend_services_operation():
+    """
+    Test case: ceph-rook backend services testing.
+
+    Test Steps:
+        - Make sure ceph-rook is storage backend
+        - Test service block and ecblock could exist one only
+        - Test take off service is not support
+        - Add service if possible
+    """
+    ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
+    system_storage_backend_keywords = SystemStorageBackendKeywords(ssh_connection)
+
+    get_logger().log_test_case_step("Check whether rook-ceph is configured as storage backend.")
+
+    if not system_storage_backend_keywords.get_system_storage_backend_list().is_backend_configured("ceph-rook"):
+        get_logger().log_test_case_step("Add rook-ceph as storage backend.")
+        system_storage_backend_keywords.system_storage_backend_add(backend="ceph-rook", confirmed=True)
+
+    backend_name = system_storage_backend_keywords.get_system_storage_backend_list().get_system_storage_backend("ceph-rook").get_backend()
+    get_logger().log_info(f"Backend is: {backend_name}")
+
+    get_logger().log_test_case_step("Get current services.")
+    curr_services = system_storage_backend_keywords.get_system_storage_backend_list().get_system_storage_backend("ceph-rook").get_services()
+    get_logger().log_info(f"Current Services are {curr_services}")
+
+    get_logger().log_test_case_step("Testing service block and ecblock could only exist one in system.")
+    if "ecblock" in curr_services:
+        new_services = f"block,{curr_services}"
+    elif "block," in curr_services:
+        new_services = f"ecblock,{curr_services}"
+    else:
+        new_services = f"block,ecblock,{curr_services}"
+
+    msg = system_storage_backend_keywords.system_storage_backend_modify_with_error(backend="ceph-rook", services=new_services)
+    validate_str_contains(msg, "not supported for the ceph-rook backend in same time", f"system backend modify should be rejected: {msg}")
+
+    get_logger().log_test_case_step("Removing service is not supported.")
+    # finding any one of the existing services, and then attempting to remove it
+    if "," not in curr_services:
+        # there is only one service
+        sub_str = curr_services
+    elif "object," in curr_services:
+        sub_str = "object,"
+    elif "block," in curr_services:
+        sub_str = "block,"
+    elif "ecblock," in curr_services:
+        sub_str = "ecblock,"
+    elif "filesystem," in curr_services:
+        sub_str = "filesystem,"
+    else:
+        raise ValueError(f"invalid services {curr_services}")
+    # remove sub_str
+    new_services = curr_services.replace(sub_str, "")
+    msg = system_storage_backend_keywords.system_storage_backend_modify_with_error(backend="ceph-rook", services=new_services)
+    validate_str_contains(msg, "is not supported", f"system backend modify should be rejected: {msg}")
+
+    if "block" not in curr_services and "ecblock" not in curr_services:
+        new_service = "block"
+    elif "filesystem" not in curr_services:
+        new_service = "filesystem"
+    elif "object" not in curr_services:
+        new_service = "object"
+    else:
+        raise ValueError("This system just has no new service available to add for this testing and it is not an issue")
+
+    get_logger().log_info("Make sure rook-ceph app in applied status")
+    app_name = "rook-ceph"
+    app_status_list = ["applied"]
+    SystemApplicationListKeywords(ssh_connection).validate_app_status_in_list(app_name, app_status_list, timeout=360, polling_sleep_time=10)
+    get_logger().log_test_case_step(f"Add new service: {new_service}.")
+    new_services = f"{new_service},{curr_services}"
+    system_storage_backend_keywords.system_storage_backend_modify(backend="ceph-rook", services=new_services)
+    get_logger().log_info("Wait for rook-ceph app auto reapply starts")
+    app_status_list = ["applying"]
+    SystemApplicationListKeywords(ssh_connection).validate_app_status_in_list(app_name, app_status_list, timeout=360, polling_sleep_time=5)
+    app_status_list = ["applied"]
+    SystemApplicationListKeywords(ssh_connection).validate_app_status_in_list(app_name, app_status_list, timeout=360, polling_sleep_time=10)
+
+    curr_services = system_storage_backend_keywords.get_system_storage_backend_list().get_system_storage_backend("ceph-rook").get_services()
+    get_logger().log_info(f"Current Services are {curr_services}")
+
+
+@mark.p2
+@mark.lab_rook_ceph
+def test_ceph_rook_backend_delete_negative_testing():
+    """
+    Test case: ceph-rook backend delete cmd testing.
+
+    Test Steps:
+        - Make sure ceph-rook is storage backend
+        - storage-backend-delete cmd negative test
+    """
+    ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
+    system_storage_backend_keywords = SystemStorageBackendKeywords(ssh_connection)
+
+    get_logger().log_test_case_step("Check whether rook-ceph is configured as storage backend.")
+
+    if not system_storage_backend_keywords.get_system_storage_backend_list().is_backend_configured("ceph-rook"):
+        get_logger().log_test_case_step("Add rook-ceph as storage backend.")
+        system_storage_backend_keywords.system_storage_backend_add(backend="ceph-rook", confirmed=True)
+
+    backend_name = system_storage_backend_keywords.get_system_storage_backend_list().get_system_storage_backend("ceph-rook").get_backend()
+    get_logger().log_info(f"Backend is: {backend_name}")
+
+    get_logger().log_test_case_step("backend delete cmd negative testing")
+    get_logger().log_info("storage-backend-delete --force should not delete backend when rook-ceph app is applied.")
+    msg = system_storage_backend_keywords.system_storage_backend_delete_with_error(backend="ceph-rook")
+    validate_str_contains(msg, "Cannot delete", f"storage-backend-delete --force should not delete backend when app is applied: {msg}")
+
+    get_logger().log_info("storage-backend-delete missing argument --force should be not supported.")
+    msg = system_storage_backend_keywords.system_storage_backend_delete_without_force(backend="ceph-rook")
+    validate_str_contains(msg, "not supported", f"storage-delete missing --force not supported: {msg}")
+
+
+@mark.p2
+@mark.lab_rook_ceph
+def test_rook_ceph_applying_host_lock_reject_testing():
+    """
+    Test case: rook-ceph app applying host-lock reject testing (SX).
+
+    Test Steps:
+        - Make sure rook-ceph app in applied status
+        - Apply rook-ceph application to make storage_backend task in "applying" status.
+        - host-lock should be rejected.
+    """
+    ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
+    lab_type = ConfigurationManager.get_lab_config().get_lab_type()
+
+    get_logger().log_test_case_step("Check whether rook-ceph app in applied status")
+    app_name = "rook-ceph"
+    app_status_list = ["applied"]
+    SystemApplicationListKeywords(ssh_connection).validate_app_status_in_list(app_name, app_status_list, timeout=360, polling_sleep_time=10)
+
+    get_logger().log_test_case_step("Re-apply rook-ceph and make the app in 'applying' status.")
+    app_status_list = ["applying"]
+    SystemApplicationApplyKeywords(ssh_connection).system_application_apply(app_name, wait_for_applied=False)
+    app_status = SystemApplicationListKeywords(ssh_connection).validate_app_status_in_list(app_name, app_status_list, timeout=60, polling_sleep_time=5)
+
+    get_logger().log_test_case_step(f"Host-lock should be rejected when app in '{app_status}' status.")
+    if lab_type == "Simplex":
+        lock_host = SystemHostListKeywords(ssh_connection).get_active_controller()
+    else:
+        lock_host = SystemHostListKeywords(ssh_connection).get_standby_controller()
+    lock_host_name = lock_host.get_host_name()
+    msg = SystemHostLockKeywords(ssh_connection).lock_host_with_error(lock_host_name)
+    validate_str_contains(msg, "Rejected: The application rook-ceph is in transition", f"system host-lock rejected: {msg}")
+
+    # Wait for rook-ceph in applied status
+    app_status_list = ["applied"]
+    SystemApplicationListKeywords(ssh_connection).validate_app_status_in_list(app_name, app_status_list, timeout=360, polling_sleep_time=10)

testcases/pytest.ini

@@ -6,6 +6,7 @@ addopts = --tb=no -s
 markers=
 # Lab Specific Requirement Markers
     lab_is_simplex: mark tests that run on simplex labs
+    lab_is_duplex: mark tests that run on duplex labs
     lab_has_standby_controller: mark tests that run using a standby controller
     lab_has_low_latency: mark tests that need low latency
     lab_has_non_low_latency: marks tests that need non low latency
@@ -34,6 +35,8 @@ markers=
     subcloud_lab_has_compute: mark tests that require at least one subcloud containing at least one compute node
     lab_has_secondary_system_controller: mark tests that require a secondary system controller
     lab_has_ptp_configuration_compute: mark tests that requred ptp_configuration_expectation_compute.json5
+    lab_rook_ceph: mark tests that require rook-ceph application applied
+    lab_is_aio: mark labs without worker nodes
     #TODO: add 'lab_has_bmc_ipmi', 'lab_has_bmc_redfish', 'lab_has_bmc_dynamic', and 'lab_bmc_sensor'