diff --git a/src/python_testing/TC_ECOINFO_2_1.py b/src/python_testing/TC_ECOINFO_2_1.py index 202d1073779cb3..52a3938f0ce0e0 100644 --- a/src/python_testing/TC_ECOINFO_2_1.py +++ b/src/python_testing/TC_ECOINFO_2_1.py @@ -15,20 +15,104 @@ # 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 os +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 + + if self.is_pics_sdk_ci_only: + await self._setup_ci_prerequisites() + + 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() + + 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: - # 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 +169,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 +204,34 @@ 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): 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_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") + "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/TC_ECOINFO_2_2.py b/src/python_testing/TC_ECOINFO_2_2.py index 24f39ab93f2a04..dd5f138530505a 100644 --- a/src/python_testing/TC_ECOINFO_2_2.py +++ b/src/python_testing/TC_ECOINFO_2_2.py @@ -15,32 +15,110 @@ # 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 os +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 + + if self.is_pics_sdk_ci_only: + self._setup_ci_prerequisites() + + 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 _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]: - 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 @@ -53,7 +131,9 @@ async def test_TC_ECOINFO_2_2(self): 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 +154,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_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. + 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 +193,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_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. + 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/TC_MCORE_FS_1_1.py b/src/python_testing/TC_MCORE_FS_1_1.py index c2f38a766670fc..0723c7ac494f62 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 @@ -47,7 +56,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 +65,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/TC_MCORE_FS_1_2.py b/src/python_testing/TC_MCORE_FS_1_2.py index f1cba9a4c5db5e..d7bcbc0ea009b7 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 9dbdabb1abbca7..005a33e911a3de 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 26e5d470ce4e92..c9244cf196d2f3 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 b8794c400356c1..03173168aca293 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 diff --git a/src/python_testing/execute_python_tests.py b/src/python_testing/execute_python_tests.py index 9e19aa142552dc..57bd117cf198f5 100644 --- a/src/python_testing/execute_python_tests.py +++ b/src/python_testing/execute_python_tests.py @@ -62,8 +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 "TC_OpstateCommon.py", # Shared code for TC_OPSTATE, not a standalone test 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..39d1be3ec872e6 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, pattern: Union[str, re.Pattern]): + if isinstance(pattern, str): + self.output_match = re.compile(re.escape(pattern.encode())) + else: + self.output_match = pattern 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