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.
Merge branch 'master' into feature/add-validation-logic
- 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.