Add TC_CCTRL_2_1, TC_CCTRL_2_2, TC_CCTRL_2_3 to CI

scripts/tests/run_python_test.py

@@ -71,9 +71,9 @@ def process_test_script_output(line, is_stderr):
 @click.command()
 @click.option("--app", type=click.Path(exists=True), default=None,
               help='Path to local application to use, omit to use external apps.')
-@click.option("--factoryreset", is_flag=True,
+@click.option("--factory-reset/--no-factory-reset", default=None,
               help='Remove app config and repl configs (/tmp/chip* and /tmp/repl*) before running the tests.')
-@click.option("--factoryreset-app-only", is_flag=True,
+@click.option("--factory-reset-app-only/--no-factory-reset-app-only", default=None,
               help='Remove app config and repl configs (/tmp/chip* and /tmp/repl*) before running the tests, but not the controller config')
 @click.option("--app-args", type=str, default='',
               help='The extra arguments passed to the device. Can use placeholders like {SCRIPT_BASE_NAME}')
@@ -90,9 +90,10 @@ def process_test_script_output(line, is_stderr):
               help='Script arguments, can use placeholders like {SCRIPT_BASE_NAME}.')
 @click.option("--script-gdb", is_flag=True,
               help='Run script through gdb')
-@click.option("--quiet", is_flag=True, help="Do not print output from passing tests. Use this flag in CI to keep github log sizes manageable.")
+@click.option("--quiet/--no-quiet", default=None,
+              help="Do not print output from passing tests. Use this flag in CI to keep GitHub log size manageable.")
 @click.option("--load-from-env", default=None, help="YAML file that contains values for environment variables.")
-def main(app: str, factoryreset: bool, factoryreset_app_only: bool, app_args: str,
+def main(app: str, factory_reset: bool, factory_reset_app_only: bool, app_args: str,
          app_ready_pattern: str, script: str, script_args: str, script_gdb: bool, quiet: bool, load_from_env):
     if load_from_env:
         reader = MetadataReader(load_from_env)
@@ -106,18 +107,23 @@ def main(app: str, factoryreset: bool, factoryreset_app_only: bool, app_args: st
                 app_args=app_args,
                 app_ready_pattern=app_ready_pattern,
                 script_args=script_args,
-                factory_reset=factoryreset,
-                factory_reset_app_only=factoryreset_app_only,
                 script_gdb=script_gdb,
-                quiet=quiet
             )
         ]
 
     if not runs:
-        raise Exception(
-            "No valid runs were found. Make sure you add runs to your file, see https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md document for reference/example.")
+        raise click.ClickException(
+            "No valid runs were found. Make sure you add runs to your file, see "
+            "https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md document for reference/example.")
 
-    coloredlogs.install(level='INFO')
+    # Override runs Metadata with the command line arguments
+    for run in runs:
+        if factory_reset is not None:
+            run.factory_reset = factory_reset
+        if factory_reset_app_only is not None:
+            run.factory_reset_app_only = factory_reset_app_only
+        if quiet is not None:
+            run.quiet = quiet
 
     for run in runs:
         logging.info("Executing %s %s", run.py_script_path.split('/')[-1], run.run)
@@ -215,4 +221,5 @@ def main_impl(app: str, factory_reset: bool, factory_reset_app_only: bool, app_a
 
 
 if __name__ == '__main__':
+    coloredlogs.install(level='INFO')
     main(auto_envvar_prefix='CHIP')

src/app/tests/suites/certification/PICS.yaml

@@ -339,6 +339,13 @@ PICS:
           "Does commissionee provide a Firmware Information field in the
           AttestationResponse?"
       id: MCORE.DA.ATTESTELEMENT_FW_INFO
+
+    #
+    # Fabric Synchronization
+    #
+    - label: "Does the device implement Fabric Synchronization capabilities?"
+      id: MCORE.FS
+
     #
     #IDM
     #

src/app/tests/suites/certification/ci-pics-values

@@ -922,6 +922,9 @@ MCORE.BDX.Sender=0
 MCORE.BDX.SynchronousReceiver=0
 MCORE.BDX.SynchronousSender=0
 
+# Fabric Synchronization
+MCORE.FS=1
+
 # General Diagnostics Cluster
 
 DGGEN.S=1

src/python_testing/TC_CCTRL_2_1.py

@@ -15,6 +15,28 @@
 #    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
+#     app-ready-pattern: "Successfully opened pairing window on the device"
+#     script-args: >
+#       --PICS src/app/tests/suites/certification/ci-pics-values
+#       --storage-path admin_storage.json
+#       --commissioning-method on-network
+#       --discriminator 1234
+#       --passcode 20202021
+#       --endpoint 0
+#       --trace-to json:${TRACE_TEST_JSON}.json
+#       --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
+#     factoryreset: true
+#     quiet: true
+# === END CI TEST ARGUMENTS ===
+
 import chip.clusters as Clusters
 from matter_testing_support import MatterBaseTest, TestStep, default_matter_test_main, has_cluster, run_if_endpoint_matches
 from mobly import asserts

src/python_testing/TC_CCTRL_2_2.py

@@ -18,50 +18,74 @@
 # See https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md#defining-the-ci-test-arguments
 # for details about the block below.
 #
-# TODO: Skip CI for now, we don't have any way to run this. Needs setup. See test_TC_CCTRL.py
+# === 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
+#     app-ready-pattern: "Successfully opened pairing window on the device"
+#     script-args: >
+#       --PICS src/app/tests/suites/certification/ci-pics-values
+#       --storage-path admin_storage.json
+#       --commissioning-method on-network
+#       --discriminator 1234
+#       --passcode 20202021
+#       --endpoint 0
+#       --string-arg th_server_app_path:${ALL_CLUSTERS_APP}
+#       --trace-to json:${TRACE_TEST_JSON}.json
+#       --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
+#     factoryreset: true
+#     quiet: true
+# === END CI TEST ARGUMENTS ===
 
 # This test requires a TH_SERVER application. Please specify with --string-arg th_server_app_path:<path_to_app>
 
 import logging
 import os
 import random
-import signal
-import subprocess
+import tempfile
 import time
-import uuid
 
 import chip.clusters as Clusters
 from chip import ChipDeviceCtrl
 from chip.interaction_model import InteractionModelError, Status
 from matter_testing_support import (MatterBaseTest, TestStep, async_test_body, default_matter_test_main, has_cluster,
                                     run_if_endpoint_matches)
 from mobly import asserts
+from TC_MCORE_FS_1_1 import AppServer
 
 
 class TC_CCTRL_2_2(MatterBaseTest):
 
     @async_test_body
     async def setup_class(self):
         super().setup_class()
-        self.app_process = None
-        app = self.user_params.get("th_server_app_path", None)
-        if not app:
-            asserts.fail('This test requires a TH_SERVER app. Specify app path with --string-arg th_server_app_path:<path_to_app>')
-
-        self.kvs = f'kvs_{str(uuid.uuid4())}'
-        self.port = 5543
-        discriminator = random.randint(0, 4095)
-        passcode = 20202021
-        cmd = [app]
-        cmd.extend(['--secured-device-port', str(5543)])
-        cmd.extend(['--discriminator', str(discriminator)])
-        cmd.extend(['--passcode', str(passcode)])
-        cmd.extend(['--KVS', self.kvs])
-        # TODO: Determine if we want these logs cooked or pushed to somewhere else
-        logging.info("Starting TH_SERVER")
-        self.app_process = subprocess.Popen(cmd)
-        logging.info("TH_SERVER started")
-        time.sleep(3)
+
+        self.th_server = None
+        self.storage = None
+
+        th_server_app = self.user_params.get("th_server_app_path", None)
+        if not th_server_app:
+            asserts.fail("This test requires a TH_SERVER app. Specify app path with --string-arg th_server_app_path:<path_to_app>")
+        if not os.path.exists(th_server_app):
+            asserts.fail(f"The path {th_server_app} does not exist")
+
+        # 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 = 5543
+        self.th_server_discriminator = random.randint(0, 4095)
+        self.th_server_passcode = 20202021
+
+        # Start the TH_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()
 
         logging.info("Commissioning from separate fabric")
 
@@ -71,20 +95,18 @@ async def setup_class(self):
         paa_path = str(self.matter_test_config.paa_trust_store_path)
         self.TH_server_controller = new_fabric_admin.NewController(nodeId=112233, paaTrustStorePath=paa_path)
         self.server_nodeid = 1111
-        await self.TH_server_controller.CommissionOnNetwork(nodeId=self.server_nodeid, setupPinCode=passcode, filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=discriminator)
+        await self.TH_server_controller.CommissionOnNetwork(
+            nodeId=self.server_nodeid,
+            setupPinCode=self.th_server_passcode,
+            filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR,
+            filter=self.th_server_discriminator)
         logging.info("Commissioning TH_SERVER complete")
 
     def teardown_class(self):
-        # In case the th_server_app_path does not exist, then we failed the test
-        # and there is nothing to remove
-        if self.app_process is not None:
-            logging.warning("Stopping app with SIGTERM")
-            self.app_process.send_signal(signal.SIGTERM.value)
-            self.app_process.wait()
-
-            if os.path.exists(self.kvs):
-                os.remove(self.kvs)
-
+        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_CCTRL_2_2(self) -> list[TestStep]:

src/python_testing/TC_CCTRL_2_3.py

@@ -18,50 +18,74 @@
 # See https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md#defining-the-ci-test-arguments
 # for details about the block below.
 #
-# TODO: Skip CI for now, we don't have any way to run this. Needs setup. See test_TC_CCTRL.py
+# === 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
+#     app-ready-pattern: "Successfully opened pairing window on the device"
+#     script-args: >
+#       --PICS src/app/tests/suites/certification/ci-pics-values
+#       --storage-path admin_storage.json
+#       --commissioning-method on-network
+#       --discriminator 1234
+#       --passcode 20202021
+#       --endpoint 0
+#       --string-arg th_server_app_path:${ALL_CLUSTERS_APP}
+#       --trace-to json:${TRACE_TEST_JSON}.json
+#       --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
+#     factoryreset: true
+#     quiet: true
+# === END CI TEST ARGUMENTS ===
 
 # This test requires a TH_SERVER application. Please specify with --string-arg th_server_app_path:<path_to_app>
 
 import logging
 import os
 import random
-import signal
-import subprocess
+import tempfile
 import time
-import uuid
 
 import chip.clusters as Clusters
 from chip import ChipDeviceCtrl
 from chip.interaction_model import InteractionModelError, Status
 from matter_testing_support import (MatterBaseTest, TestStep, async_test_body, default_matter_test_main, has_cluster,
                                     run_if_endpoint_matches)
 from mobly import asserts
+from TC_MCORE_FS_1_1 import AppServer
 
 
 class TC_CCTRL_2_3(MatterBaseTest):
 
     @async_test_body
     async def setup_class(self):
         super().setup_class()
-        self.app_process = None
-        app = self.user_params.get("th_server_app_path", None)
-        if not app:
-            asserts.fail('This test requires a TH_SERVER app. Specify app path with --string-arg th_server_app_path:<path_to_app>')
-
-        self.kvs = f'kvs_{str(uuid.uuid4())}'
-        self.port = 5543
-        discriminator = random.randint(0, 4095)
-        passcode = 20202021
-        cmd = [app]
-        cmd.extend(['--secured-device-port', str(5543)])
-        cmd.extend(['--discriminator', str(discriminator)])
-        cmd.extend(['--passcode', str(passcode)])
-        cmd.extend(['--KVS', self.kvs])
-        # TODO: Determine if we want these logs cooked or pushed to somewhere else
-        logging.info("Starting TH_SERVER")
-        self.app_process = subprocess.Popen(cmd)
-        logging.info("TH_SERVER started")
-        time.sleep(3)
+
+        self.th_server = None
+        self.storage = None
+
+        th_server_app = self.user_params.get("th_server_app_path", None)
+        if not th_server_app:
+            asserts.fail("This test requires a TH_SERVER app. Specify app path with --string-arg th_server_app_path:<path_to_app>")
+        if not os.path.exists(th_server_app):
+            asserts.fail(f"The path {th_server_app} does not exist")
+
+        # 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 = 5543
+        self.th_server_discriminator = random.randint(0, 4095)
+        self.th_server_passcode = 20202021
+
+        # Start the TH_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()
 
         logging.info("Commissioning from separate fabric")
 
@@ -71,20 +95,18 @@ async def setup_class(self):
         paa_path = str(self.matter_test_config.paa_trust_store_path)
         self.TH_server_controller = new_fabric_admin.NewController(nodeId=112233, paaTrustStorePath=paa_path)
         self.server_nodeid = 1111
-        await self.TH_server_controller.CommissionOnNetwork(nodeId=self.server_nodeid, setupPinCode=passcode, filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=discriminator)
+        await self.TH_server_controller.CommissionOnNetwork(
+            nodeId=self.server_nodeid,
+            setupPinCode=self.th_server_passcode,
+            filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR,
+            filter=self.th_server_discriminator)
         logging.info("Commissioning TH_SERVER complete")
 
     def teardown_class(self):
-        # In case the th_server_app_path does not exist, then we failed the test
-        # and there is nothing to remove
-        if self.app_process is not None:
-            logging.warning("Stopping app with SIGTERM")
-            self.app_process.send_signal(signal.SIGTERM.value)
-            self.app_process.wait()
-
-            if os.path.exists(self.kvs):
-                os.remove(self.kvs)
-
+        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_CCTRL_2_3(self) -> list[TestStep]:
@@ -172,7 +194,7 @@ async def test_TC_CCTRL_2_3(self):
         await self.send_single_cmd(cmd, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0, timedRequestTimeoutMs=5000)
 
         self.step(11)
-        time.sleep(30)
+        time.sleep(5 if self.is_pics_sdk_ci_only else 30)
 
         self.step(12)
         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)

src/python_testing/TC_MCORE_FS_1_1.py

@@ -95,7 +95,7 @@ async def setup_class(self):
         self.th_server_discriminator = random.randint(0, 4095)
         self.th_server_passcode = 20202021
 
-        # Start the TH_SERVER_NO_UID app.
+        # Start the TH_SERVER app.
         self.th_server = AppServer(
             th_server_app,
             storage_dir=self.storage.name,

src/python_testing/TC_MCORE_FS_1_2.py

@@ -110,7 +110,7 @@ async def setup_class(self):
             discriminator=3840,
             passcode=20202021)
 
-        # Start the TH_SERVER_NO_UID app.
+        # Start the TH_SERVER app.
         self.th_server = AppServer(
             th_server_app,
             storage_dir=self.storage.name,