feat: Coordinator, Geraet und Licht-Entitaet
This commit is contained in:
Michael Sauer
@@ -3,11 +3,91 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Generator
|
||||
from copy import deepcopy
|
||||
from typing import Any
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from homeassistant.const import CONF_HOST, CONF_TOKEN
|
||||
from homeassistant.core import HomeAssistant
|
||||
import pytest
|
||||
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
||||
|
||||
from custom_components.ninlight.api import NinlightClient
|
||||
from custom_components.ninlight.const import DOMAIN
|
||||
from tests_unit.samples import (
|
||||
HOST_RGB,
|
||||
INFO_RGB,
|
||||
MAC_RGB,
|
||||
SETTINGS_COMPILED,
|
||||
STATE_RGB,
|
||||
TOKEN,
|
||||
)
|
||||
|
||||
CLIENT_PATCH_TARGETS = ["custom_components.ninlight.NinlightClient"]
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def auto_enable_custom_integrations(enable_custom_integrations: None) -> Generator[None]:
|
||||
"""Load custom_components/ in every Home Assistant test."""
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def lamp_state() -> dict[str, Any]:
|
||||
return deepcopy(STATE_RGB)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def lamp_info() -> dict[str, Any]:
|
||||
return deepcopy(INFO_RGB)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def lamp_settings() -> dict[str, Any]:
|
||||
return deepcopy(SETTINGS_COMPILED)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_client(
|
||||
lamp_state: dict[str, Any], lamp_info: dict[str, Any], lamp_settings: dict[str, Any]
|
||||
) -> Generator[MagicMock]:
|
||||
"""A lamp whose answers follow the mutable lamp_* fixtures."""
|
||||
client = MagicMock(spec=NinlightClient)
|
||||
client.host = HOST_RGB
|
||||
client.get_state = AsyncMock(side_effect=lambda: deepcopy(lamp_state))
|
||||
client.get_info = AsyncMock(side_effect=lambda: deepcopy(lamp_info))
|
||||
client.get_settings = AsyncMock(side_effect=lambda: deepcopy(lamp_settings))
|
||||
|
||||
async def set_state(**fields: Any) -> dict[str, Any]:
|
||||
lamp_state.update(fields)
|
||||
return deepcopy(lamp_state)
|
||||
|
||||
client.set_state = AsyncMock(side_effect=set_state)
|
||||
client.reboot = AsyncMock(return_value=None)
|
||||
client.fw_update = AsyncMock(return_value=None)
|
||||
patches = [patch(target, return_value=client) for target in CLIENT_PATCH_TARGETS]
|
||||
for p in patches:
|
||||
p.start()
|
||||
yield client
|
||||
for p in patches:
|
||||
p.stop()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config_entry() -> MockConfigEntry:
|
||||
return MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title="Klyqa GU10 RGB CW/WW 06b8",
|
||||
unique_id=MAC_RGB,
|
||||
data={CONF_HOST: HOST_RGB, CONF_TOKEN: TOKEN},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def init_integration(
|
||||
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_client: MagicMock
|
||||
) -> MockConfigEntry:
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
return mock_config_entry
|
||||
|
||||
@@ -0,0 +1,16 @@
|
||||
"""Helpers for Home Assistant tests."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from custom_components.ninlight.const import DOMAIN
|
||||
from tests_unit.samples import MAC_RGB
|
||||
|
||||
|
||||
def entity_id_for(hass: HomeAssistant, platform: str, key: str, mac: str = MAC_RGB) -> str:
|
||||
"""Look entities up by unique_id instead of guessing slugs."""
|
||||
entity_id = er.async_get(hass).async_get_entity_id(platform, DOMAIN, f"{mac}_{key}")
|
||||
assert entity_id is not None, f"no {platform} entity with key {key}"
|
||||
return entity_id
|
||||
@@ -0,0 +1,61 @@
|
||||
"""Coordinator behaviour that is not specific to one platform."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from freezegun.api import FrozenDateTimeFactory
|
||||
from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
|
||||
from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_ON, STATE_ON
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
import pytest
|
||||
from pytest_homeassistant_custom_component.common import async_fire_time_changed
|
||||
|
||||
from custom_components.ninlight.api import (
|
||||
NinlightBusyError,
|
||||
NinlightConnectionError,
|
||||
NinlightRequestError,
|
||||
)
|
||||
from tests.helpers import entity_id_for
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("error", "key", "placeholders"),
|
||||
[
|
||||
(
|
||||
NinlightRequestError(400, "invalid brightness"),
|
||||
"request_failed",
|
||||
{"message": "invalid brightness"},
|
||||
),
|
||||
(NinlightBusyError(409, "an update is already running"), "update_busy", None),
|
||||
(NinlightConnectionError("reset"), "cannot_connect", None),
|
||||
],
|
||||
)
|
||||
async def test_write_errors_are_translated(
|
||||
hass: HomeAssistant, init_integration, mock_client: MagicMock, error, key, placeholders
|
||||
) -> None:
|
||||
mock_client.set_state.side_effect = error
|
||||
with pytest.raises(HomeAssistantError) as raised:
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_ENTITY_ID: entity_id_for(hass, "light", "light")},
|
||||
blocking=True,
|
||||
)
|
||||
assert raised.value.translation_key == key
|
||||
assert raised.value.translation_placeholders == placeholders
|
||||
|
||||
|
||||
async def test_polling_pauses_while_an_update_runs(
|
||||
hass: HomeAssistant, init_integration, mock_client: MagicMock, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
init_integration.runtime_data.ota_in_progress = True
|
||||
mock_client.get_state.side_effect = NinlightConnectionError("rebooting")
|
||||
reads = mock_client.get_state.await_count
|
||||
freezer.tick(timedelta(seconds=31))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert mock_client.get_state.await_count == reads
|
||||
assert hass.states.get(entity_id_for(hass, "light", "light")).state == STATE_ON
|
||||
@@ -0,0 +1,101 @@
|
||||
"""Setup, unload, repair issues and the device registry."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from freezegun.api import FrozenDateTimeFactory
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.const import CONF_HOST, CONF_TOKEN
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
from homeassistant.helpers import issue_registry as ir
|
||||
from pytest_homeassistant_custom_component.common import MockConfigEntry, async_fire_time_changed
|
||||
|
||||
from custom_components.ninlight.api import NinlightConnectionError
|
||||
from custom_components.ninlight.const import DEFAULT_TOKEN, DOMAIN
|
||||
from custom_components.ninlight.entity import device_name
|
||||
from tests_unit.samples import HOST_RGB, INFO_CW, INFO_RGB, MAC_RGB
|
||||
|
||||
|
||||
def test_device_name_prefers_lamp_name() -> None:
|
||||
assert device_name(INFO_RGB) == "Klyqa GU10 RGB CW/WW 06b8"
|
||||
assert device_name(INFO_CW) == "Wohnzimmer"
|
||||
assert device_name({**INFO_RGB, "name": " ", "product_name": ""}) == "gu10_rgb 06b8"
|
||||
|
||||
|
||||
async def test_setup_and_unload(hass: HomeAssistant, init_integration: MockConfigEntry) -> None:
|
||||
assert init_integration.state is ConfigEntryState.LOADED
|
||||
device = dr.async_get(hass).async_get_device_by_identifier(
|
||||
(DOMAIN, MAC_RGB), init_integration.entry_id
|
||||
)
|
||||
assert device is not None
|
||||
assert (device.name, device.model, device.model_id, device.sw_version) == (
|
||||
"Klyqa GU10 RGB CW/WW 06b8",
|
||||
"Klyqa GU10 RGB CW/WW",
|
||||
"gu10_rgb",
|
||||
"0.3.8-36-gca9bb",
|
||||
)
|
||||
assert (dr.CONNECTION_NETWORK_MAC, MAC_RGB) in device.connections
|
||||
assert await hass.config_entries.async_unload(init_integration.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert init_integration.state is ConfigEntryState.NOT_LOADED
|
||||
|
||||
|
||||
async def test_unreachable_at_startup_retries(
|
||||
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_client: MagicMock
|
||||
) -> None:
|
||||
mock_client.get_state.side_effect = NinlightConnectionError("down")
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
async def test_default_token_raises_repair_issue(
|
||||
hass: HomeAssistant, mock_client: MagicMock
|
||||
) -> None:
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title="Lampe",
|
||||
unique_id=MAC_RGB,
|
||||
data={CONF_HOST: HOST_RGB, CONF_TOKEN: DEFAULT_TOKEN},
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
issue = ir.async_get(hass).async_get_issue(DOMAIN, f"default_token_{entry.entry_id}")
|
||||
assert issue is not None
|
||||
assert issue.translation_key == "default_token"
|
||||
assert issue.translation_placeholders == {"name": "Lampe"}
|
||||
|
||||
assert await hass.config_entries.async_remove(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert ir.async_get(hass).async_get_issue(DOMAIN, f"default_token_{entry.entry_id}") is None
|
||||
|
||||
|
||||
async def test_own_token_raises_no_issue(
|
||||
hass: HomeAssistant, init_integration: MockConfigEntry
|
||||
) -> None:
|
||||
assert (
|
||||
ir.async_get(hass).async_get_issue(DOMAIN, f"default_token_{init_integration.entry_id}")
|
||||
is None
|
||||
)
|
||||
|
||||
|
||||
async def test_device_registry_follows_firmware_version(
|
||||
hass: HomeAssistant,
|
||||
init_integration: MockConfigEntry,
|
||||
lamp_info: dict[str, Any],
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
lamp_info["fw_version"] = "0.3.8-47-g71a57"
|
||||
freezer.tick(timedelta(seconds=31))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
device = dr.async_get(hass).async_get_device_by_identifier(
|
||||
(DOMAIN, MAC_RGB), init_integration.entry_id
|
||||
)
|
||||
assert device.sw_version == "0.3.8-47-g71a57"
|
||||
@@ -0,0 +1,212 @@
|
||||
"""The light entity."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from copy import deepcopy
|
||||
from datetime import timedelta
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from freezegun.api import FrozenDateTimeFactory
|
||||
from homeassistant.components.light import (
|
||||
ATTR_BRIGHTNESS,
|
||||
ATTR_COLOR_MODE,
|
||||
ATTR_COLOR_TEMP_KELVIN,
|
||||
ATTR_MAX_COLOR_TEMP_KELVIN,
|
||||
ATTR_MIN_COLOR_TEMP_KELVIN,
|
||||
ATTR_RGB_COLOR,
|
||||
ATTR_SUPPORTED_COLOR_MODES,
|
||||
ATTR_TRANSITION,
|
||||
ColorMode,
|
||||
)
|
||||
from homeassistant.components.light import (
|
||||
DOMAIN as LIGHT_DOMAIN,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID,
|
||||
CONF_HOST,
|
||||
CONF_TOKEN,
|
||||
SERVICE_TURN_OFF,
|
||||
SERVICE_TURN_ON,
|
||||
STATE_OFF,
|
||||
STATE_ON,
|
||||
STATE_UNAVAILABLE,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from pytest_homeassistant_custom_component.common import MockConfigEntry, async_fire_time_changed
|
||||
|
||||
from custom_components.ninlight.api import NinlightConnectionError
|
||||
from custom_components.ninlight.const import DOMAIN
|
||||
from custom_components.ninlight.light import (
|
||||
ha_to_lamp_brightness,
|
||||
lamp_to_ha_brightness,
|
||||
transition_ms,
|
||||
)
|
||||
from tests.helpers import entity_id_for
|
||||
from tests_unit.samples import HOST_CW, INFO_CW, MAC_CW, STATE_CW, STATE_RGB, TOKEN
|
||||
|
||||
|
||||
def test_brightness_never_zero_and_round_trips() -> None:
|
||||
for ha_value in range(1, 256):
|
||||
assert 1 <= ha_to_lamp_brightness(ha_value) <= 100, ha_value
|
||||
for lamp_value in range(1, 101):
|
||||
assert 1 <= lamp_to_ha_brightness(lamp_value) <= 255, lamp_value
|
||||
assert ha_to_lamp_brightness(lamp_to_ha_brightness(lamp_value)) == lamp_value, lamp_value
|
||||
|
||||
|
||||
def test_transition_ms_rounds_and_clamps() -> None:
|
||||
assert transition_ms(2.5) == 2500
|
||||
assert transition_ms(0.0014) == 1
|
||||
assert transition_ms(-1) == 0
|
||||
assert transition_ms(90) == 60000
|
||||
|
||||
|
||||
async def test_state_attributes_of_color_lamp(hass: HomeAssistant, init_integration) -> None:
|
||||
state = hass.states.get(entity_id_for(hass, "light", "light"))
|
||||
assert state.state == STATE_ON
|
||||
assert state.attributes[ATTR_BRIGHTNESS] == 94 # lamp 37 %
|
||||
assert state.attributes[ATTR_COLOR_MODE] == ColorMode.RGB
|
||||
assert state.attributes[ATTR_RGB_COLOR] == (12, 200, 90)
|
||||
assert set(state.attributes[ATTR_SUPPORTED_COLOR_MODES]) == {
|
||||
ColorMode.RGB,
|
||||
ColorMode.COLOR_TEMP,
|
||||
}
|
||||
assert state.attributes[ATTR_MIN_COLOR_TEMP_KELVIN] == 2700
|
||||
assert state.attributes[ATTR_MAX_COLOR_TEMP_KELVIN] == 6500
|
||||
|
||||
|
||||
async def test_white_lamp_offers_only_color_temp(
|
||||
hass: HomeAssistant,
|
||||
mock_client: MagicMock,
|
||||
lamp_state: dict[str, Any],
|
||||
lamp_info: dict[str, Any],
|
||||
) -> None:
|
||||
lamp_state.clear()
|
||||
lamp_state.update(deepcopy(STATE_CW))
|
||||
lamp_info.clear()
|
||||
lamp_info.update(deepcopy(INFO_CW))
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN, unique_id=MAC_CW, data={CONF_HOST: HOST_CW, CONF_TOKEN: TOKEN}
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
state = hass.states.get(entity_id_for(hass, "light", "light", mac=MAC_CW))
|
||||
assert state.state == STATE_OFF
|
||||
assert state.attributes[ATTR_SUPPORTED_COLOR_MODES] == [ColorMode.COLOR_TEMP]
|
||||
assert state.attributes[ATTR_MIN_COLOR_TEMP_KELVIN] == 2000
|
||||
assert state.attributes[ATTR_MAX_COLOR_TEMP_KELVIN] == 5000
|
||||
|
||||
|
||||
async def test_turn_on_sends_everything_in_one_request(
|
||||
hass: HomeAssistant, init_integration, mock_client: MagicMock
|
||||
) -> None:
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
SERVICE_TURN_ON,
|
||||
{
|
||||
ATTR_ENTITY_ID: entity_id_for(hass, "light", "light"),
|
||||
ATTR_BRIGHTNESS: 200,
|
||||
ATTR_COLOR_TEMP_KELVIN: 4100,
|
||||
ATTR_TRANSITION: 2.5,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
mock_client.set_state.assert_awaited_once_with(
|
||||
on=True, brightness=78, color_temp=4100, transition=2500
|
||||
)
|
||||
|
||||
|
||||
async def test_turn_on_with_rgb_and_without_transition(
|
||||
hass: HomeAssistant, init_integration, mock_client: MagicMock
|
||||
) -> None:
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_ENTITY_ID: entity_id_for(hass, "light", "light"), ATTR_RGB_COLOR: (40, 10, 220)},
|
||||
blocking=True,
|
||||
)
|
||||
mock_client.set_state.assert_awaited_once_with(on=True, color={"r": 40, "g": 10, "b": 220})
|
||||
|
||||
|
||||
async def test_turn_off_clamps_transition(
|
||||
hass: HomeAssistant, init_integration, mock_client: MagicMock
|
||||
) -> None:
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
SERVICE_TURN_OFF,
|
||||
{ATTR_ENTITY_ID: entity_id_for(hass, "light", "light"), ATTR_TRANSITION: 90},
|
||||
blocking=True,
|
||||
)
|
||||
mock_client.set_state.assert_awaited_once_with(on=False, transition=60000)
|
||||
|
||||
|
||||
async def test_write_adopts_lamp_answer_without_reading_again(
|
||||
hass: HomeAssistant, init_integration, mock_client: MagicMock
|
||||
) -> None:
|
||||
async def lamp_clamps(**fields: Any) -> dict[str, Any]:
|
||||
return {**deepcopy(STATE_RGB), "mode": "white", "brightness": 55}
|
||||
|
||||
mock_client.set_state.side_effect = lamp_clamps
|
||||
reads_before = mock_client.get_state.await_count
|
||||
entity_id = entity_id_for(hass, "light", "light")
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS: 200},
|
||||
blocking=True,
|
||||
)
|
||||
state = hass.states.get(entity_id)
|
||||
assert state.attributes[ATTR_BRIGHTNESS] == 140 # the lamp's 55 %, not the requested 78 %
|
||||
assert state.attributes[ATTR_COLOR_MODE] == ColorMode.COLOR_TEMP
|
||||
assert mock_client.get_state.await_count == reads_before
|
||||
|
||||
|
||||
async def test_requests_to_one_lamp_never_overlap(
|
||||
hass: HomeAssistant, init_integration, mock_client: MagicMock, lamp_state: dict[str, Any]
|
||||
) -> None:
|
||||
active = 0
|
||||
peak = 0
|
||||
|
||||
async def slow(**fields: Any) -> dict[str, Any]:
|
||||
nonlocal active, peak
|
||||
active += 1
|
||||
peak = max(peak, active)
|
||||
await asyncio.sleep(0.01)
|
||||
active -= 1
|
||||
lamp_state.update(fields)
|
||||
return deepcopy(lamp_state)
|
||||
|
||||
mock_client.set_state.side_effect = slow
|
||||
entity_id = entity_id_for(hass, "light", "light")
|
||||
await asyncio.gather(
|
||||
*(
|
||||
hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS: value},
|
||||
blocking=True,
|
||||
)
|
||||
for value in (60, 120, 180)
|
||||
)
|
||||
)
|
||||
assert mock_client.set_state.await_count == 3
|
||||
assert peak == 1
|
||||
|
||||
|
||||
async def test_unreachable_lamp_becomes_unavailable_and_recovers(
|
||||
hass: HomeAssistant, init_integration, mock_client: MagicMock, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
entity_id = entity_id_for(hass, "light", "light")
|
||||
mock_client.get_state.side_effect = NinlightConnectionError("down")
|
||||
freezer.tick(timedelta(seconds=31))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
|
||||
|
||||
mock_client.get_state.side_effect = lambda: deepcopy(STATE_RGB)
|
||||
freezer.tick(timedelta(seconds=31))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_ON
|
||||
Reference in New Issue
Block a user