comments addressed, incl TH_SERVER configurable

src/python_testing/TC_BRBINFO_4_1.py

@@ -18,42 +18,32 @@
 # 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
-# test-runner-run/run1/app: ${ALL_CLUSTERS_APP}
-# test-runner-run/run1/factoryreset: True
-# test-runner-run/run1/quiet: True
-# test-runner-run/run1/app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json
-# test-runner-run/run1/script-args: --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --PICS src/app/tests/suites/certification/ci-pics-values --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
-# === END CI TEST ARGUMENTS ===
+
+# This test requires a TH_SERVER application. Please specify with --string-arg th_server_app_path:<path_to_app>
+# TH_SERVER must support following arguments: --secured-device-port --discriminator --passcode --KVS --secured-device-port
+# E.g: python3 src/python_testing/TC_BRBINFO_4_1.py --commissioning-method on-network --qr-code MT:-24J042C00KA0648G00 \
+#      --string-arg th_server_app_path:out/linux-x64-lit-icd/lit-icd-app
 
 import logging
 import os
 import queue
-import pathlib
 import uuid
 import subprocess
 import time
 import signal
 
 
 import chip.clusters as Clusters
-from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main, SimpleEventCallback
+from chip import ChipDeviceCtrl
+from matter_testing_support import (MatterBaseTest, TestStep, async_test_body, default_matter_test_main, SimpleEventCallback)
 from mobly import asserts
 
-from chip import ChipDeviceCtrl  # Needed before chip.FabricAdmin
 
 logger = logging.getLogger(__name__)
 kRootEndpointId = 0
-kICDMEndpointId = 0
-kICDBridgedEndpointId = 2
 kMaxUserActiveModeBitmap = 0x1FFFF
 kMaxUserActiveModeTriggerInstructionByteLength = 128
 
-#uat = cluster.Bitmaps.UserActiveModeTriggerBitmap
-#modes = cluster.Enums.OperatingModeEnum
-#features = cluster.Bitmaps.Feature
-
 class TC_BRBINFO_4_1(MatterBaseTest):
 
     #
@@ -67,6 +57,12 @@ async def _read_attribute_expect_success(self, endpoint, cluster, attribute, nod
     # Test Harness Helpers
     #
 
+    # Override default timeout to support a 60 min wait
+    @property
+    def default_timeout(self) -> int:
+        return 63*60
+
+
     def desc_TC_BRBINFO_4_1(self) -> str:
         """Returns a description of this test"""
         return "[TC_BRBINFO_4_1] Verification of KeepActive Command [DUT-Server]"
@@ -82,13 +78,6 @@ def steps_TC_BRBINFO_4_1(self) -> list[TestStep]:
         return steps
 
 
-    def pics_TC_BRBINFO_4_1(self) -> list[str]:
-        """ This function returns a list of PICS for this test case that must be True for the test to be run"""
-        pics = [
-            "BRBINFO.C.80",
-        ]
-        return pics
-
     def _ask_for_vendor_commissioniong_ux_operation(self, discriminator, setupPinCode):
         self.wait_for_user_input(
             prompt_msg=f"Using the DUT vendor's provided interface, commission the ICD device using the following parameters:\n"
@@ -97,12 +86,9 @@ def _ask_for_vendor_commissioniong_ux_operation(self, discriminator, setupPinCod
                         f"If using FabricSync Admin, you may type:\n"
                         f">>> pairing onnetwork 111 {setupPinCode}")
 
-    def _display_message_and_wait_for_user_input(self, message):
-        self.wait_for_user_input(prompt_msg=message)
-
-    async def _send_keep_active_command(self, duration) -> int:
-        logging.info("Mock sending keep active command")
-        keep_active = await self.default_controller.SendCommand(nodeid=self.dut_node_id, endpoint=kICDBridgedEndpointId, payload=Clusters.Objects.BridgedDeviceBasicInformation.Commands.KeepActive(stayActiveDuration=duration))
+    async def _send_keep_active_command(self, duration, endpoint_id) -> int:
+        logging.info("Sending keep active command")
+        keep_active = await self.default_controller.SendCommand(nodeid=self.dut_node_id, endpoint=endpoint_id, payload=Clusters.Objects.BridgedDeviceBasicInformation.Commands.KeepActive(stayActiveDuration=duration))
         return keep_active
 
     async def _wait_for_active_changed_event(self, timeout) -> int:
@@ -111,13 +97,30 @@ async def _wait_for_active_changed_event(self, timeout) -> int:
             logging.info(f"PromisedActiveDuration: {promised_active_duration}")
             return promised_active_duration
         except queue.Empty:
-            asserts.fail("Timeout on event")
+            asserts.fail("Timeout on event ActiveChanged")
+
+    async def _get_dynamic_endpoint(self) -> int:
+        root_part_list = await self.read_single_attribute_check_success(cluster=Clusters.Descriptor, attribute=Clusters.Descriptor.Attributes.PartsList, endpoint=kRootEndpointId)
+        set_of_endpoints_after_adding_device = set(root_part_list)
+
+        asserts.assert_true(set_of_endpoints_after_adding_device.issuperset(self.set_of_dut_endpoints_before_adding_device), "Expected only new endpoints to be added")
+        unique_endpoints_set = set_of_endpoints_after_adding_device - self.set_of_dut_endpoints_before_adding_device
+        asserts.assert_equal(len(unique_endpoints_set), 1, "Expected only one new endpoint")
+        newly_added_endpoint = list(unique_endpoints_set)[0]
+        return newly_added_endpoint
 
 
     @async_test_body
     async def setup_class(self):
+        # These steps are not explicitly, but they help identify the dynamically added endpoint
+        # The second part of this process happens on _get_dynamic_endpoint()
+        root_part_list = await self.read_single_attribute_check_success(cluster=Clusters.Descriptor, attribute=Clusters.Descriptor.Attributes.PartsList, endpoint=kRootEndpointId)
+        self.set_of_dut_endpoints_before_adding_device = set(root_part_list)
+
         super().setup_class()
-        app = os.path.join(pathlib.Path(__file__).resolve().parent, '..','..','out', 'linux-x64-lit-icd', 'lit-icd-app')
+        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
@@ -162,12 +165,15 @@ async def test_TC_BRBINFO_4_1(self):
         basic_info_cluster = Clusters.Objects.BasicInformation
         basic_info_attributes = basic_info_cluster.Attributes
 
+        dynamic_endpoint_id = await self._get_dynamic_endpoint()
+        logging.info(f"Dynamic endpoint is {dynamic_endpoint_id}")
+
         # Preconditions
         self.step("0")
 
         logging.info("Ensuring DUT is commissioned to TH")
 
-        # Confirms commissioning of DUT on TH as it reads basic info
+        # Confirms commissioning of DUT on TH as it reads its fature map
         await self._read_attribute_expect_success(
           kRootEndpointId,
           basic_info_cluster,
@@ -177,7 +183,7 @@ async def test_TC_BRBINFO_4_1(self):
 
         logging.info("Ensuring ICD is commissioned to TH")
 
-        # Confirms commissioning of ICD on TH as it reads basic info
+        # Confirms commissioning of ICD on TH as it reads its feature map
         await self._read_attribute_expect_success(
           kRootEndpointId,
           basic_info_cluster,
@@ -188,15 +194,15 @@ async def test_TC_BRBINFO_4_1(self):
         self.step("1a")
 
         idle_mode_duration = await self._read_attribute_expect_success(
-          kICDMEndpointId,
+          kRootEndpointId,
           icdm_cluster,
           icdm_attributes.IdleModeDuration,
           self.icd_nodeid
         )
         logging.info(f"IdleModeDuration: {idle_mode_duration}")
 
         active_mode_duration = await self._read_attribute_expect_success(
-          kICDMEndpointId,
+          kRootEndpointId,
           icdm_cluster,
           icdm_attributes.ActiveModeDuration,
           self.icd_nodeid
@@ -210,12 +216,12 @@ async def test_TC_BRBINFO_4_1(self):
         self.q = queue.Queue()
         urgent = 1
         cb = SimpleEventCallback("ActiveChanged", event.cluster_id, event.event_id, self.q)
-        subscription = await self.default_controller.ReadEvent(nodeid=self.dut_node_id, events=[(kICDBridgedEndpointId, event, urgent)], reportInterval=[1, 3])
+        subscription = await self.default_controller.ReadEvent(nodeid=self.dut_node_id, events=[(dynamic_endpoint_id, event, urgent)], reportInterval=[1, 3])
         subscription.SetEventUpdateCallback(callback=cb)
 
         stay_active_duration = 1000
         logging.info(f"Sending KeepActiveCommand({stay_active_duration}ms)")
-        self._send_keep_active_command(stay_active_duration)
+        self._send_keep_active_command(stay_active_duration, dynamic_endpoint_id)
 
         logging.info("Waiting for ActiveChanged from DUT...")
         promised_active_duration = await self._wait_for_active_changed_event((idle_mode_duration + max(active_mode_duration, stay_active_duration))/1000)
@@ -232,18 +238,18 @@ async def test_TC_BRBINFO_4_1(self):
         promised_active_duration = await self._wait_for_active_changed_event((idle_mode_duration + max(active_mode_duration, stay_active_duration))/1000)
 
         # wait for active time duration
-        time.sleep(max(stay_active_duration/1000, active_mode_duration))
+        time.sleep(max(stay_active_duration/1000, promised_active_duration))
         # ICD now should be in idle mode
 
         # sends 3x keep active commands
         logging.info(f"Sending KeepActiveCommand({stay_active_duration})")
-        self._send_keep_active_command(stay_active_duration)
+        self._send_keep_active_command(stay_active_duration, dynamic_endpoint_id)
         time.sleep(100)
         logging.info(f"Sending KeepActiveCommand({stay_active_duration})")
-        self._send_keep_active_command(stay_active_duration)
+        self._send_keep_active_command(stay_active_duration, dynamic_endpoint_id)
         time.sleep(100)
         logging.info(f"Sending KeepActiveCommand({stay_active_duration})")
-        self._send_keep_active_command(stay_active_duration)
+        self._send_keep_active_command(stay_active_duration, dynamic_endpoint_id)
         time.sleep(100)
 
         logging.info("Waiting for ActiveChanged from DUT...")
@@ -255,16 +261,17 @@ async def test_TC_BRBINFO_4_1(self):
 
         stay_active_duration = 10000
         logging.info(f"Sending KeepActiveCommand({stay_active_duration})")
-        self._send_keep_active_command(stay_active_duration)
+        self._send_keep_active_command(stay_active_duration, dynamic_endpoint_id)
 
-        ## halt the ICD process
+        ## stops (halts) the ICD server process by sending a SIGTOP signal
         self.app_process.send_signal(signal.SIGSTOP.value)
 
         if not self.is_ci:
             logging.info("Waiting for 60 minutes")
+            self._timeout
             time.sleep(60*60)
 
-        ## resume the ICD
+        ## resumes (continues) the ICD server process by sending a SIGCONT signal
         self.app_process.send_signal(signal.SIGCONT.value)
 
         # wait for active changed event, expect no event will be sent