[PICS-Removal] Removing automatable PICS from ICDM 2 1 test module

src/python_testing/TC_ICDM_2_1.py

@@ -39,7 +39,7 @@
 import re
 
 import chip.clusters as Clusters
-from chip.testing.matter_testing import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
+from chip.testing.matter_testing import MatterBaseTest, TestStep, async_function_runner, async_test_body, default_matter_test_main
 from mobly import asserts
 
 logger = logging.getLogger(__name__)
@@ -135,6 +135,8 @@ def pics_TC_ICDM_2_1(self) -> list[str]:
 
     @async_test_body
     async def test_TC_ICDM_2_1(self):
+        self.endpoint = self.get_endpoint()
+        self.timeout = self.matter_test_config.timeout if self.matter_test_config.timeout is not None else self.default_timeout
 
         cluster = Clusters.Objects.IcdManagement
         attributes = cluster.Attributes
@@ -148,53 +150,39 @@ async def test_TC_ICDM_2_1(self):
 
         # Validate ActiveModeThreshold
         self.step(2)
-        if self.check_pics("ICDM.S.A0002"):
+        activeModeThreshold = await self._read_icdm_attribute_expect_success(
+            attributes.ActiveModeThreshold)
+        # Verify ActiveModeThreshold is not bigger than uint16
+        asserts.assert_true(self.is_valid_uint16_value(activeModeThreshold),
+                            "ActiveModeThreshold attribute does not fit in a uint16.")
 
-            activeModeThreshold = await self._read_icdm_attribute_expect_success(
-                attributes.ActiveModeThreshold)
-            # Verify ActiveModeThreshold is not bigger than uint16
-            asserts.assert_true(self.is_valid_uint16_value(activeModeThreshold),
-                                "ActiveModeThreshold attribute does not fit in a uint16.")
-
-            if featureMap > 0 and features.kLongIdleTimeSupport in features(featureMap):
-                asserts.assert_greater_equal(
-                    activeModeThreshold, 5000, "Minimum ActiveModeThreshold is 5s for a LIT ICD.")
-
-        else:
-            asserts.assert_true(
-                False, "ActiveModeThreshold is a mandatory attribute and must be present in the PICS file")
+        if featureMap > 0 and features.kLongIdleTimeSupport in features(featureMap):
+            asserts.assert_greater_equal(
+                activeModeThreshold, 5000, "Minimum ActiveModeThreshold is 5s for a LIT ICD.")
 
         # Validate ActiveModeDuration
         self.step(3)
-        if self.check_pics("ICDM.S.A0001"):
-            activeModeDuration = await self._read_icdm_attribute_expect_success(
-                attributes.ActiveModeDuration)
-            # Verify ActiveModeDuration is not bigger than uint32
-            asserts.assert_true(self.is_valid_uint32_value(activeModeDuration),
-                                "ActiveModeDuration attribute does not fit in a uint32")
-        else:
-            asserts.assert_true(
-                False, "ActiveModeDuration is a mandatory attribute and must be present in the PICS file")
+        activeModeDuration = await self._read_icdm_attribute_expect_success(
+            attributes.ActiveModeDuration)
+        # Verify ActiveModeDuration is not bigger than uint32
+        asserts.assert_true(self.is_valid_uint32_value(activeModeDuration),
+                            "ActiveModeDuration attribute does not fit in a uint32")
 
         # Validate IdleModeDuration
         self.step(4)
-        if self.check_pics("ICDM.S.A0000"):
-            idleModeDuration = await self._read_icdm_attribute_expect_success(
-                attributes.IdleModeDuration)
-            # Verify IdleModeDuration is not bigger than uint32
-            asserts.assert_greater_equal(
-                idleModeDuration, 1, "IdleModeDuration attribute is smaller than minimum value (1).")
-            asserts.assert_less_equal(
-                idleModeDuration, 64800, "IdleModeDuration attribute is greater than maximum value (64800).")
-            asserts.assert_greater_equal(idleModeDuration * 1000, activeModeDuration,
-                                         "ActiveModeDuration attribute is greater than the IdleModeDuration attrbiute.")
-        else:
-            asserts.assert_true(
-                False, "IdleModeDuration is a mandatory attribute and must be present in the PICS file")
+        idleModeDuration = await self._read_icdm_attribute_expect_success(
+            attributes.IdleModeDuration)
+        # Verify IdleModeDuration is not bigger than uint32
+        asserts.assert_greater_equal(
+            idleModeDuration, 1, "IdleModeDuration attribute is smaller than minimum value (1).")
+        asserts.assert_less_equal(
+            idleModeDuration, 64800, "IdleModeDuration attribute is greater than maximum value (64800).")
+        asserts.assert_greater_equal(idleModeDuration * 1000, activeModeDuration,
+                                     "ActiveModeDuration attribute is greater than the IdleModeDuration attrbiute.")
 
         # Validate ClientsSupportedPerFabric
         self.step(5)
-        if self.pics_guard(self.check_pics("ICDM.S.A0005")):
+        if async_function_runner(self.attribute_guard(endpoint=self.endpoint, attribute=attributes.ClientsSupportedPerFabric), timeout=self.timeout):
             clientsSupportedPerFabric = await self._read_icdm_attribute_expect_success(
                 attributes.ClientsSupportedPerFabric)
 
@@ -207,7 +195,7 @@ async def test_TC_ICDM_2_1(self):
 
         # Validate RegisteredClients
         self.step(6)
-        if self.pics_guard(self.check_pics("ICDM.S.A0003")):
+        if async_function_runner(self.attribute_guard(endpoint=self.endpoint, attribute=attributes.RegisteredClients), timeout=self.timeout):
             registeredClients = await self._read_icdm_attribute_expect_success(
                 attributes.RegisteredClients)
 
@@ -216,8 +204,7 @@ async def test_TC_ICDM_2_1(self):
 
         # Validate ICDCounter
         self.step(7)
-        if self.pics_guard(self.check_pics("ICDM.S.A0004")):
-
+        if async_function_runner(self.attribute_guard(endpoint=self.endpoint, attribute=attributes.ICDCounter), timeout=self.timeout):
             icdCounter = await self._read_icdm_attribute_expect_success(
                 attributes.ICDCounter)
             # Verify ICDCounter is not bigger than uint32
@@ -226,7 +213,7 @@ async def test_TC_ICDM_2_1(self):
 
         # Validate UserActiveModeTriggerHint
         self.step(8)
-        if self.pics_guard(self.check_pics("ICDM.S.A0006")):
+        if async_function_runner(self.attribute_guard(endpoint=self.endpoint, attribute=attributes.UserActiveModeTriggerHint), timeout=self.timeout):
             userActiveModeTriggerHint = await self._read_icdm_attribute_expect_success(
                 attributes.UserActiveModeTriggerHint)
 
@@ -243,7 +230,7 @@ async def test_TC_ICDM_2_1(self):
 
         # Valdate UserActiveModeTriggerInstruction
         self.step(9)
-        if self.check_pics("ICDM.S.A0007"):
+        if async_function_runner(self.attribute_guard(endpoint=self.endpoint, attribute=attributes.UserActiveModeTriggerInstruction), timeout=self.timeout):
             userActiveModeTriggerInstruction = await self._read_icdm_attribute_expect_success(
                 attributes.UserActiveModeTriggerInstruction)
 
@@ -268,14 +255,15 @@ async def test_TC_ICDM_2_1(self):
                 pattern = re.compile(r'^[0-9A-F]{6}$')
                 asserts.assert_true(pattern.match(userActiveModeTriggerInstruction),
                                     "UserActiveModeTriggerInstruction is not in the correct format for the associated UserActiveModeTriggerHint")
-        elif self.check_pics("ICDM.S.A0006"):
+
+        elif async_function_runner(self.attribute_guard(endpoint=self.endpoint, attribute=attributes.UserActiveModeTriggerHint), timeout=self.timeout):
             # Check if the UserActiveModeTriggerInstruction was required
             asserts.assert_false(uatHintInstructionDepedentBitmap in kUatInstructionMandatoryBitMask,
                                  "UserActiveModeTriggerHint requires the UserActiveModeTriggerInstruction")
 
         # Verify OperatingMode
         self.step(10)
-        if self.pics_guard(self.check_pics("ICDM.S.A0008")):
+        if async_function_runner(self.attribute_guard(endpoint=self.endpoint, attribute=attributes.OperatingMode), timeout=self.timeout):
             operatingMode = await self._read_icdm_attribute_expect_success(
                 attributes.OperatingMode)
 
@@ -284,8 +272,9 @@ async def test_TC_ICDM_2_1(self):
 
             asserts.assert_less(
                 operatingMode, modes.kUnknownEnumValue, "OperatingMode can only have 0 and 1 as valid values")
+
         self.step(11)
-        if self.pics_guard(self.check_pics("ICDM.S.A0009")):
+        if async_function_runner(self.attribute_guard(endpoint=self.endpoint, attribute=attributes.MaximumCheckInBackOff), timeout=self.timeout):
             maximumCheckInBackOff = await self._read_icdm_attribute_expect_success(attributes.MaximumCheckInBackOff)
 
             asserts.assert_true(self.is_valid_uint32_value(maximumCheckInBackOff),

src/python_testing/matter_testing_infrastructure/chip/testing/matter_testing.py

@@ -33,6 +33,7 @@
 import typing
 import uuid
 from binascii import hexlify, unhexlify
+from concurrent.futures import ThreadPoolExecutor
 from dataclasses import asdict as dataclass_asdict
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
@@ -2203,6 +2204,19 @@ def parse_matter_test_args(argv: Optional[List[str]] = None) -> MatterTestConfig
     return convert_args_to_matter_config(parser.parse_known_args(argv)[0])
 
 
+def async_function_runner(function, timeout):
+    """
+    Runs an async function synchronously, even if an event loop is already running.
+    Follows similar below flow for _async_runner() 
+    """
+    async def frapper():
+        return await asyncio.wait_for(function, timeout=timeout)
+
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        future = executor.submit(lambda: asyncio.run(frapper()))
+        return future.result()
+
+
 def _async_runner(body, self: MatterBaseTest, *args, **kwargs):
     timeout = self.matter_test_config.timeout if self.matter_test_config.timeout is not None else self.default_timeout
     return self.event_loop.run_until_complete(asyncio.wait_for(body(self, *args, **kwargs), timeout=timeout))