Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix MSR flag issues in TC-SWTCH-* #35327

Merged
merged 3 commits into from
Aug 30, 2024
Merged

Fix MSR flag issues in TC-SWTCH-* #35327

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
72e7d0e
Fix MSR flag issues in TC-SWTCH-*
tcarmelveilleux Aug 30, 2024
2c1267e
Restyled by clang-format
restyled-commits Aug 30, 2024
399f023
Restyled by autopep8
restyled-commits Aug 30, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 32 additions & 0 deletions examples/all-clusters-app/linux/AllClustersCommandDelegate.cpp
View file
Open in desktop
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
Open in desktop
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
Loading