check response instead of reading on endpoint 0

src/python_testing/TC_CNET_1_4.py

@@ -33,12 +33,15 @@
 from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
 from mobly import asserts
 
+kRootEndpointId = 0
+kSecondaryNetworkInterfaceDeviceTypeId = 0x0019
+
 
 class TC_CNET_1_4(MatterBaseTest):
     def steps_TC_CNET_1_4(self):
         return [TestStep(1, "TH is commissioned", is_commissioning=True),
                 TestStep(2, 'TH performs a wildcard read of Network Commissioning clusters across all endpoints, and save the number of Network Commissioning clusters as `NumNetworkCommissioning` for future use. If `NumNetworkCommissioning` is 0, skip the remaining steps in this test case'),
-                TestStep(3, 'TH reads from the DUT the Descriptor Cluster DeviceTypeList attribute from Endpoint 0, verify that the Network Commissioning cluster id (0x0031) is listed in the ServerList'),
+                TestStep(3, 'TH checks whether endpoint 0 is contained in the response of the previous response to determine whether there is a Network Commissioning cluster on endpoint 0'),
                 TestStep(4, 'TH reads from the DUT the Descriptor Cluster DeviceTypeList attribute on each endpoint (except for Endpoint 0) that hosts a Network Commissioning cluster, verify that the Secondary Network Interface device type id (0x0019) is listed in the DeviceTypeList'),
                 TestStep(5, 'TH reads from the DUT the General Commissioning Cluster SupportsConcurrentConnection attribute if NumNetworkCommissioning is greater than 1, verify that it is true')]
 
@@ -61,17 +64,17 @@ async def test_TC_CNET_1_4(self):
         self.step(2)
         # Read FeatureMap attribute with wildcard endpoint
         feature_map_results = await self.default_controller.ReadAttribute(self.dut_node_id, [(Clusters.NetworkCommissioning.Attributes.FeatureMap)], fabricFiltered=True)
-
         NumNetworkCommissioning = len(feature_map_results)
         if NumNetworkCommissioning == 0:
             logging.info('No endpoint has Network Commissioning Cluster, skipping remaining steps')
             self.skip_all_remaining_steps(3)
             return
 
         self.step(3)
-        cdesc = Clusters.Descriptor
-        server_list = await self.read_single_attribute_check_success(cluster=cdesc, attribute=cdesc.Attributes.ServerList, endpoint=0)
-        if 49 not in server_list:
+        endpoints = []
+        for endpoint, _ in feature_map_results.items():
+            endpoints.append(endpoint)
+        if kRootEndpointId not in endpoints:
             asserts.assert_true(False, "There is no Network Commissioning Cluster on endpoint 0")
 
         if NumNetworkCommissioning == 1:
@@ -80,20 +83,19 @@ async def test_TC_CNET_1_4(self):
             return
 
         self.step(4)
-        adevt = cdesc.Attributes.DeviceTypeList
-        for endpoint, _ in feature_map_results.items():
-            if endpoint == 0:
+        for endpoint in endpoints:
+            if endpoint == kRootEndpointId:
                 continue
-            device_type_list = await self.read_single_attribute_check_success(cluster=Clusters.Descriptor, attribute=adevt, endpoint=endpoint)
-            required_device_types = {25}
+            device_type_list = await self.read_single_attribute_check_success(cluster=Clusters.Descriptor,
+                                                                              attribute=Clusters.Descriptor.Attributes.DeviceTypeList, endpoint=endpoint)
+            required_device_types = { kSecondaryNetworkInterfaceDeviceTypeId }
             found_device_types = {device.deviceType for device in device_type_list}
             asserts.assert_true(required_device_types.intersection(found_device_types),
                                 "Network Commissioning Cluster is not on Root Node or Secondary Network Interface")
 
         self.step(5)
-        cgen = Clusters.GeneralCommissioning
-        ascc = cgen.Attributes.SupportsConcurrentConnection
-        concurrent_connection = await self.read_single_attribute_check_success(cluster=cgen, attribute=ascc)
+        concurrent_connection = await self.read_single_attribute_check_success(cluster=Clusters.GeneralCommissioning,
+                                                                               attribute=Clusters.GeneralCommissioning.Attributes.SupportsConcurrentConnection)
         asserts.assert_true(concurrent_connection, "The device does not support concurrent connection commissioning")