Step 2 progress

Author: raul-marquez-csa

src/python_testing/TC_IDM_4_3.py

@@ -27,7 +27,7 @@
 from chip.clusters.Attribute import AttributePath, TypedAttributePath
 from chip.exceptions import ChipStackError
 from chip.interaction_model import Status
-from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
+from matter_testing_support import AttributeChangeCallback, MatterBaseTest, TestStep, async_test_body, default_matter_test_main, EventChangeCallback
 from mobly import asserts
 
 '''
@@ -97,7 +97,6 @@ async def test_TC_IDM_4_3(self):
 
         # Test setup
         node_label_attr = Clusters.BasicInformation.Attributes.NodeLabel
-        # node_label_attr = Clusters.OnOff
         node_label_attr_path = [(0, node_label_attr)]
         TH: ChipDeviceController = self.default_controller
 
@@ -112,12 +111,17 @@ async def test_TC_IDM_4_3(self):
             reportInterval=(3, 5),
             keepSubscriptions=False
         )
-                
-
 
 
 
 
+        # secs = 60
+        # print(f"\n\n\n\n\nTime to sleep {secs} second(s)")
+        # time.sleep(secs)
+        # print(f"Rise and shine after {secs} second(s)\n\n\n\n\n")
+
+        
+        
 
 
 
@@ -135,20 +139,26 @@ async def test_TC_IDM_4_3(self):
         sub_th_step1a_min_interval_sec, sub_th_step1a_max_interval_sec = sub_th_step1a.GetReportingIntervalsSeconds()
         asserts.assert_is_not_none(sub_th_step1a_max_interval_sec, "MaxInterval field not present")
 
-        sub_th_step1a.Shutdown()
+        # sub_th_step1a.Shutdown()
 
         # *** Step 1b ***
         # Change the value of the attribute which has been subscribed on the DUT by manually changing some
         # settings on the device. Example: Temperature sensor may update the value of the room temperature.
         # Turning on/off on a light bulb.
         self.step("1b")
 
-        # # Modify attribute value
-        # new_node_label_write = "NewNodeLabel_11001100"
-        # await TH.WriteAttribute(
-        #     self.dut_node_id,
-        #     [(0, node_label_attr(value=new_node_label_write))]
-        # )
+        # Set Attribute Update Callback
+        node_label_update_cb = AttributeChangeCallback(node_label_attr)
+        sub_th_step1a.SetAttributeUpdateCallback(node_label_update_cb)
+    
+        # Modify attribute value
+        new_node_label_write = "NewNodeLabel_11001100"
+        await TH.WriteAttribute(
+            self.dut_node_id,
+            [(0, node_label_attr(value=new_node_label_write))]
+        )
+        
+        node_label_update_cb.wait_for_report()
 
 
 if __name__ == "__main__":

src/python_testing/matter_testing_support.py

@@ -54,7 +54,7 @@
 from chip import discovery
 from chip.ChipStack import ChipStack
 from chip.clusters import ClusterObjects as ClusterObjects
-from chip.clusters.Attribute import EventReadResult, SubscriptionTransaction
+from chip.clusters.Attribute import EventReadResult, SubscriptionTransaction, TypedAttributePath
 from chip.exceptions import ChipStackError
 from chip.interaction_model import InteractionModelError, Status
 from chip.setup_payload import SetupPayload
@@ -310,6 +310,35 @@ def wait_for_event_report(self, expected_event: ClusterObjects.ClusterEvent, tim
         asserts.assert_equal(res.Header.EventId, expected_event.event_id, "Expected event ID not found in event report")
         return res.Data
 
+class AttributeChangeCallback:
+    def __init__(self, expected_attribute: ClusterObjects.ClusterAttributeDescriptor):
+        self._output = queue.Queue()
+        self._expected_attribute = expected_attribute
+
+    def __call__(self, path: TypedAttributePath, transaction: SubscriptionTransaction):
+        """This is the subscription callback when an attribute is updated.
+           It checks the passed in attribute is the same as the subscribed to attribute and
+           then posts it into the queue for later processing."""
+           
+        asserts.assert_equal(path.AttributeType, self._expected_attribute,
+                             f"[AttributeChangeCallback] Attribute mismatch. Expected: {self._expected_attribute}, received: {path.AttributeType}")
+        logging.info(f"[AttributeChangeCallback] Attribute update callback for {path.AttributeType}")
+        q = (path, transaction)
+        self._output.put(q)
+
+    def wait_for_report(self):
+        try:
+            path, transaction = self._output.get(block=True, timeout=10)
+        except queue.Empty:
+            asserts.fail(f"[AttributeChangeCallback] Failed to receive a report for the {self._expected_attribute} attribute change")
+
+        asserts.assert_equal(path.AttributeType, self._expected_attribute,
+                             f"[AttributeChangeCallback] Received incorrect report. Expected: {self._expected_attribute}, received: {path.AttributeType}")
+        try:
+            attribute_value = transaction.GetAttribute(path)
+            logging.info(f"[AttributeChangeCallback] Got attribute subscription report. Attribute {path.AttributeType}. Updated value: {attribute_value}. SubscriptionId: {transaction.subscriptionId}")
+        except KeyError:
+            asserts.fail("[AttributeChangeCallback] Attribute {expected_attribute} not found in returned report")
 
 class InternalTestRunnerHooks(TestRunnerHooks):