jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-19 06:47:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

Kulersky refactor to new Bluetooth subsystem (#142309)

Browse Source 
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Emily Love Watson 2025-04-14 08:38:34 -05:00 committed by GitHub
parent bc683ce6ee
commit d44d07ffcf
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 495 additions and 176 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@ -291,6 +291,7 @@ homeassistant.components.kaleidescape.*
homeassistant.components.knocki.*
homeassistant.components.knx.*
homeassistant.components.kraken.*
homeassistant.components.kulersky.*
homeassistant.components.lacrosse.*
homeassistant.components.lacrosse_view.*
homeassistant.components.lamarzocco.*

79
homeassistant/components/kulersky/__init__.py
View File

@ -1,21 +1,31 @@
"""Kuler Sky lights integration."""
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
import logging
from .const import DATA_ADDRESSES, DATA_DISCOVERY_SUBSCRIPTION, DOMAIN
from homeassistant.components.bluetooth import async_ble_device_from_address
from homeassistant.config_entries import SOURCE_INTEGRATION_DISCOVERY, ConfigEntry
from homeassistant.const import CONF_ADDRESS, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from .const import DOMAIN
PLATFORMS = [Platform.LIGHT]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Kuler Sky from a config entry."""
if DOMAIN not in hass.data:
hass.data[DOMAIN] = {}
if DATA_ADDRESSES not in hass.data[DOMAIN]:
hass.data[DOMAIN][DATA_ADDRESSES] = set()
ble_device = async_ble_device_from_address(
hass, entry.data[CONF_ADDRESS], connectable=True
)
if not ble_device:
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="cannot_connect",
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
@ -23,11 +33,48 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
# Stop discovery
unregister_discovery = hass.data[DOMAIN].pop(DATA_DISCOVERY_SUBSCRIPTION, None)
if unregister_discovery:
unregister_discovery()
hass.data.pop(DOMAIN, None)
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Migrate old entry."""
_LOGGER.debug("Migrating from version %s", config_entry.version)
# Version 1 was a single entry instance that started a bluetooth discovery
# thread to add devices. Version 2 has one config entry per device, and
# supports core bluetooth discovery
if config_entry.version == 1:
dev_reg = dr.async_get(hass)
devices = dev_reg.devices.get_devices_for_config_entry_id(config_entry.entry_id)
if len(devices) == 0:
_LOGGER.error("Unable to migrate; No devices registered")
return False
first_device = devices[0]
domain_identifiers = [i for i in first_device.identifiers if i[0] == DOMAIN]
address = next(iter(domain_identifiers))[1]
hass.config_entries.async_update_entry(
config_entry,
title=first_device.name or address,
data={CONF_ADDRESS: address},
unique_id=address,
version=2,
)
# Create new config flows for the remaining devices
for device in devices[1:]:
domain_identifiers = [i for i in device.identifiers if i[0] == DOMAIN]
address = next(iter(domain_identifiers))[1]
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_INTEGRATION_DISCOVERY},
data={CONF_ADDRESS: address},
)
)
_LOGGER.debug("Migration to version %s successful", config_entry.version)
return True

143
homeassistant/components/kulersky/config_flow.py
View File

@ -1,26 +1,143 @@
"""Config flow for Kuler Sky."""
import logging
from typing import Any
from bluetooth_data_tools import human_readable_name
import pykulersky
import voluptuous as vol
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_flow
from homeassistant.components.bluetooth import (
BluetoothServiceInfoBleak,
async_discovered_service_info,
async_last_service_info,
)
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_ADDRESS
from .const import DOMAIN
from .const import DOMAIN, EXPECTED_SERVICE_UUID
_LOGGER = logging.getLogger(__name__)
async def _async_has_devices(hass: HomeAssistant) -> bool:
"""Return if there are devices that can be discovered."""
# Check if there are any devices that can be discovered in the network.
try:
devices = await pykulersky.discover()
except pykulersky.PykulerskyException as exc:
_LOGGER.error("Unable to discover nearby Kuler Sky devices: %s", exc)
return False
return len(devices) > 0
class KulerskyConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Kulersky."""
VERSION = 2
config_entry_flow.register_discovery_flow(DOMAIN, "Kuler Sky", _async_has_devices)
def __init__(self) -> None:
"""Initialize the config flow."""
self._discovery_info: BluetoothServiceInfoBleak | None = None
self._discovered_devices: dict[str, BluetoothServiceInfoBleak] = {}
async def async_step_integration_discovery(
self, discovery_info: dict[str, str]
) -> ConfigFlowResult:
"""Handle the integration discovery step.
The old version of the integration used to have multiple
device in a single config entry. This is now deprecated.
The integration discovery step is used to create config
entries for each device beyond the first one.
"""
address: str = discovery_info[CONF_ADDRESS]
if service_info := async_last_service_info(self.hass, address):
title = human_readable_name(None, service_info.name, service_info.address)
else:
title = address
await self.async_set_unique_id(address)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=title,
data={CONF_ADDRESS: address},
)
async def async_step_bluetooth(
self, discovery_info: BluetoothServiceInfoBleak
) -> ConfigFlowResult:
"""Handle the bluetooth discovery step."""
await self.async_set_unique_id(discovery_info.address)
self._abort_if_unique_id_configured()
self._discovery_info = discovery_info
self.context["title_placeholders"] = {
"name": human_readable_name(
None, discovery_info.name, discovery_info.address
)
}
return await self.async_step_user()
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the user step to pick discovered device."""
errors: dict[str, str] = {}
if user_input is not None:
address = user_input[CONF_ADDRESS]
discovery_info = self._discovered_devices[address]
local_name = human_readable_name(
None, discovery_info.name, discovery_info.address
)
await self.async_set_unique_id(
discovery_info.address, raise_on_progress=False
)
self._abort_if_unique_id_configured()
kulersky_light = None
try:
kulersky_light = pykulersky.Light(discovery_info.address)
await kulersky_light.connect()
except pykulersky.PykulerskyException:
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected error")
errors["base"] = "unknown"
else:
return self.async_create_entry(
title=local_name,
data={
CONF_ADDRESS: discovery_info.address,
},
)
finally:
if kulersky_light:
await kulersky_light.disconnect()
if discovery := self._discovery_info:
self._discovered_devices[discovery.address] = discovery
else:
current_addresses = self._async_current_ids()
for discovery in async_discovered_service_info(self.hass):
if (
discovery.address in current_addresses
or discovery.address in self._discovered_devices
or EXPECTED_SERVICE_UUID not in discovery.service_uuids
):
continue
self._discovered_devices[discovery.address] = discovery
if not self._discovered_devices:
return self.async_abort(reason="no_devices_found")
if self._discovery_info:
data_schema = vol.Schema(
{vol.Required(CONF_ADDRESS): self._discovery_info.address}
)
else:
data_schema = vol.Schema(
{
vol.Required(CONF_ADDRESS): vol.In(
{
service_info.address: (
f"{service_info.name} ({service_info.address})"
)
for service_info in self._discovered_devices.values()
}
),
}
)
return self.async_show_form(
step_id="user",
data_schema=data_schema,
errors=errors,
)

2
homeassistant/components/kulersky/const.py
View File

@ -4,3 +4,5 @@ DOMAIN = "kulersky"
DATA_ADDRESSES = "addresses"
DATA_DISCOVERY_SUBSCRIPTION = "discovery_subscription"
EXPECTED_SERVICE_UUID = "8d96a001-0002-64c2-0001-9acc4838521c"

75
homeassistant/components/kulersky/light.py
View File

@ -2,12 +2,12 @@
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any
import pykulersky
from homeassistant.components import bluetooth
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_RGBW_COLOR,
@ -15,18 +15,15 @@ from homeassistant.components.light import (
LightEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.const import CONF_ADDRESS
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH, DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.event import async_track_time_interval
from .const import DATA_ADDRESSES, DATA_DISCOVERY_SUBSCRIPTION, DOMAIN
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
DISCOVERY_INTERVAL = timedelta(seconds=60)
async def async_setup_entry(
hass: HomeAssistant,
@ -34,32 +31,15 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Kuler sky light devices."""
async def discover(*args):
"""Attempt to discover new lights."""
lights = await pykulersky.discover()
# Filter out already discovered lights
new_lights = [
light
for light in lights
if light.address not in hass.data[DOMAIN][DATA_ADDRESSES]
]
new_entities = []
for light in new_lights:
hass.data[DOMAIN][DATA_ADDRESSES].add(light.address)
new_entities.append(KulerskyLight(light))
async_add_entities(new_entities, update_before_add=True)
# Start initial discovery
hass.async_create_task(discover())
# Perform recurring discovery of new devices
hass.data[DOMAIN][DATA_DISCOVERY_SUBSCRIPTION] = async_track_time_interval(
hass, discover, DISCOVERY_INTERVAL
ble_device = bluetooth.async_ble_device_from_address(
hass, config_entry.data[CONF_ADDRESS], connectable=True
)
entity = KulerskyLight(
config_entry.title,
config_entry.data[CONF_ADDRESS],
pykulersky.Light(ble_device),
)
async_add_entities([entity], update_before_add=True)
class KulerskyLight(LightEntity):
@ -71,37 +51,30 @@ class KulerskyLight(LightEntity):
_attr_supported_color_modes = {ColorMode.RGBW}
_attr_color_mode = ColorMode.RGBW
def __init__(self, light: pykulersky.Light) -> None:
def __init__(self, name: str, address: str, light: pykulersky.Light) -> None:
"""Initialize a Kuler Sky light."""
self._light = light
self._attr_unique_id = light.address
self._attr_unique_id = address
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, light.address)},
identifiers={(DOMAIN, address)},
connections={(CONNECTION_BLUETOOTH, address)},
manufacturer="Brightech",
name=light.name,
name=name,
)
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
self.async_on_remove(
self.hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_STOP, self.async_will_remove_from_hass
)
)
async def async_will_remove_from_hass(self, *args) -> None:
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
try:
await self._light.disconnect()
except pykulersky.PykulerskyException:
_LOGGER.debug(
"Exception disconnected from %s", self._light.address, exc_info=True
"Exception disconnected from %s", self._attr_unique_id, exc_info=True
)
@property
def is_on(self):
def is_on(self) -> bool | None:
"""Return true if light is on."""
return self.brightness > 0
return self.brightness is not None and self.brightness > 0
async def async_turn_on(self, **kwargs: Any) -> None:
"""Instruct the light to turn on."""
@ -133,11 +106,13 @@ class KulerskyLight(LightEntity):
rgbw = await self._light.get_color()
except pykulersky.PykulerskyException as exc:
if self._attr_available:
_LOGGER.warning("Unable to connect to %s: %s", self._light.address, exc)
_LOGGER.warning(
"Unable to connect to %s: %s", self._attr_unique_id, exc
)
self._attr_available = False
return
if self._attr_available is False:
_LOGGER.warning("Reconnected to %s", self._light.address)
_LOGGER.info("Reconnected to %s", self._attr_unique_id)
self._attr_available = True
brightness = max(rgbw)

6
homeassistant/components/kulersky/manifest.json
View File

@ -1,8 +1,14 @@
{
"domain": "kulersky",
"name": "Kuler Sky",
"bluetooth": [
{
"service_uuid": "8d96a001-0002-64c2-0001-9acc4838521c"
}
],
"codeowners": ["@emlove"],
"config_flow": true,
"dependencies": ["bluetooth_adapters"],
"documentation": "https://www.home-assistant.io/integrations/kulersky",
"iot_class": "local_polling",
"loggers": ["bleak", "pykulersky"],

16
homeassistant/components/kulersky/strings.json
View File

@ -1,13 +1,23 @@
{
"config": {
"step": {
"confirm": {
"description": "[%key:common::config_flow::description::confirm_setup%]"
"user": {
"data": {
"address": "[%key:common::config_flow::data::device%]"
}
}
},
"abort": {
"single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]",
"no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
}
},
"exceptions": {
"cannot_connect": {
"message": "[%key:common::config_flow::error::cannot_connect%]"
}
}
}

4
homeassistant/generated/bluetooth.py generated
View File

@ -404,6 +404,10 @@ BLUETOOTH: Final[list[dict[str, bool | str | int | list[int]]]] = [
"domain": "keymitt_ble",
"local_name": "mib*",
},
{
"domain": "kulersky",
"service_uuid": "8d96a001-0002-64c2-0001-9acc4838521c",
},
{
"domain": "lamarzocco",
"local_name": "MICRA_*",

10
mypy.ini generated
View File

@ -2666,6 +2666,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.kulersky.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.lacrosse.*]
check_untyped_defs = true
disallow_incomplete_defs = true

213
tests/components/kulersky/test_config_flow.py
View File

@ -1,105 +1,182 @@
"""Test the Kuler Sky config flow."""
from unittest.mock import MagicMock, patch
from unittest.mock import AsyncMock, Mock, patch
import pykulersky
from homeassistant import config_entries
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from homeassistant.components.kulersky.config_flow import DOMAIN
from homeassistant.config_entries import (
SOURCE_BLUETOOTH,
SOURCE_INTEGRATION_DISCOVERY,
SOURCE_USER,
)
from homeassistant.const import CONF_ADDRESS
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.components.bluetooth import generate_advertisement_data, generate_ble_device
async def test_flow_success(hass: HomeAssistant) -> None:
"""Test we get the form."""
KULERSKY_SERVICE_INFO = BluetoothServiceInfoBleak(
name="KulerLight",
manufacturer_data={},
service_data={},
service_uuids=["8d96a001-0002-64c2-0001-9acc4838521c"],
address="AA:BB:CC:DD:EE:FF",
rssi=-60,
source="local",
advertisement=generate_advertisement_data(
local_name="KulerLight",
manufacturer_data={},
service_data={},
service_uuids=["8d96a001-0002-64c2-0001-9acc4838521c"],
),
device=generate_ble_device("AA:BB:CC:DD:EE:FF", "KulerLight"),
time=0,
connectable=True,
tx_power=-127,
)
async def test_bluetooth_discovery(hass: HomeAssistant) -> None:
"""Test discovery via bluetooth with a valid device."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN,
context={"source": SOURCE_BLUETOOTH},
data=KULERSKY_SERVICE_INFO,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] is None
assert result["step_id"] == "user"
light = MagicMock(spec=pykulersky.Light)
light.address = "AA:BB:CC:11:22:33"
light.name = "Bedroom"
with (
patch(
"homeassistant.components.kulersky.config_flow.pykulersky.discover",
return_value=[light],
),
patch(
"homeassistant.components.kulersky.async_setup_entry",
return_value=True,
) as mock_setup_entry,
):
result2 = await hass.config_entries.flow.async_configure(
with patch("pykulersky.Light", Mock(return_value=AsyncMock())):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{},
{CONF_ADDRESS: "AA:BB:CC:DD:EE:FF"},
)
await hass.async_block_till_done()
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert result2["title"] == "Kuler Sky"
assert result2["data"] == {}
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "KulerLight (EEFF)"
assert result["data"] == {
CONF_ADDRESS: "AA:BB:CC:DD:EE:FF",
}
async def test_flow_no_devices_found(hass: HomeAssistant) -> None:
"""Test we get the form."""
async def test_integration_discovery(hass: HomeAssistant) -> None:
"""Test discovery via bluetooth with a valid device."""
with patch(
"homeassistant.components.kulersky.config_flow.async_last_service_info",
return_value=KULERSKY_SERVICE_INFO,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_INTEGRATION_DISCOVERY},
data={CONF_ADDRESS: "AA:BB:CC:DD:EE:FF"},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "KulerLight (EEFF)"
assert result["data"] == {
CONF_ADDRESS: "AA:BB:CC:DD:EE:FF",
}
async def test_integration_discovery_no_last_service_info(hass: HomeAssistant) -> None:
"""Test discovery via bluetooth with a valid device."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN,
context={"source": SOURCE_INTEGRATION_DISCOVERY},
data={CONF_ADDRESS: "AA:BB:CC:DD:EE:FF"},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "AA:BB:CC:DD:EE:FF"
assert result["data"] == {
CONF_ADDRESS: "AA:BB:CC:DD:EE:FF",
}
async def test_user_setup(hass: HomeAssistant) -> None:
"""Test the user manually setting up the integration."""
with patch(
"homeassistant.components.kulersky.config_flow.async_discovered_service_info",
return_value=[
KULERSKY_SERVICE_INFO,
KULERSKY_SERVICE_INFO,
], # Pass twice to test duplicate logic
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
with patch("pykulersky.Light", Mock(return_value=AsyncMock())):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_ADDRESS: "AA:BB:CC:DD:EE:FF"},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "KulerLight (EEFF)"
assert result["data"] == {
CONF_ADDRESS: "AA:BB:CC:DD:EE:FF",
}
async def test_user_setup_no_devices(hass: HomeAssistant) -> None:
"""Test the user manually setting up the integration."""
with patch(
"homeassistant.components.kulersky.config_flow.async_discovered_service_info",
return_value=[],
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "no_devices_found"
async def test_connection_error(hass: HomeAssistant) -> None:
"""Test a connection error trying to set up."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_BLUETOOTH},
data=KULERSKY_SERVICE_INFO,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] is None
assert result["step_id"] == "user"
with (
patch(
"homeassistant.components.kulersky.config_flow.pykulersky.discover",
return_value=[],
),
patch(
"homeassistant.components.kulersky.async_setup_entry",
return_value=True,
) as mock_setup_entry,
):
result2 = await hass.config_entries.flow.async_configure(
with patch("pykulersky.Light", Mock(side_effect=pykulersky.PykulerskyException)):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{},
{CONF_ADDRESS: "AA:BB:CC:DD:EE:FF"},
)
await hass.async_block_till_done()
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "no_devices_found"
assert len(mock_setup_entry.mock_calls) == 0
assert result["type"] is FlowResultType.FORM
assert result["errors"]["base"] == "cannot_connect"
async def test_flow_exceptions_caught(hass: HomeAssistant) -> None:
"""Test we get the form."""
async def test_unexpected_error(hass: HomeAssistant) -> None:
"""Test an unexpected error trying to set up."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN,
context={"source": SOURCE_BLUETOOTH},
data=KULERSKY_SERVICE_INFO,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] is None
assert result["step_id"] == "user"
with (
patch(
"homeassistant.components.kulersky.config_flow.pykulersky.discover",
side_effect=pykulersky.PykulerskyException("TEST"),
),
patch(
"homeassistant.components.kulersky.async_setup_entry",
return_value=True,
) as mock_setup_entry,
):
result2 = await hass.config_entries.flow.async_configure(
with patch("pykulersky.Light", Mock(side_effect=Exception)):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{},
{CONF_ADDRESS: "AA:BB:CC:DD:EE:FF"},
)
await hass.async_block_till_done()
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "no_devices_found"
assert len(mock_setup_entry.mock_calls) == 0
assert result["type"] is FlowResultType.FORM
assert result["errors"]["base"] == "unknown"

65
tests/components/kulersky/test_init.py Normal file
View File

@ -0,0 +1,65 @@
"""Tests for init methods."""
from homeassistant.components.kulersky.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_ADDRESS
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from tests.common import MockConfigEntry
async def test_migrate_entry(
hass: HomeAssistant,
) -> None:
"""Test migrate config entry from v1 to v2."""
mock_config_entry_v1 = MockConfigEntry(
version=1,
domain=DOMAIN,
title="KulerSky",
)
mock_config_entry_v1.add_to_hass(hass)
dev_reg = dr.async_get(hass)
# Create device registry entries for old integration
dev_reg.async_get_or_create(
config_entry_id=mock_config_entry_v1.entry_id,
identifiers={(DOMAIN, "AA:BB:CC:11:22:33")},
name="KuLight 1",
)
dev_reg.async_get_or_create(
config_entry_id=mock_config_entry_v1.entry_id,
identifiers={(DOMAIN, "AA:BB:CC:44:55:66")},
name="KuLight 2",
)
await hass.config_entries.async_setup(mock_config_entry_v1.entry_id)
await hass.async_block_till_done()
assert mock_config_entry_v1.state is ConfigEntryState.SETUP_RETRY
assert mock_config_entry_v1.version == 2
assert mock_config_entry_v1.unique_id == "AA:BB:CC:11:22:33"
assert mock_config_entry_v1.data == {
CONF_ADDRESS: "AA:BB:CC:11:22:33",
}
async def test_migrate_entry_no_devices_found(
hass: HomeAssistant,
) -> None:
"""Test migrate config entry from v1 to v2."""
mock_config_entry_v1 = MockConfigEntry(
version=1,
domain=DOMAIN,
title="KulerSky",
)
mock_config_entry_v1.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry_v1.entry_id)
await hass.async_block_till_done()
assert mock_config_entry_v1.state is ConfigEntryState.MIGRATION_ERROR
assert mock_config_entry_v1.version == 1

57
tests/components/kulersky/test_light.py
View File

@ -1,16 +1,13 @@
"""Test the Kuler Sky lights."""
from collections.abc import AsyncGenerator
from unittest.mock import MagicMock, patch
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch
from bleak.backends.device import BLEDevice
import pykulersky
import pytest
from homeassistant.components.kulersky.const import (
DATA_ADDRESSES,
DATA_DISCOVERY_SUBSCRIPTION,
DOMAIN,
)
from homeassistant.components.kulersky.const import DOMAIN
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_COLOR_MODE,
@ -26,6 +23,7 @@ from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_FRIENDLY_NAME,
ATTR_SUPPORTED_FEATURES,
CONF_ADDRESS,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
@ -37,26 +35,43 @@ from homeassistant.util import dt as dt_util
from tests.common import MockConfigEntry, async_fire_time_changed
@pytest.fixture
def mock_ble_device() -> Generator[MagicMock]:
"""Mock BLEDevice."""
with patch(
"homeassistant.components.kulersky.async_ble_device_from_address",
return_value=BLEDevice(
address="AA:BB:CC:11:22:33", name="Bedroom", rssi=-50, details={}
),
) as ble_device:
yield ble_device
@pytest.fixture
async def mock_entry() -> MockConfigEntry:
"""Create a mock light entity."""
return MockConfigEntry(domain=DOMAIN)
return MockConfigEntry(
domain=DOMAIN,
data={CONF_ADDRESS: "AA:BB:CC:11:22:33"},
title="Bedroom",
version=2,
)
@pytest.fixture
async def mock_light(
hass: HomeAssistant, mock_entry: MockConfigEntry
) -> AsyncGenerator[MagicMock]:
"""Create a mock light entity."""
light = MagicMock(spec=pykulersky.Light)
hass: HomeAssistant, mock_entry: MockConfigEntry, mock_ble_device: MagicMock
) -> Generator[AsyncMock]:
"""Mock pykulersky light."""
light = AsyncMock()
light.address = "AA:BB:CC:11:22:33"
light.name = "Bedroom"
light.connect.return_value = True
light.get_color.return_value = (0, 0, 0, 0)
with patch(
"homeassistant.components.kulersky.light.pykulersky.discover",
return_value=[light],
"pykulersky.Light",
return_value=light,
):
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
@ -67,7 +82,7 @@ async def mock_light(
yield light
async def test_init(hass: HomeAssistant, mock_light: MagicMock) -> None:
async def test_init(hass: HomeAssistant, mock_light: AsyncMock) -> None:
"""Test platform setup."""
state = hass.states.get("light.bedroom")
assert state.state == STATE_OFF
@ -83,24 +98,14 @@ async def test_init(hass: HomeAssistant, mock_light: MagicMock) -> None:
ATTR_RGBW_COLOR: None,
}
with patch.object(hass.loop, "stop"):
await hass.async_stop()
await hass.async_block_till_done()
assert mock_light.disconnect.called
async def test_remove_entry(
hass: HomeAssistant, mock_light: MagicMock, mock_entry: MockConfigEntry
) -> None:
"""Test platform setup."""
assert hass.data[DOMAIN][DATA_ADDRESSES] == {"AA:BB:CC:11:22:33"}
assert DATA_DISCOVERY_SUBSCRIPTION in hass.data[DOMAIN]
await hass.config_entries.async_remove(mock_entry.entry_id)
assert mock_light.disconnect.called
assert DOMAIN not in hass.data
async def test_remove_entry_exceptions_caught(