From 231ed8e9df9ad993d74967219b49c1f532e6f0c5 Mon Sep 17 00:00:00 2001 From: Arkadiusz Bokowy Date: Thu, 26 Sep 2024 16:05:19 +0200 Subject: [PATCH 1/7] Make Subprocess wrapper class even more generic --- src/python_testing/TC_MCORE_FS_1_1.py | 4 +- .../chip/testing/tasks.py | 85 +++++++++++++------ 2 files changed, 60 insertions(+), 29 deletions(-) diff --git a/src/python_testing/TC_MCORE_FS_1_1.py b/src/python_testing/TC_MCORE_FS_1_1.py index 0d77bad8c6f268..7b11d5bae440e7 100755 --- a/src/python_testing/TC_MCORE_FS_1_1.py +++ b/src/python_testing/TC_MCORE_FS_1_1.py @@ -47,7 +47,7 @@ class AppServer(Subprocess): """Wrapper class for starting an application server in a subprocess.""" # Prefix for log messages from the application server. - PREFIX = "[SERVER]" + PREFIX = b"[SERVER]" def __init__(self, app: str, storage_dir: str, discriminator: int, passcode: int, port: int = 5540): storage_kvs_dir = tempfile.mkstemp(dir=storage_dir, prefix="kvs-app-")[1] @@ -56,7 +56,7 @@ def __init__(self, app: str, storage_dir: str, discriminator: int, passcode: int '--secured-device-port', str(port), "--discriminator", str(discriminator), "--passcode", str(passcode), - prefix=self.PREFIX) + output_cb=lambda line, is_stderr: self.PREFIX + line) def start(self): # Start process and block until it prints the expected output. diff --git a/src/python_testing/matter_testing_infrastructure/chip/testing/tasks.py b/src/python_testing/matter_testing_infrastructure/chip/testing/tasks.py index a73e73fbeb2bf2..64c3347537235c 100644 --- a/src/python_testing/matter_testing_infrastructure/chip/testing/tasks.py +++ b/src/python_testing/matter_testing_infrastructure/chip/testing/tasks.py @@ -13,57 +13,70 @@ # limitations under the License. import logging +import re import subprocess import sys import threading -import typing +from typing import BinaryIO, Callable, List, Optional, Union -def forward_f(prefix: bytes, - f_in: typing.BinaryIO, - f_out: typing.BinaryIO, - cb: typing.Optional[typing.Callable[[bytes, bool], None]] = None, +def forward_f(f_in: BinaryIO, + f_out: BinaryIO, + cb: Optional[Callable[[bytes, bool], bytes]] = None, is_stderr: bool = False): - """Forward f_in to f_out with a prefix attached. + """Forward f_in to f_out. - This function can optionally feed received lines to a callback function. + This function can optionally post-process received lines using a callback + function. """ while line := f_in.readline(): if cb is not None: - cb(line, is_stderr) - f_out.buffer.write(prefix) - f_out.buffer.write(line) + line = cb(line, is_stderr) + f_out.write(line) f_out.flush() class Subprocess(threading.Thread): - """Run a subprocess and optionally prefix its output.""" + """Run a subprocess in a thread.""" - def __init__(self, program: str, *args: typing.List[str], prefix: str = "", - output_cb: typing.Optional[typing.Callable[[bytes, bool], None]] = None): + def __init__(self, program: str, *args: List[str], + output_cb: Optional[Callable[[bytes, bool], bytes]] = None, + f_stdout: BinaryIO = sys.stdout.buffer, + f_stderr: BinaryIO = sys.stderr.buffer): """Initialize the subprocess. Args: program: The program to run. args: The arguments to the program. - prefix: A prefix to attach to the output. output_cb: A callback function to process the output. It should take two arguments: the output line bytes and the boolean indicating if the - output comes from stderr. + output comes from stderr. It should return the processed output. + f_stdout: The file to forward the stdout to. + f_stderr: The file to forward the stderr to. """ super().__init__() self.event = threading.Event() - self.prefix = prefix.encode() + self.event_started = threading.Event() self.program = program self.args = args self.output_cb = output_cb - self.expected_output = None + self.f_stdout = f_stdout + self.f_stderr = f_stderr + self.output_match = None + self.returncode = None + + def _set_output_match(self, patter: Union[str, re.Pattern]): + if isinstance(patter, str): + self.output_match = re.compile(re.escape(patter.encode())) + else: + self.output_match = patter def _check_output(self, line: bytes, is_stderr: bool): - if self.output_cb is not None: - self.output_cb(line, is_stderr) - if self.expected_output is not None and self.expected_output in line: + if self.output_match is not None and self.output_match.search(line): self.event.set() + if self.output_cb is not None: + line = self.output_cb(line, is_stderr) + return line def run(self): """Thread entry point.""" @@ -73,40 +86,53 @@ def run(self): stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + self.event_started.set() # Forward stdout and stderr with a tag attached. forwarding_stdout_thread = threading.Thread( target=forward_f, - args=(self.prefix, self.p.stdout, sys.stdout, self._check_output)) + args=(self.p.stdout, self.f_stdout, self._check_output)) forwarding_stdout_thread.start() forwarding_stderr_thread = threading.Thread( target=forward_f, - args=(self.prefix, self.p.stderr, sys.stderr, self._check_output, True)) + args=(self.p.stderr, self.f_stderr, self._check_output, True)) forwarding_stderr_thread.start() # Wait for the process to finish. - self.p.wait() + self.returncode = self.p.wait() forwarding_stdout_thread.join() forwarding_stderr_thread.join() - def start(self, expected_output: str = None, timeout: float = None): + def start(self, + expected_output: Optional[Union[str, re.Pattern]] = None, + timeout: Optional[float] = None): """Start a subprocess and optionally wait for a specific output.""" + if expected_output is not None: - self.expected_output = expected_output.encode() + self._set_output_match(expected_output) self.event.clear() + super().start() + # Wait for the thread to start, so the self.p attribute is available. + self.event_started.wait() + if expected_output is not None: if self.event.wait(timeout) is False: + # Terminate the process, so the Python interpreter will not + # hang on the join call in our thread entry point in case of + # Python process termination (not-caught exception). + self.p.terminate() raise TimeoutError("Expected output not found") self.expected_output = None def send(self, message: str, end: str = "\n", - expected_output: str = None, timeout: float = None): + expected_output: Optional[Union[str, re.Pattern]] = None, + timeout: Optional[float] = None): """Send a message to a process and optionally wait for a response.""" if expected_output is not None: - self.expected_output = expected_output.encode() + self._set_output_match(expected_output) self.event.clear() self.p.stdin.write((message + end).encode()) @@ -121,3 +147,8 @@ def terminate(self): """Terminate the subprocess and wait for it to finish.""" self.p.terminate() self.join() + + def wait(self, timeout: Optional[float] = None) -> int: + """Wait for the subprocess to finish.""" + self.join(timeout) + return self.returncode From aa6654da5e54f0957f6876b40e60676837f6d664 Mon Sep 17 00:00:00 2001 From: Arkadiusz Bokowy Date: Wed, 25 Sep 2024 15:15:12 +0200 Subject: [PATCH 2/7] Add ECOINFO_2_1 to CI --- src/python_testing/TC_ECOINFO_2_1.py | 103 ++++++++++++++++++--- src/python_testing/execute_python_tests.py | 1 - 2 files changed, 90 insertions(+), 14 deletions(-) diff --git a/src/python_testing/TC_ECOINFO_2_1.py b/src/python_testing/TC_ECOINFO_2_1.py index 202d1073779cb3..4d846f20e8e84a 100644 --- a/src/python_testing/TC_ECOINFO_2_1.py +++ b/src/python_testing/TC_ECOINFO_2_1.py @@ -15,20 +15,92 @@ # limitations under the License. # +# See https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md#defining-the-ci-test-arguments +# for details about the block below. +# +# === BEGIN CI TEST ARGUMENTS === +# test-runner-runs: +# run1: +# app: examples/fabric-admin/scripts/fabric-sync-app.py +# app-args: --app-admin=${FABRIC_ADMIN_APP} --app-bridge=${FABRIC_BRIDGE_APP} --stdin-pipe=dut-fsa-stdin --discriminator=1234 +# script-args: > +# --PICS src/app/tests/suites/certification/ci-pics-values +# --storage-path admin_storage.json +# --commissioning-method on-network +# --discriminator 1234 +# --passcode 20202021 +# --string-arg th_server_app_path:${ALL_CLUSTERS_APP} dut_fsa_stdin_pipe:dut-fsa-stdin +# --trace-to json:${TRACE_TEST_JSON}.json +# --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto +# script-start-delay: 5 +# factoryreset: true +# quiet: false +# === END CI TEST ARGUMENTS === + +import asyncio +import logging +import random +import tempfile + import chip.clusters as Clusters from chip.clusters.Types import NullValue from chip.interaction_model import Status from chip.tlv import uint from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main, type_matches from mobly import asserts +from TC_MCORE_FS_1_1 import AppServer class TC_ECOINFO_2_1(MatterBaseTest): + @async_test_body + async def setup_class(self): + super().setup_class() + + self.th_server = None + self.storage = None + + # Get the path to the TH_SERVER app from the user params. + if th_server_app := self.user_params.get("th_server_app_path"): + # Create a temporary storage directory for keeping KVS files. + self.storage = tempfile.TemporaryDirectory(prefix=self.__class__.__name__) + logging.info("Temporary storage directory: %s", self.storage.name) + + # Get the named pipe path for the DUT_FSA app input from the user params. + if dut_fsa_stdin_pipe := self.user_params.get("dut_fsa_stdin_pipe"): + self.dut_fsa_stdin = open(dut_fsa_stdin_pipe, "w") + + self.th_server_port = 5544 + self.th_server_discriminator = random.randint(0, 4095) + self.th_server_passcode = 20202021 + + # Start the server app. + self.th_server = AppServer( + th_server_app, + storage_dir=self.storage.name, + port=self.th_server_port, + discriminator=self.th_server_discriminator, + passcode=self.th_server_passcode) + self.th_server.start() + + # Add some server to the DUT_FSA's Aggregator/Bridge. + self.dut_fsa_stdin.write( + f"pairing onnetwork 2 {self.th_server_passcode}\n") + self.dut_fsa_stdin.flush() + # Wait for the commissioning to complete. + await asyncio.sleep(5) + + def teardown_class(self): + if self.th_server is not None: + self.th_server.terminate() + if self.storage is not None: + self.storage.cleanup() + super().teardown_class() + def _validate_device_directory(self, current_fabric_index, device_directory): for device in device_directory: if current_fabric_index != device.fabricIndex: - # Fabric sensitve field still exist in python, just that they have default values + # Fabric sensitive field still exist in python, just that they have default values asserts.assert_equal(device.deviceName, None, "Unexpected value in deviceName") asserts.assert_equal(device.deviceNameLastEdit, None, "Unexpected value in deviceNameLastEdit") asserts.assert_equal(device.bridgedEndpoint, 0, "Unexpected value in bridgedEndpoint") @@ -85,7 +157,7 @@ def _validate_device_directory(self, current_fabric_index, device_directory): def _validate_location_directory(self, current_fabric_index, location_directory): for location in location_directory: if current_fabric_index != location.fabricIndex: - # Fabric sensitve field still exist in python, just that they have default values + # Fabric sensitive field still exist in python, just that they have default values asserts.assert_equal(location.uniqueLocationID, "", "Unexpected value in uniqueLocationID") asserts.assert_equal(location.locationDescriptor.locationName, "", "Unexpected value in locationDescriptor.locationName") @@ -120,30 +192,35 @@ def _validate_location_directory(self, current_fabric_index, location_directory) asserts.assert_greater(location.locationDescriptorLastEdit, 0, "LocationDescriptorLastEdit must be non-zero") def steps_TC_ECOINFO_2_1(self) -> list[TestStep]: - steps = [TestStep(1, "Identify endpoints with Ecosystem Information Cluster", is_commissioning=True), - TestStep(2, "Reading DeviceDirectory Attribute"), - TestStep(3, "Reading LocationDirectory Attribute"), - TestStep(4, "Try Writing to DeviceDirectory Attribute"), - TestStep(5, "Try Writing to LocationDirectory Attribute"), - TestStep(6, "Repeating steps 2 to 5 for each endpoint identified in step 1")] - return steps + return [ + TestStep(0, "Commission DUT if not done", is_commissioning=True), + TestStep(1, "Identify endpoints with Ecosystem Information Cluster"), + TestStep(2, "Reading DeviceDirectory Attribute"), + TestStep(3, "Reading LocationDirectory Attribute"), + TestStep(4, "Try Writing to DeviceDirectory Attribute"), + TestStep(5, "Try Writing to LocationDirectory Attribute"), + TestStep(6, "Repeating steps 2 to 5 for each endpoint identified in step 1"), + ] @async_test_body async def test_TC_ECOINFO_2_1(self): + self.is_ci = self.check_pics('PICS_SDK_CI_ONLY') dev_ctrl = self.default_controller dut_node_id = self.dut_node_id - self.print_step(0, "Commissioning, already done") + # Commissioning - done + self.step(0) - pause_for_pre_condition = self.user_params.get("pause_for_pre_condition", False) - if pause_for_pre_condition: + if not self.is_ci: self.wait_for_user_input( - "Paused test to allow for manufacturer to satisfy precondition where one or more bridged devices of a supported type is connected to DUT") + "Paused test to allow for manufacturer to satisfy precondition where " + "one or more bridged devices of a supported type is connected to DUT") current_fabric_index = await self.read_single_attribute_check_success(cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.CurrentFabricIndex) self.step(1) endpoint_wild_card_read = await dev_ctrl.ReadAttribute(dut_node_id, [(Clusters.EcosystemInformation.Attributes.ClusterRevision)]) list_of_endpoints = list(endpoint_wild_card_read.keys()) + asserts.assert_greater(len(list_of_endpoints), 0, "Expecting at least one endpoint with Ecosystem Information Cluster") for idx, cluster_endpoint in enumerate(list_of_endpoints): if idx == 0: diff --git a/src/python_testing/execute_python_tests.py b/src/python_testing/execute_python_tests.py index 9e19aa142552dc..f2367335159bad 100644 --- a/src/python_testing/execute_python_tests.py +++ b/src/python_testing/execute_python_tests.py @@ -62,7 +62,6 @@ def main(search_directory, env_file): "TC_CCTRL_2_3.py", # They rely on example applications that inter-communicate and there is no example app that works right now "TC_DGGEN_3_2.py", # src/python_testing/test_testing/test_TC_DGGEN_3_2.py is the Unit test of this test "TC_EEVSE_Utils.py", # Shared code for TC_EEVSE, not a standalone test - "TC_ECOINFO_2_1.py", # They rely on example applications that inter-communicate and there is no example app that works right now "TC_ECOINFO_2_2.py", # They rely on example applications that inter-communicate and there is no example app that works right now "TC_EWATERHTRBase.py", # Shared code for TC_EWATERHTR, not a standalone test "TC_EnergyReporting_Utils.py", # Shared code for TC_EEM and TC_EPM, not a standalone test From 72d686fe8341127f9d0aa65d6083e1a6ded8218c Mon Sep 17 00:00:00 2001 From: Arkadiusz Bokowy Date: Wed, 25 Sep 2024 15:52:55 +0200 Subject: [PATCH 3/7] Add ECOINFO_2_2 to CI --- src/python_testing/TC_ECOINFO_2_2.py | 118 +++++++++++++++++---- src/python_testing/execute_python_tests.py | 1 - 2 files changed, 100 insertions(+), 19 deletions(-) diff --git a/src/python_testing/TC_ECOINFO_2_2.py b/src/python_testing/TC_ECOINFO_2_2.py index 24f39ab93f2a04..64ead03a2d0d4d 100644 --- a/src/python_testing/TC_ECOINFO_2_2.py +++ b/src/python_testing/TC_ECOINFO_2_2.py @@ -15,32 +15,97 @@ # limitations under the License. # +# See https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md#defining-the-ci-test-arguments +# for details about the block below. +# +# === BEGIN CI TEST ARGUMENTS === +# test-runner-runs: +# run1: +# app: examples/fabric-admin/scripts/fabric-sync-app.py +# app-args: --app-admin=${FABRIC_ADMIN_APP} --app-bridge=${FABRIC_BRIDGE_APP} --stdin-pipe=dut-fsa-stdin --discriminator=1234 +# script-args: > +# --PICS src/app/tests/suites/certification/ci-pics-values +# --storage-path admin_storage.json +# --commissioning-method on-network +# --discriminator 1234 +# --passcode 20202021 +# --string-arg th_server_app_path:${ALL_CLUSTERS_APP} dut_fsa_stdin_pipe:dut-fsa-stdin +# --trace-to json:${TRACE_TEST_JSON}.json +# --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto +# script-start-delay: 5 +# factoryreset: true +# quiet: false +# === END CI TEST ARGUMENTS === + +import asyncio +import logging +import random +import tempfile + import chip.clusters as Clusters from chip.interaction_model import Status from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main from mobly import asserts +from TC_MCORE_FS_1_1 import AppServer _DEVICE_TYPE_AGGREGGATOR = 0x000E class TC_ECOINFO_2_2(MatterBaseTest): + def setup_class(self): + super().setup_class() + + self.th_server = None + self.storage = None + + # Get the path to the TH_SERVER app from the user params. + if th_server_app := self.user_params.get("th_server_app_path"): + # Create a temporary storage directory for keeping KVS files. + self.storage = tempfile.TemporaryDirectory(prefix=self.__class__.__name__) + logging.info("Temporary storage directory: %s", self.storage.name) + + # Get the named pipe path for the DUT_FSA app input from the user params. + if dut_fsa_stdin_pipe := self.user_params.get("dut_fsa_stdin_pipe"): + self.dut_fsa_stdin = open(dut_fsa_stdin_pipe, "w") + + self.th_server_port = 5544 + self.th_server_discriminator = random.randint(0, 4095) + self.th_server_passcode = 20202021 + + # Start the server app. + self.th_server = AppServer( + th_server_app, + storage_dir=self.storage.name, + port=self.th_server_port, + discriminator=self.th_server_discriminator, + passcode=self.th_server_passcode) + self.th_server.start() + + def teardown_class(self): + if self.th_server is not None: + self.th_server.terminate() + if self.storage is not None: + self.storage.cleanup() + super().teardown_class() + def steps_TC_ECOINFO_2_2(self) -> list[TestStep]: - steps = [TestStep(1, "Prepare", is_commissioning=True), - TestStep("1a", "Read root endpoint's PartsList"), - TestStep("1b", "For each endpoint in 1a read DeviceType list confirming aggregator endpoint exists"), - TestStep(2, "Add a bridged device"), - TestStep("2a", "(Manual Step) Add a bridged device using method indicated by the manufacturer"), - TestStep("2b", "Read root endpoint's PartsList, validate exactly one endpoint added"), - TestStep("2c", "On newly added endpoint detected in 2b read DeviceDirectory Ecosystem Information Attribute and validate success"), - TestStep("2d", "On newly added endpoint detected in 2b read LocationDirectory Ecosystem Information Attribute and validate success"), - TestStep(3, "Remove bridged device"), - TestStep("3a", "(Manual Step) Removed bridged device added in step 2a using method indicated by the manufacturer"), - TestStep("3b", "Verify that PartsList equals what was read in 1a"), - TestStep("3c", "On endpoint detected in 2b, read DeviceDirectory Ecosystem Information Attribute and validate failure"), - TestStep("3d", "On endpoint detected in 2b, read LocationDirectory Ecosystem Information Attribute and validate failure")] - - return steps + return [ + TestStep(0, "Commission DUT if not done", is_commissioning=True), + TestStep(1, "Prepare"), + TestStep("1a", "Read root endpoint's PartsList"), + TestStep("1b", "For each endpoint in 1a read DeviceType list confirming aggregator endpoint exists"), + TestStep(2, "Add a bridged device"), + TestStep("2a", "(Manual Step) Add a bridged device using method indicated by the manufacturer"), + TestStep("2b", "Read root endpoint's PartsList, validate exactly one endpoint added"), + TestStep("2c", "On newly added endpoint detected in 2b read DeviceDirectory Ecosystem Information Attribute and validate success"), + TestStep("2d", "On newly added endpoint detected in 2b read LocationDirectory Ecosystem Information Attribute and validate success"), + TestStep(3, "Remove bridged device"), + TestStep("3a", "(Manual Step) Removed bridged device added in step 2a using method indicated by the manufacturer"), + TestStep("3b", "Verify that PartsList equals what was read in 1a"), + TestStep("3c", "On endpoint detected in 2b, read DeviceDirectory Ecosystem Information Attribute and validate failure"), + TestStep("3d", "On endpoint detected in 2b, read LocationDirectory Ecosystem Information Attribute and validate failure"), + ] # This test has some manual steps, so we need a longer timeout. Test typically runs under 1 mins so 3 mins should # be enough time for test to run @@ -50,10 +115,13 @@ def default_timeout(self) -> int: @async_test_body async def test_TC_ECOINFO_2_2(self): + self.is_ci = self.check_pics('PICS_SDK_CI_ONLY') dev_ctrl = self.default_controller dut_node_id = self.dut_node_id - self.print_step(0, "Commissioning, already done") + # Commissioning - done + self.step(0) + self.step(1) self.step("1a") root_node_endpoint = 0 @@ -74,7 +142,14 @@ async def test_TC_ECOINFO_2_2(self): self.step(2) self.step("2a") - self.wait_for_user_input(prompt_msg="Add a bridged device using method indicated by the manufacturer") + if not self.is_ci: + self.wait_for_user_input("Add a bridged device using method indicated by the manufacturer") + else: + # Add some server to the DUT_FSA's Aggregator/Bridge. + self.dut_fsa_stdin.write(f"pairing onnetwork 2 {self.th_server_passcode}\n") + self.dut_fsa_stdin.flush() + # Wait for the commissioning to complete. + await asyncio.sleep(5) self.step("2b") root_part_list_step_2 = await dev_ctrl.ReadAttribute(dut_node_id, [(root_node_endpoint, Clusters.Descriptor.Attributes.PartsList)]) @@ -106,7 +181,14 @@ async def test_TC_ECOINFO_2_2(self): self.step(3) self.step("3a") - self.wait_for_user_input(prompt_msg="Removed bridged device added in step 2a using method indicated by the manufacturer") + if not self.is_ci: + self.wait_for_user_input("Removed bridged device added in step 2a using method indicated by the manufacturer") + else: + # Remove previously added server from the DUT_FSA's Aggregator/Bridge. + self.dut_fsa_stdin.write("pairing unpair 2\n") + self.dut_fsa_stdin.flush() + # Wait for the command to complete. + await asyncio.sleep(2) self.step("3b") root_part_list_step_3 = await dev_ctrl.ReadAttribute(dut_node_id, [(root_node_endpoint, Clusters.Descriptor.Attributes.PartsList)]) diff --git a/src/python_testing/execute_python_tests.py b/src/python_testing/execute_python_tests.py index f2367335159bad..57bd117cf198f5 100644 --- a/src/python_testing/execute_python_tests.py +++ b/src/python_testing/execute_python_tests.py @@ -62,7 +62,6 @@ def main(search_directory, env_file): "TC_CCTRL_2_3.py", # They rely on example applications that inter-communicate and there is no example app that works right now "TC_DGGEN_3_2.py", # src/python_testing/test_testing/test_TC_DGGEN_3_2.py is the Unit test of this test "TC_EEVSE_Utils.py", # Shared code for TC_EEVSE, not a standalone test - "TC_ECOINFO_2_2.py", # They rely on example applications that inter-communicate and there is no example app that works right now "TC_EWATERHTRBase.py", # Shared code for TC_EWATERHTR, not a standalone test "TC_EnergyReporting_Utils.py", # Shared code for TC_EEM and TC_EPM, not a standalone test "TC_OpstateCommon.py", # Shared code for TC_OPSTATE, not a standalone test From 78c31b27d6370ae0d5d28826505f8cadf89c271f Mon Sep 17 00:00:00 2001 From: Arkadiusz Bokowy Date: Fri, 27 Sep 2024 09:05:25 +0200 Subject: [PATCH 4/7] Rewrite TC_MCORE_FS CI arguments to YAML --- src/python_testing/TC_MCORE_FS_1_1.py | 23 ++++++++++++++++------- src/python_testing/TC_MCORE_FS_1_2.py | 23 ++++++++++++++++------- src/python_testing/TC_MCORE_FS_1_3.py | 23 ++++++++++++++++------- src/python_testing/TC_MCORE_FS_1_4.py | 22 +++++++++++++++------- src/python_testing/TC_MCORE_FS_1_5.py | 23 ++++++++++++++++------- 5 files changed, 79 insertions(+), 35 deletions(-) diff --git a/src/python_testing/TC_MCORE_FS_1_1.py b/src/python_testing/TC_MCORE_FS_1_1.py index 7b11d5bae440e7..bd6d327390a7cb 100755 --- a/src/python_testing/TC_MCORE_FS_1_1.py +++ b/src/python_testing/TC_MCORE_FS_1_1.py @@ -21,13 +21,22 @@ # for details about the block below. # # === BEGIN CI TEST ARGUMENTS === -# test-runner-runs: run1 -# test-runner-run/run1/app: examples/fabric-admin/scripts/fabric-sync-app.py -# test-runner-run/run1/app-args: --app-admin=${FABRIC_ADMIN_APP} --app-bridge=${FABRIC_BRIDGE_APP} --stdin-pipe=dut-fsa-stdin --discriminator=1234 -# test-runner-run/run1/factoryreset: true -# test-runner-run/run1/script-args: --PICS src/app/tests/suites/certification/ci-pics-values --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --string-arg th_server_app_path:${ALL_CLUSTERS_APP} --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto -# test-runner-run/run1/script-start-delay: 5 -# test-runner-run/run1/quiet: true +# test-runner-runs: +# run1: +# app: examples/fabric-admin/scripts/fabric-sync-app.py +# app-args: --app-admin=${FABRIC_ADMIN_APP} --app-bridge=${FABRIC_BRIDGE_APP} --stdin-pipe=dut-fsa-stdin --discriminator=1234 +# script-args: > +# --PICS src/app/tests/suites/certification/ci-pics-values +# --storage-path admin_storage.json +# --commissioning-method on-network +# --discriminator 1234 +# --passcode 20202021 +# --string-arg th_server_app_path:${ALL_CLUSTERS_APP} +# --trace-to json:${TRACE_TEST_JSON}.json +# --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto +# script-start-delay: 5 +# factoryreset: true +# quiet: false # === END CI TEST ARGUMENTS === import logging diff --git a/src/python_testing/TC_MCORE_FS_1_2.py b/src/python_testing/TC_MCORE_FS_1_2.py index 7be1bf96063c2e..fe6d53faae6b22 100644 --- a/src/python_testing/TC_MCORE_FS_1_2.py +++ b/src/python_testing/TC_MCORE_FS_1_2.py @@ -19,13 +19,22 @@ # for details about the block below. # # === BEGIN CI TEST ARGUMENTS === -# test-runner-runs: run1 -# test-runner-run/run1/app: examples/fabric-admin/scripts/fabric-sync-app.py -# test-runner-run/run1/app-args: --app-admin=${FABRIC_ADMIN_APP} --app-bridge=${FABRIC_BRIDGE_APP} --stdin-pipe=dut-fsa-stdin --discriminator=1234 -# test-runner-run/run1/factoryreset: true -# test-runner-run/run1/script-args: --PICS src/app/tests/suites/certification/ci-pics-values --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --string-arg th_server_app_path:${ALL_CLUSTERS_APP} dut_fsa_stdin_pipe:dut-fsa-stdin --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto -# test-runner-run/run1/script-start-delay: 5 -# test-runner-run/run1/quiet: false +# test-runner-runs: +# run1: +# app: examples/fabric-admin/scripts/fabric-sync-app.py +# app-args: --app-admin=${FABRIC_ADMIN_APP} --app-bridge=${FABRIC_BRIDGE_APP} --stdin-pipe=dut-fsa-stdin --discriminator=1234 +# script-args: > +# --PICS src/app/tests/suites/certification/ci-pics-values +# --storage-path admin_storage.json +# --commissioning-method on-network +# --discriminator 1234 +# --passcode 20202021 +# --string-arg th_server_app_path:${ALL_CLUSTERS_APP} dut_fsa_stdin_pipe:dut-fsa-stdin +# --trace-to json:${TRACE_TEST_JSON}.json +# --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto +# script-start-delay: 5 +# factoryreset: true +# quiet: false # === END CI TEST ARGUMENTS === import asyncio diff --git a/src/python_testing/TC_MCORE_FS_1_3.py b/src/python_testing/TC_MCORE_FS_1_3.py index e57487a24966a2..f4c7546c196ef1 100644 --- a/src/python_testing/TC_MCORE_FS_1_3.py +++ b/src/python_testing/TC_MCORE_FS_1_3.py @@ -23,13 +23,22 @@ # for details about the block below. # # === BEGIN CI TEST ARGUMENTS === -# test-runner-runs: run1 -# test-runner-run/run1/app: examples/fabric-admin/scripts/fabric-sync-app.py -# test-runner-run/run1/app-args: --app-admin=${FABRIC_ADMIN_APP} --app-bridge=${FABRIC_BRIDGE_APP} --stdin-pipe=dut-fsa-stdin --discriminator=1234 -# test-runner-run/run1/factoryreset: true -# test-runner-run/run1/script-args: --PICS src/app/tests/suites/certification/ci-pics-values --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --string-arg th_server_no_uid_app_path:${LIGHTING_APP_NO_UNIQUE_ID} --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto -# test-runner-run/run1/script-start-delay: 5 -# test-runner-run/run1/quiet: true +# test-runner-runs: +# run1: +# app: examples/fabric-admin/scripts/fabric-sync-app.py +# app-args: --app-admin=${FABRIC_ADMIN_APP} --app-bridge=${FABRIC_BRIDGE_APP} --stdin-pipe=dut-fsa-stdin --discriminator=1234 +# script-args: > +# --PICS src/app/tests/suites/certification/ci-pics-values +# --storage-path admin_storage.json +# --commissioning-method on-network +# --discriminator 1234 +# --passcode 20202021 +# --string-arg th_server_no_uid_app_path:${LIGHTING_APP_NO_UNIQUE_ID} +# --trace-to json:${TRACE_TEST_JSON}.json +# --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto +# script-start-delay: 5 +# factoryreset: true +# quiet: true # === END CI TEST ARGUMENTS === import asyncio diff --git a/src/python_testing/TC_MCORE_FS_1_4.py b/src/python_testing/TC_MCORE_FS_1_4.py index d76b0625202e1f..596929ba3872af 100644 --- a/src/python_testing/TC_MCORE_FS_1_4.py +++ b/src/python_testing/TC_MCORE_FS_1_4.py @@ -23,13 +23,21 @@ # for details about the block below. # # === BEGIN CI TEST ARGUMENTS === -# test-runner-runs: run1 -# test-runner-run/run1/app: examples/fabric-admin/scripts/fabric-sync-app.py -# test-runner-run/run1/app-args: --app-admin=${FABRIC_ADMIN_APP} --app-bridge=${FABRIC_BRIDGE_APP} --stdin-pipe=dut-fsa-stdin --discriminator=1234 -# test-runner-run/run1/factoryreset: true -# test-runner-run/run1/script-args: --PICS src/app/tests/suites/certification/ci-pics-values --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --string-arg th_fsa_app_path:examples/fabric-admin/scripts/fabric-sync-app.py th_fsa_admin_path:${FABRIC_ADMIN_APP} th_fsa_bridge_path:${FABRIC_BRIDGE_APP} th_server_no_uid_app_path:${LIGHTING_APP_NO_UNIQUE_ID} dut_fsa_stdin_pipe:dut-fsa-stdin --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto -# test-runner-run/run1/script-start-delay: 5 -# test-runner-run/run1/quiet: true +# test-runner-runs: +# run1: +# app: examples/fabric-admin/scripts/fabric-sync-app.py +# app-args: --app-admin=${FABRIC_ADMIN_APP} --app-bridge=${FABRIC_BRIDGE_APP} --stdin-pipe=dut-fsa-stdin --discriminator=1234 +# script-args: > +# --PICS src/app/tests/suites/certification/ci-pics-values +# --storage-path admin_storage.json +# --commissioning-method on-network +# --discriminator 1234 --passcode 20202021 +# --string-arg th_fsa_app_path:examples/fabric-admin/scripts/fabric-sync-app.py th_fsa_admin_path:${FABRIC_ADMIN_APP} th_fsa_bridge_path:${FABRIC_BRIDGE_APP} th_server_no_uid_app_path:${LIGHTING_APP_NO_UNIQUE_ID} dut_fsa_stdin_pipe:dut-fsa-stdin +# --trace-to json:${TRACE_TEST_JSON}.json +# --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto +# script-start-delay: 5 +# factoryreset: true +# quiet: true # === END CI TEST ARGUMENTS === import asyncio diff --git a/src/python_testing/TC_MCORE_FS_1_5.py b/src/python_testing/TC_MCORE_FS_1_5.py index 4a395f787496e9..fd1d38a5402541 100755 --- a/src/python_testing/TC_MCORE_FS_1_5.py +++ b/src/python_testing/TC_MCORE_FS_1_5.py @@ -19,13 +19,22 @@ # for details about the block below. # # === BEGIN CI TEST ARGUMENTS === -# test-runner-runs: run1 -# test-runner-run/run1/app: examples/fabric-admin/scripts/fabric-sync-app.py -# test-runner-run/run1/app-args: --app-admin=${FABRIC_ADMIN_APP} --app-bridge=${FABRIC_BRIDGE_APP} --stdin-pipe=dut-fsa-stdin --discriminator=1234 -# test-runner-run/run1/factoryreset: true -# test-runner-run/run1/script-args: --PICS src/app/tests/suites/certification/ci-pics-values --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --string-arg th_server_app_path:${ALL_CLUSTERS_APP} dut_fsa_stdin_pipe:dut-fsa-stdin --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto -# test-runner-run/run1/script-start-delay: 5 -# test-runner-run/run1/quiet: false +# test-runner-runs: +# run1: +# app: examples/fabric-admin/scripts/fabric-sync-app.py +# app-args: --app-admin=${FABRIC_ADMIN_APP} --app-bridge=${FABRIC_BRIDGE_APP} --stdin-pipe=dut-fsa-stdin --discriminator=1234 +# script-args: > +# --PICS src/app/tests/suites/certification/ci-pics-values +# --storage-path admin_storage.json +# --commissioning-method on-network +# --discriminator 1234 +# --passcode 20202021 +# --string-arg th_server_app_path:${ALL_CLUSTERS_APP} dut_fsa_stdin_pipe:dut-fsa-stdin +# --trace-to json:${TRACE_TEST_JSON}.json +# --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto +# script-start-delay: 5 +# factoryreset: true +# quiet: false # === END CI TEST ARGUMENTS === import asyncio From 4f1ae8e8794b4b36d677d6caca32b851b974ac4d Mon Sep 17 00:00:00 2001 From: Arkadiusz Bokowy Date: Tue, 1 Oct 2024 10:26:38 +0200 Subject: [PATCH 5/7] [Testing] Convenience property for checking PICS_SDK_CI_ONLY --- docs/testing/python.md | 5 ++++- src/python_testing/MinimalRepresentation.py | 3 +-- src/python_testing/TC_BRBINFO_4_1.py | 1 - src/python_testing/TC_CCTRL_2_2.py | 12 +++++------- src/python_testing/TC_CCTRL_2_3.py | 3 +-- src/python_testing/TC_DA_1_2.py | 5 ++--- src/python_testing/TC_DeviceBasicComposition.py | 3 +-- src/python_testing/TC_DeviceConformance.py | 3 +-- src/python_testing/TC_MCORE_FS_1_1.py | 7 +++---- src/python_testing/TC_MCORE_FS_1_2.py | 3 +-- src/python_testing/TC_MCORE_FS_1_3.py | 3 +-- src/python_testing/TC_MCORE_FS_1_4.py | 5 ++--- src/python_testing/TC_MCORE_FS_1_5.py | 3 +-- src/python_testing/TC_pics_checker.py | 2 +- src/python_testing/drlk_2_x_common.py | 3 +-- src/python_testing/matter_testing_support.py | 10 ++++++---- 16 files changed, 31 insertions(+), 40 deletions(-) diff --git a/docs/testing/python.md b/docs/testing/python.md index 9c2edcfc85caf0..07e157c459f8cd 100644 --- a/docs/testing/python.md +++ b/docs/testing/python.md @@ -628,7 +628,10 @@ example DUT on the host and includes factory reset support - Don’t forget to set the PICS file to the ci-pics-values - If there are steps in your test that will fail on CI (e.g. test vendor checks), gate them on the PICS_SDK_CI_ONLY - - `is_ci = self.check_pics('PICS_SDK_CI_ONLY')` + - ```python + if not self.is_pics_sdk_ci_only: + ... # Step that will fail on CI + ``` The CI test runner uses a structured environment setup that can be declared using structured comments at the top of the test file. To use this structured diff --git a/src/python_testing/MinimalRepresentation.py b/src/python_testing/MinimalRepresentation.py index c4a19b76cbbf9f..c99919a9080a8b 100644 --- a/src/python_testing/MinimalRepresentation.py +++ b/src/python_testing/MinimalRepresentation.py @@ -129,8 +129,7 @@ def test_MinimalRepresentation(self): # Before we can generate a minimal representation, we need to make sure that the device is conformant. # Otherwise, the values we extract aren't fully informative. ignore_in_progress = self.user_params.get("ignore_in_progress", False) - is_ci = self.check_pics('PICS_SDK_CI_ONLY') - representation = self.GenerateMinimals(ignore_in_progress, is_ci) + representation = self.GenerateMinimals(ignore_in_progress, self.is_pics_sdk_ci_only) print(type(representation[0])) self.PrettyPrintRepresentation(representation) diff --git a/src/python_testing/TC_BRBINFO_4_1.py b/src/python_testing/TC_BRBINFO_4_1.py index 8b064a775f358b..32dd541d66f676 100644 --- a/src/python_testing/TC_BRBINFO_4_1.py +++ b/src/python_testing/TC_BRBINFO_4_1.py @@ -198,7 +198,6 @@ def resume_th_icd_server(self, check_state): @async_test_body async def test_TC_BRBINFO_4_1(self): - self.is_ci = self.check_pics('PICS_SDK_CI_ONLY') icdm_cluster = Clusters.Objects.IcdManagement icdm_attributes = icdm_cluster.Attributes brb_info_cluster = Clusters.Objects.BridgedDeviceBasicInformation diff --git a/src/python_testing/TC_CCTRL_2_2.py b/src/python_testing/TC_CCTRL_2_2.py index 87849e40443d01..20a03e3bd6ef94 100644 --- a/src/python_testing/TC_CCTRL_2_2.py +++ b/src/python_testing/TC_CCTRL_2_2.py @@ -127,8 +127,6 @@ def default_timeout(self) -> int: @run_if_endpoint_matches(has_cluster(Clusters.CommissionerControl)) async def test_TC_CCTRL_2_2(self): - self.is_ci = self.check_pics('PICS_SDK_CI_ONLY') - self.step(1) th_server_fabrics = await self.read_single_attribute_check_success(cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.Fabrics, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0, fabric_filtered=False) self.step(2) @@ -192,7 +190,7 @@ async def test_TC_CCTRL_2_2(self): await self.send_single_cmd(cmd) self.step(12) - if not self.is_ci: + if not self.is_pics_sdk_ci_only: self.wait_for_user_input("Approve Commissioning approval request using manufacturer specified mechanism") self.step(13) @@ -248,7 +246,7 @@ async def test_TC_CCTRL_2_2(self): self.step(19) logging.info("Test now waits for 30 seconds") - if not self.is_ci: + if not self.is_pics_sdk_ci_only: time.sleep(30) self.step(20) @@ -267,7 +265,7 @@ async def test_TC_CCTRL_2_2(self): await self.send_single_cmd(cmd) self.step(23) - if not self.is_ci: + if not self.is_pics_sdk_ci_only: self.wait_for_user_input("Approve Commissioning approval request using manufacturer specified mechanism") self.step(24) @@ -295,13 +293,13 @@ async def test_TC_CCTRL_2_2(self): await self.send_single_cmd(cmd, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0, timedRequestTimeoutMs=5000) self.step(27) - if not self.is_ci: + if not self.is_pics_sdk_ci_only: time.sleep(30) self.step(28) th_server_fabrics_new = await self.read_single_attribute_check_success(cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.Fabrics, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0, fabric_filtered=False) # TODO: this should be mocked too. - if not self.is_ci: + if not self.is_pics_sdk_ci_only: asserts.assert_equal(len(th_server_fabrics) + 1, len(th_server_fabrics_new), "Unexpected number of fabrics on TH_SERVER") diff --git a/src/python_testing/TC_CCTRL_2_3.py b/src/python_testing/TC_CCTRL_2_3.py index 2c1c20e17f309f..95bd546c8a6b0c 100644 --- a/src/python_testing/TC_CCTRL_2_3.py +++ b/src/python_testing/TC_CCTRL_2_3.py @@ -111,7 +111,6 @@ def default_timeout(self) -> int: @run_if_endpoint_matches(has_cluster(Clusters.CommissionerControl)) async def test_TC_CCTRL_2_3(self): - self.is_ci = self.check_pics('PICS_SDK_CI_ONLY') self.step(1) th_server_fabrics = await self.read_single_attribute_check_success(cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.Fabrics, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0, fabric_filtered=False) @@ -129,7 +128,7 @@ async def test_TC_CCTRL_2_3(self): await self.send_single_cmd(cmd=cmd) self.step(5) - if not self.is_ci: + if not self.is_pics_sdk_ci_only: self.wait_for_user_input("Approve Commissioning approval request using manufacturer specified mechanism") self.step(6) diff --git a/src/python_testing/TC_DA_1_2.py b/src/python_testing/TC_DA_1_2.py index 30a5285a4a4b68..a8b9dbf3cd414a 100644 --- a/src/python_testing/TC_DA_1_2.py +++ b/src/python_testing/TC_DA_1_2.py @@ -175,7 +175,6 @@ def steps_TC_DA_1_2(self): @async_test_body async def test_TC_DA_1_2(self): - is_ci = self.check_pics('PICS_SDK_CI_ONLY') cd_cert_dir = self.user_params.get("cd_cert_dir", 'credentials/development/cd-certs') post_cert_test = self.user_params.get("post_cert_test", False) @@ -311,7 +310,7 @@ async def test_TC_DA_1_2(self): asserts.assert_equal(format_version, 1, "Format version is incorrect") self.step("6.2") asserts.assert_equal(vendor_id, basic_info_vendor_id, "Vendor ID is incorrect") - if not is_ci: + if not self.is_pics_sdk_ci_only: asserts.assert_in(vendor_id, range(1, 0xfff0), "Vendor ID is out of range") self.step("6.3") asserts.assert_true(basic_info_product_id in product_id_array, "Product ID not found in CD product array") @@ -328,7 +327,7 @@ async def test_TC_DA_1_2(self): self.step("6.9") if post_cert_test: asserts.assert_equal(certification_type, 2, "Certification declaration is not marked as production.") - elif is_ci: + elif self.is_pics_sdk_ci_only: asserts.assert_in(certification_type, [0, 1, 2], "Certification type is out of range") else: asserts.assert_in(certification_type, [1, 2], "Certification type is out of range") diff --git a/src/python_testing/TC_DeviceBasicComposition.py b/src/python_testing/TC_DeviceBasicComposition.py index a4ab495114a245..31d8bfb76c545e 100644 --- a/src/python_testing/TC_DeviceBasicComposition.py +++ b/src/python_testing/TC_DeviceBasicComposition.py @@ -455,8 +455,7 @@ class RequiredMandatoryAttribute: self.print_step( 6, "Validate that none of the global attribute IDs contain values with prefixes outside of the allowed standard or MEI prefix range") - is_ci = self.check_pics('PICS_SDK_CI_ONLY') - if is_ci: + if self.is_pics_sdk_ci_only: # test vendor prefixes are allowed in the CI because we use them internally in examples bad_prefix_min = 0xFFF5_0000 else: diff --git a/src/python_testing/TC_DeviceConformance.py b/src/python_testing/TC_DeviceConformance.py index 86a981c86cde15..2d03e8f326e09a 100644 --- a/src/python_testing/TC_DeviceConformance.py +++ b/src/python_testing/TC_DeviceConformance.py @@ -356,9 +356,8 @@ def test_TC_IDM_10_2(self): # TODO: Turn this off after TE2 # https://github.com/project-chip/connectedhomeip/issues/34615 ignore_in_progress = self.user_params.get("ignore_in_progress", True) - is_ci = self.check_pics('PICS_SDK_CI_ONLY') allow_provisional = self.user_params.get("allow_provisional", False) - success, problems = self.check_conformance(ignore_in_progress, is_ci, allow_provisional) + success, problems = self.check_conformance(ignore_in_progress, self.is_pics_sdk_ci_only, allow_provisional) self.problems.extend(problems) if not success: self.fail_current_test("Problems with conformance") diff --git a/src/python_testing/TC_MCORE_FS_1_1.py b/src/python_testing/TC_MCORE_FS_1_1.py index 0d77bad8c6f268..c2f38a766670fc 100755 --- a/src/python_testing/TC_MCORE_FS_1_1.py +++ b/src/python_testing/TC_MCORE_FS_1_1.py @@ -133,7 +133,6 @@ def default_timeout(self) -> int: @async_test_body async def test_TC_MCORE_FS_1_1(self): - self.is_ci = self.check_pics('PICS_SDK_CI_ONLY') # TODO this value should either be determined or passed in from command line dut_commissioning_control_endpoint = 0 self.step(1) @@ -152,7 +151,7 @@ async def test_TC_MCORE_FS_1_1(self): requestID=good_request_id, vendorID=th_fsa_server_vid, productID=th_fsa_server_pid, label="Test Ecosystem") await self.send_single_cmd(cmd, endpoint=dut_commissioning_control_endpoint) - if not self.is_ci: + if not self.is_pics_sdk_ci_only: self.wait_for_user_input("Approve Commissioning approval request on DUT using manufacturer specified mechanism") if not events: @@ -181,12 +180,12 @@ async def test_TC_MCORE_FS_1_1(self): await self.send_single_cmd(cmd, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0, timedRequestTimeoutMs=5000) self.step("3c") - if not self.is_ci: + if not self.is_pics_sdk_ci_only: time.sleep(30) th_fsa_server_fabrics_new = await self.read_single_attribute_check_success(cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.Fabrics, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0, fabric_filtered=False) # TODO: this should be mocked too. - if not self.is_ci: + if not self.is_pics_sdk_ci_only: asserts.assert_equal(len(th_fsa_server_fabrics) + 1, len(th_fsa_server_fabrics_new), "Unexpected number of fabrics on TH_SERVER") diff --git a/src/python_testing/TC_MCORE_FS_1_2.py b/src/python_testing/TC_MCORE_FS_1_2.py index 7be1bf96063c2e..f1cba9a4c5db5e 100644 --- a/src/python_testing/TC_MCORE_FS_1_2.py +++ b/src/python_testing/TC_MCORE_FS_1_2.py @@ -148,7 +148,6 @@ def default_timeout(self) -> int: @async_test_body async def test_TC_MCORE_FS_1_2(self): - self.is_ci = self.check_pics('PICS_SDK_CI_ONLY') min_report_interval_sec = self.user_params.get("min_report_interval_sec", 0) max_report_interval_sec = self.user_params.get("max_report_interval_sec", 30) @@ -176,7 +175,7 @@ async def test_TC_MCORE_FS_1_2(self): asserts.assert_true(type_matches(step_1_dut_parts_list, list), "PartsList is expected to be a list") self.step(2) - if not self.is_ci: + if not self.is_pics_sdk_ci_only: self._ask_for_vendor_commissioning_ux_operation(self.th_server_setup_params) else: self.dut_fsa_stdin.write( diff --git a/src/python_testing/TC_MCORE_FS_1_3.py b/src/python_testing/TC_MCORE_FS_1_3.py index e57487a24966a2..9dbdabb1abbca7 100644 --- a/src/python_testing/TC_MCORE_FS_1_3.py +++ b/src/python_testing/TC_MCORE_FS_1_3.py @@ -124,7 +124,7 @@ async def commission_via_commissioner_control(self, controller_node_id: int, dev ), ) - if not self.is_ci: + if not self.is_pics_sdk_ci_only: self.wait_for_user_input("Approve Commissioning Approval Request on DUT using manufacturer specified mechanism") resp = await self.send_single_cmd( @@ -152,7 +152,6 @@ async def commission_via_commissioner_control(self, controller_node_id: int, dev @async_test_body async def test_TC_MCORE_FS_1_3(self): - self.is_ci = self.check_pics('PICS_SDK_CI_ONLY') # Commissioning - done self.step(0) diff --git a/src/python_testing/TC_MCORE_FS_1_4.py b/src/python_testing/TC_MCORE_FS_1_4.py index d76b0625202e1f..26e5d470ce4e92 100644 --- a/src/python_testing/TC_MCORE_FS_1_4.py +++ b/src/python_testing/TC_MCORE_FS_1_4.py @@ -195,7 +195,6 @@ def steps_TC_MCORE_FS_1_4(self) -> list[TestStep]: @async_test_body async def test_TC_MCORE_FS_1_4(self): - self.is_ci = self.check_pics('PICS_SDK_CI_ONLY') # Commissioning - done self.step(0) @@ -298,7 +297,7 @@ async def test_TC_MCORE_FS_1_4(self): self.step(4) # Commissioning TH_FSA_BRIDGE to DUT_FSA fabric. - if not self.is_ci: + if not self.is_pics_sdk_ci_only: self.wait_for_user_input( f"Commission TH_FSA's aggregator on DUT using manufacturer specified mechanism.\n" f"Use the following parameters:\n" @@ -326,7 +325,7 @@ async def test_TC_MCORE_FS_1_4(self): )) # Synchronize TH_SERVER_NO_UID from TH_FSA to DUT_FSA fabric. - if not self.is_ci: + if not self.is_pics_sdk_ci_only: self.wait_for_user_input( f"Synchronize endpoint from TH_FSA's aggregator to DUT using manufacturer specified mechanism.\n" f"Use the following parameters:\n" diff --git a/src/python_testing/TC_MCORE_FS_1_5.py b/src/python_testing/TC_MCORE_FS_1_5.py index 4a395f787496e9..b8794c400356c1 100755 --- a/src/python_testing/TC_MCORE_FS_1_5.py +++ b/src/python_testing/TC_MCORE_FS_1_5.py @@ -155,7 +155,6 @@ def default_timeout(self) -> int: @async_test_body async def test_TC_MCORE_FS_1_5(self): - self.is_ci = self.check_pics('PICS_SDK_CI_ONLY') min_report_interval_sec = 0 max_report_interval_sec = 30 @@ -183,7 +182,7 @@ async def test_TC_MCORE_FS_1_5(self): asserts.assert_true(type_matches(step_1_dut_parts_list, list), "PartsList is expected to be a list") self.step(2) - if not self.is_ci: + if not self.is_pics_sdk_ci_only: self._ask_for_vendor_commissioning_ux_operation(self.th_server_setup_params) else: self.dut_fsa_stdin.write( diff --git a/src/python_testing/TC_pics_checker.py b/src/python_testing/TC_pics_checker.py index f63628fd4eb426..91e2102b5ec46d 100644 --- a/src/python_testing/TC_pics_checker.py +++ b/src/python_testing/TC_pics_checker.py @@ -177,7 +177,7 @@ def test_TC_IDM_10_4(self): self._check_and_record_errors(location, required, pics) self.step(7) - if self.check_pics('PICS_SDK_CI_ONLY'): + if self.is_pics_sdk_ci_only: self.record_error("PICS check", location=ProblemLocation(), problem="PICS PICS_SDK_CI_ONLY found in PICS list. This PICS is disallowed for certification.") self.success = False diff --git a/src/python_testing/drlk_2_x_common.py b/src/python_testing/drlk_2_x_common.py index c25365bf993d8f..044128d0d383e0 100644 --- a/src/python_testing/drlk_2_x_common.py +++ b/src/python_testing/drlk_2_x_common.py @@ -121,7 +121,6 @@ async def teardown(self): credentials=self.createdCredential, userIndex=self.user_index) async def run_drlk_test_common(self, lockUnlockCommand, lockUnlockCmdRspPICS, lockUnlockText, doAutoRelockTest): - is_ci = self.check_pics('PICS_SDK_CI_ONLY') self.clear_credential_and_user_flag = True # Allow for user overrides of these values @@ -131,7 +130,7 @@ async def run_drlk_test_common(self, lockUnlockCommand, lockUnlockCmdRspPICS, lo userCodeTemporaryDisableTime = self.user_params.get("user_code_temporary_disable_time", 15) wrongCodeEntryLimit = self.user_params.get("wrong_code_entry_limit", 3) autoRelockTime = self.user_params.get("auto_relock_time", 60) - if is_ci: + if self.is_pics_sdk_ci_only: autoRelockTime = 10 cluster = Clusters.Objects.DoorLock diff --git a/src/python_testing/matter_testing_support.py b/src/python_testing/matter_testing_support.py index c5d46e2306dae2..2b9ccb0f286613 100644 --- a/src/python_testing/matter_testing_support.py +++ b/src/python_testing/matter_testing_support.py @@ -1008,7 +1008,7 @@ def write_to_app_pipe(self, command_dict: dict, app_pipe_name: Optional[str] = N Use the following environment variables: - - LINUX_DUT_IP + - 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 @@ -1128,9 +1128,11 @@ def teardown_class(self): super().teardown_class() def check_pics(self, pics_key: str) -> bool: - picsd = self.matter_test_config.pics - pics_key = pics_key.strip() - return pics_key in picsd and picsd[pics_key] + return self.matter_test_config.pics.get(pics_key.strip(), False) + + @property + def is_pics_sdk_ci_only(self) -> bool: + return self.check_pics('PICS_SDK_CI_ONLY') async def openCommissioningWindow(self, dev_ctrl: ChipDeviceCtrl, node_id: int) -> CustomCommissioningParameters: rnd_discriminator = random.randint(0, 4095) From a91448d6a2f26a2569c5007f57b1af708dbe2ce3 Mon Sep 17 00:00:00 2001 From: Arkadiusz Bokowy Date: Tue, 1 Oct 2024 15:37:37 +0200 Subject: [PATCH 6/7] Guard CI prerequisites setup with PICS SDK CI --- src/python_testing/TC_ECOINFO_2_1.py | 73 ++++++++++++++++------------ src/python_testing/TC_ECOINFO_2_2.py | 62 +++++++++++++---------- 2 files changed, 79 insertions(+), 56 deletions(-) diff --git a/src/python_testing/TC_ECOINFO_2_1.py b/src/python_testing/TC_ECOINFO_2_1.py index 4d846f20e8e84a..52a3938f0ce0e0 100644 --- a/src/python_testing/TC_ECOINFO_2_1.py +++ b/src/python_testing/TC_ECOINFO_2_1.py @@ -39,6 +39,7 @@ import asyncio import logging +import os import random import tempfile @@ -60,35 +61,8 @@ async def setup_class(self): self.th_server = None self.storage = None - # Get the path to the TH_SERVER app from the user params. - if th_server_app := self.user_params.get("th_server_app_path"): - # Create a temporary storage directory for keeping KVS files. - self.storage = tempfile.TemporaryDirectory(prefix=self.__class__.__name__) - logging.info("Temporary storage directory: %s", self.storage.name) - - # Get the named pipe path for the DUT_FSA app input from the user params. - if dut_fsa_stdin_pipe := self.user_params.get("dut_fsa_stdin_pipe"): - self.dut_fsa_stdin = open(dut_fsa_stdin_pipe, "w") - - self.th_server_port = 5544 - self.th_server_discriminator = random.randint(0, 4095) - self.th_server_passcode = 20202021 - - # Start the server app. - self.th_server = AppServer( - th_server_app, - storage_dir=self.storage.name, - port=self.th_server_port, - discriminator=self.th_server_discriminator, - passcode=self.th_server_passcode) - self.th_server.start() - - # Add some server to the DUT_FSA's Aggregator/Bridge. - self.dut_fsa_stdin.write( - f"pairing onnetwork 2 {self.th_server_passcode}\n") - self.dut_fsa_stdin.flush() - # Wait for the commissioning to complete. - await asyncio.sleep(5) + if self.is_pics_sdk_ci_only: + await self._setup_ci_prerequisites() def teardown_class(self): if self.th_server is not None: @@ -97,6 +71,44 @@ def teardown_class(self): self.storage.cleanup() super().teardown_class() + async def _setup_ci_prerequisites(self): + asserts.assert_true(self.is_pics_sdk_ci_only, "This method is only for PICS SDK CI") + + th_server_app = self.user_params.get("th_server_app_path", None) + if not th_server_app: + asserts.fail("CI setup requires a TH_SERVER app. Specify app path with --string-arg th_server_app_path:") + if not os.path.exists(th_server_app): + asserts.fail(f"The path {th_server_app} does not exist") + + # Get the named pipe path for the DUT_FSA app input from the user params. + dut_fsa_stdin_pipe = self.user_params.get("dut_fsa_stdin_pipe") + if not dut_fsa_stdin_pipe: + asserts.fail("CI setup requires --string-arg dut_fsa_stdin_pipe:") + self.dut_fsa_stdin = open(dut_fsa_stdin_pipe, "w") + + # Create a temporary storage directory for keeping KVS files. + self.storage = tempfile.TemporaryDirectory(prefix=self.__class__.__name__) + logging.info("Temporary storage directory: %s", self.storage.name) + + self.th_server_port = 5544 + self.th_server_discriminator = random.randint(0, 4095) + self.th_server_passcode = 20202021 + + # Start the server app. + self.th_server = AppServer( + th_server_app, + storage_dir=self.storage.name, + port=self.th_server_port, + discriminator=self.th_server_discriminator, + passcode=self.th_server_passcode) + self.th_server.start() + + # Add some server to the DUT_FSA's Aggregator/Bridge. + self.dut_fsa_stdin.write(f"pairing onnetwork 2 {self.th_server_passcode}\n") + self.dut_fsa_stdin.flush() + # Wait for the commissioning to complete. + await asyncio.sleep(5) + def _validate_device_directory(self, current_fabric_index, device_directory): for device in device_directory: if current_fabric_index != device.fabricIndex: @@ -204,14 +216,13 @@ def steps_TC_ECOINFO_2_1(self) -> list[TestStep]: @async_test_body async def test_TC_ECOINFO_2_1(self): - self.is_ci = self.check_pics('PICS_SDK_CI_ONLY') dev_ctrl = self.default_controller dut_node_id = self.dut_node_id # Commissioning - done self.step(0) - if not self.is_ci: + if not self.is_pics_sdk_ci_only: self.wait_for_user_input( "Paused test to allow for manufacturer to satisfy precondition where " "one or more bridged devices of a supported type is connected to DUT") diff --git a/src/python_testing/TC_ECOINFO_2_2.py b/src/python_testing/TC_ECOINFO_2_2.py index 64ead03a2d0d4d..dd5f138530505a 100644 --- a/src/python_testing/TC_ECOINFO_2_2.py +++ b/src/python_testing/TC_ECOINFO_2_2.py @@ -39,6 +39,7 @@ import asyncio import logging +import os import random import tempfile @@ -59,28 +60,8 @@ def setup_class(self): self.th_server = None self.storage = None - # Get the path to the TH_SERVER app from the user params. - if th_server_app := self.user_params.get("th_server_app_path"): - # Create a temporary storage directory for keeping KVS files. - self.storage = tempfile.TemporaryDirectory(prefix=self.__class__.__name__) - logging.info("Temporary storage directory: %s", self.storage.name) - - # Get the named pipe path for the DUT_FSA app input from the user params. - if dut_fsa_stdin_pipe := self.user_params.get("dut_fsa_stdin_pipe"): - self.dut_fsa_stdin = open(dut_fsa_stdin_pipe, "w") - - self.th_server_port = 5544 - self.th_server_discriminator = random.randint(0, 4095) - self.th_server_passcode = 20202021 - - # Start the server app. - self.th_server = AppServer( - th_server_app, - storage_dir=self.storage.name, - port=self.th_server_port, - discriminator=self.th_server_discriminator, - passcode=self.th_server_passcode) - self.th_server.start() + if self.is_pics_sdk_ci_only: + self._setup_ci_prerequisites() def teardown_class(self): if self.th_server is not None: @@ -89,6 +70,38 @@ def teardown_class(self): self.storage.cleanup() super().teardown_class() + def _setup_ci_prerequisites(self): + asserts.assert_true(self.is_pics_sdk_ci_only, "This method is only for PICS SDK CI") + + th_server_app = self.user_params.get("th_server_app_path", None) + if not th_server_app: + asserts.fail("CI setup requires a TH_SERVER app. Specify app path with --string-arg th_server_app_path:") + if not os.path.exists(th_server_app): + asserts.fail(f"The path {th_server_app} does not exist") + + # Get the named pipe path for the DUT_FSA app input from the user params. + dut_fsa_stdin_pipe = self.user_params.get("dut_fsa_stdin_pipe") + if not dut_fsa_stdin_pipe: + asserts.fail("CI setup requires --string-arg dut_fsa_stdin_pipe:") + self.dut_fsa_stdin = open(dut_fsa_stdin_pipe, "w") + + # Create a temporary storage directory for keeping KVS files. + self.storage = tempfile.TemporaryDirectory(prefix=self.__class__.__name__) + logging.info("Temporary storage directory: %s", self.storage.name) + + self.th_server_port = 5544 + self.th_server_discriminator = random.randint(0, 4095) + self.th_server_passcode = 20202021 + + # Start the server app. + self.th_server = AppServer( + th_server_app, + storage_dir=self.storage.name, + port=self.th_server_port, + discriminator=self.th_server_discriminator, + passcode=self.th_server_passcode) + self.th_server.start() + def steps_TC_ECOINFO_2_2(self) -> list[TestStep]: return [ TestStep(0, "Commission DUT if not done", is_commissioning=True), @@ -115,7 +128,6 @@ def default_timeout(self) -> int: @async_test_body async def test_TC_ECOINFO_2_2(self): - self.is_ci = self.check_pics('PICS_SDK_CI_ONLY') dev_ctrl = self.default_controller dut_node_id = self.dut_node_id @@ -142,7 +154,7 @@ async def test_TC_ECOINFO_2_2(self): self.step(2) self.step("2a") - if not self.is_ci: + if not self.is_pics_sdk_ci_only: self.wait_for_user_input("Add a bridged device using method indicated by the manufacturer") else: # Add some server to the DUT_FSA's Aggregator/Bridge. @@ -181,7 +193,7 @@ async def test_TC_ECOINFO_2_2(self): self.step(3) self.step("3a") - if not self.is_ci: + if not self.is_pics_sdk_ci_only: self.wait_for_user_input("Removed bridged device added in step 2a using method indicated by the manufacturer") else: # Remove previously added server from the DUT_FSA's Aggregator/Bridge. From 9bc22ea57a8bbc18fe5c56b7db3b7483441fc7c3 Mon Sep 17 00:00:00 2001 From: Arkadiusz Bokowy Date: Tue, 1 Oct 2024 15:40:29 +0200 Subject: [PATCH 7/7] Fix typo --- .../matter_testing_infrastructure/chip/testing/tasks.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/python_testing/matter_testing_infrastructure/chip/testing/tasks.py b/src/python_testing/matter_testing_infrastructure/chip/testing/tasks.py index 64c3347537235c..39d1be3ec872e6 100644 --- a/src/python_testing/matter_testing_infrastructure/chip/testing/tasks.py +++ b/src/python_testing/matter_testing_infrastructure/chip/testing/tasks.py @@ -65,11 +65,11 @@ def __init__(self, program: str, *args: List[str], self.output_match = None self.returncode = None - def _set_output_match(self, patter: Union[str, re.Pattern]): - if isinstance(patter, str): - self.output_match = re.compile(re.escape(patter.encode())) + def _set_output_match(self, pattern: Union[str, re.Pattern]): + if isinstance(pattern, str): + self.output_match = re.compile(re.escape(pattern.encode())) else: - self.output_match = patter + self.output_match = pattern def _check_output(self, line: bytes, is_stderr: bool): if self.output_match is not None and self.output_match.search(line):