From 506d973496e92cd4e57faa39d6a40708954098ad Mon Sep 17 00:00:00 2001 From: Tennessee Carmel-Veilleux Date: Wed, 28 Aug 2024 17:27:48 -0400 Subject: [PATCH] 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 https://github.com/CHIP-Specifications/chip-test-plans/issues/4493 * Restyled by autopep8 * Update prompt support --------- Co-authored-by: Restyled.io --- .../linux/ButtonEventsSimulator.cpp | 2 ++ .../matter_yamltests/hooks.py | 4 ++- src/python_testing/TC_SWTCH.py | 12 ++++++--- src/python_testing/matter_testing_support.py | 25 ++++++++++++++++--- 4 files changed, 34 insertions(+), 9 deletions(-) diff --git a/examples/all-clusters-app/linux/ButtonEventsSimulator.cpp b/examples/all-clusters-app/linux/ButtonEventsSimulator.cpp index 53a08672fdbc07..0122eba4336340 100644 --- a/examples/all-clusters-app/linux/ButtonEventsSimulator.cpp +++ b/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(Clusters::Switch::Feature::kActionSwitch)) { @@ -268,6 +269,7 @@ void ButtonEventsSimulator::Next() { EmitShortRelease(mEndpointId, mPressedButtonId); } + SetButtonPosition(mEndpointId, mIdleButtonId); StartTimer(mMultiPressReleasedTimeMillis); break; } diff --git a/scripts/py_matter_yamltests/matter_yamltests/hooks.py b/scripts/py_matter_yamltests/matter_yamltests/hooks.py index 78905826f55757..ca739b8ea27633 100644 --- a/scripts/py_matter_yamltests/matter_yamltests/hooks.py +++ b/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. """ diff --git a/src/python_testing/TC_SWTCH.py b/src/python_testing/TC_SWTCH.py index d6c4694560b16d..8f9a694db7c496 100644 --- a/src/python_testing/TC_SWTCH.py +++ b/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) diff --git a/src/python_testing/matter_testing_support.py b/src/python_testing/matter_testing_support.py index 4127ead64c3083..20c33a0d36b785 100644 --- a/src/python_testing/matter_testing_support.py +++ b/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))