Skip to content

Commit

Permalink
Fix MSR flag issues in TC-SWTCH-* (#35327)
Browse files Browse the repository at this point in the history 
* 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 <[email protected]>
  • Loading branch information
@tcarmelveilleux @restyled-commits
tcarmelveilleux and restyled-commits authored Aug 30, 2024
1 parent 6fb9997 commit e5d9431
Show file tree
Hide file tree
Showing 2 changed files with 64 additions and 23 deletions.
32 changes: 32 additions & 0 deletions examples/all-clusters-app/linux/AllClustersCommandDelegate.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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{};
Expand Down Expand Up @@ -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());
Expand Down
55 changes: 32 additions & 23 deletions src/python_testing/TC_SWTCH.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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():
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand All @@ -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.")
Expand Down Expand Up @@ -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.
"""),
Expand All @@ -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
Expand All @@ -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.
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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")
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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")
Expand Down

0 comments on commit e5d9431

Please sign in to comment.