Improvements in Python testing support (#27116)

* Improvements in Python testing support

Problem:
    - It was no longer possible to run `matter_testing_support.py`-based
      tests on already commissioned devices due to changes in command
      line logic that impacted fabric ID selection and broke stack init.
    - Some variables that contained multiple entries in
      `matter_test_support.py` were singular which was somewhat confusing
      and a change from the initial semantics where unique values only
      were allowed.

Changes in this PR:
    - Pluralize required fields
    - Fix running tests without commissioning
    - Improve setup_payload decoder module API
    - Give access to raw TLV results for reads, not just subs

Testing done:
    - Re-ran `TC_DA_1_7.py` with updates (thanks cecille@)
    - Re-ran `TC_CGEN_2_4.py` with updates.
    - Used updated APIs in a seperate branch (upcoming)

* Restyled by isort

* Fix TC_RR_1_1.py

* Fix indent

* Restyled by autopep8

---------

Co-authored-by: Restyled.io <[email protected]>
Co-authored-by: [email protected] <[email protected]>

src/controller/python/chip/clusters/Attribute.py

@@ -505,7 +505,7 @@ def OverrideLivenessTimeoutMs(self, timeoutMs: int):
 
     def GetReportingIntervalsSeconds(self) -> Tuple[int, int]:
         '''
-        Retrieve the reporting intervals associated with an active subscription. 
+        Retrieve the reporting intervals associated with an active subscription.
         This should only be called if we're of subscription interaction type and after a subscription has been established.
         '''
         handle = chip.native.GetLibraryHandle()
@@ -650,8 +650,9 @@ def _BuildEventIndex():
 class AsyncReadTransaction:
     @dataclass
     class ReadResponse:
-        attributes: AttributeCache = None
-        events: List[ClusterEvent] = None
+        attributes: dict[Any, Any]
+        events: list[ClusterEvent]
+        tlvAttributes: dict[int, Any]
 
     def __init__(self, future: Future, eventLoop, devCtrl, returnClusterObject: bool):
         self._event_loop = eventLoop
@@ -796,7 +797,7 @@ def _handleDone(self):
                     self._future.set_exception(chip.exceptions.ChipStackError(self._resultError))
             else:
                 self._future.set_result(AsyncReadTransaction.ReadResponse(
-                    attributes=self._cache.attributeCache, events=self._events))
+                    attributes=self._cache.attributeCache, events=self._events, tlvAttributes=self._cache.attributeTLVCache))
 
         #
         # Decrement the ref on ourselves to match the increment that happened at allocation.

src/controller/python/chip/setup_payload/setup_payload.py

@@ -1,5 +1,5 @@
 #
-#    Copyright (c) 2021 Project CHIP Authors
+#    Copyright (c) 2021-2023 Project CHIP Authors
 #
 #    Licensed under the Apache License, Version 2.0 (the "License");
 #    you may not use this file except in compliance with the License.
@@ -15,6 +15,7 @@
 #
 
 from ctypes import CFUNCTYPE, c_char_p, c_int32, c_uint8
+from typing import Optional
 
 from chip.exceptions import ChipStackError
 from chip.native import GetLibraryHandle, NativeLibraryHandleMethodArguments
@@ -65,6 +66,7 @@ def ParseManualPairingCode(self, manualPairingCode: str):
 
         return self
 
+    # DEPRECATED
     def PrintOnboardingCodes(self, passcode, vendorId, productId, discriminator, customFlow, capabilities, version):
         self.Clear()
         err = self.chipLib.pychip_SetupPayload_PrintOnboardingCodes(
@@ -73,6 +75,7 @@ def PrintOnboardingCodes(self, passcode, vendorId, productId, discriminator, cus
         if err != 0:
             raise ChipStackError(err)
 
+    # DEPRECATED
     def Print(self):
         for name, value in self.attributes.items():
             decorated_value = self.__DecorateValue(name, value)
@@ -83,10 +86,12 @@ def Print(self):
             print(
                 f"Vendor attribute '{tag:>3}': {self.vendor_attributes[tag]}")
 
+    # DEPRECATED
     def Clear(self):
         self.attributes.clear()
         self.vendor_attributes.clear()
 
+    # DEPRECATED
     def __DecorateValue(self, name, value):
         if name == "RendezvousInformation":
             rendezvous_methods = []
@@ -113,3 +118,51 @@ def __InitNativeFunctions(self, chipLib):
         setter.Set("pychip_SetupPayload_PrintOnboardingCodes",
                    c_int32,
                    [c_uint32, c_uint16, c_uint16, c_uint16, uint8_t, uint8_t, uint8_t])
+
+    # Getters from parsed contents.
+    # Prefer using the methods below to access setup payload information once parse.
+
+    @property
+    def vendor_id(self) -> int:
+        return int(self.attributes.get("VendorID", "0"))
+
+    @property
+    def product_id(self) -> int:
+        return int(self.attributes.get("ProductID", "0"))
+
+    @property
+    def setup_passcode(self) -> int:
+        if "SetUpPINCode" not in self.attributes:
+            raise KeyError("Missing setup passcode in setup payload: parsing likely not yet done")
+
+        return int(self.attributes["SetUpPINCode"])
+
+    @property
+    def long_discriminator(self) -> Optional[int]:
+        if "Long discriminator" not in self.attributes:
+            return None
+
+        return int(self.attributes["Long discriminator"])
+
+    @property
+    def short_discriminator(self) -> Optional[int]:
+        if "Short discriminator" not in self.attributes:
+            return None
+
+        return int(self.attributes["Short discriminator"])
+
+    @property
+    def commissioning_flow(self) -> int:
+        return int(self.attributes.get("CommissioningFlow", "0"))
+
+    @property
+    def rendezvous_information(self) -> int:
+        return int(self.attributes.get("RendezvousInformation", "0"))
+
+    @property
+    def supports_ble_commissioning(self) -> bool:
+        return (self.rendezvous_information & 0b010) != 0
+
+    @property
+    def supports_on_network_commissioning(self) -> bool:
+        return (self.rendezvous_information & 0b100) != 0

src/python_testing/TC_CGEN_2_4.py

@@ -32,7 +32,7 @@ class TC_CGEN_2_4(MatterBaseTest):
     def OpenCommissioningWindow(self) -> int:
         try:
             pin, code = self.th1.OpenCommissioningWindow(
-                nodeid=self.dut_node_id, timeout=600, iteration=10000, discriminator=self.matter_test_config.discriminator[0], option=1)
+                nodeid=self.dut_node_id, timeout=600, iteration=10000, discriminator=self.matter_test_config.discriminators[0], option=1)
             time.sleep(5)
             return pin, code
 
@@ -51,7 +51,7 @@ async def CommissionToStageSendCompleteAndCleanup(
         self.th2.SetTestCommissionerPrematureCompleteAfter(stage)
         success, errcode = self.th2.CommissionOnNetwork(
             nodeId=self.dut_node_id, setupPinCode=pin,
-            filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=self.matter_test_config.discriminator[0])
+            filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=self.matter_test_config.discriminators[0])
         logging.info('Commissioning complete done. Successful? {}, errorcode = {}'.format(success, errcode))
         asserts.assert_false(success, 'Commissioning complete did not error as expected')
         asserts.assert_true(errcode.sdk_part == expectedErrorPart, 'Unexpected error type returned from CommissioningComplete')
@@ -91,7 +91,7 @@ async def test_TC_CGEN_2_4(self):
         self.th2.ResetTestCommissioner()
         success, errcode = self.th2.CommissionOnNetwork(
             nodeId=self.dut_node_id, setupPinCode=pin,
-            filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=self.matter_test_config.discriminator[0])
+            filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=self.matter_test_config.discriminators[0])
         logging.info('Commissioning complete done. Successful? {}, errorcode = {}'.format(success, errcode))
 
         logging.info('Step 17 - TH1 sends an arm failsafe')

src/python_testing/TC_DA_1_7.py

@@ -113,11 +113,13 @@ async def test_TC_DA_1_7(self):
         # On the CI, this doesn't make sense to do since all the examples use the same DAC
         # To specify more than 1 DUT, use a list of discriminators and passcodes
         allow_sdk_dac = self.user_params.get("allow_sdk_dac", False)
+        if allow_sdk_dac:
+            asserts.assert_equal(len(self.matter_test_config.discriminators), 1, "Only one device can be tested with SDK DAC")
         if not allow_sdk_dac:
-            asserts.assert_equal(len(self.matter_test_config.discriminator), 2, "This test requires 2 DUTs")
+            asserts.assert_equal(len(self.matter_test_config.discriminators), 2, "This test requires 2 DUTs")
         pk = []
-        for i in range(len(self.matter_test_config.dut_node_id)):
-            pk.append(await self.single_DUT(i, self.matter_test_config.dut_node_id[i]))
+        for i in range(len(self.matter_test_config.dut_node_ids)):
+            pk.append(await self.single_DUT(i, self.matter_test_config.dut_node_ids[i]))
 
         asserts.assert_equal(len(pk), len(set(pk)), "Found matching public keys in different DUTs")
 

src/python_testing/TC_RR_1_1.py

@@ -434,6 +434,8 @@ async def test_TC_RR_1_1(self):
                 node_id=self.dut_node_id,
                 endpoint=0,
                 attribute=Clusters.GroupKeyManagement.Attributes.MaxGroupsPerFabric)
+            logging.info(
+                f"MaxGroupsPerFabric value: {indicated_max_groups_per_fabric}, number of endpoints with Groups clusters cluster: {counted_groups_clusters}, which are: {list(groups_cluster_endpoints.keys())}")
             if indicated_max_groups_per_fabric < 4 * counted_groups_clusters:
                 asserts.fail("Failed Step 11: MaxGroupsPerFabric < 4 * counted_groups_clusters")