Python test scripts: enable remote use of named pipe (project-chip#34950

)

* initial commit

* fix issue with the previous commit

* edit change

* Restyled by autopep8

* fix linter issues

* Restyled by isort

* Apply suggestions from code review

Co-authored-by: William <[email protected]>

* address code review feedback

* Added documentation about the write_to_app_pipe command to the testing docs.

* Restyled by prettier-markdown

* Fixed typos.

* Apply suggestions from code review

Pleasing restyler.

* Apply suggestions from code review

Co-authored-by: Kiel Oleson <[email protected]>

* Restyled by prettier-markdown

* address code review comments

* Restyled by autopep8

* fix bug introduced by the previous commit

* Restyled by autopep8

* remove unused import

* reset to 5359c7b. refactor the code to use a suggestion from the code review.

* fix conflict

* fix conflict

* fix conflict

* Restyled by autopep8

* Restyled by isort

---------

Co-authored-by: Restyled.io <[email protected]>
Co-authored-by: William <[email protected]>
Co-authored-by: William Hicklin <[email protected]>
Co-authored-by: Kiel Oleson <[email protected]>

.github/.wordlist.txt

@@ -454,6 +454,7 @@ DS
 duplicative
 DUT
 DUTS
+DUT's
 DV
 DVK
 dynload

docs/testing/python.md

@@ -497,6 +497,61 @@ Fabric admin for default controller:
   second_ctrl = fa.new_fabric_admin.NewController(nodeId=node_id)
 ```
 
+## Automating manual steps
+
+Some test plans have manual steps that require the tester to manually change the
+state of the DUT. To run these tests in a CI environment, specific example apps
+can be built such that these manual steps can be achieved by Matter or
+named-pipe commands.
+
+In the case that all the manual steps in a test script can be achieved just
+using Matter commands, you can check if the `PICS_SDK_CI_ONLY` PICS is set to
+decide if the test script should send the required Matter commands to perform
+the manual step.
+
+```python
+self.is_ci = self.check_pics("PICS_SDK_CI_ONLY")
+```
+
+In the case that a test script requires the use of named-pipe commands to
+achieve the manual steps, you can use the `write_to_app_pipe(command)` to send
+these commands. This command requires the test class to define a `self.app_pipe`
+string value with the name of the pipe. This depends on how the app is set up.
+
+If the name of the pipe is dynamic and based on the app's PID, the following
+snippet can be added to the start of tests that use the `write_to_app_pipe`
+method.
+
+```python
+app_pid = self.matter_test_config.app_pid
+if app_pid != 0:
+	self.is_ci = true
+	self.app_pipe = "/tmp/chip_<app name>_fifo_" + str(app_pid)
+```
+
+This requires the test to be executed with the `--app-pid` flag set if the
+manual steps should be executed by the script. This flag sets the process ID of
+the DUT's matter application.
+
+### Running on a separate machines
+
+If the DUT and test script are running on different machines, the
+`write_to_app_pipe` method can send named-pipe commands to the DUT via ssh. This
+requires two additional environment variables:
+
+-   `LINUX_DUT_IP` sets the DUT's IP address
+-   `LINUX_DUT_UNAME` sets the DUT's ssh username. If not set, this is assumed
+    to be `root`.
+
+The `write_to_app_pipe` also requires that ssh-keys are set up to access the DUT
+from the machine running the test script without a password. You can follow
+these steps to set this up:
+
+1. If you do not have a key, create one using `ssh-keygen`.
+2. Authorize this key on the remote host: run `ssh-copy-id user@ip` once, using
+   your password.
+3. From now on `ssh user@ip` will no longer ask for your password.
+
 ## Other support utilities
 
 -   `basic_composition_support`

src/python_testing/TC_OpstateCommon.py

@@ -15,7 +15,6 @@
 #    limitations under the License.
 #
 
-import json
 import logging
 import queue
 import time
@@ -112,12 +111,7 @@ 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):
+    def send_raw_manual_or_pipe_command(self, command: dict):
         if self.is_ci:
             self.write_to_app_pipe(command)
             time.sleep(0.1)
@@ -134,7 +128,7 @@ def send_manual_or_pipe_command(self, device: str, name: str, operation: str, pa
         if param is not None:
             command["Param"] = param
 
-        self.send_raw_manual_or_pipe_command(json.dumps(command))
+        self.send_raw_manual_or_pipe_command(command)
 
     async def send_cmd(self, endpoint, cmd, timedRequestTimeoutMs=None):
         logging.info(f"##### Command {cmd}")

src/python_testing/TC_RVCCLEANM_2_1.py

@@ -28,7 +28,6 @@
 # === END CI TEST ARGUMENTS ===
 
 import logging
-from time import sleep
 
 import chip.clusters as Clusters
 from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main, type_matches
@@ -65,14 +64,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"]
 
@@ -107,7 +98,7 @@ async def test_TC_RVCCLEANM_2_1(self):
 
         # Ensure that the device is in the correct state
         if self.is_ci:
-            self.write_to_app_pipe('{"Name": "Reset"}')
+            self.write_to_app_pipe({"Name": "Reset"})
 
         self.print_step(2, "Read SupportedModes attribute")
         supported_modes = await self.read_mod_attribute_expect_success(endpoint=self.endpoint, attribute=attributes.SupportedModes)

src/python_testing/TC_RVCCLEANM_2_2.py

@@ -28,7 +28,6 @@
 # === END CI TEST ARGUMENTS ===
 
 import enum
-from time import sleep
 
 import chip.clusters as Clusters
 from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main, type_matches
@@ -94,14 +93,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):
         # TODO Replace 0x8000 with python object of RVCCLEAN FEATURE bit map when implemented
@@ -123,7 +114,7 @@ async def test_TC_RVCCLEANM_2_2(self):
 
         # Ensure that the device is in the correct state
         if self.is_ci:
-            self.write_to_app_pipe('{"Name": "Reset"}')
+            self.write_to_app_pipe({"Name": "Reset"})
 
         self.print_step(
             2, "Manually put the device in a state in which the RVC Run Mode cluster’s CurrentMode attribute is set to a mode without the Idle mode tag.")

src/python_testing/TC_RVCOPSTATE_2_1.py

@@ -28,7 +28,6 @@
 # === END CI TEST ARGUMENTS ===
 
 import logging
-from time import sleep
 
 import chip.clusters as Clusters
 from chip.clusters.Types import NullValue
@@ -72,14 +71,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"]
 
@@ -100,7 +91,7 @@ async def test_TC_RVCOPSTATE_2_1(self):
 
         # Ensure that the device is in the correct state
         if self.is_ci:
-            self.write_to_app_pipe('{"Name": "Reset"}')
+            self.write_to_app_pipe({"Name": "Reset"})
 
         if self.check_pics("RVCOPSTATE.S.A0000"):
             self.print_step(2, "Read PhaseList attribute")
@@ -195,15 +186,15 @@ async def test_TC_RVCOPSTATE_2_1(self):
                 test_step = "Manually put the device in the error state"
                 self.print_step("6g", test_step)
                 if self.is_ci:
-                    self.write_to_app_pipe('{"Name": "ErrorEvent", "Error": "UnableToStartOrResume"}')
+                    self.write_to_app_pipe({"Name": "ErrorEvent", "Error": "UnableToStartOrResume"})
                 else:
                     self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
                 await self.read_and_validate_opstate(step="6h", expected_state=Clusters.OperationalState.Enums.OperationalStateEnum.kError)
             if self.check_pics("RVCOPSTATE.S.M.ST_SEEKING_CHARGER"):
                 test_step = "Manually put the device in the seeking charger state"
                 self.print_step("6i", test_step)
                 if self.is_ci:
-                    self.write_to_app_pipe('{"Name": "Reset"}')
+                    self.write_to_app_pipe({"Name": "Reset"})
                     await self.send_run_change_to_mode_cmd(1)
                     await self.send_run_change_to_mode_cmd(0)
                 else:
@@ -213,15 +204,15 @@ async def test_TC_RVCOPSTATE_2_1(self):
                 test_step = "Manually put the device in the charging state"
                 self.print_step("6k", test_step)
                 if self.is_ci:
-                    self.write_to_app_pipe('{"Name": "ChargerFound"}')
+                    self.write_to_app_pipe({"Name": "ChargerFound"})
                 else:
                     self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
                 await self.read_and_validate_opstate(step="6l", expected_state=Clusters.RvcOperationalState.Enums.OperationalStateEnum.kCharging)
             if self.check_pics("RVCOPSTATE.S.M.ST_DOCKED"):
                 test_step = "Manually put the device in the docked state"
                 self.print_step("6m", test_step)
                 if self.is_ci:
-                    self.write_to_app_pipe('{"Name": "Charged"}')
+                    self.write_to_app_pipe({"Name": "Charged"})
                 else:
                     self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
                 await self.read_and_validate_opstate(step="6n", expected_state=Clusters.RvcOperationalState.Enums.OperationalStateEnum.kDocked)
@@ -254,87 +245,87 @@ async def test_TC_RVCOPSTATE_2_1(self):
                 test_step = "Manually put the device in the unable to start or resume error state"
                 self.print_step("7c", test_step)
                 if self.is_ci:
-                    self.write_to_app_pipe('{"Name": "ErrorEvent", "Error": "UnableToStartOrResume"}')
+                    self.write_to_app_pipe({"Name": "ErrorEvent", "Error": "UnableToStartOrResume"})
                 else:
                     self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
                 await self.read_and_validate_operror(step="7d", expected_error=Clusters.OperationalState.Enums.ErrorStateEnum.kUnableToStartOrResume)
             if self.check_pics("RVCOPSTATE.S.M.ERR_UNABLE_TO_COMPLETE_OPERATION"):
                 test_step = "Manually put the device in the unable to complete operation error state"
                 self.print_step("7e", test_step)
                 if self.is_ci:
-                    self.write_to_app_pipe('{"Name": "ErrorEvent", "Error": "UnableToCompleteOperation"}')
+                    self.write_to_app_pipe({"Name": "ErrorEvent", "Error": "UnableToCompleteOperation"})
                 else:
                     self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
                 await self.read_and_validate_operror(step="7f", expected_error=Clusters.OperationalState.Enums.ErrorStateEnum.kUnableToCompleteOperation)
             if self.check_pics("RVCOPSTATE.S.M.ERR_COMMAND_INVALID_IN_STATE"):
                 test_step = "Manually put the device in the command invalid error state"
                 self.print_step("7g", test_step)
                 if self.is_ci:
-                    self.write_to_app_pipe('{"Name": "ErrorEvent", "Error": "CommandInvalidInState"}')
+                    self.write_to_app_pipe({"Name": "ErrorEvent", "Error": "CommandInvalidInState"})
                 else:
                     self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
                 await self.read_and_validate_operror(step="7h", expected_error=Clusters.OperationalState.Enums.ErrorStateEnum.kCommandInvalidInState)
             if self.check_pics("RVCOPSTATE.S.M.ERR_FAILED_TO_FIND_CHARGING_DOCK"):
                 test_step = "Manually put the device in the failed to find dock error state"
                 self.print_step("7i", test_step)
                 if self.is_ci:
-                    self.write_to_app_pipe('{"Name": "ErrorEvent", "Error": "FailedToFindChargingDock"}')
+                    self.write_to_app_pipe({"Name": "ErrorEvent", "Error": "FailedToFindChargingDock"})
                 else:
                     self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
                 await self.read_and_validate_operror(step="7j", expected_error=Clusters.RvcOperationalState.Enums.ErrorStateEnum.kFailedToFindChargingDock)
             if self.check_pics("RVCOPSTATE.S.M.ERR_STUCK"):
                 test_step = "Manually put the device in the stuck error state"
                 self.print_step("7k", test_step)
                 if self.is_ci:
-                    self.write_to_app_pipe('{"Name": "ErrorEvent", "Error": "Stuck"}')
+                    self.write_to_app_pipe({"Name": "ErrorEvent", "Error": "Stuck"})
                 else:
                     self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
                 await self.read_and_validate_operror(step="7l", expected_error=Clusters.RvcOperationalState.Enums.ErrorStateEnum.kStuck)
             if self.check_pics("RVCOPSTATE.S.M.ERR_DUST_BIN_MISSING"):
                 test_step = "Manually put the device in the dust bin missing error state"
                 self.print_step("7m", test_step)
                 if self.is_ci:
-                    self.write_to_app_pipe('{"Name": "ErrorEvent", "Error": "DustBinMissing"}')
+                    self.write_to_app_pipe({"Name": "ErrorEvent", "Error": "DustBinMissing"})
                 else:
                     self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
                 await self.read_and_validate_operror(step="7n", expected_error=Clusters.RvcOperationalState.Enums.ErrorStateEnum.kDustBinMissing)
             if self.check_pics("RVCOPSTATE.S.M.ERR_DUST_BIN_FULL"):
                 test_step = "Manually put the device in the dust bin full error state"
                 self.print_step("7o", test_step)
                 if self.is_ci:
-                    self.write_to_app_pipe('{"Name": "ErrorEvent", "Error": "DustBinFull"}')
+                    self.write_to_app_pipe({"Name": "ErrorEvent", "Error": "DustBinFull"})
                 else:
                     self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
                 await self.read_and_validate_operror(step="7p", expected_error=Clusters.RvcOperationalState.Enums.ErrorStateEnum.kDustBinFull)
             if self.check_pics("RVCOPSTATE.S.M.ERR_WATER_TANK_EMPTY"):
                 test_step = "Manually put the device in the water tank empty error state"
                 self.print_step("7q", test_step)
                 if self.is_ci:
-                    self.write_to_app_pipe('{"Name": "ErrorEvent", "Error": "WaterTankEmpty"}')
+                    self.write_to_app_pipe({"Name": "ErrorEvent", "Error": "WaterTankEmpty"})
                 else:
                     self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
                 await self.read_and_validate_operror(step="7r", expected_error=Clusters.RvcOperationalState.Enums.ErrorStateEnum.kWaterTankEmpty)
             if self.check_pics("RVCOPSTATE.S.M.ERR_WATER_TANK_MISSING"):
                 test_step = "Manually put the device in the water tank missing error state"
                 self.print_step("7s", test_step)
                 if self.is_ci:
-                    self.write_to_app_pipe('{"Name": "ErrorEvent", "Error": "WaterTankMissing"}')
+                    self.write_to_app_pipe({"Name": "ErrorEvent", "Error": "WaterTankMissing"})
                 else:
                     self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
                 await self.read_and_validate_operror(step="7t", expected_error=Clusters.RvcOperationalState.Enums.ErrorStateEnum.kWaterTankMissing)
             if self.check_pics("RVCOPSTATE.S.M.ERR_WATER_TANK_LID_OPEN"):
                 test_step = "Manually put the device in the water tank lid open error state"
                 self.print_step("7u", test_step)
                 if self.is_ci:
-                    self.write_to_app_pipe('{"Name": "ErrorEvent", "Error": "WaterTankLidOpen"}')
+                    self.write_to_app_pipe({"Name": "ErrorEvent", "Error": "WaterTankLidOpen"})
                 else:
                     self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
                 await self.read_and_validate_operror(step="7v", expected_error=Clusters.RvcOperationalState.Enums.ErrorStateEnum.kWaterTankLidOpen)
             if self.check_pics("RVCOPSTATE.S.M.ERR_MOP_CLEANING_PAD_MISSING"):
                 test_step = "Manually put the device in the mop cleaning pad missing error state"
                 self.print_step("7w", test_step)
                 if self.is_ci:
-                    self.write_to_app_pipe('{"Name": "ErrorEvent", "Error": "MopCleaningPadMissing"}')
+                    self.write_to_app_pipe({"Name": "ErrorEvent", "Error": "MopCleaningPadMissing"})
                 else:
                     self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
                 await self.read_and_validate_operror(step="7x", expected_error=Clusters.RvcOperationalState.Enums.ErrorStateEnum.kMopCleaningPadMissing)