diff --git a/src/python_testing/TC_IDM_4_3.py b/src/python_testing/TC_IDM_4_3.py index 3bd8595c0b0105..3c72a4a07b8e1a 100644 --- a/src/python_testing/TC_IDM_4_3.py +++ b/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__":