Clean up subscriptions from fabric sync tests (#35485)

Author: tehampson

src/python_testing/TC_BRBINFO_4_1.py

@@ -123,6 +123,7 @@ async def setup_class(self):
         self.set_of_dut_endpoints_before_adding_device = set(root_part_list)
 
         super().setup_class()
+        self._active_change_event_subscription = None
         self.app_process = None
         self.app_process_paused = False
         app = self.user_params.get("th_icd_server_app_path", None)
@@ -156,6 +157,10 @@ async def setup_class(self):
                                                          params.commissioningParameters.setupManualCode, params.commissioningParameters.setupQRCode)
 
     def teardown_class(self):
+        if self._active_change_event_subscription is not None:
+            self._active_change_event_subscription.Shutdown()
+            self._active_change_event_subscription = None
+
         # In case the th_icd_server_app_path does not exist, then we failed the test
         # and there is nothing to remove
         if self.app_process is not None:
@@ -239,8 +244,8 @@ async def test_TC_BRBINFO_4_1(self):
         self.q = queue.Queue()
         urgent = 1
         cb = SimpleEventCallback("ActiveChanged", event.cluster_id, event.event_id, self.q)
-        subscription = await self.default_controller.ReadEvent(nodeid=self.dut_node_id, events=[(dynamic_endpoint_id, event, urgent)], reportInterval=[1, 3])
-        subscription.SetEventUpdateCallback(callback=cb)
+        self._active_change_event_subscription = await self.default_controller.ReadEvent(nodeid=self.dut_node_id, events=[(dynamic_endpoint_id, event, urgent)], reportInterval=[1, 3])
+        self._active_change_event_subscription.SetEventUpdateCallback(callback=cb)
 
         self.step("3")
         stay_active_duration_ms = 1000

src/python_testing/TC_MCORE_FS_1_2.py

@@ -63,10 +63,15 @@ class TC_MCORE_FS_1_2(MatterBaseTest):
     @async_test_body
     async def setup_class(self):
         super().setup_class()
+        self._partslist_subscription = None
         self._app_th_server_process = None
         self._th_server_kvs = None
 
     def teardown_class(self):
+        if self._partslist_subscription is not None:
+            self._partslist_subscription.Shutdown()
+            self._partslist_subscription = None
+
         if self._app_th_server_process is not None:
             logging.warning("Stopping app with SIGTERM")
             self._app_th_server_process.send_signal(signal.SIGTERM.value)
@@ -142,7 +147,7 @@ async def test_TC_MCORE_FS_1_2(self):
         subscription_contents = [
             (root_endpoint, Clusters.Descriptor.Attributes.PartsList)
         ]
-        sub = await self.default_controller.ReadAttribute(
+        self._partslist_subscription = await self.default_controller.ReadAttribute(
             nodeid=self.dut_node_id,
             attributes=subscription_contents,
             reportInterval=(min_report_interval_sec, max_report_interval_sec),
@@ -152,8 +157,8 @@ async def test_TC_MCORE_FS_1_2(self):
         parts_list_queue = queue.Queue()
         attribute_handler = AttributeChangeAccumulator(
             name=self.default_controller.name, expected_attribute=Clusters.Descriptor.Attributes.PartsList, output=parts_list_queue)
-        sub.SetAttributeUpdateCallback(attribute_handler)
-        cached_attributes = sub.GetAttributes()
+        self._partslist_subscription.SetAttributeUpdateCallback(attribute_handler)
+        cached_attributes = self._partslist_subscription.GetAttributes()
         step_1_dut_parts_list = cached_attributes[root_endpoint][Clusters.Descriptor][Clusters.Descriptor.Attributes.PartsList]
 
         asserts.assert_true(type_matches(step_1_dut_parts_list, list), "PartsList is expected to be a list")

src/python_testing/TC_MCORE_FS_1_5.py

@@ -61,10 +61,20 @@ class TC_MCORE_FS_1_5(MatterBaseTest):
     @async_test_body
     async def setup_class(self):
         super().setup_class()
+        self._partslist_subscription = None
+        self._cadmin_subscription = None
         self._app_th_server_process = None
         self._th_server_kvs = None
 
     def teardown_class(self):
+        if self._partslist_subscription is not None:
+            self._partslist_subscription.Shutdown()
+            self._partslist_subscription = None
+
+        if self._cadmin_subscription is not None:
+            self._cadmin_subscription.Shutdown()
+            self._cadmin_subscription = None
+
         if self._app_th_server_process is not None:
             logging.warning("Stopping app with SIGTERM")
             self._app_th_server_process.send_signal(signal.SIGTERM.value)
@@ -142,7 +152,7 @@ async def test_TC_MCORE_FS_1_5(self):
         parts_list_subscription_contents = [
             (root_endpoint, Clusters.Descriptor.Attributes.PartsList)
         ]
-        parts_list_sub = await self.default_controller.ReadAttribute(
+        self._partslist_subscription = await self.default_controller.ReadAttribute(
             nodeid=self.dut_node_id,
             attributes=parts_list_subscription_contents,
             reportInterval=(min_report_interval_sec, max_report_interval_sec),
@@ -152,8 +162,8 @@ async def test_TC_MCORE_FS_1_5(self):
         parts_list_queue = queue.Queue()
         parts_list_attribute_handler = AttributeChangeAccumulator(
             name=self.default_controller.name, expected_attribute=Clusters.Descriptor.Attributes.PartsList, output=parts_list_queue)
-        parts_list_sub.SetAttributeUpdateCallback(parts_list_attribute_handler)
-        parts_list_cached_attributes = parts_list_sub.GetAttributes()
+        self._partslist_subscription.SetAttributeUpdateCallback(parts_list_attribute_handler)
+        parts_list_cached_attributes = self._partslist_subscription.GetAttributes()
         step_1_dut_parts_list = parts_list_cached_attributes[root_endpoint][Clusters.Descriptor][Clusters.Descriptor.Attributes.PartsList]
 
         asserts.assert_true(type_matches(step_1_dut_parts_list, list), "PartsList is expected to be a list")
@@ -219,7 +229,7 @@ async def test_TC_MCORE_FS_1_5(self):
         cadmin_subscription_contents = [
             (newly_added_endpoint, Clusters.AdministratorCommissioning)
         ]
-        cadmin_sub = await self.default_controller.ReadAttribute(
+        self._cadmin_subscription = await self.default_controller.ReadAttribute(
             nodeid=self.dut_node_id,
             attributes=cadmin_subscription_contents,
             reportInterval=(min_report_interval_sec, max_report_interval_sec),
@@ -230,7 +240,7 @@ async def test_TC_MCORE_FS_1_5(self):
         # This AttributeChangeAccumulator is really just to let us know when new subscription came in
         cadmin_attribute_handler = AttributeChangeAccumulator(
             name=self.default_controller.name, expected_attribute=Clusters.AdministratorCommissioning.Attributes.WindowStatus, output=cadmin_queue)
-        cadmin_sub.SetAttributeUpdateCallback(cadmin_attribute_handler)
+        self._cadmin_subscription.SetAttributeUpdateCallback(cadmin_attribute_handler)
         time.sleep(1)
 
         self.step(7)