Add TC_BRBINFO_4_1 to CI (project-chip#35940)

* Use async sleep instead of blocking one

* Keep AppServer wrapper class in chip.testing.tasks

* Fix typos

* Arguments for CI workflow

* Require dut_fsa_stdin_pipe options in CI run

* Start ICD server on CI run

* Read all attributes defined in test plan

* Wrap ICD server with IcdAppServerSubprocess

* Move app testing wrappers to chip.testing.apps

* Add TODO for fixing imports in unit test

* Restyled by isort

* Remove test from exception list

---------

Co-authored-by: Restyled.io <commits@restyled.io>

src/python_testing/TC_CCTRL_2_2.py

@@ -49,10 +49,10 @@
 import chip.clusters as Clusters
 from chip import ChipDeviceCtrl
 from chip.interaction_model import InteractionModelError, Status
+from chip.testing.apps import AppServerSubprocess
 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):
@@ -79,13 +79,15 @@ async def setup_class(self):
         self.th_server_passcode = 20202021
 
         # Start the TH_SERVER app.
-        self.th_server = AppServer(
+        self.th_server = AppServerSubprocess(
             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()
+        self.th_server.start(
+            expected_output="Server initialization complete",
+            timeout=30)
 
         logging.info("Commissioning from separate fabric")
 

src/python_testing/TC_CCTRL_2_3.py

@@ -49,10 +49,10 @@
 import chip.clusters as Clusters
 from chip import ChipDeviceCtrl
 from chip.interaction_model import InteractionModelError, Status
+from chip.testing.apps import AppServerSubprocess
 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):
@@ -79,13 +79,15 @@ async def setup_class(self):
         self.th_server_passcode = 20202021
 
         # Start the TH_SERVER app.
-        self.th_server = AppServer(
+        self.th_server = AppServerSubprocess(
             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()
+        self.th_server.start(
+            expected_output="Server initialization complete",
+            timeout=30)
 
         logging.info("Commissioning from separate fabric")
 

src/python_testing/TC_ECOINFO_2_1.py

@@ -46,10 +46,10 @@
 import chip.clusters as Clusters
 from chip.clusters.Types import NullValue
 from chip.interaction_model import Status
+from chip.testing.apps import AppServerSubprocess
 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):
@@ -95,13 +95,15 @@ async def _setup_ci_prerequisites(self):
         self.th_server_passcode = 20202021
 
         # Start the server app.
-        self.th_server = AppServer(
+        self.th_server = AppServerSubprocess(
             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()
+        self.th_server.start(
+            expected_output="Server initialization complete",
+            timeout=30)
 
         # Add some server to the DUT_FSA's Aggregator/Bridge.
         self.dut_fsa_stdin.write(f"pairing onnetwork 2 {self.th_server_passcode}\n")

src/python_testing/TC_ECOINFO_2_2.py

@@ -45,9 +45,9 @@
 
 import chip.clusters as Clusters
 from chip.interaction_model import Status
+from chip.testing.apps import AppServerSubprocess
 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
 
@@ -94,13 +94,15 @@ def _setup_ci_prerequisites(self):
         self.th_server_passcode = 20202021
 
         # Start the server app.
-        self.th_server = AppServer(
+        self.th_server = AppServerSubprocess(
             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()
+        self.th_server.start(
+            expected_output="Server initialization complete",
+            timeout=30)
 
     def steps_TC_ECOINFO_2_2(self) -> list[TestStep]:
         return [

src/python_testing/TC_MCORE_FS_1_1.py

@@ -47,31 +47,11 @@
 
 import chip.clusters as Clusters
 from chip import ChipDeviceCtrl
-from chip.testing.tasks import Subprocess
+from chip.testing.apps import AppServerSubprocess
 from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
 from mobly import asserts
 
 
-class AppServer(Subprocess):
-    """Wrapper class for starting an application server in a subprocess."""
-
-    # Prefix for log messages from the application 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]
-        # Start the server application with dedicated KVS storage.
-        super().__init__(app, "--KVS", storage_kvs_dir,
-                         '--secured-device-port', str(port),
-                         "--discriminator", str(discriminator),
-                         "--passcode", str(passcode),
-                         output_cb=lambda line, is_stderr: self.PREFIX + line)
-
-    def start(self):
-        # Start process and block until it prints the expected output.
-        super().start(expected_output="Server initialization complete")
-
-
 class TC_MCORE_FS_1_1(MatterBaseTest):
 
     @async_test_body
@@ -96,13 +76,15 @@ async def setup_class(self):
         self.th_server_passcode = 20202021
 
         # Start the TH_SERVER app.
-        self.th_server = AppServer(
+        self.th_server = AppServerSubprocess(
             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()
+        self.th_server.start(
+            expected_output="Server initialization complete",
+            timeout=30)
 
         logging.info("Commissioning from separate fabric")
         # Create a second controller on a new fabric to communicate to the server

src/python_testing/TC_MCORE_FS_1_2.py

@@ -50,10 +50,10 @@
 
 import chip.clusters as Clusters
 from chip import ChipDeviceCtrl
+from chip.testing.apps import AppServerSubprocess
 from ecdsa.curves import NIST256p
 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
 from TC_SC_3_6 import AttributeChangeAccumulator
 
 # Length of `w0s` and `w1s` elements
@@ -97,9 +97,11 @@ async def setup_class(self):
         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.
-        dut_fsa_stdin_pipe = self.user_params.get("dut_fsa_stdin_pipe", None)
-        if dut_fsa_stdin_pipe is not None:
+        if self.is_pics_sdk_ci_only:
+            # 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:<path_to_pipe>")
             self.dut_fsa_stdin = open(dut_fsa_stdin_pipe, "w")
 
         self.th_server_port = th_server_port
@@ -111,13 +113,15 @@ async def setup_class(self):
             passcode=20202021)
 
         # Start the TH_SERVER app.
-        self.th_server = AppServer(
+        self.th_server = AppServerSubprocess(
             th_server_app,
             storage_dir=self.storage.name,
             port=self.th_server_port,
             discriminator=self.th_server_setup_params.discriminator,
             passcode=self.th_server_setup_params.passcode)
-        self.th_server.start()
+        self.th_server.start(
+            expected_output="Server initialization complete",
+            timeout=30)
 
     def teardown_class(self):
         if self._partslist_subscription is not None:
@@ -135,7 +139,7 @@ def _ask_for_vendor_commissioning_ux_operation(self, setup_params: _SetupParamet
             f"- discriminator: {setup_params.discriminator}\n"
             f"- setupPinCode: {setup_params.passcode}\n"
             f"- setupQRCode: {setup_params.setup_qr_code}\n"
-            f"- setupManualcode: {setup_params.manual_code}\n"
+            f"- setupManualCode: {setup_params.manual_code}\n"
             f"If using FabricSync Admin test app, you may type:\n"
             f">>> pairing onnetwork 111 {setup_params.passcode}")
 

src/python_testing/TC_MCORE_FS_1_3.py

@@ -50,9 +50,9 @@
 import chip.clusters as Clusters
 from chip import ChipDeviceCtrl
 from chip.interaction_model import Status
+from chip.testing.apps import AppServerSubprocess
 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_MCORE_FS_1_3(MatterBaseTest):
@@ -84,13 +84,15 @@ def setup_class(self):
         self.th_server_passcode = 20202021
 
         # Start the TH_SERVER_NO_UID app.
-        self.th_server = AppServer(
+        self.th_server = AppServerSubprocess(
             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()
+        self.th_server.start(
+            expected_output="Server initialization complete",
+            timeout=30)
 
     def teardown_class(self):
         if self.th_server is not None:

src/python_testing/TC_MCORE_FS_1_4.py

@@ -49,10 +49,10 @@
 import chip.clusters as Clusters
 from chip import ChipDeviceCtrl
 from chip.interaction_model import Status
+from chip.testing.apps import AppServerSubprocess
 from chip.testing.tasks import Subprocess
 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 FabricSyncApp(Subprocess):
@@ -160,23 +160,27 @@ def setup_class(self):
             vendor_id=0xFFF1)
         self.th_fsa_controller.start()
 
-        # 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", None)
-        if dut_fsa_stdin_pipe is not None:
+        if self.is_pics_sdk_ci_only:
+            # 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:<path_to_pipe>")
             self.dut_fsa_stdin = open(dut_fsa_stdin_pipe, "w")
 
         self.th_server_port = 5544
         self.th_server_discriminator = self.th_fsa_bridge_discriminator + 1
         self.th_server_passcode = 20202022
 
         # Start the TH_SERVER_NO_UID app.
-        self.th_server = AppServer(
+        self.th_server = AppServerSubprocess(
             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()
+        self.th_server.start(
+            expected_output="Server initialization complete",
+            timeout=30)
 
     def teardown_class(self):
         if self.th_fsa_controller is not None:

src/python_testing/TC_MCORE_FS_1_5.py

@@ -50,10 +50,10 @@
 
 import chip.clusters as Clusters
 from chip import ChipDeviceCtrl
+from chip.testing.apps import AppServerSubprocess
 from ecdsa.curves import NIST256p
 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
 from TC_SC_3_6 import AttributeChangeAccumulator
 
 # Length of `w0s` and `w1s` elements
@@ -98,9 +98,11 @@ async def setup_class(self):
         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.
-        dut_fsa_stdin_pipe = self.user_params.get("dut_fsa_stdin_pipe", None)
-        if dut_fsa_stdin_pipe is not None:
+        if self.is_pics_sdk_ci_only:
+            # 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:<path_to_pipe>")
             self.dut_fsa_stdin = open(dut_fsa_stdin_pipe, "w")
 
         self.th_server_port = th_server_port
@@ -112,13 +114,15 @@ async def setup_class(self):
             passcode=20202021)
 
         # Start the TH_SERVER app.
-        self.th_server = AppServer(
+        self.th_server = AppServerSubprocess(
             th_server_app,
             storage_dir=self.storage.name,
             port=self.th_server_port,
             discriminator=self.th_server_setup_params.discriminator,
             passcode=self.th_server_setup_params.passcode)
-        self.th_server.start()
+        self.th_server.start(
+            expected_output="Server initialization complete",
+            timeout=30)
 
     def teardown_class(self):
         if self._partslist_subscription is not None:
@@ -139,7 +143,7 @@ def _ask_for_vendor_commissioning_ux_operation(self, setup_params: _SetupParamet
             f"- discriminator: {setup_params.discriminator}\n"
             f"- setupPinCode: {setup_params.passcode}\n"
             f"- setupQRCode: {setup_params.setup_qr_code}\n"
-            f"- setupManualcode: {setup_params.manual_code}\n"
+            f"- setupManualCode: {setup_params.manual_code}\n"
             f"If using FabricSync Admin test app, you may type:\n"
             f">>> pairing onnetwork 111 {setup_params.passcode}")
 

src/python_testing/execute_python_tests.py

@@ -66,7 +66,6 @@ def main(search_directory, env_file):
         "TC_TMP_2_1.py",  # src/python_testing/test_testing/test_TC_TMP_2_1.py is the Unit test of this test
         "TC_OCC_3_1.py",  # There are CI issues for the test cases that implements manually controlling sensor device for the occupancy state ON/OFF change
         "TC_OCC_3_2.py",  # There are CI issues for the test cases that implements manually controlling sensor device for the occupancy state ON/OFF change
-        "TC_BRBINFO_4_1.py",  # This test requires a TH_ICD_SERVER application, hence not ready run with CI
         "TestCommissioningTimeSync.py",  # Code/Test not being used or not shared code for any other tests
         "TestConformanceSupport.py",  # Unit test - does not run against an app
         "TestChoiceConformanceSupport.py",  # Unit test - does not run against an app

src/python_testing/matter_testing_infrastructure/BUILD.gn

@@ -30,6 +30,7 @@ pw_python_package("chip-testing") {
 
   sources = [
     "chip/testing/__init__.py",
+    "chip/testing/apps.py",
     "chip/testing/metadata.py",
     "chip/testing/tasks.py",
   ]