From e89783c8a81d729cfc7dae83e02718c5753e0511 Mon Sep 17 00:00:00 2001 From: Petru Lauric Date: Tue, 13 Aug 2024 14:03:04 +0300 Subject: [PATCH] initial commit --- src/python_testing/TC_OpstateCommon.py | 5 -- src/python_testing/TC_RVCCLEANM_2_1.py | 8 --- src/python_testing/TC_RVCCLEANM_2_2.py | 8 --- src/python_testing/TC_RVCOPSTATE_2_1.py | 8 --- src/python_testing/TC_RVCOPSTATE_2_3.py | 8 --- src/python_testing/TC_RVCOPSTATE_2_4.py | 8 --- src/python_testing/TC_RVCRUNM_2_1.py | 8 --- src/python_testing/TC_RVCRUNM_2_2.py | 8 --- src/python_testing/TC_SEAR_1_2.py | 26 +++++----- src/python_testing/TC_SEAR_1_3.py | 8 --- src/python_testing/TC_SEAR_1_5.py | 9 ---- src/python_testing/TC_SEAR_1_6.py | 8 --- src/python_testing/matter_testing_support.py | 54 ++++++++++++++++++++ 13 files changed, 67 insertions(+), 99 deletions(-) diff --git a/src/python_testing/TC_OpstateCommon.py b/src/python_testing/TC_OpstateCommon.py index edc72f253eead5..16a1035f5b84fa 100644 --- a/src/python_testing/TC_OpstateCommon.py +++ b/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) diff --git a/src/python_testing/TC_RVCCLEANM_2_1.py b/src/python_testing/TC_RVCCLEANM_2_1.py index cac18f9601f6ff..49a4dc4181105f 100644 --- a/src/python_testing/TC_RVCCLEANM_2_1.py +++ b/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"] diff --git a/src/python_testing/TC_RVCCLEANM_2_2.py b/src/python_testing/TC_RVCCLEANM_2_2.py index 8aaae7ff78738f..1dce7a43c139ff 100644 --- a/src/python_testing/TC_RVCCLEANM_2_2.py +++ b/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 diff --git a/src/python_testing/TC_RVCOPSTATE_2_1.py b/src/python_testing/TC_RVCOPSTATE_2_1.py index 42a2d02d6fae9c..3deb0c5a39675e 100644 --- a/src/python_testing/TC_RVCOPSTATE_2_1.py +++ b/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"] diff --git a/src/python_testing/TC_RVCOPSTATE_2_3.py b/src/python_testing/TC_RVCOPSTATE_2_3.py index 1d915641260d3b..55340ebd967b59 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) diff --git a/src/python_testing/TC_RVCOPSTATE_2_4.py b/src/python_testing/TC_RVCOPSTATE_2_4.py index b680453b06c287..63fb8b2e555372 100644 --- a/src/python_testing/TC_RVCOPSTATE_2_4.py +++ b/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"] diff --git a/src/python_testing/TC_RVCRUNM_2_1.py b/src/python_testing/TC_RVCRUNM_2_1.py index e98a1840629864..966b7355614039 100644 --- a/src/python_testing/TC_RVCRUNM_2_1.py +++ b/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"] diff --git a/src/python_testing/TC_RVCRUNM_2_2.py b/src/python_testing/TC_RVCRUNM_2_2.py index dca866adb9f62e..1a7e4fe2e7e7b2 100644 --- a/src/python_testing/TC_RVCRUNM_2_2.py +++ b/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"] diff --git a/src/python_testing/TC_SEAR_1_2.py b/src/python_testing/TC_SEAR_1_2.py index 939f4be638df5b..ac7d006ce26422 100644 --- a/src/python_testing/TC_SEAR_1_2.py +++ b/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,7 +307,9 @@ 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) @@ -319,7 +317,7 @@ async def test_TC_SEAR_1_2(self): 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) diff --git a/src/python_testing/TC_SEAR_1_3.py b/src/python_testing/TC_SEAR_1_3.py index 2ea2e86b7d198d..fafdddb5c5e30b 100644 --- a/src/python_testing/TC_SEAR_1_3.py +++ b/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"] diff --git a/src/python_testing/TC_SEAR_1_5.py b/src/python_testing/TC_SEAR_1_5.py index a90c1feaf9c329..5879120bf1b6e0 100644 --- a/src/python_testing/TC_SEAR_1_5.py +++ b/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"] diff --git a/src/python_testing/TC_SEAR_1_6.py b/src/python_testing/TC_SEAR_1_6.py index aaa6ac788dd427..ac72204eca1f2d 100644 --- a/src/python_testing/TC_SEAR_1_6.py +++ b/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"] diff --git a/src/python_testing/matter_testing_support.py b/src/python_testing/matter_testing_support.py index d44754f379d92e..07be9a565fbabe 100644 --- a/src/python_testing/matter_testing_support.py +++ b/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__fifo_ (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