TC_DEMTestBase.py

#
#    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 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")

    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)