[Implementation] Adding get_endpoint functionality to matter_testing support module (#36568)

* Incorporating new get_endpoint functionality in matter_testing support module:
- Added prior previous get_endpoint functionality from Cecille back to matter_testing support module.
- Updating test modules to use new get_endpoint functionality mentioned in issue: project-chip/matter-test-scripts#416

* Restyled by autopep8

* Updated TC_OCC_3_2 to resolve linting errors

---------

Co-authored-by: Restyled.io <commits@restyled.io>

Author: j-ororke

src/python_testing/TC_BOOLCFG_2_1.py

@@ -72,7 +72,7 @@ def pics_TC_BOOLCFG_2_1(self) -> list[str]:
     @async_test_body
     async def test_TC_BOOLCFG_2_1(self):
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
         all_alarm_mode_bitmap_bits = functools.reduce(
             ior, [b.value for b in Clusters.BooleanStateConfiguration.Bitmaps.AlarmModeBitmap])
         all_sensor_fault_bitmap_bits = functools.reduce(

src/python_testing/TC_BOOLCFG_3_1.py

@@ -75,7 +75,7 @@ def pics_TC_BOOLCFG_3_1(self) -> list[str]:
     @async_test_body
     async def test_TC_BOOLCFG_3_1(self):
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
 
         self.step(1)
         attributes = Clusters.BooleanStateConfiguration.Attributes

src/python_testing/TC_BOOLCFG_4_1.py

@@ -67,7 +67,7 @@ def pics_TC_BOOLCFG_4_1(self) -> list[str]:
     @async_test_body
     async def test_TC_BOOLCFG_4_1(self):
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
 
         self.step(1)
         attributes = Clusters.BooleanStateConfiguration.Attributes

src/python_testing/TC_BOOLCFG_4_2.py

@@ -89,7 +89,7 @@ async def test_TC_BOOLCFG_4_2(self):
                             "the --hex-arg flag as PIXIT.BOOLCFG.TEST_EVENT_TRIGGER_KEY:<key>, "
                             "e.g. --hex-arg PIXIT.BOOLCFG.TEST_EVENT_TRIGGER_KEY:000102030405060708090a0b0c0d0e0f")
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
         enableKey = self.matter_test_config.global_test_params['PIXIT.BOOLCFG.TEST_EVENT_TRIGGER_KEY']
 
         self.step(1)

src/python_testing/TC_BOOLCFG_4_3.py

@@ -101,7 +101,7 @@ async def test_TC_BOOLCFG_4_3(self):
                             "the --hex-arg flag as PIXIT.BOOLCFG.TEST_EVENT_TRIGGER_KEY:<key>, "
                             "e.g. --hex-arg PIXIT.BOOLCFG.TEST_EVENT_TRIGGER_KEY:000102030405060708090a0b0c0d0e0f")
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
         enableKey = self.matter_test_config.global_test_params['PIXIT.BOOLCFG.TEST_EVENT_TRIGGER_KEY']
 
         self.step(1)

src/python_testing/TC_BOOLCFG_4_4.py

@@ -93,7 +93,7 @@ async def test_TC_BOOLCFG_4_4(self):
                             "the --hex-arg flag as PIXIT.BOOLCFG.TEST_EVENT_TRIGGER_KEY:<key>, "
                             "e.g. --hex-arg PIXIT.BOOLCFG.TEST_EVENT_TRIGGER_KEY:000102030405060708090a0b0c0d0e0f")
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
         enableKey = self.matter_test_config.global_test_params['PIXIT.BOOLCFG.TEST_EVENT_TRIGGER_KEY']
 
         self.step(1)

src/python_testing/TC_BOOLCFG_5_1.py

@@ -88,7 +88,7 @@ async def test_TC_BOOLCFG_5_1(self):
                             "the --hex-arg flag as PIXIT.BOOLCFG.TEST_EVENT_TRIGGER_KEY:<key>, "
                             "e.g. --hex-arg PIXIT.BOOLCFG.TEST_EVENT_TRIGGER_KEY:000102030405060708090a0b0c0d0e0f")
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
         enableKey = self.matter_test_config.global_test_params['PIXIT.BOOLCFG.TEST_EVENT_TRIGGER_KEY']
 
         self.step(1)

src/python_testing/TC_BOOLCFG_5_2.py

@@ -87,7 +87,7 @@ async def test_TC_BOOLCFG_5_2(self):
                             "the --hex-arg flag as PIXIT.BOOLCFG.TEST_EVENT_TRIGGER_KEY:<key>, "
                             "e.g. --hex-arg PIXIT.BOOLCFG.TEST_EVENT_TRIGGER_KEY:000102030405060708090a0b0c0d0e0f")
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
         enableKey = self.matter_test_config.global_test_params['PIXIT.BOOLCFG.TEST_EVENT_TRIGGER_KEY']
 
         self.step(1)

src/python_testing/TC_CADMIN_1_11.py

@@ -28,7 +28,7 @@
 #       --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
 #       --PICS src/app/tests/suites/certification/ci-pics-values
 #     factory-reset: true
-#     quiet: false
+#     quiet: true
 # === END CI TEST ARGUMENTS ===
 
 

src/python_testing/TC_DEM_2_10.py

@@ -162,7 +162,7 @@ async def test_TC_DEM_2_10(self):
         self.step("4")
         sub_handler = ClusterAttributeChangeAccumulator(Clusters.DeviceEnergyManagement)
         await sub_handler.start(self.default_controller, self.dut_node_id,
-                                self.matter_test_config.endpoint,
+                                self.get_endpoint(),
                                 min_interval_sec=0,
                                 max_interval_sec=10, keepSubscriptions=False)
 

src/python_testing/TC_DEM_2_2.py

@@ -181,7 +181,7 @@ async def test_TC_DEM_2_2(self):
         events_callback = EventChangeCallback(Clusters.DeviceEnergyManagement)
         await events_callback.start(self.default_controller,
                                     self.dut_node_id,
-                                    self.matter_test_config.endpoint)
+                                    self.get_endpoint(default=1))
 
         self.step("4")
         await self.check_test_event_triggers_enabled()

src/python_testing/TC_DEM_2_4.py

@@ -186,7 +186,7 @@ async def test_TC_DEM_2_4(self):
         events_callback = EventChangeCallback(Clusters.DeviceEnergyManagement)
         await events_callback.start(self.default_controller,
                                     self.dut_node_id,
-                                    self.matter_test_config.endpoint)
+                                    self.get_endpoint())
 
         self.step("4")
         await self.check_test_event_triggers_enabled()

src/python_testing/TC_DGSW_2_1.py

@@ -74,7 +74,7 @@ def steps_TC_DGSW_2_1(self) -> list[TestStep]:
     @async_test_body
     async def test_TC_DGSW_2_1(self):
 
-        endpoint = self.user_params.get("endpoint", 0)
+        endpoint = self.get_endpoint(default=0)
 
         # STEP 1: Commission DUT (already done)
         self.step(1)

src/python_testing/TC_DRLK_2_13.py

@@ -368,7 +368,7 @@ async def test_TC_DRLK_2_13(self):
         self.groupIdentifier = bytes.fromhex("89d085fc302ca53e279bfcdecdf3c4ad")
         self.groupResolvingKey = bytes.fromhex("89d0859bfcdecdf3c4adfc302ca53e27")
         self.common_cluster_endpoint = 0
-        self.app_cluster_endpoint = self.matter_test_config.endpoint
+        self.app_cluster_endpoint = self.get_endpoint(default=1)
         self.alirouser = "AliroUser"
         self.alirocredentialissuerkey = bytes.fromhex(
             "047a4c882d753924cdf3779a3c84fec2debaa6f0b3084450878acc7ddcce7856ae57b1ebbe2561015103dd7474c2a183675378ec55f1e465ac3436bf3dd5ca54d4")

src/python_testing/TC_DRLK_2_9.py

@@ -379,7 +379,7 @@ async def test_TC_DRLK_2_9(self):
         self.maxrfidcodelength = None
         self.minrfidcodelength = None
 
-        self.endpoint = self.user_params.get("endpoint", 1)
+        self.endpoint = self.get_endpoint(default=1)
         print("endpoint", self.endpoint)
 
         # Aliro Keys for setting Aliro configuration and credential

src/python_testing/TC_EEVSE_2_2.py

@@ -172,7 +172,7 @@ async def test_TC_EEVSE_2_2(self):
         events_callback = EventChangeCallback(Clusters.EnergyEvse)
         await events_callback.start(self.default_controller,
                                     self.dut_node_id,
-                                    self.matter_test_config.endpoint)
+                                    self.get_endpoint())
 
         self.step("2")
         await self.check_test_event_triggers_enabled()

src/python_testing/TC_EEVSE_2_3.py

@@ -211,7 +211,7 @@ async def test_TC_EEVSE_2_3(self):
         events_callback = EventChangeCallback(Clusters.EnergyEvse)
         await events_callback.start(self.default_controller,
                                     self.dut_node_id,
-                                    self.matter_test_config.endpoint)
+                                    self.get_endpoint())
 
         self.step("2")
         await self.check_test_event_triggers_enabled()

src/python_testing/TC_EEVSE_2_4.py

@@ -127,7 +127,7 @@ async def test_TC_EEVSE_2_4(self):
         events_callback = EventChangeCallback(Clusters.EnergyEvse)
         await events_callback.start(self.default_controller,
                                     self.dut_node_id,
-                                    self.matter_test_config.endpoint)
+                                    self.get_endpoint())
 
         self.step("2")
         await self.check_test_event_triggers_enabled()

src/python_testing/TC_EEVSE_2_5.py

@@ -119,7 +119,7 @@ async def test_TC_EEVSE_2_5(self):
         events_callback = EventChangeCallback(Clusters.EnergyEvse)
         await events_callback.start(self.default_controller,
                                     self.dut_node_id,
-                                    self.matter_test_config.endpoint)
+                                    self.get_endpoint())
 
         self.step("2")
         await self.check_test_event_triggers_enabled()

src/python_testing/TC_EEVSE_2_6.py

@@ -149,7 +149,7 @@ async def test_TC_EEVSE_2_6(self):
         events_callback = EventChangeCallback(Clusters.EnergyEvse)
         await events_callback.start(self.default_controller,
                                     self.dut_node_id,
-                                    self.matter_test_config.endpoint)
+                                    self.get_endpoint())
 
         self.step("4")
         await self.check_test_event_triggers_enabled()

src/python_testing/TC_EEVSE_Utils.py

@@ -59,7 +59,7 @@ async def get_supported_energy_evse_attributes(self, endpoint: int = None):
 
     async def write_user_max_charge(self, endpoint: int = None, user_max_charge: int = 0):
         if endpoint is None:
-            endpoint = self.matter_test_config.endpoint
+            endpoint = self.get_endpoint()
         result = await self.default_controller.WriteAttribute(self.dut_node_id,
                                                               [(endpoint,
                                                                Clusters.EnergyEvse.Attributes.UserMaximumChargeCurrent(user_max_charge))])

src/python_testing/TC_EWATERHTR_2_2.py

@@ -213,7 +213,7 @@ async def test_TC_EWATERHTR_2_2(self):
         events_callback = EventChangeCallback(Clusters.WaterHeaterManagement)
         await events_callback.start(self.default_controller,
                                     self.dut_node_id,
-                                    self.matter_test_config.endpoint)
+                                    self.get_endpoint())
 
         self.step("3")
         await self.check_test_event_triggers_enabled()

src/python_testing/TC_EWATERHTR_2_3.py

@@ -166,7 +166,7 @@ async def test_TC_EWATERHTR_2_3(self):
         events_callback = EventChangeCallback(Clusters.WaterHeaterManagement)
         await events_callback.start(self.default_controller,
                                     self.dut_node_id,
-                                    self.matter_test_config.endpoint)
+                                    self.get_endpoint())
 
         self.step("3")
         await self.check_test_event_triggers_enabled()

src/python_testing/TC_FAN_3_1.py

@@ -73,7 +73,7 @@ def pics_TC_FAN_3_1(self) -> list[str]:
 
     @async_test_body
     async def test_TC_FAN_3_1(self):
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
 
         self.print_step(1, "Commissioning, already done")
 

src/python_testing/TC_FAN_3_2.py

@@ -72,8 +72,7 @@ async def test_TC_FAN_3_2(self):
             logger.info("Test skipped because PICS FAN.S.F00 is not set")
             return
 
-        endpoint = self.user_params.get("endpoint", 1)
-
+        endpoint = self.get_endpoint(default=1)
         self.print_step(1, "Commissioning, already done")
 
         self.print_step(2, "Read from the DUT the SpeedSetting attribute and store")

src/python_testing/TC_FAN_3_3.py

@@ -94,7 +94,7 @@ async def test_TC_FAN_3_3(self):
             self.skip_all_remaining_steps(1)
             return
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
 
         self.step(1)
 

src/python_testing/TC_FAN_3_4.py

@@ -94,7 +94,7 @@ async def test_TC_FAN_3_4(self):
             logger.info("Test skipped because PICS FAN.S.F03 is not set")
             return
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
 
         self.step(1)
 

src/python_testing/TC_FAN_3_5.py

@@ -86,7 +86,7 @@ async def test_TC_FAN_3_5(self):
             logger.info("Test skipped because PICS FAN.S.F04 is not set")
             return
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
 
         # Part 1
 

src/python_testing/TC_LVL_2_3.py

@@ -131,7 +131,7 @@ async def test_TC_LVL_2_3(self):
 
         self.step(6)
         sub_handler = ClusterAttributeChangeAccumulator(lvl)
-        await sub_handler.start(self.default_controller, self.dut_node_id, self.matter_test_config.endpoint)
+        await sub_handler.start(self.default_controller, self.dut_node_id, self.get_endpoint())
 
         self.step(7)
         cmd = lvl.Commands.MoveToLevelWithOnOff(level=max_level, transitionTime=100, optionsMask=0, optionsOverride=0)

src/python_testing/TC_MWOCTRL_2_1.py

@@ -107,7 +107,7 @@ def pics_TC_MWOCTRL_2_1(self) -> list[str]:
     @async_test_body
     async def test_TC_MWOCTRL_2_1(self):
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
 
         self.step(1)
         attributes = Clusters.MicrowaveOvenControl.Attributes

src/python_testing/TC_MWOCTRL_2_2.py

@@ -106,7 +106,7 @@ def pics_TC_MWOCTRL_2_2(self) -> list[str]:
     @async_test_body
     async def test_TC_MWOCTRL_2_2(self):
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
 
         self.step(1)
         attributes = Clusters.MicrowaveOvenControl.Attributes

src/python_testing/TC_MWOCTRL_2_4.py

@@ -75,7 +75,7 @@ def pics_TC_MWOCTRL_2_4(self) -> list[str]:
     @async_test_body
     async def test_TC_MWOCTRL_2_4(self):
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
 
         self.step(1)
         attributes = Clusters.MicrowaveOvenControl.Attributes

src/python_testing/TC_MWOM_1_2.py

@@ -67,7 +67,7 @@ def pics_TC_MWOM_1_2(self) -> list[str]:
     @async_test_body
     async def test_TC_MWOM_1_2(self):
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
 
         attributes = Clusters.MicrowaveOvenMode.Attributes
 

src/python_testing/TC_OCC_2_1.py

@@ -83,7 +83,7 @@ def pics_TC_OCC_2_1(self) -> list[str]:
 
     @async_test_body
     async def test_TC_OCC_2_1(self):
-        endpoint = self.matter_test_config.endpoint
+        endpoint = self.get_endpoint()
         cluster = Clusters.Objects.OccupancySensing
         attributes = cluster.Attributes
 

src/python_testing/TC_OCC_2_2.py

@@ -65,7 +65,7 @@ def pics_TC_OCC_2_2(self) -> list[str]:
 
     @async_test_body
     async def test_TC_OCC_2_2(self):
-        endpoint = self.matter_test_config.endpoint
+        endpoint = self.get_endpoint()
 
         self.step(1)  # Already done, immediately go to step 2
 

src/python_testing/TC_OCC_2_3.py

@@ -46,7 +46,7 @@
 class TC_OCC_2_3(MatterBaseTest):
     async def read_occ_attribute_expect_success(self, attribute):
         cluster = Clusters.Objects.OccupancySensing
-        endpoint_id = self.matter_test_config.endpoint
+        endpoint_id = self.get_endpoint()
         return await self.read_single_attribute_check_success(endpoint=endpoint_id, cluster=cluster, attribute=attribute)
 
     def desc_TC_OCC_2_3(self) -> str:

src/python_testing/TC_OCC_3_1.py

@@ -55,13 +55,13 @@ def setup_test(self):
 
     async def read_occ_attribute_expect_success(self, attribute):
         cluster = Clusters.Objects.OccupancySensing
-        endpoint = self.matter_test_config.endpoint
+        endpoint = self.get_endpoint()
         return await self.read_single_attribute_check_success(endpoint=endpoint, cluster=cluster, attribute=attribute)
 
     async def write_hold_time(self, hold_time: Optional[Any]) -> Status:
         dev_ctrl = self.default_controller
         node_id = self.dut_node_id
-        endpoint = self.matter_test_config.endpoint
+        endpoint = self.get_endpoint()
 
         cluster = Clusters.OccupancySensing
         write_result = await dev_ctrl.WriteAttribute(node_id, [(endpoint, cluster.Attributes.HoldTime(hold_time))])
@@ -144,7 +144,7 @@ async def test_TC_OCC_3_1(self):
 
         self.step(4)
         # Setup Occupancy attribute subscription here
-        endpoint_id = self.matter_test_config.endpoint
+        endpoint_id = self.get_endpoint()
         node_id = self.dut_node_id
         dev_ctrl = self.default_controller
         attrib_listener = ClusterAttributeChangeAccumulator(cluster)

src/python_testing/TC_OCC_3_2.py

@@ -56,7 +56,7 @@ def setup_test(self):
 
     async def read_occ_attribute_expect_success(self, attribute):
         cluster = Clusters.Objects.OccupancySensing
-        endpoint_id = self.matter_test_config.endpoint
+        endpoint_id = self.get_endpoint()
         return await self.read_single_attribute_check_success(endpoint=endpoint_id, cluster=cluster, attribute=attribute)
 
     def desc_TC_OCC_3_2(self) -> str:
@@ -100,7 +100,7 @@ def write_to_app_pipe(self, command):
 
     @async_test_body
     async def test_TC_OCC_3_2(self):
-        endpoint_id = self.matter_test_config.endpoint
+        endpoint_id = self.get_endpoint()
         node_id = self.dut_node_id
         dev_ctrl = self.default_controller
 

src/python_testing/TC_OPSTATE_2_1.py

@@ -60,7 +60,7 @@ def pics_TC_OPSTATE_2_1(self) -> list[str]:
 
     @async_test_body
     async def test_TC_OPSTATE_2_1(self):
-        endpoint = self.matter_test_config.endpoint
+        endpoint = self.get_endpoint(default=1)
         await self.TEST_TC_OPSTATE_BASE_2_1(endpoint)
 
 

src/python_testing/TC_OPSTATE_2_2.py

@@ -62,7 +62,7 @@ def pics_TC_OPSTATE_2_2(self) -> list[str]:
 
     @async_test_body
     async def test_TC_OPSTATE_2_2(self):
-        endpoint = self.matter_test_config.endpoint
+        endpoint = self.get_endpoint(default=1)
 
         await self.TEST_TC_OPSTATE_BASE_2_2(endpoint=endpoint)
 

src/python_testing/TC_OPSTATE_2_3.py

@@ -62,7 +62,7 @@ def pics_TC_OPSTATE_2_3(self) -> list[str]:
 
     @async_test_body
     async def test_TC_OPSTATE_2_3(self):
-        endpoint = self.matter_test_config.endpoint
+        endpoint = self.get_endpoint(default=1)
 
         await self.TEST_TC_OPSTATE_BASE_2_3(endpoint=endpoint)