Measuring time

Author: raul-marquez-csa

src/python_testing/TC_IDM_4_3.py

@@ -45,6 +45,23 @@
 
 class TC_IDM_4_3(MatterBaseTest):
 
+    # ANSI escape codes for background colors
+    BACKGROUND_COLORS = {
+        'black': '\033[40m',
+        'red': '\033[41m',
+        'green': '\033[42m',
+        'yellow': '\033[43m',
+        'blue': '\033[44m',
+        'magenta': '\033[45m',
+        'cyan': '\033[46m',
+        'white': '\033[47m',
+        'reset': '\033[0m'
+    }
+
+    # Function to print text with a specific background color
+    def fprint(self, text: str, background_color: str):
+        print(f"{self.BACKGROUND_COLORS.get(background_color, self.BACKGROUND_COLORS['reset'])}{text}{self.BACKGROUND_COLORS['reset']}")
+
     def steps_TC_IDM_4_3(self):
         return [TestStep("1a", "DUT and TH activate the subscription.",
                          "Verify on the TH, a report data message is received. Verify on the TH the Subscribe Response has the following fields: SubscriptionId and MaxInterval In the following Steps 2, 3, 5-10, 13, and 15, the MaxInterval time reference in each step is the MaxInterval presented in the Subscribe Response of the subscription."),
@@ -93,11 +110,45 @@ def steps_TC_IDM_4_3(self):
                 ]
 
     def on_notify_subscription_still_active(self):
-        print("NotifyLogic")
+
+        self.previous_report_data_time = self.current_report_data_time
+        self.current_report_data_time = time.time()
+
+        if self.previous_report_data_time > 0:
+            diff = self.current_report_data_time - self.previous_report_data_time
+        else:
+            diff = 0
+
+        self.fprint(f"NotifyLogic {time.time()} >>> t: {diff}", "red")
+
+    def wait_for_attribute_update_report(self, expected_attribute, output):
+        try:
+            path, transaction = output.get(block=True, timeout=10)
+        except queue.Empty:
+            asserts.fail(
+                f"[AttributeChangeCallback | Local] Failed to receive a report for the {expected_attribute} attribute change")
+
+        asserts.assert_equal(path.AttributeType, expected_attribute,
+                             f"[AttributeChangeCallback | Local] Received incorrect report. Expected: {expected_attribute}, received: {path.AttributeType}")
+        try:
+            attribute_value = transaction.GetAttribute(path)
+
+            self.attr_update_report_data_time = time.time()
+
+            logging.info(
+                f"[AttributeChangeCallback | Local] Got attribute subscription report. Attribute {path.AttributeType}. Updated value: {attribute_value}. SubscriptionId: {transaction.subscriptionId}")
+        except KeyError:
+            asserts.fail("[AttributeChangeCallback | Local] Attribute {expected_attribute} not found in returned report")
+
+    current_report_data_time: time = 0
+    previous_report_data_time: time = 0
+    attr_update_report_data_time: time = 0
+
+    min_interval_floor_sec: int = 2
+    max_interval_ceiling_sec: int = 3
 
     @async_test_body
     async def test_TC_IDM_4_3(self):
-        print("Hey there")
 
         # Test setup
         node_label_attr = Clusters.BasicInformation.Attributes.NodeLabel
@@ -109,44 +160,30 @@ async def test_TC_IDM_4_3(self):
         self.step("1a")
 
         # Subscribe to attribute
-        sub_th_step1a = await TH.ReadAttribute(
+        sub_th_step1ab = await TH.ReadAttribute(
             nodeid=self.dut_node_id,
             attributes=node_label_attr_path,
-            reportInterval=(3, 5),
+            reportInterval=(self.min_interval_floor_sec, self.max_interval_ceiling_sec),
             keepSubscriptions=False
         )
 
+        sub_th_step1ab.SetNotifySubscriptionStillActiveCallback(self.on_notify_subscription_still_active)
 
-
-
-
-
-
-        sub_th_step1a.SetNotifySubscriptionStillActiveCallback(self.on_notify_subscription_still_active)
-
-        secs = 60
+        secs = 45
         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")
 
-
-
-
-
-
-
         # Verify that the subscription is activated between TH and DUT
         # Verify on the TH, a report data message is received.
-        asserts.assert_true(sub_th_step1a.subscriptionId, "Subscription not activated")
+        asserts.assert_true(sub_th_step1ab.subscriptionId, "Subscription not activated")
 
         # Verify subscriptionId field is present
-        asserts.assert_is_not_none(sub_th_step1a.subscriptionId, "SubscriptionId field not present")
+        asserts.assert_is_not_none(sub_th_step1ab.subscriptionId, "SubscriptionId field not present")
 
         # Verify MaxInterval field is present
-        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_step1ab_min_interval_sec, sub_th_step1ab_max_interval_sec = sub_th_step1ab.GetReportingIntervalsSeconds()
+        asserts.assert_is_not_none(sub_th_step1ab_max_interval_sec, "MaxInterval field not present")
 
         # *** Step 1b ***
         # Change the value of the attribute which has been subscribed on the DUT by manually changing some
@@ -156,16 +193,29 @@ async def test_TC_IDM_4_3(self):
 
         # Set Attribute Update Callback
         node_label_update_cb = AttributeChangeCallback(node_label_attr)
-        sub_th_step1a.SetAttributeUpdateCallback(node_label_update_cb)
+        sub_th_step1ab.SetAttributeUpdateCallback(node_label_update_cb)
 
-        # Modify attribute value
+        # Update 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()
+        self.wait_for_attribute_update_report(node_label_attr, node_label_update_cb._output)
+
+        # Number of seconds elapsed between the last report data event
+        # and the arrival of the attribute update report data
+        elapsed_time_since_report = self.attr_update_report_data_time - self.previous_report_data_time
+
+        # Verify that the attribute update report data is sent
+        # after MinInterval time and before MaxInterval time
+        asserts.assert_greater(elapsed_time_since_report, self.min_interval_floor_sec,
+                               f"Attribute update report data must be sent after the MinInterval")
+        asserts.assert_less(elapsed_time_since_report, self.max_interval_ceiling_sec,
+                        f"Attribute update report data must be sent before the MaxInterval")
+
+        sub_th_step1ab.Shutdown()
 
 
 if __name__ == "__main__":