diff --git a/src/python_testing/TC_OCC_3_1.py b/src/python_testing/TC_OCC_3_1.py index 2d7fb0e2de05e6..9fa5f144b5d430 100644 --- a/src/python_testing/TC_OCC_3_1.py +++ b/src/python_testing/TC_OCC_3_1.py @@ -27,6 +27,7 @@ # --trace-to json:${TRACE_TEST_JSON}.json # --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto # --endpoint 1 +# --app-pipe /tmp/chip_all_clusters_fifo_ # --bool-arg simulate_occupancy:true # factory-reset: true # quiet: true @@ -91,25 +92,9 @@ def pics_TC_OCC_3_1(self) -> list[str]: ] return pics - # Sends and out-of-band command to the all-clusters-app - def write_to_app_pipe(self, command): - # CI app pipe id creation - self.app_pipe = "/tmp/chip_all_clusters_fifo_" - if self.is_ci: - app_pid = self.matter_test_config.app_pid - if app_pid == 0: - asserts.fail("The --app-pid flag must be set when using named pipe") - self.app_pipe = self.app_pipe + str(app_pid) - - 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) - time.sleep(0.001) - @async_test_body async def test_TC_OCC_3_1(self): hold_time = 10 if not self.is_ci else 1.0 # 10 seconds for occupancy state hold time - self.step(1) # Commissioning already done self.step(2) @@ -133,7 +118,7 @@ async def test_TC_OCC_3_1(self): if self.is_ci: # CI call to trigger unoccupied. - self.write_to_app_pipe('{"Name":"SetOccupancy", "EndpointId": 1, "Occupancy": 0}') + self.write_to_app_pipe({"Name": "SetOccupancy", "EndpointId": 1, "Occupancy": 0}) else: self.wait_for_user_input( prompt_msg="Type any letter and press ENTER after the sensor occupancy is unoccupied state (occupancy attribute = 0)") @@ -157,7 +142,7 @@ async def test_TC_OCC_3_1(self): self.step("5a") # CI call to trigger on if self.is_ci: - self.write_to_app_pipe('{"Name":"SetOccupancy", "EndpointId": 1, "Occupancy": 1}') + self.write_to_app_pipe({"Name": "SetOccupancy", "EndpointId": 1, "Occupancy": 1}) else: # Trigger occupancy sensor to change Occupancy attribute value to 1 => TESTER ACTION on DUT self.wait_for_user_input(prompt_msg="Type any letter and press ENTER after a sensor occupancy is triggered.") @@ -183,7 +168,7 @@ async def test_TC_OCC_3_1(self): self.step(6) if self.is_ci: # CI call to trigger unoccupied. - self.write_to_app_pipe('{"Name":"SetOccupancy", "EndpointId": 1, "Occupancy": 0}') + self.write_to_app_pipe({"Name": "SetOccupancy", "EndpointId": 1, "Occupancy": 0}) if has_hold_time: time.sleep(hold_time + 2.0) # add some extra 2 seconds to ensure hold time has passed. diff --git a/src/python_testing/TC_OCC_3_2.py b/src/python_testing/TC_OCC_3_2.py index 10051631436efd..0eb7c907aa16ff 100644 --- a/src/python_testing/TC_OCC_3_2.py +++ b/src/python_testing/TC_OCC_3_2.py @@ -29,6 +29,7 @@ # --passcode 20202021 # --trace-to json:${TRACE_TEST_JSON}.json # --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto +# --app-pipe /tmp/chip_all_clusters_fifo_ # --endpoint 1 # --bool-arg simulate_occupancy:true # factory-reset: true @@ -41,7 +42,6 @@ # [TC-OCC-3.2] test precedure step 3a, 3c import logging -import time import chip.clusters as Clusters from chip.testing.matter_testing import (ClusterAttributeChangeAccumulator, MatterBaseTest, TestStep, async_test_body, @@ -83,21 +83,6 @@ def pics_TC_OCC_3_2(self) -> list[str]: ] return pics - # Sends and out-of-band command to the all-clusters-app - def write_to_app_pipe(self, command): - # CI app pipe id creation - self.app_pipe = "/tmp/chip_all_clusters_fifo_" - if self.is_ci: - app_pid = self.matter_test_config.app_pid - if app_pid == 0: - asserts.fail("The --app-pid flag must be set when using named pipe") - self.app_pipe = self.app_pipe + str(app_pid) - - 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) - time.sleep(0.001) - @async_test_body async def test_TC_OCC_3_2(self): endpoint_id = self.get_endpoint() @@ -136,7 +121,7 @@ async def test_TC_OCC_3_2(self): self.step("3a") # CI call to trigger off if self.is_ci: - self.write_to_app_pipe('{"Name":"SetOccupancy", "EndpointId": 1, "Occupancy": 0}') + self.write_to_app_pipe({"Name": "SetOccupancy", "EndpointId": 1, "Occupancy": 0}) else: self.wait_for_user_input(prompt_msg="Type any letter and press ENTER after DUT goes back to unoccupied state.") @@ -150,7 +135,7 @@ async def test_TC_OCC_3_2(self): # CI call to trigger on if self.is_ci: - self.write_to_app_pipe('{"Name":"SetOccupancy", "EndpointId": 1, "Occupancy": 1}') + self.write_to_app_pipe({"Name": "SetOccupancy", "EndpointId": 1, "Occupancy": 1}) else: self.wait_for_user_input( prompt_msg="Type any letter and press ENTER after the sensor occupancy is triggered and its occupancy state changed.") diff --git a/src/python_testing/TC_OpstateCommon.py b/src/python_testing/TC_OpstateCommon.py index 1a4cfbe7150e30..81147c3e3db6ee 100644 --- a/src/python_testing/TC_OpstateCommon.py +++ b/src/python_testing/TC_OpstateCommon.py @@ -109,11 +109,11 @@ def init_test(self): app_pid = get_pid("chip-all-clusters-app") if app_pid is None: 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) + self.app_pid = app_pid def send_raw_manual_or_pipe_command(self, command: dict, msg: str): if self.is_ci: - self.write_to_app_pipe(command) + self.write_to_app_pipe(command, app_pipe_name=self.app_pipe, app_pid=self.app_pid) time.sleep(0.1) else: prompt = msg if msg is not None else "Press Enter when ready." diff --git a/src/python_testing/TC_RVCCLEANM_2_1.py b/src/python_testing/TC_RVCCLEANM_2_1.py index 5c5271e0be01ae..eae59b6a30926d 100644 --- a/src/python_testing/TC_RVCCLEANM_2_1.py +++ b/src/python_testing/TC_RVCCLEANM_2_1.py @@ -32,6 +32,7 @@ # --int-arg PIXIT.RVCCLEANM.MODE_CHANGE_FAIL:1 # --int-arg PIXIT.RVCCLEANM.MODE_CHANGE_OK:2 # --endpoint 1 +# --app-pipe /tmp/chip_rvc_fifo_ # --trace-to json:${TRACE_TEST_JSON}.json # --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto # factory-reset: true @@ -57,7 +58,6 @@ def __init__(self, *args): self.mode_ok = 0 self.mode_fail = 0 self.is_ci = False - self.app_pipe = "/tmp/chip_rvc_fifo_" async def read_mod_attribute_expect_success(self, endpoint, attribute): cluster = Clusters.Objects.RvcCleanMode @@ -92,11 +92,6 @@ async def test_TC_RVCCLEANM_2_1(self): self.mode_ok = self.matter_test_config.global_test_params['PIXIT.RVCCLEANM.MODE_CHANGE_OK'] self.mode_fail = self.matter_test_config.global_test_params['PIXIT.RVCCLEANM.MODE_CHANGE_FAIL'] self.is_ci = self.check_pics("PICS_SDK_CI_ONLY") - if self.is_ci: - app_pid = self.matter_test_config.app_pid - if app_pid == 0: - 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) asserts.assert_true(self.check_pics("RVCCLEANM.S.A0000"), "RVCCLEANM.S.A0000 must be supported") asserts.assert_true(self.check_pics("RVCCLEANM.S.A0001"), "RVCCLEANM.S.A0001 must be supported") diff --git a/src/python_testing/TC_RVCCLEANM_2_2.py b/src/python_testing/TC_RVCCLEANM_2_2.py index a34d0cda2a12a5..1bfe8ea37365a1 100644 --- a/src/python_testing/TC_RVCCLEANM_2_2.py +++ b/src/python_testing/TC_RVCCLEANM_2_2.py @@ -29,6 +29,7 @@ # --discriminator 1234 # --passcode 20202021 # --PICS examples/rvc-app/rvc-common/pics/rvc-app-pics-values +# --app-pipe /tmp/chip_rvc_fifo_ # --endpoint 1 # --trace-to json:${TRACE_TEST_JSON}.json # --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto @@ -63,7 +64,6 @@ def __init__(self, *args): self.old_clean_mode_dut = 0 self.new_clean_mode_th = 0 self.is_ci = False - self.app_pipe = "/tmp/chip_rvc_fifo_" async def read_mod_attribute_expect_success(self, cluster, attribute): return await self.read_single_attribute_check_success( @@ -107,11 +107,6 @@ async def test_TC_RVCCLEANM_2_2(self): self.directmodech_bit_mask = Clusters.RvcCleanMode.Bitmaps.Feature.kDirectModeChange self.endpoint = self.get_endpoint() self.is_ci = self.check_pics("PICS_SDK_CI_ONLY") - if self.is_ci: - app_pid = self.matter_test_config.app_pid - if app_pid == 0: - asserts.fail("The --app-pid flag must be set when PICS_SDK_CI_ONLY is set.c") - self.app_pipe = self.app_pipe + str(app_pid) asserts.assert_true(self.check_pics("RVCCLEANM.S"), "RVCCLEANM.S must be supported") asserts.assert_true(self.check_pics("RVCRUNM.S.A0000"), "RVCRUNM.S.A0000 must be supported") diff --git a/src/python_testing/TC_RVCOPSTATE_2_1.py b/src/python_testing/TC_RVCOPSTATE_2_1.py index 4fc63fbe236632..f90f90e8d4d572 100644 --- a/src/python_testing/TC_RVCOPSTATE_2_1.py +++ b/src/python_testing/TC_RVCOPSTATE_2_1.py @@ -30,6 +30,7 @@ # --passcode 20202021 # --PICS examples/rvc-app/rvc-common/pics/rvc-app-pics-values # --endpoint 1 +# --app-pipe /tmp/chip_rvc_fifo_ # --trace-to json:${TRACE_TEST_JSON}.json # --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto # factory-reset: true @@ -49,7 +50,6 @@ def __init__(self, *args): super().__init__(*args) self.endpoint = None self.is_ci = False - self.app_pipe = "/tmp/chip_rvc_fifo_" async def read_mod_attribute_expect_success(self, endpoint, attribute): cluster = Clusters.Objects.RvcOperationalState @@ -90,11 +90,6 @@ async def test_TC_RVCOPSTATE_2_1(self): self.endpoint = self.get_endpoint() asserts.assert_false(self.endpoint is None, "--endpoint must be included on the command line in.") self.is_ci = self.check_pics("PICS_SDK_CI_ONLY") - if self.is_ci: - app_pid = self.matter_test_config.app_pid - if app_pid == 0: - 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) cluster = Clusters.RvcOperationalState attributes = cluster.Attributes diff --git a/src/python_testing/TC_RVCOPSTATE_2_3.py b/src/python_testing/TC_RVCOPSTATE_2_3.py index ce583a7f527e90..ecab70d255b145 100644 --- a/src/python_testing/TC_RVCOPSTATE_2_3.py +++ b/src/python_testing/TC_RVCOPSTATE_2_3.py @@ -30,6 +30,7 @@ # --passcode 20202021 # --PICS examples/rvc-app/rvc-common/pics/rvc-app-pics-values # --endpoint 1 +# --app-pipe /tmp/chip_rvc_fifo_ # --trace-to json:${TRACE_TEST_JSON}.json # --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto # factory-reset: true @@ -99,7 +100,6 @@ def __init__(self, *args): super().__init__(*args) self.endpoint = None self.is_ci = False - self.app_pipe = "/tmp/chip_rvc_fifo_" async def read_mod_attribute_expect_success(self, endpoint, attribute): cluster = Clusters.Objects.RvcOperationalState @@ -161,11 +161,6 @@ async def test_TC_RVCOPSTATE_2_3(self): self.endpoint = self.get_endpoint() asserts.assert_false(self.endpoint is None, "--endpoint must be included on the command line in.") self.is_ci = self.check_pics("PICS_SDK_CI_ONLY") - if self.is_ci: - app_pid = self.matter_test_config.app_pid - if app_pid == 0: - asserts.fail("The --app-pid flag must be set when PICS_SDK_CI_ONLY is set.c") - self.app_pipe = self.app_pipe + str(app_pid) asserts.assert_true(self.check_pics("RVCOPSTATE.S.A0003"), "RVCOPSTATE.S.A0003 must be supported") asserts.assert_true(self.check_pics("RVCOPSTATE.S.A0004"), "RVCOPSTATE.S.A0004 must be supported") diff --git a/src/python_testing/TC_RVCOPSTATE_2_4.py b/src/python_testing/TC_RVCOPSTATE_2_4.py index c1bbb5509c8f8d..1ce70cc9d2bbd6 100644 --- a/src/python_testing/TC_RVCOPSTATE_2_4.py +++ b/src/python_testing/TC_RVCOPSTATE_2_4.py @@ -30,6 +30,7 @@ # --passcode 20202021 # --PICS examples/rvc-app/rvc-common/pics/rvc-app-pics-values # --endpoint 1 +# --app-pipe /tmp/chip_rvc_fifo_ # --trace-to json:${TRACE_TEST_JSON}.json # --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto # factory-reset: true @@ -64,7 +65,6 @@ def __init__(self, *args): super().__init__(*args) self.endpoint = None self.is_ci = False - self.app_pipe = "/tmp/chip_rvc_fifo_" async def read_mod_attribute_expect_success(self, endpoint, attribute): cluster = Clusters.Objects.RvcOperationalState @@ -106,11 +106,6 @@ async def test_TC_RVCOPSTATE_2_4(self): self.endpoint = self.get_endpoint() asserts.assert_false(self.endpoint is None, "--endpoint must be included on the command line in.") self.is_ci = self.check_pics("PICS_SDK_CI_ONLY") - if self.is_ci: - app_pid = self.matter_test_config.app_pid - if app_pid == 0: - asserts.fail("The --app-pid flag must be set when PICS_SDK_CI_ONLY is set.c") - self.app_pipe = self.app_pipe + str(app_pid) asserts.assert_true(self.check_pics("RVCOPSTATE.S.A0004"), "RVCOPSTATE.S.A0004 must be supported") asserts.assert_true(self.check_pics("RVCOPSTATE.S.C04.Tx"), "RVCOPSTATE.S.C04.Tx must be supported") diff --git a/src/python_testing/TC_RVCRUNM_2_1.py b/src/python_testing/TC_RVCRUNM_2_1.py index 802c87fad52f0f..71ef7cd920fd37 100644 --- a/src/python_testing/TC_RVCRUNM_2_1.py +++ b/src/python_testing/TC_RVCRUNM_2_1.py @@ -32,6 +32,7 @@ # --endpoint 1 # --int-arg PIXIT.RVCRUNM.MODE_CHANGE_OK:0 # --int-arg PIXIT.RVCRUNM.MODE_CHANGE_FAIL:2 +# --app-pipe /tmp/chip_rvc_fifo_ # --trace-to json:${TRACE_TEST_JSON}.json # --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto # factory-reset: true @@ -58,7 +59,6 @@ def __init__(self, *args): self.mode_ok = 0 self.mode_fail = 0 self.is_ci = False - self.app_pipe = "/tmp/chip_rvc_fifo_" async def read_mod_attribute_expect_success(self, endpoint, attribute): cluster = Clusters.Objects.RvcRunMode @@ -87,11 +87,6 @@ async def test_TC_RVCRUNM_2_1(self): self.mode_ok = self.matter_test_config.global_test_params['PIXIT.RVCRUNM.MODE_CHANGE_OK'] self.mode_fail = self.matter_test_config.global_test_params['PIXIT.RVCRUNM.MODE_CHANGE_FAIL'] self.is_ci = self.check_pics("PICS_SDK_CI_ONLY") - if self.is_ci: - app_pid = self.matter_test_config.app_pid - if app_pid == 0: - asserts.fail("The --app-pid flag must be set when PICS_SDK_CI_ONLY is set.c") - self.app_pipe = self.app_pipe + str(app_pid) asserts.assert_true(self.check_pics("RVCRUNM.S.A0000"), "RVCRUNM.S.A0000 must be supported") asserts.assert_true(self.check_pics("RVCRUNM.S.A0001"), "RVCRUNM.S.A0001 must be supported") diff --git a/src/python_testing/TC_RVCRUNM_2_2.py b/src/python_testing/TC_RVCRUNM_2_2.py index 55c5a6ef17d61c..c88f7c8bbc3d86 100644 --- a/src/python_testing/TC_RVCRUNM_2_2.py +++ b/src/python_testing/TC_RVCRUNM_2_2.py @@ -30,6 +30,7 @@ # --discriminator 1234 # --passcode 20202021 # --endpoint 1 +# --app-pipe /tmp/chip_rvc_fifo_ # --int-arg PIXIT.RVCRUNM.MODE_A:1 # --int-arg PIXIT.RVCRUNM.MODE_B:2 # --trace-to json:${TRACE_TEST_JSON}.json @@ -84,7 +85,6 @@ def __init__(self, *args): self.supported_run_modes_dut = [] self.idle_mode_dut = 0 self.is_ci = False - self.app_pipe = "/tmp/chip_rvc_fifo_" async def read_mod_attribute_expect_success(self, cluster, attribute): return await self.read_single_attribute_check_success( @@ -138,11 +138,6 @@ async def test_TC_RVCRUNM_2_2(self): self.is_ci = self.check_pics("PICS_SDK_CI_ONLY") self.mode_a = self.matter_test_config.global_test_params['PIXIT.RVCRUNM.MODE_A'] self.mode_b = self.matter_test_config.global_test_params['PIXIT.RVCRUNM.MODE_B'] - if self.is_ci: - app_pid = self.matter_test_config.app_pid - if app_pid == 0: - asserts.fail("The --app-pid flag must be set when PICS_SDK_CI_ONLY is set.c") - self.app_pipe = self.app_pipe + str(app_pid) asserts.assert_true(self.check_pics("RVCRUNM.S"), "RVCRUNM.S must be supported") # I think that the following PICS should be listed in the preconditions section in the test plan as if either diff --git a/src/python_testing/TC_SEAR_1_2.py b/src/python_testing/TC_SEAR_1_2.py index 039a2ff9d071af..9670a4231fff70 100644 --- a/src/python_testing/TC_SEAR_1_2.py +++ b/src/python_testing/TC_SEAR_1_2.py @@ -31,6 +31,7 @@ # --passcode 20202021 # --PICS examples/rvc-app/rvc-common/pics/rvc-app-pics-values # --endpoint 1 +# --app-pipe /tmp/chip_rvc_fifo_ # --trace-to json:${TRACE_TEST_JSON}.json # --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto # factory-reset: true @@ -50,7 +51,6 @@ def __init__(self, *args): super().__init__(*args) self.endpoint = None self.is_ci = False - self.app_pipe = "/tmp/chip_rvc_fifo_" self.mapid_list = [] # this must be kept in sync with the definitions from the Common Landmark Semantic Tag Namespace @@ -207,12 +207,6 @@ async def test_TC_SEAR_1_2(self): self.endpoint = self.get_endpoint(default=1) asserts.assert_false(self.endpoint is None, "--endpoint must be included on the command line in.") self.is_ci = self.check_pics("PICS_SDK_CI_ONLY") - if self.is_ci: - app_pid = self.matter_test_config.app_pid - if app_pid == 0: - 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) - self.print_step(1, "Commissioning, already done") # Ensure that the device is in the correct state diff --git a/src/python_testing/TC_SEAR_1_3.py b/src/python_testing/TC_SEAR_1_3.py index 50818c3c6033fb..ccd5998233921d 100644 --- a/src/python_testing/TC_SEAR_1_3.py +++ b/src/python_testing/TC_SEAR_1_3.py @@ -31,6 +31,7 @@ # --passcode 20202021 # --PICS examples/rvc-app/rvc-common/pics/rvc-app-pics-values # --endpoint 1 +# --app-pipe /tmp/chip_rvc_fifo_ # --json-arg PIXIT.SEAR.VALID_AREAS:'[7, 1234567]' # --trace-to json:${TRACE_TEST_JSON}.json # --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto @@ -50,7 +51,6 @@ def __init__(self, *args): super().__init__(*args) self.endpoint = None self.is_ci = False - self.app_pipe = "/tmp/chip_rvc_fifo_" async def read_sear_attribute_expect_success(self, endpoint, attribute): cluster = Clusters.Objects.ServiceArea @@ -91,12 +91,6 @@ async def test_TC_SEAR_1_3(self): self.endpoint = self.get_endpoint() asserts.assert_false(self.endpoint is None, "--endpoint must be included on the command line in.") self.is_ci = self.check_pics("PICS_SDK_CI_ONLY") - if self.is_ci: - app_pid = self.matter_test_config.app_pid - if app_pid == 0: - 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) - self.print_step(1, "Commissioning, already done") # Ensure that the device is in the correct state diff --git a/src/python_testing/TC_SEAR_1_5.py b/src/python_testing/TC_SEAR_1_5.py index 96209191e6458b..f39908a29994fa 100644 --- a/src/python_testing/TC_SEAR_1_5.py +++ b/src/python_testing/TC_SEAR_1_5.py @@ -31,6 +31,7 @@ # --passcode 20202021 # --PICS examples/rvc-app/rvc-common/pics/rvc-app-pics-values # --endpoint 1 +# --app-pipe /tmp/chip_rvc_fifo_ # --trace-to json:${TRACE_TEST_JSON}.json # --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto # factory-reset: true @@ -50,7 +51,6 @@ def __init__(self, *args): super().__init__(*args) self.endpoint = None self.is_ci = False - self.app_pipe = "/tmp/chip_rvc_fifo_" async def read_sear_attribute_expect_success(self, endpoint, attribute): cluster = Clusters.Objects.ServiceArea @@ -105,11 +105,6 @@ async def test_TC_SEAR_1_5(self): self.endpoint = self.get_endpoint() asserts.assert_false(self.endpoint is None, "--endpoint must be included on the command line in.") self.is_ci = self.check_pics("PICS_SDK_CI_ONLY") - if self.is_ci: - app_pid = self.matter_test_config.app_pid - if app_pid == 0: - 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) self.print_step(1, "Commissioning, already done") diff --git a/src/python_testing/TC_SEAR_1_6.py b/src/python_testing/TC_SEAR_1_6.py index 6e5c2157d0b35e..88e380fab228ed 100644 --- a/src/python_testing/TC_SEAR_1_6.py +++ b/src/python_testing/TC_SEAR_1_6.py @@ -31,6 +31,7 @@ # --passcode 20202021 # --PICS examples/rvc-app/rvc-common/pics/rvc-app-pics-values # --endpoint 1 +# --app-pipe /tmp/chip_rvc_fifo_ # --trace-to json:${TRACE_TEST_JSON}.json # --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto # factory-reset: true @@ -50,7 +51,6 @@ def __init__(self, *args): super().__init__(*args) self.endpoint = None self.is_ci = False - self.app_pipe = "/tmp/chip_rvc_fifo_" async def read_sear_attribute_expect_success(self, endpoint, attribute): cluster = Clusters.Objects.ServiceArea @@ -88,11 +88,6 @@ async def test_TC_SEAR_1_6(self): self.endpoint = self.get_endpoint() asserts.assert_false(self.endpoint is None, "--endpoint must be included on the command line in.") self.is_ci = self.check_pics("PICS_SDK_CI_ONLY") - if self.is_ci: - app_pid = self.matter_test_config.app_pid - if app_pid == 0: - 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) self.print_step(1, "Commissioning, already done") diff --git a/src/python_testing/matter_testing_infrastructure/chip/testing/matter_testing.py b/src/python_testing/matter_testing_infrastructure/chip/testing/matter_testing.py index 4625f9ca3b32b3..39a8cfcea3f987 100644 --- a/src/python_testing/matter_testing_infrastructure/chip/testing/matter_testing.py +++ b/src/python_testing/matter_testing_infrastructure/chip/testing/matter_testing.py @@ -653,6 +653,7 @@ class MatterTestConfig: timeout: typing.Union[int, None] = None endpoint: typing.Union[int, None] = 0 app_pid: int = 0 + pipe_name: typing.Union[str, None] = None fail_on_skipped_tests: bool = False commissioning_method: Optional[str] = None @@ -966,8 +967,6 @@ 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 @@ -1031,36 +1030,43 @@ 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): + def write_to_app_pipe(self, command_dict: dict, app_pipe_name: Optional[str] = None, app_pid: Optional[int] = None): """ - Sends an out-of-band command to a Matter app. + Send an out-of-band command to a Matter app. + Args: + command_dict (dict): dictionary with the command and data + app_pipe_name (Optional[str], optional): Name of the cluster pipe file (i.e. /tmp/chip_all_clusters_fifo_ or /tmp/chip_rvc_fifo_). Defaults to None. + app_pid (Optional[uint], optional): pid of the process for app_pipe_name. Defaults to None. + + Extra: + By default app_pipe_name is taken from CI argument --app_pipe. Parameter (app_pipe_name) overwrites this value. + By default app_pid is taken from CI argument --app-pid. Parameter (app_pid) overwrites this value. - Use the following environment variables: + This method 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() + app_pipe_name = self.matter_test_config.app_pipe + if app_pid is None: + app_pid = self.matter_test_config.app_pid if not isinstance(app_pipe_name, str): - raise TypeError("the named pipe must be provided as a string value") + 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") + raise TypeError("The command must be passed as a dictionary value") import json command = json.dumps(command_dict) @@ -1068,13 +1074,22 @@ def write_to_app_pipe(self, command_dict: dict, app_pipe_name: Optional[str] = N import os dut_ip = os.getenv('LINUX_DUT_IP') + # Checks for concatenate app_pipe and app_pid + if not isinstance(app_pid, int): + raise TypeError("The --app-pid flag is not instance of int") + # Verify we have a valid app-id + if app_pid == 0: + asserts.fail("app_pid is 0 , is the flag --app-pid set?. app-id flag must be set in order to write to pipe.") + app_pipe_name = app_pipe_name + str(app_pid) + if dut_ip is None: if not os.path.exists(app_pipe_name): # Named pipes are unique, so we MUST have consistent PID/paths - # set up for them to work. + # Set up for them to work. logging.error("Named pipe %r does NOT exist" % app_pipe_name) raise FileNotFoundError("CANNOT FIND %r" % app_pipe_name) with open(app_pipe_name, "w") as app_pipe: + logger.info(f"Sending out-of-band command: {command} to file: {app_pipe_name}") app_pipe.write(command + "\n") # TODO(#31239): remove the need for sleep sleep(0.001) @@ -1960,6 +1975,7 @@ def convert_args_to_matter_config(args: argparse.Namespace) -> MatterTestConfig: config.timeout = args.timeout # This can be none, we pull the default from the test if it's unspecified config.endpoint = args.endpoint # This can be None, the get_endpoint function allows the tests to supply a default config.app_pid = 0 if args.app_pid is None else args.app_pid + config.app_pipe = args.app_pipe config.fail_on_skipped_tests = args.fail_on_skipped config.controller_node_id = args.controller_node_id @@ -2017,6 +2033,7 @@ def parse_matter_test_args(argv: Optional[List[str]] = None) -> MatterTestConfig 'and NodeID to assign if commissioning (default: %d)' % _DEFAULT_DUT_NODE_ID, nargs="+") basic_group.add_argument('--endpoint', type=int, default=None, help="Endpoint under test") basic_group.add_argument('--app-pid', type=int, default=0, help="The PID of the app against which the test is going to run") + basic_group.add_argument('--app-pipe', type=str, default=None, help="The path of the app to send an out-of-band command") basic_group.add_argument('--timeout', type=int, help="Test timeout in seconds") basic_group.add_argument("--PICS", help="PICS file path", type=str)