core/tests/components/zha/test_light.py

"""Test zha light."""
import asyncio
from unittest.mock import MagicMock, call, patch, sentinel

import zigpy.profiles.zha
import zigpy.types
import zigpy.zcl.clusters.general as general
import zigpy.zcl.foundation as zcl_f

from homeassistant.components.light import DOMAIN
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE

from .common import (
    async_enable_traffic,
    async_init_zigpy_device,
    async_test_device_join,
    find_entity_id,
    make_attribute,
    make_zcl_header,
)

from tests.common import mock_coro

ON = 1
OFF = 0


async def test_light(hass, config_entry, zha_gateway, monkeypatch):
    """Test zha light platform."""

    # create zigpy devices
    zigpy_device_on_off = await async_init_zigpy_device(
        hass,
        [general.OnOff.cluster_id, general.Basic.cluster_id],
        [],
        zigpy.profiles.zha.DeviceType.ON_OFF_LIGHT,
        zha_gateway,
        ieee="00:0d:6f:11:0a:90:69:e6",
    )

    zigpy_device_level = await async_init_zigpy_device(
        hass,
        [
            general.OnOff.cluster_id,
            general.LevelControl.cluster_id,
            general.Basic.cluster_id,
        ],
        [],
        zigpy.profiles.zha.DeviceType.ON_OFF_LIGHT,
        zha_gateway,
        ieee="00:0d:6f:11:0a:90:69:e7",
        manufacturer="FakeLevelManufacturer",
        model="FakeLevelModel",
    )

    # load up light domain
    await hass.config_entries.async_forward_entry_setup(config_entry, DOMAIN)
    await hass.async_block_till_done()

    # on off light
    on_off_device_on_off_cluster = zigpy_device_on_off.endpoints.get(1).on_off
    on_off_zha_device = zha_gateway.get_device(zigpy_device_on_off.ieee)
    on_off_entity_id = await find_entity_id(DOMAIN, on_off_zha_device, hass)
    assert on_off_entity_id is not None

    # dimmable light
    level_device_on_off_cluster = zigpy_device_level.endpoints.get(1).on_off
    level_device_level_cluster = zigpy_device_level.endpoints.get(1).level
    on_off_mock = MagicMock(
        side_effect=asyncio.coroutine(
            MagicMock(return_value=[sentinel.data, zcl_f.Status.SUCCESS])
        )
    )
    level_mock = MagicMock(
        side_effect=asyncio.coroutine(
            MagicMock(return_value=[sentinel.data, zcl_f.Status.SUCCESS])
        )
    )
    monkeypatch.setattr(level_device_on_off_cluster, "request", on_off_mock)
    monkeypatch.setattr(level_device_level_cluster, "request", level_mock)
    level_zha_device = zha_gateway.get_device(zigpy_device_level.ieee)
    level_entity_id = await find_entity_id(DOMAIN, level_zha_device, hass)
    assert level_entity_id is not None

    # test that the lights were created and that they are unavailable
    assert hass.states.get(on_off_entity_id).state == STATE_UNAVAILABLE
    assert hass.states.get(level_entity_id).state == STATE_UNAVAILABLE

    # allow traffic to flow through the gateway and device
    await async_enable_traffic(hass, zha_gateway, [on_off_zha_device, level_zha_device])

    # test that the lights were created and are off
    assert hass.states.get(on_off_entity_id).state == STATE_OFF
    assert hass.states.get(level_entity_id).state == STATE_OFF

    # test turning the lights on and off from the light
    await async_test_on_off_from_light(
        hass, on_off_device_on_off_cluster, on_off_entity_id
    )

    await async_test_on_off_from_light(
        hass, level_device_on_off_cluster, level_entity_id
    )

    # test turning the lights on and off from the HA
    await async_test_on_off_from_hass(
        hass, on_off_device_on_off_cluster, on_off_entity_id
    )

    await async_test_level_on_off_from_hass(
        hass, level_device_on_off_cluster, level_device_level_cluster, level_entity_id
    )

    # test turning the lights on and off from the light
    await async_test_on_from_light(hass, level_device_on_off_cluster, level_entity_id)

    # test getting a brightness change from the network
    await async_test_dimmer_from_light(
        hass, level_device_level_cluster, level_entity_id, 150, STATE_ON
    )

    # test adding a new light to the network and HA
    await async_test_device_join(
        hass,
        zha_gateway,
        general.OnOff.cluster_id,
        on_off_entity_id,
        device_type=zigpy.profiles.zha.DeviceType.ON_OFF_LIGHT,
    )


async def async_test_on_off_from_light(hass, cluster, entity_id):
    """Test on off functionality from the light."""
    # turn on at light
    attr = make_attribute(0, 1)
    hdr = make_zcl_header(zcl_f.Command.Report_Attributes)
    cluster.handle_message(hdr, [[attr]])
    await hass.async_block_till_done()
    assert hass.states.get(entity_id).state == STATE_ON

    # turn off at light
    attr.value.value = 0
    cluster.handle_message(hdr, [[attr]])
    await hass.async_block_till_done()
    assert hass.states.get(entity_id).state == STATE_OFF


async def async_test_on_from_light(hass, cluster, entity_id):
    """Test on off functionality from the light."""
    # turn on at light
    attr = make_attribute(0, 1)
    hdr = make_zcl_header(zcl_f.Command.Report_Attributes)
    cluster.handle_message(hdr, [[attr]])
    await hass.async_block_till_done()
    assert hass.states.get(entity_id).state == STATE_ON


async def async_test_on_off_from_hass(hass, cluster, entity_id):
    """Test on off functionality from hass."""
    with patch(
        "zigpy.zcl.Cluster.request",
        return_value=mock_coro([0x00, zcl_f.Status.SUCCESS]),
    ):
        # turn on via UI
        await hass.services.async_call(
            DOMAIN, "turn_on", {"entity_id": entity_id}, blocking=True
        )
        assert cluster.request.call_count == 1
        assert cluster.request.call_args == call(
            False, ON, (), expect_reply=True, manufacturer=None
        )

    await async_test_off_from_hass(hass, cluster, entity_id)


async def async_test_off_from_hass(hass, cluster, entity_id):
    """Test turning off the light from Home Assistant."""
    with patch(
        "zigpy.zcl.Cluster.request",
        return_value=mock_coro([0x01, zcl_f.Status.SUCCESS]),
    ):
        # turn off via UI
        await hass.services.async_call(
            DOMAIN, "turn_off", {"entity_id": entity_id}, blocking=True
        )
        assert cluster.request.call_count == 1
        assert cluster.request.call_args == call(
            False, OFF, (), expect_reply=True, manufacturer=None
        )


async def async_test_level_on_off_from_hass(
    hass, on_off_cluster, level_cluster, entity_id
):
    """Test on off functionality from hass."""

    # turn on via UI
    await hass.services.async_call(
        DOMAIN, "turn_on", {"entity_id": entity_id}, blocking=True
    )
    assert on_off_cluster.request.call_count == 1
    assert level_cluster.request.call_count == 0
    assert on_off_cluster.request.call_args == call(
        False, 1, (), expect_reply=True, manufacturer=None
    )
    on_off_cluster.request.reset_mock()
    level_cluster.request.reset_mock()

    await hass.services.async_call(
        DOMAIN, "turn_on", {"entity_id": entity_id, "transition": 10}, blocking=True
    )
    assert on_off_cluster.request.call_count == 1
    assert level_cluster.request.call_count == 1
    assert on_off_cluster.request.call_args == call(
        False, 1, (), expect_reply=True, manufacturer=None
    )
    assert level_cluster.request.call_args == call(
        False,
        4,
        (zigpy.types.uint8_t, zigpy.types.uint16_t),
        254,
        100.0,
        expect_reply=True,
        manufacturer=None,
    )
    on_off_cluster.request.reset_mock()
    level_cluster.request.reset_mock()

    await hass.services.async_call(
        DOMAIN, "turn_on", {"entity_id": entity_id, "brightness": 10}, blocking=True
    )
    assert on_off_cluster.request.call_count == 1
    assert level_cluster.request.call_count == 1
    assert on_off_cluster.request.call_args == call(
        False, 1, (), expect_reply=True, manufacturer=None
    )
    assert level_cluster.request.call_args == call(
        False,
        4,
        (zigpy.types.uint8_t, zigpy.types.uint16_t),
        10,
        0,
        expect_reply=True,
        manufacturer=None,
    )
    on_off_cluster.request.reset_mock()
    level_cluster.request.reset_mock()

    await async_test_off_from_hass(hass, on_off_cluster, entity_id)


async def async_test_dimmer_from_light(hass, cluster, entity_id, level, expected_state):
    """Test dimmer functionality from the light."""
    attr = make_attribute(0, level)
    hdr = make_zcl_header(zcl_f.Command.Report_Attributes)
    cluster.handle_message(hdr, [[attr]])
    await hass.async_block_till_done()
    assert hass.states.get(entity_id).state == expected_state
    # hass uses None for brightness of 0 in state attributes
    if level == 0:
        level = None
    assert hass.states.get(entity_id).attributes.get("brightness") == level