Add ECOINFO_2_1 and ECOINFO_2_2 to CI

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:

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)])

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.

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