From dd875401ae2f5c2f39b9e69f156cd6fabebe2aa8 Mon Sep 17 00:00:00 2001 From: Petru Lauric <81822411+plauric@users.noreply.github.com> Date: Thu, 29 Aug 2024 22:18:49 -0400 Subject: [PATCH] Python test scripts: enable remote use of named pipe (#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 * 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 * 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 5359c7b6a829cbcfdcb7b51d35d54fcbfd9958ea. 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 Co-authored-by: William Co-authored-by: William Hicklin Co-authored-by: Kiel Oleson --- .github/.wordlist.txt | 1 + docs/testing/python.md | 55 +++++++++++++++++++ src/python_testing/TC_OpstateCommon.py | 10 +--- src/python_testing/TC_RVCCLEANM_2_1.py | 11 +--- src/python_testing/TC_RVCCLEANM_2_2.py | 11 +--- src/python_testing/TC_RVCOPSTATE_2_1.py | 41 ++++++-------- src/python_testing/TC_RVCOPSTATE_2_3.py | 20 ++----- src/python_testing/TC_RVCOPSTATE_2_4.py | 21 ++----- src/python_testing/TC_RVCRUNM_2_1.py | 11 +--- src/python_testing/TC_RVCRUNM_2_2.py | 13 +---- src/python_testing/TC_SEAR_1_2.py | 19 ++----- src/python_testing/TC_SEAR_1_3.py | 15 +---- src/python_testing/TC_SEAR_1_5.py | 16 +----- src/python_testing/TC_SEAR_1_6.py | 11 +--- src/python_testing/matter_testing_support.py | 58 ++++++++++++++++++++ 15 files changed, 161 insertions(+), 152 deletions(-) diff --git a/.github/.wordlist.txt b/.github/.wordlist.txt index c4f48058b70416..caad2a8f8462e3 100644 --- a/.github/.wordlist.txt +++ b/.github/.wordlist.txt @@ -454,6 +454,7 @@ DS duplicative DUT DUTS +DUT's DV DVK dynload diff --git a/docs/testing/python.md b/docs/testing/python.md index 3b5c8ffc74abf7..ec358b8796403e 100644 --- a/docs/testing/python.md +++ b/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__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` diff --git a/src/python_testing/TC_OpstateCommon.py b/src/python_testing/TC_OpstateCommon.py index 45d5b0918a76f2..e0e252ea180eb0 100644 --- a/src/python_testing/TC_OpstateCommon.py +++ b/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}") diff --git a/src/python_testing/TC_RVCCLEANM_2_1.py b/src/python_testing/TC_RVCCLEANM_2_1.py index cac18f9601f6ff..8e5013ad4c2dbd 100644 --- a/src/python_testing/TC_RVCCLEANM_2_1.py +++ b/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) diff --git a/src/python_testing/TC_RVCCLEANM_2_2.py b/src/python_testing/TC_RVCCLEANM_2_2.py index e3a92033b3bdf4..6c4885ff4efdfa 100644 --- a/src/python_testing/TC_RVCCLEANM_2_2.py +++ b/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.") diff --git a/src/python_testing/TC_RVCOPSTATE_2_1.py b/src/python_testing/TC_RVCOPSTATE_2_1.py index 42a2d02d6fae9c..b15fc927430374 100644 --- a/src/python_testing/TC_RVCOPSTATE_2_1.py +++ b/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,7 +186,7 @@ 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) @@ -203,7 +194,7 @@ async def test_TC_RVCOPSTATE_2_1(self): 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,7 +204,7 @@ 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) @@ -221,7 +212,7 @@ async def test_TC_RVCOPSTATE_2_1(self): 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,7 +245,7 @@ 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) @@ -262,7 +253,7 @@ async def test_TC_RVCOPSTATE_2_1(self): 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) @@ -270,7 +261,7 @@ async def test_TC_RVCOPSTATE_2_1(self): 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) @@ -278,7 +269,7 @@ async def test_TC_RVCOPSTATE_2_1(self): 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) @@ -286,7 +277,7 @@ async def test_TC_RVCOPSTATE_2_1(self): 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) @@ -294,7 +285,7 @@ async def test_TC_RVCOPSTATE_2_1(self): 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) @@ -302,7 +293,7 @@ async def test_TC_RVCOPSTATE_2_1(self): 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) @@ -310,7 +301,7 @@ async def test_TC_RVCOPSTATE_2_1(self): 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) @@ -318,7 +309,7 @@ async def test_TC_RVCOPSTATE_2_1(self): 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) @@ -326,7 +317,7 @@ async def test_TC_RVCOPSTATE_2_1(self): 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) @@ -334,7 +325,7 @@ async def test_TC_RVCOPSTATE_2_1(self): 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) diff --git a/src/python_testing/TC_RVCOPSTATE_2_3.py b/src/python_testing/TC_RVCOPSTATE_2_3.py index 1d915641260d3b..e9dc801cc11071 100644 --- a/src/python_testing/TC_RVCOPSTATE_2_3.py +++ b/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) @@ -182,7 +174,7 @@ async def test_TC_RVCOPSTATE_2_3(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"}) test_step = "Manually put the device in a state where it can receive a Pause command" self.print_step(2, test_step) @@ -269,7 +261,7 @@ async def test_TC_RVCOPSTATE_2_3(self): test_step = "Manually put the device in the Stopped(0x00) operational state" self.print_step(24, test_step) if self.is_ci: - self.write_to_app_pipe('{"Name": "Reset"}') + self.write_to_app_pipe({"Name": "Reset"}) else: self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n") @@ -283,7 +275,7 @@ async def test_TC_RVCOPSTATE_2_3(self): test_step = "Manually put the device in the Error(0x03) operational state" self.print_step(28, 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") @@ -297,10 +289,10 @@ async def test_TC_RVCOPSTATE_2_3(self): test_step = "Manually put the device in the Charging(0x41) operational state" self.print_step(32, 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) - 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") @@ -320,7 +312,7 @@ async def test_TC_RVCOPSTATE_2_3(self): test_step = "Manually put the device in the Docked(0x42) operational state" self.print_step(38, 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") diff --git a/src/python_testing/TC_RVCOPSTATE_2_4.py b/src/python_testing/TC_RVCOPSTATE_2_4.py index 7f03d33b196833..8180cb696296e9 100644 --- a/src/python_testing/TC_RVCOPSTATE_2_4.py +++ b/src/python_testing/TC_RVCOPSTATE_2_4.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 @@ -90,14 +89,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"] @@ -128,13 +119,13 @@ async def test_TC_RVCOPSTATE_2_4(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.M.ST_ERROR"): 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"}') + self.write_to_app_pipe({"Name": "ErrorEvent", "Error": "UnableToStartOrResume"}) else: self.wait_for_user_input(prompt_msg=f"{step_name}, and press Enter when ready.") @@ -146,9 +137,9 @@ async def test_TC_RVCOPSTATE_2_4(self): 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"}') + self.write_to_app_pipe({"Name": "Reset"}) + self.write_to_app_pipe({"Name": "Docked"}) + self.write_to_app_pipe({"Name": "Charging"}) else: self.wait_for_user_input(prompt_msg=f"{step_name}, and press Enter when ready.") @@ -160,7 +151,7 @@ async def test_TC_RVCOPSTATE_2_4(self): 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"}') + self.write_to_app_pipe({"Name": "Charged"}) else: self.wait_for_user_input(prompt_msg=f"{step_name}, and press Enter when ready.") diff --git a/src/python_testing/TC_RVCRUNM_2_1.py b/src/python_testing/TC_RVCRUNM_2_1.py index e98a1840629864..2693cb041a16b9 100644 --- a/src/python_testing/TC_RVCRUNM_2_1.py +++ b/src/python_testing/TC_RVCRUNM_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 @@ -60,14 +59,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"] @@ -102,7 +93,7 @@ async def test_TC_RVCRUNM_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) diff --git a/src/python_testing/TC_RVCRUNM_2_2.py b/src/python_testing/TC_RVCRUNM_2_2.py index b0d1233010f932..3fd4268f2c24f5 100644 --- a/src/python_testing/TC_RVCRUNM_2_2.py +++ b/src/python_testing/TC_RVCRUNM_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 @@ -110,14 +109,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"] @@ -158,7 +149,7 @@ async def test_TC_RVCRUNM_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"}) test_step = ("Manually put the device in a RVC Run Mode cluster mode with " "the Idle(0x4000) mode tag and in a device state that allows changing to either " "of these modes: %i, %i" % (self.mode_a, self.mode_b)) @@ -237,7 +228,7 @@ async def test_TC_RVCRUNM_2_2(self): if op_state not in valid_op_states: self.print_step(9, "Manually put the device in one of Stopped(0x00), Paused(0x02), Charging(0x41) or Docked(0x42)") 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="Manually put the device in one of Stopped(0x00), Paused(0x02), Charging(0x41) or Docked(0x42), and press Enter when ready.\n") diff --git a/src/python_testing/TC_SEAR_1_2.py b/src/python_testing/TC_SEAR_1_2.py index 1c08b87cad2bf5..2c253efa860038 100644 --- a/src/python_testing/TC_SEAR_1_2.py +++ b/src/python_testing/TC_SEAR_1_2.py @@ -29,7 +29,6 @@ # === END CI TEST ARGUMENTS === import logging -from time import sleep import chip.clusters as Clusters from chip.clusters.Types import NullValue @@ -191,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"] @@ -217,7 +208,7 @@ async def test_TC_SEAR_1_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"}) if self.check_pics("SEAR.S.F02"): await self.read_and_validate_supported_maps(step=2) @@ -247,7 +238,7 @@ 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 self.is_ci: - self.write_to_app_pipe('{"Name": "RemoveMap", "MapId": 3}') + 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") @@ -282,7 +273,7 @@ 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 self.is_ci: - self.write_to_app_pipe('{"Name": "AddMap", "MapId": 1, "MapName": "NewTestMap1"}') + self.write_to_app_pipe({"Name": "AddMap", "MapId": 1, "MapName": "NewTestMap1"}) else: self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n") @@ -317,7 +308,7 @@ 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 self.is_ci: - self.write_to_app_pipe('{"Name": "RemoveArea", "AreaId": 10050}') + self.write_to_app_pipe({"Name": "RemoveArea", "AreaId": 10050}) else: self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n") @@ -351,7 +342,7 @@ 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 self.is_ci: - self.write_to_app_pipe('{"Name": "AddArea", "AreaId": 42, "MapId": 1, "LocationName": "NewTestArea1"}') + self.write_to_app_pipe({"Name": "AddArea", "AreaId": 42, "MapId": 1, "LocationName": "NewTestArea1"}) else: self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n") diff --git a/src/python_testing/TC_SEAR_1_3.py b/src/python_testing/TC_SEAR_1_3.py index 867cdcbe234353..9613e583fa1a74 100644 --- a/src/python_testing/TC_SEAR_1_3.py +++ b/src/python_testing/TC_SEAR_1_3.py @@ -29,7 +29,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 @@ -74,14 +73,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"] @@ -100,7 +91,7 @@ async def test_TC_SEAR_1_3(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"}) supported_area_ids = await self.read_supported_areas(step=2) asserts.assert_true(len(self.supported_areas) > 0, "SupportedAreas is empty") @@ -135,7 +126,7 @@ async def test_TC_SEAR_1_3(self): test_step = f"Manually intervene to put the device in a state that allows it to execute the SelectAreas({supported_area_ids}) command" self.print_step("9", test_step) if self.is_ci: - self.write_to_app_pipe('{"Name": "Reset"}') + self.write_to_app_pipe({"Name": "Reset"}) else: self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n") @@ -151,7 +142,7 @@ async def test_TC_SEAR_1_3(self): test_step = f"Manually intervene to put the device in a state that allows it to execute the SelectAreas({valid_area_id}) command, and put the device in a non-idle state" self.print_step("14", test_step) if self.is_ci: - self.write_to_app_pipe('{"Name": "Reset"}') + self.write_to_app_pipe({"Name": "Reset"}) await self.send_single_cmd(cmd=Clusters.Objects.RvcRunMode.Commands.ChangeToMode(newMode=1), endpoint=self.endpoint) else: self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n") diff --git a/src/python_testing/TC_SEAR_1_5.py b/src/python_testing/TC_SEAR_1_5.py index a90c1feaf9c329..50b9080dff08c2 100644 --- a/src/python_testing/TC_SEAR_1_5.py +++ b/src/python_testing/TC_SEAR_1_5.py @@ -29,7 +29,6 @@ # === END CI TEST ARGUMENTS === import logging -from time import sleep import chip.clusters as Clusters from chip.clusters.Types import NullValue @@ -89,15 +88,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"] @@ -116,7 +106,7 @@ async def test_TC_SEAR_1_5(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"}) supported_area_ids = await self.read_supported_areas(step=2) asserts.assert_true(len(supported_area_ids) > 0, "SupportedAreas is empty") @@ -150,7 +140,7 @@ async def test_TC_SEAR_1_5(self): test_step = "Manually intervene to put the device in a state that allows it to execute the SkipArea command" self.print_step("7", test_step) if self.is_ci: - self.write_to_app_pipe('{"Name": "Reset"}') + self.write_to_app_pipe({"Name": "Reset"}) await self.send_single_cmd(cmd=Clusters.Objects.ServiceArea.Commands.SelectAreas(newAreas=[7, 1234567]), endpoint=self.endpoint) await self.send_single_cmd(cmd=Clusters.Objects.RvcRunMode.Commands.ChangeToMode(newMode=1), @@ -234,7 +224,7 @@ async def test_TC_SEAR_1_5(self): test_step = "Manually intervene to put the device in a state that allows it to execute the SkipArea command" self.print_step("18", test_step) if self.is_ci: - self.write_to_app_pipe('{"Name": "Reset"}') + self.write_to_app_pipe({"Name": "Reset"}) await self.send_single_cmd(cmd=Clusters.Objects.ServiceArea.Commands.SelectAreas(newAreas=[7, 1234567]), endpoint=self.endpoint) await self.send_single_cmd(cmd=Clusters.Objects.RvcRunMode.Commands.ChangeToMode(newMode=1), diff --git a/src/python_testing/TC_SEAR_1_6.py b/src/python_testing/TC_SEAR_1_6.py index aaa6ac788dd427..8e94971c8bc599 100644 --- a/src/python_testing/TC_SEAR_1_6.py +++ b/src/python_testing/TC_SEAR_1_6.py @@ -29,7 +29,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 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"] @@ -98,7 +89,7 @@ async def test_TC_SEAR_1_6(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"}) test_step = "Manually intervene to put the device in the idle state and ensure SupportedAreas and SelectedAreas are not empty" self.print_step("2", test_step) diff --git a/src/python_testing/matter_testing_support.py b/src/python_testing/matter_testing_support.py index c5e92482836b67..9b5a30e326e4ce 100644 --- a/src/python_testing/matter_testing_support.py +++ b/src/python_testing/matter_testing_support.py @@ -50,6 +50,8 @@ from chip.ChipDeviceCtrl import CommissioningParameters # isort: on +from time import sleep + import chip.clusters as Clusters import chip.logging import chip.native @@ -908,6 +910,8 @@ def __init__(self, *args): # List of accumulated problems across all tests self.problems = [] self.is_commissioning = False + # The named pipe name must be set in the derived classes + self.app_pipe = None def get_test_steps(self, test: str) -> list[TestStep]: ''' Retrieves the test step list for the given test @@ -971,6 +975,60 @@ def get_test_desc(self, test: str) -> str: except AttributeError: return test + def get_default_app_pipe_name(self) -> str: + return self.app_pipe + + def write_to_app_pipe(self, command_dict: dict, app_pipe_name: Optional[str] = None): + """ + Sends an out-of-band command to a Matter app. + + Use the following environment variables: + + - LINUX_DUT_IP + * if not provided, the Matter app is assumed to run on the same machine as the 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_USER + + * if LINUX_DUT_IP is provided, use this for the DUT user name + * If a remote password is needed, set up ssh keys to ensure that this script can log in to the DUT without a password: + + 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 + """ + + if app_pipe_name is None: + app_pipe_name = self.get_default_app_pipe_name() + + if not isinstance(app_pipe_name, str): + raise TypeError("the named pipe must be provided as a string value") + + if not isinstance(command_dict, dict): + raise TypeError("the command must be passed as a dictionary value") + + import json + command = json.dumps(command_dict) + + import os + dut_ip = os.getenv('LINUX_DUT_IP') + + if dut_ip is None: + with open(app_pipe_name, "w") as app_pipe: + app_pipe.write(command + "\n") + # TODO(#31239): remove the need for sleep + sleep(0.001) + else: + logging.info(f"Using DUT IP address: {dut_ip}") + + dut_uname = os.getenv('LINUX_DUT_USER') + asserts.assert_true(dut_uname is not None, "The LINUX_DUT_USER environment variable must be set") + + logging.info(f"Using DUT user name: {dut_uname}") + + command_fixed = command.replace('\"', '\\"') + cmd = "echo \"%s\" | ssh %s@%s \'cat > %s\'" % (command_fixed, dut_uname, dut_ip, app_pipe_name) + 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