Support user input for python tests in TH (#32299) (#32325)

* Add supporting functions

* Update TC-RVCOPSTATE-2.4

* Fix linting errors

* Restyled by isort

* Update src/python_testing/matter_testing_support.py

---------

Co-authored-by: Carolina Lopes <116589288+ccruzagralopes@users.noreply.github.com>
Co-authored-by: Restyled.io <commits@restyled.io>

Author: 3 people

scripts/py_matter_yamltests/matter_yamltests/hooks.py

@@ -13,6 +13,8 @@
 #    See the License for the specific language governing permissions and
 #    limitations under the License.
 
+from typing import Optional
+
 from .parser import TestStep
 
 
@@ -213,6 +215,15 @@ async def step_manual(self):
         """
         pass
 
+    def show_prompt(self,
+                    msg: str,
+                    placeholder: Optional[str] = None,
+                    default_value: Optional[str] = None) -> None:
+        """
+        This method is called when the step needs to ask the user to perform some action or provide some value.
+        """
+        pass
+
 
 class WebSocketRunnerHooks():
     def connecting(self, url: str):

src/python_testing/TC_RVCOPSTATE_2_4.py

@@ -115,47 +115,51 @@ async def test_TC_RVCOPSTATE_2_4(self):
             self.write_to_app_pipe('{"Name": "Reset"}')
 
         if self.check_pics("RVCOPSTATE.S.M.ST_ERROR"):
-            self.print_step(2, "Manually put the device in the ERROR operational state")
+            step_name = "Manually put the device in the ERROR operational state"
+            self.print_step(2, step_name)
             if self.is_ci:
                 self.write_to_app_pipe('{"Name": "ErrorEvent", "Error": "UnableToStartOrResume"}')
             else:
-                input("Press Enter when done.\n")
+                self.wait_for_user_input(step_name)
 
             await self.read_operational_state_with_check(3, op_states.kError)
 
             await self.send_go_home_cmd_with_check(4, op_errors.kCommandInvalidInState)
 
         if self.check_pics("RVCOPSTATE.S.M.ST_CHARGING"):
-            self.print_step(5, "Manually put the device in the CHARGING operational state")
+            step_name = "Manually put the device in the CHARGING operational state"
+            self.print_step(5, step_name)
             if self.is_ci:
                 self.write_to_app_pipe('{"Name": "Reset"}')
                 self.write_to_app_pipe('{"Name": "Docked"}')
                 self.write_to_app_pipe('{"Name": "Charging"}')
             else:
-                input("Press Enter when done.\n")
+                self.wait_for_user_input(step_name)
 
             await self.read_operational_state_with_check(6, rvc_op_states.kCharging)
 
             await self.send_go_home_cmd_with_check(7, op_errors.kCommandInvalidInState)
 
         if self.check_pics("RVCOPSTATE.S.M.ST_DOCKED"):
-            self.print_step(8, "Manually put the device in the DOCKED operational state")
+            step_name = "Manually put the device in the DOCKED operational state"
+            self.print_step(8, step_name)
             if self.is_ci:
                 self.write_to_app_pipe('{"Name": "Charged"}')
             else:
-                input("Press Enter when done.\n")
+                self.wait_for_user_input(step_name)
 
             await self.read_operational_state_with_check(9, rvc_op_states.kDocked)
 
             await self.send_go_home_cmd_with_check(10, op_errors.kCommandInvalidInState)
 
         if self.check_pics("PICS_M_ST_SEEKING_CHARGER"):
-            self.print_step(8, "Manually put the device in the SEEKING CHARGER operational state")
+            step_name = "Manually put the device in the SEEKING CHARGER operational state"
+            self.print_step(8, step_name)
             if self.is_ci:
                 await self.send_run_change_to_mode_cmd(rvc_app_run_mode_cleaning)
                 await self.send_run_change_to_mode_cmd(rvc_app_run_mode_idle)
             else:
-                input("Press Enter when done.\n")
+                self.wait_for_user_input(step_name)
 
             await self.read_operational_state_with_check(9, rvc_op_states.kSeekingCharger)
 

src/python_testing/matter_testing_support.py

@@ -346,6 +346,12 @@ def step_unknown(self):
         """
         pass
 
+    def show_prompt(self,
+                    msg: str,
+                    placeholder: Optional[str] = None,
+                    default_value: Optional[str] = None) -> None:
+        pass
+
 
 @dataclass
 class MatterTestConfig:
@@ -1091,6 +1097,28 @@ def get_setup_payload_info(self) -> SetupPayloadInfo:
 
         return info
 
+    def wait_for_user_input(self,
+                            prompt_msg: str,
+                            input_msg: str = "Press Enter when done.\n",
+                            prompt_msg_placeholder: str = "Submit anything to continue",
+                            default_value: str = "y") -> str:
+        """Ask for user input and wait for it.
+
+        Args:
+            prompt_msg (str): Message for TH UI prompt. Indicates what is expected from the user.
+            input_msg (str, optional): Prompt for input function, used when running tests manually. Defaults to "Press Enter when done.\n".
+            prompt_msg_placeholder (str, optional): TH UI prompt input placeholder. Defaults to "Submit anything to continue".
+            default_value (str, optional): TH UI prompt default value. Defaults to "y".
+
+        Returns:
+            str: User input
+        """
+        if self.runner_hook:
+            self.runner_hook.show_prompt(msg=prompt_msg,
+                                         placeholder=prompt_msg_placeholder,
+                                         default_value=default_value)
+        return input(input_msg)
+
 
 def generate_mobly_test_config(matter_test_config: MatterTestConfig):
     test_run_config = TestRunConfig()