Fix MSR flag issues in TC-SWTCH-* (project-chip#35327)

tcarmelveilleux · restyled-commits · shgutte · commit 8dae5846e729 · 2024-09-10T16:03:05.000+05:30
* Fix MSR flag issues in TC-SWTCH-* - MSR is optional, but was checked as if mandatory. - See CHIP-Specifications/chip-test-plans#4593 - Fix initial conditions in CI * Restyled by clang-format * Restyled by autopep8 --------- Co-authored-by: Restyled.io <commits@restyled.io>
diff --git a/examples/all-clusters-app/linux/AllClustersCommandDelegate.cpp b/examples/all-clusters-app/linux/AllClustersCommandDelegate.cpp
@@ -248,6 +248,34 @@ void HandleSimulateLatchPosition(Json::Value & jsonValue)
     }
 }
 
+/**
+ * Named pipe handler for simulating switch is idle
+ *
+ * Usage example:
+ *   echo '{"Name": "SimulateSwitchIdle", "EndpointId": 3}' > /tmp/chip_all_clusters_fifo_1146610
+ *
+ * JSON Arguments:
+ *   - "Name": Must be "SimulateSwitchIdle"
+ *   - "EndpointId": ID of endpoint having a switch cluster
+ *
+ * @param jsonValue - JSON payload from named pipe
+ */
+
+void HandleSimulateSwitchIdle(Json::Value & jsonValue)
+{
+    bool hasEndpointId = HasNumericField(jsonValue, "EndpointId");
+
+    if (!hasEndpointId)
+    {
+        std::string inputJson = jsonValue.toStyledString();
+        ChipLogError(NotSpecified, "Missing or invalid value for one of EndpointId in %s", inputJson.c_str());
+        return;
+    }
+
+    EndpointId endpointId = static_cast<EndpointId>(jsonValue["EndpointId"].asUInt());
+    (void) Switch::Attributes::CurrentPosition::Set(endpointId, 0);
+}
+
 void EmitOccupancyChangedEvent(EndpointId endpointId, uint8_t occupancyValue)
 {
     Clusters::OccupancySensing::Events::OccupancyChanged::Type event{};
@@ -427,6 +455,10 @@ void AllClustersAppCommandHandler::HandleCommand(intptr_t context)
     {
         HandleSimulateLatchPosition(self->mJsonValue);
     }
+    else if (name == "SimulateSwitchIdle")
+    {
+        HandleSimulateSwitchIdle(self->mJsonValue);
+    }
     else if (name == "SetOccupancy")
     {
         uint8_t occupancy     = static_cast<uint8_t>(self->mJsonValue["Occupancy"].asUInt());
diff --git a/src/python_testing/TC_SWTCH.py b/src/python_testing/TC_SWTCH.py
@@ -104,9 +104,15 @@ def _send_latching_switch_named_pipe_command(self, endpoint_id: int, new_positio
         command_dict = {"Name": "SimulateLatchPosition", "EndpointId": endpoint_id, "PositionId": new_position}
         self._send_named_pipe_command(command_dict)
 
-    def _ask_for_switch_idle(self):
+    def _send_switch_idle_named_pipe_command(self, endpoint_id: int):
+        command_dict = {"Name": "SimulateSwitchIdle", "EndpointId": endpoint_id}
+        self._send_named_pipe_command(command_dict)
+
+    def _ask_for_switch_idle(self, endpoint_id: int, omit_for_simulator: bool = False):
         if not self._use_button_simulator():
             self.wait_for_user_input(prompt_msg="Ensure switch is idle")
+        elif not omit_for_simulator:
+            self._send_switch_idle_named_pipe_command(endpoint_id)
 
     def _ask_for_switch_position(self, endpoint_id: int, new_position: int):
         if not self._use_button_simulator():
@@ -283,7 +289,6 @@ async def test_TC_SWTCH_2_2(self):
         # Step 3: Operator sets switch to first position on the DUT.
         self.step(3)
         self._ask_for_switch_position(endpoint_id, new_position=0)
-        event_listener.flush_events()
 
         # Step 4: TH reads the CurrentPosition attribute from the DUT.
         # Verify that the value is 0.
@@ -369,7 +374,7 @@ async def test_TC_SWTCH_2_3(self):
         await event_listener.start(self.default_controller, self.dut_node_id, endpoint=endpoint_id)
 
         self.step(3)
-        self._ask_for_switch_idle()
+        self._ask_for_switch_idle(endpoint_id)
 
         self.step(4)
         button_val = await self.read_single_attribute_check_success(cluster=cluster, attribute=cluster.Attributes.CurrentPosition)
@@ -461,7 +466,7 @@ async def test_TC_SWTCH_2_4(self):
 
         # Step 3: Operator does not operate switch on the DUT
         self.step(3)
-        self._ask_for_switch_idle()
+        self._ask_for_switch_idle(endpoint_id)
 
         # Step 4: TH reads the CurrentPosition attribute from the DUT
         self.step(4)
@@ -486,9 +491,9 @@ async def test_TC_SWTCH_2_4(self):
         self._await_sequence_of_events(event_queue=event_listener.event_queue, endpoint_id=endpoint_id,
                                        sequence=expected_events, timeout_sec=post_prompt_settle_delay_seconds)
 
-        # - if MSL or AS feature is supported, expect to see LongPress/LongRelease in that order.
-        if not has_msl_feature and not has_as_feature:
-            logging.info("Since MSL and AS features both unsupported, skipping check for LongPress/LongRelease")
+        # - if MSL feature is supported, expect to see LongPress/LongRelease in that order.
+        if not has_msl_feature:
+            logging.info("Since MSL feature unsupported, skipping check for LongPress/LongRelease")
         else:
             # - TH expects report of LongPress, LongRelease in that order.
             logging.info(f"Starting to wait for {post_prompt_settle_delay_seconds:.1f} seconds for LongPress then LongRelease.")
@@ -538,13 +543,13 @@ def steps_TC_SWTCH_2_5(self):
                 TestStep("6b", "Operator repeat step 4a 3 times quickly",
                          """
                          * Verify that the TH receives InitialPress event with NewPosition set to 1 from the DUT
-                         * Verify that the TH receives ShortRelease event with PreviousPosition set to 1 from the DUT
+                         * If MSR supported, Verify that the TH receives ShortRelease event with PreviousPosition set to 1 from the DUT
                          * Verify that the TH receives InitialPress event with NewPosition set to 1 from the DUT
                          * Verify that the TH receives MultiPressOngoing event with NewPosition set to 1 and CurrentNumberOfPressesCounted set to 2 from the DUT
-                         * Verify that the TH receives ShortRelease event with PreviousPosition set to 1 from the DUT
+                         * If MSR supported, Verify that the TH receives ShortRelease event with PreviousPosition set to 1 from the DUT
                          * Verify that the TH receives InitialPress event with NewPosition set to 1 from the DUT
                          * Verify that the TH receives MultiPressOngoing event with NewPosition set to 1 and CurrentNumberOfPressesCounted set to 3 from the DUT
-                         * Verify that the TH receives ShortRelease event with PreviousPosition set to 1 from the DUT +
+                         * If MSR supported, Verify that the TH receives ShortRelease event with PreviousPosition set to 1 from the DUT +
 
                          The events sequence from the subscription SHALL follow the same sequence as expressed above, in the exact order of events specified.
                          """),
@@ -561,11 +566,11 @@ def steps_TC_SWTCH_2_5(self):
                          """
 
                          * Verify that the TH receives InitialPress event with NewPosition set to 1 from the DUT
-                         * Verify that the TH receives ShortRelease event with PreviousPosition set to 1 from the DUT
+                         * If MSR supported, Verify that the TH receives ShortRelease event with PreviousPosition set to 1 from the DUT
                          * Verify that the TH receives InitialPress event with NewPosition set to 1 from the DUT
 
                          * Verify that the TH receives MultiPressOngoing event with NewPosition set to 1 and CurrentNumberOfPressesCounted set to 2 from the DUT
-                         * Verify that the TH receives ShortRelease event with PreviousPosition set to 1 from the DUT
+                         * If MSR supported, verify that the TH receives ShortRelease event with PreviousPosition set to 1 from the DUT
                          * Verify that the TH does not receive LongPress event from the DUT
                          * Verify that the TH does not receive LongRelease event from the DUT
 
@@ -586,7 +591,7 @@ def steps_TC_SWTCH_2_5(self):
                          * Verify that the TH receives (one, not more than one) LongPress event with NewPosition set to 1 from the DUT
                          * Verify that the TH receives LongRelease event with PreviousPosition set to 1 from the DUT
                          * Verify that the TH receives InitialPress event with NewPosition set to 1 from the DUT
-                         * Verify that the TH receives ShortRelease event with PreviousPosition set to 1 from the DUT
+                         * If MSR supported, verify that the TH receives ShortRelease event with PreviousPosition set to 1 from the DUT
                          * Verify that the TH does not receive MultiPressOngoing event from the DUT
 
                          The events sequence from the subscription SHALL follow the same sequence as expressed above, in the exact order of events specified.
@@ -610,6 +615,7 @@ async def test_TC_SWTCH_2_5(self):
         cluster = Clusters.Switch
         feature_map = await self.read_single_attribute_check_success(cluster, attribute=cluster.Attributes.FeatureMap)
         has_msl_feature = (feature_map & cluster.Bitmaps.Feature.kMomentarySwitchLongPress)
+        has_msr_feature = (feature_map & cluster.Bitmaps.Feature.kMomentarySwitchRelease)
         multi_press_max = await self.read_single_attribute_check_success(cluster, attribute=cluster.Attributes.MultiPressMax)
 
         endpoint_id = self.matter_test_config.endpoint
@@ -620,7 +626,7 @@ async def test_TC_SWTCH_2_5(self):
         await event_listener.start(self.default_controller, self.dut_node_id, endpoint=endpoint_id)
 
         self.step(3)
-        self._ask_for_switch_idle()
+        self._ask_for_switch_idle(endpoint_id)
 
         def test_multi_press_sequence(starting_step: str, count: int, short_long: bool = False):
             step = starting_step
@@ -640,12 +646,13 @@ def test_multi_press_sequence(starting_step: str, count: int, short_long: bool =
                     asserts.assert_equal(event.newPosition, pressed_position, "Unexpected NewPosition on MultiPressOngoing")
                     asserts.assert_equal(event.currentNumberOfPressesCounted, pos_idx + 1,
                                          "Unexpected CurrentNumberOfPressesCounted on MultiPressOngoing")
-                event = event_listener.wait_for_event_report(cluster.Events.ShortRelease)
-                asserts.assert_equal(event.previousPosition, pressed_position, "Unexpected PreviousPosition on ShortRelease")
+                if has_msr_feature:
+                    event = event_listener.wait_for_event_report(cluster.Events.ShortRelease)
+                    asserts.assert_equal(event.previousPosition, pressed_position, "Unexpected PreviousPosition on ShortRelease")
 
             step = bump_substep(step)
             self.step(step)
-            self._ask_for_switch_idle()
+            self._ask_for_switch_idle(endpoint_id, omit_for_simulator=True)
             event = event_listener.wait_for_event_report(cluster.Events.MultiPressComplete)
             asserts.assert_equal(event.previousPosition, pressed_position, "Unexpected PreviousPosition on MultiPressComplete")
             asserts.assert_equal(event.totalNumberOfPressesCounted, count, "Unexpected count on MultiPressComplete")
@@ -687,13 +694,15 @@ def test_multi_press_sequence(starting_step: str, count: int, short_long: bool =
 
         event = event_listener.wait_for_event_report(cluster.Events.InitialPress)
         asserts.assert_equal(event.newPosition, pressed_position, "Unexpected NewPosition on InitialEvent")
-        event = event_listener.wait_for_event_report(cluster.Events.ShortRelease)
-        asserts.assert_equal(event.previousPosition, pressed_position, "Unexpected PreviousPosition on ShortRelease")
+
+        if has_msr_feature:
+            event = event_listener.wait_for_event_report(cluster.Events.ShortRelease)
+            asserts.assert_equal(event.previousPosition, pressed_position, "Unexpected PreviousPosition on ShortRelease")
 
         # Because this is a queue, we verify that no multipress ongoing is received by verifying that the next event is the multipress complete
 
         self.step("9b")
-        self._ask_for_switch_idle()
+        self._ask_for_switch_idle(endpoint_id, omit_for_simulator=True)
         event = event_listener.wait_for_event_report(cluster.Events.MultiPressComplete)
         asserts.assert_equal(event.previousPosition, pressed_position, "Unexpected PreviousPosition on MultiPressComplete")
         asserts.assert_equal(event.totalNumberOfPressesCounted, 1, "Unexpected count on MultiPressComplete")
@@ -795,7 +804,7 @@ async def test_TC_SWTCH_2_6(self):
         await event_listener.start(self.default_controller, self.dut_node_id, endpoint=endpoint_id)
 
         self.step(3)
-        self._ask_for_switch_idle()
+        self._ask_for_switch_idle(endpoint_id)
 
         def test_multi_press_sequence(starting_step: str, count: int, short_long: bool = False):
             step = starting_step
@@ -813,7 +822,7 @@ def test_multi_press_sequence(starting_step: str, count: int, short_long: bool =
 
             step = bump_substep(step)
             self.step(step)
-            self._ask_for_switch_idle()
+            self._ask_for_switch_idle(endpoint_id, omit_for_simulator=True)
             event = event_listener.wait_for_event_report(cluster.Events.MultiPressComplete)
             asserts.assert_equal(event.previousPosition, pressed_position, "Unexpected PreviousPosition on MultiPressComplete")
             expected_count = 0 if count > multi_press_max else count
@@ -853,7 +862,7 @@ def test_multi_press_sequence(starting_step: str, count: int, short_long: bool =
 
         # Verify that we don't receive the multi-press ongoing or short release by verifying that the next event in the sequence is the multi-press complete
         self.step("9b")
-        self._ask_for_switch_idle()
+        self._ask_for_switch_idle(endpoint_id, omit_for_simulator=True)
         event = event_listener.wait_for_event_report(cluster.Events.MultiPressComplete)
         asserts.assert_equal(event.previousPosition, pressed_position, "Unexpected PreviousPosition on MultiPressComplete")
         asserts.assert_equal(event.totalNumberOfPressesCounted, 1, "Unexpected count on MultiPressComplete")
