initial commit

src/python_testing/TC_OpstateCommon.py

@@ -112,11 +112,6 @@ def init_test(self):
                     asserts.fail("The --app-pid flag must be set when PICS_SDK_CI_ONLY is set")
             self.app_pipe = self.app_pipe + str(app_pid)
 
-    # Sends and out-of-band command to test-app
-    def write_to_app_pipe(self, command):
-        with open(self.app_pipe, "w") as app_pipe:
-            app_pipe.write(command + "\n")
-
     def send_raw_manual_or_pipe_command(self, command):
         if self.is_ci:
             self.write_to_app_pipe(command)

src/python_testing/TC_RVCCLEANM_2_1.py

@@ -65,14 +65,6 @@ async def send_run_change_to_mode_cmd(self, newMode) -> Clusters.Objects.RvcRunM
                             "Unexpected return type for RVC Run Mode ChangeToMode")
         return ret
 
-    # Sends and out-of-band command to the rvc-app
-    def write_to_app_pipe(self, command):
-        with open(self.app_pipe, "w") as app_pipe:
-            app_pipe.write(command + "\n")
-        # Delay for pipe command to be processed (otherwise tests are flaky)
-        # TODO(#31239): centralize pipe write logic and remove the need of sleep
-        sleep(0.001)
-
     def pics_TC_RVCCLEANM_2_1(self) -> list[str]:
         return ["RVCCLEANM.S"]
 

src/python_testing/TC_RVCCLEANM_2_2.py

@@ -80,14 +80,6 @@ def print_instruction(self, step_number, instruction):
     def pics_TC_RVCCLEANM_2_2(self) -> list[str]:
         return ["RVCCLEANM.S"]
 
-    # Sends and out-of-band command to the rvc-app
-    def write_to_app_pipe(self, command):
-        with open(self.app_pipe, "w") as app_pipe:
-            app_pipe.write(command + "\n")
-        # Delay for pipe command to be processed (otherwise tests are flaky)
-        # TODO(#31239): centralize pipe write logic and remove the need of sleep
-        sleep(0.001)
-
     @async_test_body
     async def test_TC_RVCCLEANM_2_2(self):
         self.endpoint = self.matter_test_config.endpoint

src/python_testing/TC_RVCOPSTATE_2_1.py

@@ -72,14 +72,6 @@ async def send_pause_cmd(self) -> Clusters.Objects.RvcOperationalState.Commands.
         ret = await self.send_single_cmd(cmd=Clusters.Objects.RvcOperationalState.Commands.Pause(), endpoint=self.endpoint)
         return ret
 
-    # Sends and out-of-band command to the rvc-app
-    def write_to_app_pipe(self, command):
-        with open(self.app_pipe, "w") as app_pipe:
-            app_pipe.write(command + "\n")
-        # Allow some time for the command to take effect.
-        # This removes the test flakyness which is very annoying for everyone in CI.
-        sleep(0.001)
-
     def TC_RVCOPSTATE_2_1(self) -> list[str]:
         return ["RVCOPSTATE.S"]
 

src/python_testing/TC_RVCOPSTATE_2_3.py

@@ -138,14 +138,6 @@ async def send_run_change_to_mode_cmd(self, new_mode) -> Clusters.Objects.RvcRun
                                          endpoint=self.endpoint)
         return ret
 
-    # Sends and out-of-band command to the rvc-app
-    def write_to_app_pipe(self, command):
-        with open(self.app_pipe, "w") as app_pipe:
-            app_pipe.write(command + "\n")
-        # Delay for pipe command to be processed (otherwise tests are flaky)
-        # TODO(#31239): centralize pipe write logic and remove the need of sleep
-        sleep(0.001)
-
     # Prints the instruction and waits for a user input to continue
     def print_instruction(self, step_number, instruction):
         self.print_step(step_number, instruction)

src/python_testing/TC_RVCOPSTATE_2_4.py

@@ -90,14 +90,6 @@ async def send_run_change_to_mode_cmd(self, new_mode):
         await self.send_single_cmd(cmd=Clusters.Objects.RvcRunMode.Commands.ChangeToMode(newMode=new_mode),
                                    endpoint=self.endpoint)
 
-    # Sends an out-of-band command to the rvc-app
-    def write_to_app_pipe(self, command):
-        with open(self.app_pipe, "w") as app_pipe:
-            app_pipe.write(command + "\n")
-        # Delay for pipe command to be processed (otherwise tests are flaky)
-        # TODO(#31239): centralize pipe write logic and remove the need of sleep
-        sleep(0.001)
-
     def pics_TC_RVCOPSTATE_2_4(self) -> list[str]:
         return ["RVCOPSTATE.S"]
 

src/python_testing/TC_RVCRUNM_2_1.py

@@ -60,14 +60,6 @@ async def send_change_to_mode_cmd(self, newMode) -> Clusters.Objects.RvcRunMode.
                             "Unexpected return type for ChangeToMode")
         return ret
 
-    # Sends and out-of-band command to the rvc-app
-    def write_to_app_pipe(self, command):
-        with open(self.app_pipe, "w") as app_pipe:
-            app_pipe.write(command + "\n")
-        # Delay for pipe command to be processed (otherwise tests are flaky)
-        # TODO(#31239): centralize pipe write logic and remove the need of sleep
-        sleep(0.001)
-
     def pics_TC_RVCRUNM_2_1(self) -> list[str]:
         return ["RVCRUNM.S"]
 

src/python_testing/TC_RVCRUNM_2_2.py

@@ -101,14 +101,6 @@ async def read_op_state_operational_state(self) -> Clusters.Objects.RvcOperation
             Clusters.RvcOperationalState.Attributes.OperationalState)
         return ret
 
-    # Sends and out-of-band command to the rvc-app
-    def write_to_app_pipe(self, command):
-        with open(self.app_pipe, "w") as app_pipe:
-            app_pipe.write(command + "\n")
-        # Delay for pipe command to be processed (otherwise tests are flaky)
-        # TODO(#31239): centralize pipe write logic and remove the need of sleep
-        sleep(0.001)
-
     def pics_TC_RVCRUNM_2_2(self) -> list[str]:
         return ["RVCRUNM.S"]
 

src/python_testing/TC_SEAR_1_2.py

@@ -190,14 +190,6 @@ async def read_and_validate_progress(self, step):
                                         f"Progress entry should have a null TotalOperationalTime value (Status is {p.status})")
                 # TODO how to check that InitialTimeEstimate is either null or uint32?
 
-    # Sends and out-of-band command to the rvc-app
-    def write_to_app_pipe(self, command):
-        with open(self.app_pipe, "w") as app_pipe:
-            app_pipe.write(command + "\n")
-        # Allow some time for the command to take effect.
-        # This removes the test flakyness which is very annoying for everyone in CI.
-        sleep(0.001)
-
     def TC_SEAR_1_2(self) -> list[str]:
         return ["SEAR.S"]
 
@@ -245,7 +237,9 @@ async def test_TC_SEAR_1_2(self):
 
             test_step = "Manually intervene to remove one or more entries in the SupportedMaps list"
             self.print_step("10", test_step)
-            if not self.is_ci:
+            if self.is_ci:
+                self.write_to_app_pipe('{"Name": "RemoveMap", "MapId": 3}')
+            else:
                 self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
 
             await self.read_and_validate_supported_maps(step=11)
@@ -278,7 +272,9 @@ async def test_TC_SEAR_1_2(self):
 
             test_step = "Manually intervene to add one or more entries to the SupportedMaps list"
             self.print_step("14", test_step)
-            if not self.is_ci:
+            if self.is_ci:
+                self.write_to_app_pipe('{"Name": "RemoveMap", "MapId": 3}')
+            else:
                 self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
 
             await self.read_and_validate_supported_maps(step=15)
@@ -311,15 +307,17 @@ async def test_TC_SEAR_1_2(self):
 
             test_step = "Manually intervene to remove one or more entries from the SupportedAreas list"
             self.print_step("18", test_step)
-            if not self.is_ci:
+            if self.is_ci:
+                self.write_to_app_pipe('{"Name": "RemoveMap", "MapId": 3}')
+            else:
                 self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
 
             await self.read_and_validate_supported_areas(step=19)
             new_supported_areas = self.areaid_list
             asserts.assert_true(len(old_supported_areas) > len(new_supported_areas), "Failed to remove area(s)")
 
             # NOTE the following operations are all part of step 19 - read all these attributes and check the data consistency
-            #     after removing areas(s)
+            #     after removing areas(s)   
 
             await self.read_and_validate_selected_areas(step=19)
 
@@ -343,7 +341,9 @@ async def test_TC_SEAR_1_2(self):
 
             test_step = "Manually intervene to add one or more entries to the SupportedAreas list"
             self.print_step("22", test_step)
-            if not self.is_ci:
+            if self.is_ci:
+                self.write_to_app_pipe('{"Name": "RemoveMap", "MapId": 3}')
+            else:
                 self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
 
             await self.read_and_validate_supported_areas(step=23)

src/python_testing/TC_SEAR_1_3.py

@@ -74,14 +74,6 @@ async def send_cmd_select_areas_expect_response(self, step, new_areas, expected_
                              expected_response,
                              f"Command response ({ret.status}) doesn't match the expected one")
 
-    # Sends and out-of-band command to the rvc-app
-    def write_to_app_pipe(self, command):
-        with open(self.app_pipe, "w") as app_pipe:
-            app_pipe.write(command + "\n")
-        # Allow some time for the command to take effect.
-        # This removes the test flakiness which is very annoying for everyone in CI.
-        sleep(0.001)
-
     def TC_SEAR_1_3(self) -> list[str]:
         return ["SEAR.S"]
 

src/python_testing/TC_SEAR_1_5.py

@@ -89,15 +89,6 @@ async def send_cmd_skip_area_expect_response(self, step, skipped_area, expected_
                              expected_response,
                              f"Command response ({ret.status}) doesn't match the expected one")
 
-    # Sends and out-of-band command to the rvc-app
-
-    def write_to_app_pipe(self, command):
-        with open(self.app_pipe, "w") as app_pipe:
-            app_pipe.write(command + "\n")
-        # Allow some time for the command to take effect.
-        # This removes the test flakiness which is very annoying for everyone in CI.
-        sleep(0.001)
-
     def TC_SEAR_1_5(self) -> list[str]:
         return ["SEAR.S", "SEAR.S.C02.Rsp"]
 

src/python_testing/TC_SEAR_1_6.py

@@ -72,14 +72,6 @@ async def read_progress(self, step):
 
         return progress
 
-    # Sends and out-of-band command to the rvc-app
-    def write_to_app_pipe(self, command):
-        with open(self.app_pipe, "w") as app_pipe:
-            app_pipe.write(command + "\n")
-        # Allow some time for the command to take effect.
-        # This removes the test flakiness which is very annoying for everyone in CI.
-        sleep(0.001)
-
     def TC_SEAR_1_6(self) -> list[str]:
         return ["SEAR.S", "SEAR.S.A0005", "SEAR.S.A0000", "SEAR.S.A0002", "SEAR.S.M.HAS_MANUAL_OPERATING_STATE_CONTROL"]
 

src/python_testing/matter_testing_support.py

@@ -877,6 +877,60 @@ def get_test_desc(self, test: str) -> str:
         except AttributeError:
             return test
 
+    # Sends an out-of-band command to a Matter app
+    def write_to_app_pipe(self, command):
+        """
+        Use the following environment variables:
+        
+         - LINUX_DUT_IP 
+            * if not provided, the Matter app is assumed to run on the same machine as this test,
+              such as during CI, and the commands are sent to it using a local named pipe
+            * if provided, the commands for writing to the named pipe are forwarded to the DUT
+        - LINUX_DUT_UNAME
+            * if LINUX_DUT_IP is provided, use this for the DUT user name
+            * if LINUX_DUT_NAME is not provided, a default value is used ("root")
+            * To ensure this script can log in to the DUT with the user name above, if a remote password is needed:
+                 + Step 1: If you do not have a key, create one using ssh-keygen
+                 + Step 2: Authorize this key on the remote host: run ssh-copy-id user@ip once, using your password
+                 + Step 3: From now on ssh user@ip will no longer ask for your password
+        - LINUX_DUT_APPNAME
+            * if LINUX_DUT_IP is provided, use this for the name of the Matter application used by the named pipe
+            * for example: "rvc", "air_quality", "all_clusters" etc
+            * the named pipe file name format is chip_<linux_dut_appname>_fifo_<pid> (e.g. chip_rvc_1009)
+            * if not provided, a default value is used ("rvc")
+        """
+        import os
+        dut_ip = os.getenv('LINUX_DUT_IP')
+
+        if dut_ip is None:
+            with open(self.app_pipe, "w") as app_pipe:
+                app_pipe.write(command + "\n")
+            # TODO(#31239): remove the need for sleep
+            sleep(0.001)
+        else:
+            print(f"DUT IP address: {dut_ip}")
+
+            dut_uname = os.getenv('LINUX_DUT_UNAME')
+            if dut_uname is None:
+                dut_uname = "root"
+                print(f"Using default DUT user name (root)")
+            else:      
+                print(f"Using DUT user name: {dut_uname}")
+
+            app_name = os.getenv('LINUX_DUT_APPNAME')
+            if app_name is None:
+                app_name = "rvc"
+                print(f"Using default DUT app name (rvc)")
+            else:      
+                print(f"Using DUT app name: {app_name}")    
+
+            pid = self.matter_test_config.app_pid
+            #use the dut's IP address below
+            ip = dut_ip
+            command_fixed = command.replace('\"','\\"')
+            cmd = "echo \"%s\" | ssh %s@%s \'cat > /tmp/chip_%s_fifo_%d\'" % (command_fixed, dut_uname, ip, app_name, pid)
+            os.system(cmd)
+
     # Override this if the test requires a different default timeout.
     # This value will be overridden if a timeout is supplied on the command line.
     @property