From 1ca8115db2d7515f73015c77f206a58a59d01824 Mon Sep 17 00:00:00 2001 From: Cecille Freeman Date: Tue, 24 Sep 2024 14:40:16 -0700 Subject: [PATCH] TC-OPSTATE-2.5,6: Fix messages, re-establish subscription after reset --- src/python_testing/TC_OpstateCommon.py | 38 +++++++++++++++++--------- 1 file changed, 25 insertions(+), 13 deletions(-) diff --git a/src/python_testing/TC_OpstateCommon.py b/src/python_testing/TC_OpstateCommon.py index 060f5d2f53ed30..ec8ae301840b28 100644 --- a/src/python_testing/TC_OpstateCommon.py +++ b/src/python_testing/TC_OpstateCommon.py @@ -111,14 +111,16 @@ def init_test(self): asserts.fail("The --app-pid flag must be set when PICS_SDK_CI_ONLY is set") self.app_pipe = self.app_pipe + str(app_pid) - def send_raw_manual_or_pipe_command(self, command: dict): + def send_raw_manual_or_pipe_command(self, command: dict, msg: str): if self.is_ci: self.write_to_app_pipe(command) time.sleep(0.1) else: - self.wait_for_user_input(prompt_msg="Press Enter when ready.\n") + prompt = msg if msg is not None else "Press Enter when ready." + prompt += '\n' + self.wait_for_user_input(prompt_msg=prompt) - def send_manual_or_pipe_command(self, device: str, name: str, operation: str, param: Any = None): + def send_manual_or_pipe_command(self, device: str, name: str, operation: str, param: Any = None, msg=None): command = { "Name": name, "Device": device, @@ -128,7 +130,7 @@ def send_manual_or_pipe_command(self, device: str, name: str, operation: str, pa if param is not None: command["Param"] = param - self.send_raw_manual_or_pipe_command(command) + self.send_raw_manual_or_pipe_command(command, msg) async def send_cmd(self, endpoint, cmd, timedRequestTimeoutMs=None): logging.info(f"##### Command {cmd}") @@ -1046,11 +1048,13 @@ async def TEST_TC_OPSTATE_BASE_2_5(self, endpoint=1): self.send_manual_or_pipe_command(name="OperationalStateChange", device=self.device, operation="OnFault", - param=cluster.Enums.ErrorStateEnum.kNoError) + param=cluster.Enums.ErrorStateEnum.kNoError, + msg="Ensure the DUT is not in an error state.") self.send_manual_or_pipe_command(name="OperationalStateChange", device=self.device, - operation="Stop") + operation="Stop", + msg="Put the DUT in a state where it can receive a start command") # STEP 4: TH sends Start command to the DUT self.step(4) @@ -1116,7 +1120,14 @@ async def TEST_TC_OPSTATE_BASE_2_5(self, endpoint=1): # reboot outside CI environment. if not self.is_ci: self.wait_for_user_input(prompt_msg="Restart DUT. Press Enter when ready.\n") + # Expire the session and re-establish the subscription self.default_controller.ExpireSessions(self.dut_node_id) + if self.check_pics(f"{self.test_info.pics_code}.S.E01"): + # Subscribe to Events and when they are sent push them to a queue for checking later + events_callback = EventSpecificChangeCallback(events.OperationCompletion) + await events_callback.start(self.default_controller, + self.dut_node_id, + endpoint) # STEP 12: TH waits for {PIXIT.WAITTIME.REBOOT} self.step(12) @@ -1259,11 +1270,12 @@ async def TEST_TC_OPSTATE_BASE_2_6(self, endpoint=1): sub_handler = ClusterAttributeChangeAccumulator(cluster) await sub_handler.start(self.default_controller, self.dut_node_id, endpoint) + self.step(3) if self.pics_guard(self.check_pics(f"{self.test_info.pics_code}.S.M.ST_RUNNING")): - self.step(3) self.send_manual_or_pipe_command(name="OperationalStateChange", device=self.device, - operation="Start") + operation="Start", + msg="Put the DUT into the running state") time.sleep(1) await self.read_and_expect_value(endpoint=endpoint, attribute=attributes.OperationalState, @@ -1272,8 +1284,6 @@ async def TEST_TC_OPSTATE_BASE_2_6(self, endpoint=1): if countdownTime is not NullValue: count = sub_handler.attribute_report_counts[attributes.CountdownTime] asserts.assert_greater(count, 0, "Did not receive any reports for CountdownTime") - else: - self.skip_step(3) sub_handler.reset() self.step(4) @@ -1311,11 +1321,12 @@ async def TEST_TC_OPSTATE_BASE_2_6(self, endpoint=1): self.skip_step(7) sub_handler.reset() + self.step(8) if self.pics_guard(self.check_pics(f"{self.test_info.pics_code}.S.M.ST_RUNNING")): - self.step(8) self.send_manual_or_pipe_command(name="OperationalStateChange", device=self.device, - operation="Start") + operation="Start", + msg="Put DUT in running state") time.sleep(1) await self.read_and_expect_value(endpoint=endpoint, attribute=attributes.OperationalState, @@ -1339,7 +1350,8 @@ async def TEST_TC_OPSTATE_BASE_2_6(self, endpoint=1): self.step(11) self.send_manual_or_pipe_command(name="OperationalStateChange", device=self.device, - operation="Pause") + operation="Pause", + msg="Put DUT in paused state") time.sleep(1) count = sub_handler.attribute_report_counts[attributes.CountdownTime] asserts.assert_greater(count, 0, "Did not receive any reports for CountdownTime")