forked from project-chip/connectedhomeip
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add DEM test scripts (project-chip#34234)
* Add DEM test scripts * Restyled by autopep8 * Restyled by isort * Renamed DEMBaseTest.py to TC_DEMTestBase.py so that it's grouped with the TC_DEM scripts. Also split the expected result in the TestSteps() into 3rd 'expected' arg. Corrected some missing Verification steps. * Updated DEM_2.4 with corrections to match latest test plan. Also corrected PICS and description from DEM 2.3 * Restyled by isort * TC_DEM_2_5.py updated with TestSteps and corrections to match latest test plan * Reformatted TC_DEM_2_9.py * Changed PICS for DEM_2_5.py to have it as a single boolean - unsure on the correct format. * Updated DEM_2_6.py test steps in new format and corrected Test description and PICS * Fixed extra " in DEM_2_6 * Reformatted TC_DEM_2_7.py - corrected PICS and description * Corrected Feature 2 for DEM_2_6 * Corrected TC number for DEM_2_6 * Updated TC_DEM_2_8 test steps, description and pics * Restyled by isort * Apply review comments from James Harrow * Apply review comments from James Harrow * Restyled by autopep8 * Restyled by isort * Apply review comments from James Harrow * Fix python lint error * Apply review comments from James Harrow * Restyled by autopep8 * Added # === BEGIN CI TEST ARGUMENTS banners to TC_DEM python scripts * Specify --featureSet option on runner command line * Address review comments from Rob Bultman * Address review comments from Rob Bultman * Restyled by isort * Update src/python_testing/TC_DEM_2_5.py Co-authored-by: Carolina Lopes <[email protected]> * Update TC_DEM_2_6.py - changed PICS conformance to comma separated list * Update TC_DEM_2_7.py - changed PICS conformance to comma separated list * Update TC_DEM_2_9.py - changed PICS conformance to comma separated list * Update TC_DEM_2_8.py - changed PICS conformance to comma separated list * Enabled DEM Test scripts as part of CI --------- Co-authored-by: Restyled.io <[email protected]> Co-authored-by: James Harrow <[email protected]> Co-authored-by: jamesharrow <[email protected]> Co-authored-by: Carolina Lopes <[email protected]>
- Loading branch information
Showing
10 changed files
with
2,290 additions
and
7 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,242 @@ | ||
| # | ||
| # Copyright (c) 2023 Project CHIP Authors | ||
| # All rights reserved. | ||
| # | ||
| # Licensed under the Apache License, Version 2.0 (the "License"); | ||
| # you may not use this file except in compliance with the License. | ||
| # You may obtain a copy of the License at | ||
| # | ||
| # http://www.apache.org/licenses/LICENSE-2.0 | ||
| # | ||
| # Unless required by applicable law or agreed to in writing, software | ||
| # distributed under the License is distributed on an "AS IS" BASIS, | ||
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| # See the License for the specific language governing permissions and | ||
| # limitations under the License. | ||
|
|
||
|
|
||
| import logging | ||
|
|
||
| import chip.clusters as Clusters | ||
| from chip.interaction_model import InteractionModelError, Status | ||
| from matter_testing_support import utc_time_in_matter_epoch | ||
| from mobly import asserts | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| class DEMTestBase: | ||
|
|
||
| async def read_dem_attribute_expect_success(self, endpoint: int = None, attribute: str = ""): | ||
| cluster = Clusters.Objects.DeviceEnergyManagement | ||
| full_attr = getattr(cluster.Attributes, attribute) | ||
| logging.info(f"endpoint {endpoint} full_attr {full_attr}") | ||
| return await self.read_single_attribute_check_success(endpoint=endpoint, cluster=cluster, attribute=full_attr) | ||
|
|
||
| async def check_dem_attribute(self, attribute, expected_value, endpoint: int = None): | ||
| value = await self.read_dem_attribute_expect_success(endpoint=endpoint, attribute=attribute) | ||
| asserts.assert_equal(value, expected_value, | ||
| f"Unexpected '{attribute}' value - expected {expected_value}, was {value}") | ||
|
|
||
| async def send_power_adjustment_command(self, power: int, duration: int, | ||
| cause: Clusters.Objects.DeviceEnergyManagement.Enums.CauseEnum, | ||
| endpoint: int = None, timedRequestTimeoutMs: int = 3000, | ||
| expected_status: Status = Status.Success): | ||
| try: | ||
| await self.send_single_cmd(cmd=Clusters.DeviceEnergyManagement.Commands.PowerAdjustRequest( | ||
| power=power, | ||
| duration=duration, | ||
| cause=cause), | ||
| endpoint=endpoint, | ||
| timedRequestTimeoutMs=timedRequestTimeoutMs) | ||
|
|
||
| asserts.assert_equal(expected_status, Status.Success) | ||
|
|
||
| except InteractionModelError as e: | ||
| asserts.assert_equal(e.status, expected_status, "Unexpected error returned") | ||
|
|
||
| async def send_cancel_power_adjustment_command(self, endpoint: int = None, timedRequestTimeoutMs: int = 3000, | ||
| expected_status: Status = Status.Success): | ||
| try: | ||
| await self.send_single_cmd(cmd=Clusters.DeviceEnergyManagement.Commands.CancelPowerAdjustRequest(), | ||
| endpoint=endpoint, | ||
| timedRequestTimeoutMs=timedRequestTimeoutMs) | ||
|
|
||
| asserts.assert_equal(expected_status, Status.Success) | ||
|
|
||
| except InteractionModelError as e: | ||
| asserts.assert_equal(e.status, expected_status, "Unexpected error returned") | ||
|
|
||
| async def send_start_time_adjust_request_command(self, requestedStartTime: int, | ||
| cause: Clusters.Objects.DeviceEnergyManagement.Enums.CauseEnum, | ||
| endpoint: int = None, timedRequestTimeoutMs: int = 3000, | ||
| expected_status: Status = Status.Success): | ||
| try: | ||
| await self.send_single_cmd(cmd=Clusters.DeviceEnergyManagement.Commands.StartTimeAdjustRequest( | ||
| requestedStartTime=requestedStartTime, | ||
| cause=cause), | ||
| endpoint=endpoint, | ||
| timedRequestTimeoutMs=timedRequestTimeoutMs) | ||
|
|
||
| asserts.assert_equal(expected_status, Status.Success) | ||
|
|
||
| except InteractionModelError as e: | ||
| asserts.assert_equal(e.status, expected_status, "Unexpected error returned") | ||
|
|
||
| async def send_start_time_adjust_clear_command(self, | ||
| endpoint: int = None, timedRequestTimeoutMs: int = 3000, | ||
| expected_status: Status = Status.Success): | ||
| try: | ||
| await self.send_single_cmd(cmd=Clusters.DeviceEnergyManagement.Commands.StartTimeAdjustClear(), # StartTimeAdjustmentClear(), | ||
| endpoint=endpoint, | ||
| timedRequestTimeoutMs=timedRequestTimeoutMs) | ||
|
|
||
| asserts.assert_equal(expected_status, Status.Success) | ||
|
|
||
| except InteractionModelError as e: | ||
| asserts.assert_equal(e.status, expected_status, "Unexpected error returned") | ||
|
|
||
| async def send_cancel_request_command(self, | ||
| endpoint: int = None, timedRequestTimeoutMs: int = 3000, | ||
| expected_status: Status = Status.Success): | ||
| try: | ||
| await self.send_single_cmd(cmd=Clusters.DeviceEnergyManagement.Commands.CancelRequest(), | ||
| endpoint=endpoint, | ||
| timedRequestTimeoutMs=timedRequestTimeoutMs) | ||
|
|
||
| asserts.assert_equal(expected_status, Status.Success) | ||
|
|
||
| except InteractionModelError as e: | ||
| asserts.assert_equal(e.status, expected_status, "Unexpected error returned") | ||
|
|
||
| async def send_pause_request_command(self, duration: int, cause: | ||
| Clusters.Objects.DeviceEnergyManagement.Enums.AdjustmentCauseEnum, | ||
| endpoint: int = None, timedRequestTimeoutMs: int = 3000, | ||
| expected_status: Status = Status.Success): | ||
| try: | ||
| await self.send_single_cmd(cmd=Clusters.DeviceEnergyManagement.Commands.PauseRequest( | ||
| duration=duration, | ||
| cause=cause), | ||
| endpoint=endpoint, | ||
| timedRequestTimeoutMs=timedRequestTimeoutMs) | ||
|
|
||
| asserts.assert_equal(expected_status, Status.Success) | ||
|
|
||
| except InteractionModelError as e: | ||
| asserts.assert_equal(e.status, expected_status, "Unexpected error returned") | ||
|
|
||
| async def send_resume_request_command(self, endpoint: int = None, timedRequestTimeoutMs: int = 3000, | ||
| expected_status: Status = Status.Success): | ||
| try: | ||
| await self.send_single_cmd(cmd=Clusters.DeviceEnergyManagement.Commands.ResumeRequest(), | ||
| endpoint=endpoint, | ||
| timedRequestTimeoutMs=timedRequestTimeoutMs) | ||
|
|
||
| asserts.assert_equal(expected_status, Status.Success) | ||
|
|
||
| except InteractionModelError as e: | ||
| asserts.assert_equal(e.status, expected_status, "Unexpected error returned") | ||
|
|
||
| async def send_modify_forecast_request_command(self, forecastID: int, | ||
| slotAdjustments: list[Clusters.DeviceEnergyManagement.Structs.SlotAdjustmentStruct], | ||
| cause: Clusters.DeviceEnergyManagement.Enums.AdjustmentCauseEnum, | ||
| endpoint: int = None, timedRequestTimeoutMs: int = 3000, | ||
| expected_status: Status = Status.Success): | ||
| try: | ||
| await self.send_single_cmd(cmd=Clusters.DeviceEnergyManagement.Commands.ModifyForecastRequest(forecastID=forecastID, | ||
| slotAdjustments=slotAdjustments, | ||
| cause=cause), | ||
| endpoint=endpoint, | ||
| timedRequestTimeoutMs=timedRequestTimeoutMs) | ||
|
|
||
| asserts.assert_equal(expected_status, Status.Success) | ||
|
|
||
| except InteractionModelError as e: | ||
| asserts.assert_equal(e.status, expected_status, "Unexpected error returned") | ||
|
|
||
| async def send_request_constraint_based_forecast(self, constraintList: list[Clusters.DeviceEnergyManagement.Structs.ConstraintsStruct], | ||
| cause: Clusters.DeviceEnergyManagement.Enums.AdjustmentCauseEnum, | ||
| endpoint: int = None, timedRequestTimeoutMs: int = 3000, | ||
| expected_status: Status = Status.Success): | ||
| try: | ||
| await self.send_single_cmd(cmd=Clusters.DeviceEnergyManagement.Commands.RequestConstraintBasedForecast(constraints=constraintList, | ||
| cause=cause), | ||
| endpoint=endpoint, | ||
| timedRequestTimeoutMs=timedRequestTimeoutMs) | ||
|
|
||
| asserts.assert_equal(expected_status, Status.Success) | ||
|
|
||
| except InteractionModelError as e: | ||
| asserts.assert_equal(e.status, expected_status, "Unexpected error returned") | ||
|
|
||
| def print_forecast(self, forecast): | ||
| for index, slot in enumerate(forecast.slots): | ||
| logging.info( | ||
| f" [{index}] MinDuration: {slot.minDuration} MaxDuration: {slot.maxDuration} DefaultDuration: {slot.defaultDuration}") | ||
| logging.info(f" ElapseSlotTime: {slot.elapsedSlotTime} RemainingSlotTime: {slot.remainingSlotTime}") | ||
| logging.info( | ||
| f" SlotIsPausable: {slot.slotIsPausable} MinPauseDuration: {slot.minPauseDuration} MaxPauseDuration: {slot.maxPauseDuration}") | ||
| logging.info(f" ManufacturerESAState: {slot.manufacturerESAState}") | ||
| logging.info(f" NominalPower: {slot.nominalPower} MinPower: {slot.minPower} MaxPower: {slot.maxPower}") | ||
| logging.info(f" MinPowerAdjustment: {slot.minPowerAdjustment} MaxPowerAdjustment: {slot.maxPowerAdjustment}") | ||
| logging.info( | ||
| f" MinDurationAdjustment: {slot.minDurationAdjustment} MaxDurationAdjustment: {slot.maxDurationAdjustment}") | ||
| if slot.costs is not None: | ||
| for cost_index, cost in enumerate(slot): | ||
| logging.info( | ||
| f" Cost: [{cost_index}] CostType:{cost.costType} Value: {cost.value} DecimalPoints: {cost.decimalPoints} Currency: {cost.currency}") | ||
|
|
||
| def get_current_utc_time_in_seconds(self): | ||
| microseconds_in_second = 1000000 | ||
| return int(utc_time_in_matter_epoch()/microseconds_in_second) | ||
|
|
||
| async def send_test_event_trigger_power_adjustment(self): | ||
| await self.send_test_event_triggers(eventTrigger=0x0098000000000000) | ||
|
|
||
| async def send_test_event_trigger_power_adjustment_clear(self): | ||
| await self.send_test_event_triggers(eventTrigger=0x0098000000000001) | ||
|
|
||
| async def send_test_event_trigger_user_opt_out_local(self): | ||
| await self.send_test_event_triggers(eventTrigger=0x0098000000000002) | ||
|
|
||
| async def send_test_event_trigger_user_opt_out_grid(self): | ||
| await self.send_test_event_triggers(eventTrigger=0x0098000000000003) | ||
|
|
||
| async def send_test_event_trigger_user_opt_out_clear_all(self): | ||
| await self.send_test_event_triggers(eventTrigger=0x0098000000000004) | ||
|
|
||
| async def send_test_event_trigger_start_time_adjustment(self): | ||
| await self.send_test_event_triggers(eventTrigger=0x0098000000000005) | ||
|
|
||
| async def send_test_event_trigger_start_time_adjustment_clear(self): | ||
| await self.send_test_event_triggers(eventTrigger=0x0098000000000006) | ||
|
|
||
| async def send_test_event_trigger_pausable(self): | ||
| await self.send_test_event_triggers(eventTrigger=0x0098000000000007) | ||
|
|
||
| async def send_test_event_trigger_pausable_next_slot(self): | ||
| await self.send_test_event_triggers(eventTrigger=0x0098000000000008) | ||
|
|
||
| async def send_test_event_trigger_pausable_clear(self): | ||
| await self.send_test_event_triggers(eventTrigger=0x0098000000000009) | ||
|
|
||
| async def send_test_event_trigger_forecast_adjustment(self): | ||
| await self.send_test_event_triggers(eventTrigger=0x009800000000000A) | ||
|
|
||
| async def send_test_event_trigger_forecast_adjustment_next_slot(self): | ||
| await self.send_test_event_triggers(eventTrigger=0x009800000000000B) | ||
|
|
||
| async def send_test_event_trigger_forecast_adjustment_clear(self): | ||
| await self.send_test_event_triggers(eventTrigger=0x009800000000000C) | ||
|
|
||
| async def send_test_event_trigger_constraint_based_adjustment(self): | ||
| await self.send_test_event_triggers(eventTrigger=0x009800000000000D) | ||
|
|
||
| async def send_test_event_trigger_constraint_based_adjustment_clear(self): | ||
| await self.send_test_event_triggers(eventTrigger=0x009800000000000E) | ||
|
|
||
| async def send_test_event_trigger_forecast(self): | ||
| await self.send_test_event_triggers(eventTrigger=0x009800000000000F) | ||
|
|
||
| async def send_test_event_trigger_forecast_clear(self): | ||
| await self.send_test_event_triggers(eventTrigger=0x0098000000000010) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.