Python: consolidate endpoint selection

Add a function to get the endpoint from the config with an explicit
default. This makes it so folks don't have to remember where
that parameter comes in.

TL;DR - all tests should now use --endpoint to set endpoint

Author: cecille

src/python_testing/TC_BOOLCFG_2_1.py

@@ -56,7 +56,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

@@ -58,7 +58,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_2.py

@@ -68,7 +68,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_EEVSE_2_2.py

@@ -101,7 +101,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_4.py

@@ -77,7 +77,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

@@ -71,7 +71,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_Utils.py

@@ -72,7 +72,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_FAN_3_4.py

@@ -51,7 +51,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.print_step(1, "Commissioning, already done")
 

src/python_testing/TC_RVCCLEANM_1_2.py

@@ -48,7 +48,7 @@ def pics_TC_RVCCLEANM_1_2(self) -> list[str]:
 
     @async_test_body
     async def test_TC_RVCCLEANM_1_2(self):
-        self.endpoint = self.matter_test_config.endpoint
+        self.endpoint = self.get_endpoint()
 
         attributes = Clusters.RvcCleanMode.Attributes
 

src/python_testing/TC_RVCCLEANM_2_1.py

@@ -57,7 +57,7 @@ async def test_TC_RVCCLEANM_2_1(self):
                             "PIXIT.RVCCLEANM.MODE_CHANGE_FAIL must be included on the command line in "
                             "the --int-arg flag as PIXIT.RVCCLEANM.MODE_CHANGE_FAIL:<mode id>")
 
-        self.endpoint = self.matter_test_config.endpoint
+        self.endpoint = self.get_endpoint()
         self.mode_ok = self.matter_test_config.global_test_params['PIXIT.RVCCLEANM.MODE_CHANGE_OK']
         self.mode_fail = self.matter_test_config.global_test_params['PIXIT.RVCCLEANM.MODE_CHANGE_FAIL']
 

src/python_testing/TC_RVCCLEANM_2_2.py

@@ -62,7 +62,7 @@ def pics_TC_RVCCLEANM_2_2(self) -> list[str]:
 
     @async_test_body
     async def test_TC_RVCCLEANM_2_2(self):
-        self.endpoint = self.matter_test_config.endpoint
+        self.endpoint = self.get_endpoint()
 
         asserts.assert_true(self.check_pics("RVCCLEANM.S"), "RVCCLEANM.S must be supported")
         asserts.assert_true(self.check_pics("RVCRUNM.S.A0000"), "RVCRUNM.S.A0000 must be supported")

src/python_testing/TC_RVCOPSTATE_2_1.py

@@ -53,7 +53,7 @@ def TC_RVCOPSTATE_2_1(self) -> list[str]:
 
     @async_test_body
     async def test_TC_RVCOPSTATE_2_1(self):
-        self.endpoint = self.matter_test_config.endpoint
+        self.endpoint = self.get_endpoint(default=None)
         asserts.assert_false(self.endpoint is None, "--endpoint <endpoint> must be included on the command line in.")
 
         attributes = Clusters.RvcOperationalState.Attributes

src/python_testing/TC_RVCOPSTATE_2_3.py

@@ -130,7 +130,7 @@ def print_instruction(self, step_number, instruction):
     @async_test_body
     async def test_TC_RVCOPSTATE_2_3(self):
 
-        self.endpoint = self.matter_test_config.endpoint
+        self.endpoint = self.get_endpoint(default=None)
         asserts.assert_false(self.endpoint is None, "--endpoint <endpoint> must be included on the command line in.")
 
         asserts.assert_true(self.check_pics("RVCOPSTATE.S.A0003"), "RVCOPSTATE.S.A0003 must be supported")

src/python_testing/TC_RVCRUNM_1_2.py

@@ -48,7 +48,7 @@ def pics_TC_RVCRUNM_1_2(self) -> list[str]:
 
     @async_test_body
     async def test_TC_RVCRUNM_1_2(self):
-        self.endpoint = self.matter_test_config.endpoint
+        self.endpoint = self.get_endpoint()
 
         attributes = Clusters.RvcRunMode.Attributes
 

src/python_testing/TC_RVCRUNM_2_1.py

@@ -57,7 +57,7 @@ async def test_TC_RVCRUNM_2_1(self):
                             "PIXIT.RVCRUNM.MODE_CHANGE_FAIL must be included on the command line in "
                             "the --int-arg flag as PIXIT.RVCRUNM.MODE_CHANGE_FAIL:<mode id>")
 
-        self.endpoint = self.matter_test_config.endpoint
+        self.endpoint = self.get_endpoint()
         self.mode_ok = self.matter_test_config.global_test_params['PIXIT.RVCRUNM.MODE_CHANGE_OK']
         self.mode_fail = self.matter_test_config.global_test_params['PIXIT.RVCRUNM.MODE_CHANGE_FAIL']
 

src/python_testing/TC_RVCRUNM_2_2.py

@@ -97,7 +97,7 @@ async def test_TC_RVCRUNM_2_2(self):
                          "PIXIT.RVCRUNM.MODE_A:<mode id> \n"
                          "PIXIT.RVCRUNM.MODE_B:<mode id>")
 
-        self.endpoint = self.matter_test_config.endpoint
+        self.endpoint = self.get_endpoint()
         self.mode_a = self.matter_test_config.global_test_params['PIXIT.RVCRUNM.MODE_A']
         self.mode_b = self.matter_test_config.global_test_params['PIXIT.RVCRUNM.MODE_B']
 

src/python_testing/TC_VALCC_3_4.py

@@ -51,7 +51,7 @@ def pics_TC_VALCC_3_4(self) -> list[str]:
     @async_test_body
     async def test_TC_VALCC_3_4(self):
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
 
         self.step(1)
         attributes = Clusters.ValveConfigurationAndControl.Attributes

src/python_testing/TC_VALCC_4_2.py

@@ -58,7 +58,7 @@ def pics_TC_VALCC_4_2(self) -> list[str]:
     @async_test_body
     async def test_TC_VALCC_4_2(self):
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
 
         self.step(1)
         attributes = Clusters.ValveConfigurationAndControl.Attributes

src/python_testing/TC_VALCC_4_3.py

@@ -61,7 +61,7 @@ def pics_TC_VALCC_4_3(self) -> list[str]:
     @async_test_body
     async def test_TC_VALCC_4_3(self):
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
 
         self.step(1)
         attributes = Clusters.ValveConfigurationAndControl.Attributes

src/python_testing/TC_VALCC_4_4.py

@@ -55,7 +55,7 @@ def pics_TC_VALCC_4_4(self) -> list[str]:
     @async_test_body
     async def test_TC_VALCC_4_4(self):
 
-        endpoint = self.user_params.get("endpoint", 1)
+        endpoint = self.get_endpoint(default=1)
 
         self.step(1)
         attributes = Clusters.ValveConfigurationAndControl.Attributes

src/python_testing/TC_pics_checker.py

@@ -112,7 +112,7 @@ def steps_TC_IDM_10_4(self):
     def test_TC_IDM_10_4(self):
         # wildcard read is done in setup_class
         self.step(1)
-        self.endpoint_id = self.matter_test_config.endpoint
+        self.endpoint_id = self.get_endpoint()
         self.endpoint = self.endpoints_tlv[self.endpoint_id]
         self.success = True
 

src/python_testing/drlk_2_x_common.py

@@ -97,7 +97,7 @@ async def run_drlk_test_common(self, lockUnlockCommand, lockUnlockCmdRspPICS, lo
 
         self.createdCredential = None
 
-        self.endpoint = self.user_params.get("endpoint", 1)
+        self.endpoint = self.get_endpoint(default=1)
 
         # Allow for user overrides of these values
         credentialIndex = self.user_params.get("credential_index", 1)

src/python_testing/matter_testing_support.py

@@ -690,6 +690,9 @@ def certificate_authority_manager(self) -> chip.CertificateAuthority.Certificate
     def dut_node_id(self) -> int:
         return self.matter_test_config.dut_node_ids[0]
 
+    def get_endpoint(self, default: Optional[int] = 0) -> int:
+        return self.matter_test_config.endpoint if self.matter_test_config.endpoint is not None else default
+
     def setup_class(self):
         super().setup_class()
 
@@ -761,7 +764,7 @@ async def read_single_attribute_check_success(
         if node_id is None:
             node_id = self.dut_node_id
         if endpoint is None:
-            endpoint = self.matter_test_config.endpoint
+            endpoint = self.get_endpoint()
 
         result = await dev_ctrl.ReadAttribute(node_id, [(endpoint, attribute)], fabricFiltered=fabric_filtered)
         attr_ret = result[endpoint][cluster][attribute]
@@ -793,7 +796,7 @@ async def read_single_attribute_expect_error(
         if node_id is None:
             node_id = self.dut_node_id
         if endpoint is None:
-            endpoint = self.matter_test_config.endpoint
+            endpoint = self.get_endpoint()
 
         result = await dev_ctrl.ReadAttribute(node_id, [(endpoint, attribute)], fabricFiltered=fabric_filtered)
         attr_ret = result[endpoint][cluster][attribute]
@@ -820,7 +823,7 @@ async def send_single_cmd(
         if node_id is None:
             node_id = self.dut_node_id
         if endpoint is None:
-            endpoint = self.matter_test_config.endpoint
+            endpoint = self.get_endpoint()
 
         result = await dev_ctrl.SendCommand(nodeid=node_id, endpoint=endpoint, payload=cmd, timedRequestTimeoutMs=timedRequestTimeoutMs)
         return result