Testing fixes for TC_SWTCH from TE2 (#34984)

* Testing fixes for TC_SWTCH from TE2

- all-clusters-app was not generating button position
  changes in some cases. This was not detected in some
  situations since the test cases don't always test for this.
- Prompts are missing endpoint ID which makes it hard when running
  per-endpoint tests to know where it applies.
- Some partials could fail on decode errors, causing test errors
  instead of fails.

This PR:

- Adds correct generation of positions on press/release.
- Adds a way to claim endpoint tested in user prompts
- Fixes failing on decode errors in partials

Testing done:

- TC_SWTCH still passes
- Manually validated button position in multi-press test/simulation
  (update to TC_SWTCH needs test plan changes). Issue is in
  all-clusters-app for CI only. See
  CHIP-Specifications/chip-test-plans#4493

* Restyled by autopep8

* Update prompt support

---------

Co-authored-by: Restyled.io <[email protected]>

examples/all-clusters-app/linux/ButtonEventsSimulator.cpp

@@ -235,6 +235,7 @@ void ButtonEventsSimulator::Next()
         break;
     }
     case ButtonEventsSimulator::State::kEmitStartOfMultiPress: {
+        SetButtonPosition(mEndpointId, mPressedButtonId);
         EmitInitialPress(mEndpointId, mPressedButtonId);
         if (mFeatureMap & static_cast<uint32_t>(Clusters::Switch::Feature::kActionSwitch))
         {
@@ -268,6 +269,7 @@ void ButtonEventsSimulator::Next()
         {
             EmitShortRelease(mEndpointId, mPressedButtonId);
         }
+        SetButtonPosition(mEndpointId, mIdleButtonId);
         StartTimer(mMultiPressReleasedTimeMillis);
         break;
     }

scripts/py_matter_yamltests/matter_yamltests/hooks.py

@@ -221,7 +221,9 @@ async def step_manual(self):
     def show_prompt(self,
                     msg: str,
                     placeholder: Optional[str] = None,
-                    default_value: Optional[str] = None) -> None:
+                    default_value: Optional[str] = None,
+                    endpoint_id: Optional[int] = None,
+                    ) -> None:
         """
         This method is called when the step needs to ask the user to perform some action or provide some value.
         """

src/python_testing/TC_SWTCH.py

@@ -44,6 +44,8 @@
 
 logger = logging.getLogger(__name__)
 
+SIMULATED_LONG_PRESS_LENGTH_SECONDS = 2.0
+
 
 def bump_substep(step: str) -> str:
     """Given a string like "5a", bump it to "5b", or "6c" to "6d" """
@@ -94,7 +96,8 @@ def _send_multi_press_named_pipe_command(self, endpoint_id: int, number_of_press
 
     def _send_long_press_named_pipe_command(self, endpoint_id: int, pressed_position: int, feature_map: int):
         command_dict = {"Name": "SimulateLongPress", "EndpointId": endpoint_id,
-                        "ButtonId": pressed_position, "LongPressDelayMillis": 5000, "LongPressDurationMillis": 5500, "FeatureMap": feature_map}
+                        "ButtonId": pressed_position, "LongPressDelayMillis": int(1000 * (SIMULATED_LONG_PRESS_LENGTH_SECONDS - 0.5)),
+                        "LongPressDurationMillis": int(1000 * SIMULATED_LONG_PRESS_LENGTH_SECONDS), "FeatureMap": feature_map}
         self._send_named_pipe_command(command_dict)
 
     def _send_latching_switch_named_pipe_command(self, endpoint_id: int, new_position: int):
@@ -168,7 +171,9 @@ def _ask_for_release(self):
                 prompt_msg="Release the button."
             )
         else:
-            time.sleep(self.keep_pressed_delay/1000)
+            # This will await for the events to be generated properly. Note that there is a bit of a
+            # race here for the button simulator, but this race is extremely unlikely to be lost.
+            time.sleep(SIMULATED_LONG_PRESS_LENGTH_SECONDS + 0.5)
 
     def _await_sequence_of_events(self, event_queue: queue.Queue, endpoint_id: int, sequence: list[ClusterObjects.ClusterEvent], timeout_sec: float):
         start_time = time.time()
@@ -373,7 +378,6 @@ async def test_TC_SWTCH_2_3(self):
         self.step(5)
         # We're using a long press here with a very long duration (in computer-land). This will let us check the intermediate values.
         # This is 1s larger than the subscription ceiling
-        self.keep_pressed_delay = 6000
         self.pressed_position = 1
         self._ask_for_keep_pressed(endpoint_id, self.pressed_position, feature_map)
         event_listener.wait_for_event_report(cluster.Events.InitialPress)
@@ -595,7 +599,7 @@ def steps_TC_SWTCH_2_5(self):
     @staticmethod
     def should_run_SWTCH_2_5(wildcard, endpoint):
         msm = has_feature(Clusters.Switch, Clusters.Switch.Bitmaps.Feature.kMomentarySwitchMultiPress)
-        asf = has_feature(Clusters.Switch, 0x20)
+        asf = has_feature(Clusters.Switch, Clusters.Switch.Bitmaps.Feature.kActionSwitch)
         return msm(wildcard, endpoint) and not asf(wildcard, endpoint)
 
     @per_endpoint_test(should_run_SWTCH_2_5)

src/python_testing/matter_testing_support.py

@@ -1380,11 +1380,21 @@ def wait_for_user_input(self,
         Returns:
             str: User input or none if input is closed.
         """
+
+        # TODO(#31928): Remove any assumptions of test params for endpoint ID.
+
+        # Get the endpoint user param instead of `--endpoint-id` result, if available, temporarily.
+        endpoint_id = self.user_params.get("endpoint", None)
+        if endpoint_id is None or not isinstance(endpoint_id, int):
+            endpoint_id = self.matter_test_config.endpoint
+
         if self.runner_hook:
+            # TODO(#31928): Add endpoint support to hooks.
             self.runner_hook.show_prompt(msg=prompt_msg,
                                          placeholder=prompt_msg_placeholder,
                                          default_value=default_value)
-        logging.info("========= USER PROMPT =========")
+
+        logging.info(f"========= USER PROMPT for Endpoint {endpoint_id} =========")
         logging.info(f">>> {prompt_msg.rstrip()} (press enter to confirm)")
         try:
             return input()
@@ -1881,6 +1891,9 @@ def _has_attribute(wildcard, endpoint, attribute: ClusterObjects.ClusterAttribut
     cluster = get_cluster_from_attribute(attribute)
     try:
         attr_list = wildcard.attributes[endpoint][cluster][cluster.Attributes.AttributeList]
+        if not isinstance(attr_list, list):
+            asserts.fail(
+                f"Failed to read mandatory AttributeList attribute value for cluster {cluster} on endpoint {endpoint}: {attr_list}.")
         return attribute.attribute_id in attr_list
     except KeyError:
         return False
@@ -1910,9 +1923,13 @@ def has_attribute(attribute: ClusterObjects.ClusterAttributeDescriptor) -> Endpo
     return partial(_has_attribute, attribute=attribute)
 
 
-def _has_feature(wildcard, endpoint, cluster: ClusterObjects.ClusterObjectDescriptor, feature: IntFlag) -> bool:
+def _has_feature(wildcard, endpoint: int, cluster: ClusterObjects.ClusterObjectDescriptor, feature: IntFlag) -> bool:
     try:
         feature_map = wildcard.attributes[endpoint][cluster][cluster.Attributes.FeatureMap]
+        if not isinstance(feature_map, int):
+            asserts.fail(
+                f"Failed to read mandatory FeatureMap attribute value for cluster {cluster} on endpoint {endpoint}: {feature_map}.")
+
         return (feature & feature_map) != 0
     except KeyError:
         return False
@@ -1926,8 +1943,8 @@ def has_feature(cluster: ClusterObjects.ClusterObjectDescriptor, feature: IntFla
 
         EP0: cluster A, B, C
         EP1: cluster D with feature F0
-        EP2, cluster D with feature F0
-        EP3, cluster D without feature F0
+        EP2: cluster D with feature F0
+        EP3: cluster D without feature F0
 
         And the following test specification:
         @per_endpoint_test(has_feature(Clusters.D.Bitmaps.Feature.F0))