Fix flux_led not generating unique ids when discovery fails (#65250)

This commit is contained in:
J. Nick Koston 2022-01-30 22:17:19 -06:00 committed by GitHub
parent 27d5be71dd
commit 73bd8db273
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
13 changed files with 268 additions and 60 deletions

View file

@ -14,7 +14,7 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, EVENT_HOMEASSISTANT_STARTED, Platform from homeassistant.const import CONF_HOST, EVENT_HOMEASSISTANT_STARTED, Platform
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.dispatcher import async_dispatcher_send from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.event import ( from homeassistant.helpers.event import (
async_track_time_change, async_track_time_change,
@ -88,6 +88,31 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
return True return True
async def _async_migrate_unique_ids(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Migrate entities when the mac address gets discovered."""
unique_id = entry.unique_id
if not unique_id:
return
entry_id = entry.entry_id
@callback
def _async_migrator(entity_entry: er.RegistryEntry) -> dict[str, Any] | None:
# Old format {entry_id}.....
# New format {unique_id}....
entity_unique_id = entity_entry.unique_id
if not entity_unique_id.startswith(entry_id):
return None
new_unique_id = f"{unique_id}{entity_unique_id[len(entry_id):]}"
_LOGGER.info(
"Migrating unique_id from [%s] to [%s]",
entity_unique_id,
new_unique_id,
)
return {"new_unique_id": new_unique_id}
await er.async_migrate_entries(hass, entry.entry_id, _async_migrator)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Flux LED/MagicLight from a config entry.""" """Set up Flux LED/MagicLight from a config entry."""
host = entry.data[CONF_HOST] host = entry.data[CONF_HOST]
@ -135,6 +160,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
# is either missing or we have verified it matches # is either missing or we have verified it matches
async_update_entry_from_discovery(hass, entry, discovery, device.model_num) async_update_entry_from_discovery(hass, entry, discovery, device.model_num)
await _async_migrate_unique_ids(hass, entry)
coordinator = FluxLedUpdateCoordinator(hass, device, entry) coordinator = FluxLedUpdateCoordinator(hass, device, entry)
hass.data[DOMAIN][entry.entry_id] = coordinator hass.data[DOMAIN][entry.entry_id] = coordinator
platforms = PLATFORMS_BY_TYPE[device.device_type] platforms = PLATFORMS_BY_TYPE[device.device_type]

View file

@ -64,8 +64,8 @@ class FluxButton(FluxBaseEntity, ButtonEntity):
self.entity_description = description self.entity_description = description
super().__init__(device, entry) super().__init__(device, entry)
self._attr_name = f"{entry.data[CONF_NAME]} {description.name}" self._attr_name = f"{entry.data[CONF_NAME]} {description.name}"
if entry.unique_id: base_unique_id = entry.unique_id or entry.entry_id
self._attr_unique_id = f"{entry.unique_id}_{description.key}" self._attr_unique_id = f"{base_unique_id}_{description.key}"
async def async_press(self) -> None: async def async_press(self) -> None:
"""Send out a command.""" """Send out a command."""

View file

@ -7,19 +7,28 @@ from typing import Any
from flux_led.aiodevice import AIOWifiLedBulb from flux_led.aiodevice import AIOWifiLedBulb
from homeassistant import config_entries from homeassistant import config_entries
from homeassistant.const import CONF_NAME from homeassistant.const import (
ATTR_CONNECTIONS,
ATTR_HW_VERSION,
ATTR_IDENTIFIERS,
ATTR_MANUFACTURER,
ATTR_MODEL,
ATTR_NAME,
ATTR_SW_VERSION,
CONF_NAME,
)
from homeassistant.core import callback from homeassistant.core import callback
from homeassistant.helpers import device_registry as dr from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import DeviceInfo, Entity from homeassistant.helpers.entity import DeviceInfo, Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import CONF_MINOR_VERSION, CONF_MODEL, SIGNAL_STATE_UPDATED from .const import CONF_MINOR_VERSION, CONF_MODEL, DOMAIN, SIGNAL_STATE_UPDATED
from .coordinator import FluxLedUpdateCoordinator from .coordinator import FluxLedUpdateCoordinator
def _async_device_info( def _async_device_info(
unique_id: str, device: AIOWifiLedBulb, entry: config_entries.ConfigEntry device: AIOWifiLedBulb, entry: config_entries.ConfigEntry
) -> DeviceInfo: ) -> DeviceInfo:
version_num = device.version_num version_num = device.version_num
if minor_version := entry.data.get(CONF_MINOR_VERSION): if minor_version := entry.data.get(CONF_MINOR_VERSION):
@ -27,14 +36,18 @@ def _async_device_info(
sw_version_str = f"{sw_version:0.2f}" sw_version_str = f"{sw_version:0.2f}"
else: else:
sw_version_str = str(device.version_num) sw_version_str = str(device.version_num)
return DeviceInfo( device_info: DeviceInfo = {
connections={(dr.CONNECTION_NETWORK_MAC, unique_id)}, ATTR_IDENTIFIERS: {(DOMAIN, entry.entry_id)},
manufacturer="Zengge", ATTR_MANUFACTURER: "Zengge",
model=device.model, ATTR_MODEL: device.model,
name=entry.data[CONF_NAME], ATTR_NAME: entry.data[CONF_NAME],
sw_version=sw_version_str, ATTR_SW_VERSION: sw_version_str,
hw_version=entry.data.get(CONF_MODEL), }
) if hw_model := entry.data.get(CONF_MODEL):
device_info[ATTR_HW_VERSION] = hw_model
if entry.unique_id:
device_info[ATTR_CONNECTIONS] = {(dr.CONNECTION_NETWORK_MAC, entry.unique_id)}
return device_info
class FluxBaseEntity(Entity): class FluxBaseEntity(Entity):
@ -50,10 +63,7 @@ class FluxBaseEntity(Entity):
"""Initialize the light.""" """Initialize the light."""
self._device: AIOWifiLedBulb = device self._device: AIOWifiLedBulb = device
self.entry = entry self.entry = entry
if entry.unique_id: self._attr_device_info = _async_device_info(self._device, entry)
self._attr_device_info = _async_device_info(
entry.unique_id, self._device, entry
)
class FluxEntity(CoordinatorEntity): class FluxEntity(CoordinatorEntity):
@ -64,7 +74,7 @@ class FluxEntity(CoordinatorEntity):
def __init__( def __init__(
self, self,
coordinator: FluxLedUpdateCoordinator, coordinator: FluxLedUpdateCoordinator,
unique_id: str | None, base_unique_id: str,
name: str, name: str,
key: str | None, key: str | None,
) -> None: ) -> None:
@ -74,13 +84,10 @@ class FluxEntity(CoordinatorEntity):
self._responding = True self._responding = True
self._attr_name = name self._attr_name = name
if key: if key:
self._attr_unique_id = f"{unique_id}_{key}" self._attr_unique_id = f"{base_unique_id}_{key}"
else: else:
self._attr_unique_id = unique_id self._attr_unique_id = base_unique_id
if unique_id: self._attr_device_info = _async_device_info(self._device, coordinator.entry)
self._attr_device_info = _async_device_info(
unique_id, self._device, coordinator.entry
)
async def _async_ensure_device_on(self) -> None: async def _async_ensure_device_on(self) -> None:
"""Turn the device on if it needs to be turned on before a command.""" """Turn the device on if it needs to be turned on before a command."""

View file

@ -177,7 +177,7 @@ async def async_setup_entry(
[ [
FluxLight( FluxLight(
coordinator, coordinator,
entry.unique_id, entry.unique_id or entry.entry_id,
entry.data[CONF_NAME], entry.data[CONF_NAME],
list(custom_effect_colors), list(custom_effect_colors),
options.get(CONF_CUSTOM_EFFECT_SPEED_PCT, DEFAULT_EFFECT_SPEED), options.get(CONF_CUSTOM_EFFECT_SPEED_PCT, DEFAULT_EFFECT_SPEED),
@ -195,14 +195,14 @@ class FluxLight(FluxOnOffEntity, CoordinatorEntity, LightEntity):
def __init__( def __init__(
self, self,
coordinator: FluxLedUpdateCoordinator, coordinator: FluxLedUpdateCoordinator,
unique_id: str | None, base_unique_id: str,
name: str, name: str,
custom_effect_colors: list[tuple[int, int, int]], custom_effect_colors: list[tuple[int, int, int]],
custom_effect_speed_pct: int, custom_effect_speed_pct: int,
custom_effect_transition: str, custom_effect_transition: str,
) -> None: ) -> None:
"""Initialize the light.""" """Initialize the light."""
super().__init__(coordinator, unique_id, name, None) super().__init__(coordinator, base_unique_id, name, None)
self._attr_min_mireds = color_temperature_kelvin_to_mired(self._device.max_temp) self._attr_min_mireds = color_temperature_kelvin_to_mired(self._device.max_temp)
self._attr_max_mireds = color_temperature_kelvin_to_mired(self._device.min_temp) self._attr_max_mireds = color_temperature_kelvin_to_mired(self._device.min_temp)
self._attr_supported_color_modes = _hass_color_modes(self._device) self._attr_supported_color_modes = _hass_color_modes(self._device)

View file

@ -51,26 +51,28 @@ async def async_setup_entry(
| FluxMusicSegmentsNumber | FluxMusicSegmentsNumber
] = [] ] = []
name = entry.data[CONF_NAME] name = entry.data[CONF_NAME]
unique_id = entry.unique_id base_unique_id = entry.unique_id or entry.entry_id
if device.pixels_per_segment is not None: if device.pixels_per_segment is not None:
entities.append( entities.append(
FluxPixelsPerSegmentNumber( FluxPixelsPerSegmentNumber(
coordinator, coordinator,
unique_id, base_unique_id,
f"{name} Pixels Per Segment", f"{name} Pixels Per Segment",
"pixels_per_segment", "pixels_per_segment",
) )
) )
if device.segments is not None: if device.segments is not None:
entities.append( entities.append(
FluxSegmentsNumber(coordinator, unique_id, f"{name} Segments", "segments") FluxSegmentsNumber(
coordinator, base_unique_id, f"{name} Segments", "segments"
)
) )
if device.music_pixels_per_segment is not None: if device.music_pixels_per_segment is not None:
entities.append( entities.append(
FluxMusicPixelsPerSegmentNumber( FluxMusicPixelsPerSegmentNumber(
coordinator, coordinator,
unique_id, base_unique_id,
f"{name} Music Pixels Per Segment", f"{name} Music Pixels Per Segment",
"music_pixels_per_segment", "music_pixels_per_segment",
) )
@ -78,12 +80,12 @@ async def async_setup_entry(
if device.music_segments is not None: if device.music_segments is not None:
entities.append( entities.append(
FluxMusicSegmentsNumber( FluxMusicSegmentsNumber(
coordinator, unique_id, f"{name} Music Segments", "music_segments" coordinator, base_unique_id, f"{name} Music Segments", "music_segments"
) )
) )
if device.effect_list and device.effect_list != [EFFECT_RANDOM]: if device.effect_list and device.effect_list != [EFFECT_RANDOM]:
entities.append( entities.append(
FluxSpeedNumber(coordinator, unique_id, f"{name} Effect Speed", None) FluxSpeedNumber(coordinator, base_unique_id, f"{name} Effect Speed", None)
) )
if entities: if entities:
@ -131,12 +133,12 @@ class FluxConfigNumber(FluxEntity, CoordinatorEntity, NumberEntity):
def __init__( def __init__(
self, self,
coordinator: FluxLedUpdateCoordinator, coordinator: FluxLedUpdateCoordinator,
unique_id: str | None, base_unique_id: str,
name: str, name: str,
key: str | None, key: str | None,
) -> None: ) -> None:
"""Initialize the flux number.""" """Initialize the flux number."""
super().__init__(coordinator, unique_id, name, key) super().__init__(coordinator, base_unique_id, name, key)
self._debouncer: Debouncer | None = None self._debouncer: Debouncer | None = None
self._pending_value: int | None = None self._pending_value: int | None = None

View file

@ -54,28 +54,28 @@ async def async_setup_entry(
| FluxWhiteChannelSelect | FluxWhiteChannelSelect
] = [] ] = []
name = entry.data[CONF_NAME] name = entry.data[CONF_NAME]
unique_id = entry.unique_id base_unique_id = entry.unique_id or entry.entry_id
if device.device_type == DeviceType.Switch: if device.device_type == DeviceType.Switch:
entities.append(FluxPowerStateSelect(coordinator.device, entry)) entities.append(FluxPowerStateSelect(coordinator.device, entry))
if device.operating_modes: if device.operating_modes:
entities.append( entities.append(
FluxOperatingModesSelect( FluxOperatingModesSelect(
coordinator, unique_id, f"{name} Operating Mode", "operating_mode" coordinator, base_unique_id, f"{name} Operating Mode", "operating_mode"
) )
) )
if device.wirings: if device.wirings:
entities.append( entities.append(
FluxWiringsSelect(coordinator, unique_id, f"{name} Wiring", "wiring") FluxWiringsSelect(coordinator, base_unique_id, f"{name} Wiring", "wiring")
) )
if device.ic_types: if device.ic_types:
entities.append( entities.append(
FluxICTypeSelect(coordinator, unique_id, f"{name} IC Type", "ic_type") FluxICTypeSelect(coordinator, base_unique_id, f"{name} IC Type", "ic_type")
) )
if device.remote_config: if device.remote_config:
entities.append( entities.append(
FluxRemoteConfigSelect( FluxRemoteConfigSelect(
coordinator, unique_id, f"{name} Remote Config", "remote_config" coordinator, base_unique_id, f"{name} Remote Config", "remote_config"
) )
) )
if FLUX_COLOR_MODE_RGBW in device.color_modes: if FLUX_COLOR_MODE_RGBW in device.color_modes:
@ -111,8 +111,8 @@ class FluxPowerStateSelect(FluxConfigAtStartSelect, SelectEntity):
"""Initialize the power state select.""" """Initialize the power state select."""
super().__init__(device, entry) super().__init__(device, entry)
self._attr_name = f"{entry.data[CONF_NAME]} Power Restored" self._attr_name = f"{entry.data[CONF_NAME]} Power Restored"
if entry.unique_id: base_unique_id = entry.unique_id or entry.entry_id
self._attr_unique_id = f"{entry.unique_id}_power_restored" self._attr_unique_id = f"{base_unique_id}_power_restored"
self._async_set_current_option_from_device() self._async_set_current_option_from_device()
@callback @callback
@ -201,12 +201,12 @@ class FluxRemoteConfigSelect(FluxConfigSelect):
def __init__( def __init__(
self, self,
coordinator: FluxLedUpdateCoordinator, coordinator: FluxLedUpdateCoordinator,
unique_id: str | None, base_unique_id: str,
name: str, name: str,
key: str, key: str,
) -> None: ) -> None:
"""Initialize the remote config type select.""" """Initialize the remote config type select."""
super().__init__(coordinator, unique_id, name, key) super().__init__(coordinator, base_unique_id, name, key)
assert self._device.remote_config is not None assert self._device.remote_config is not None
self._name_to_state = { self._name_to_state = {
_human_readable_option(option.name): option for option in RemoteConfig _human_readable_option(option.name): option for option in RemoteConfig
@ -238,8 +238,8 @@ class FluxWhiteChannelSelect(FluxConfigAtStartSelect):
"""Initialize the white channel select.""" """Initialize the white channel select."""
super().__init__(device, entry) super().__init__(device, entry)
self._attr_name = f"{entry.data[CONF_NAME]} White Channel" self._attr_name = f"{entry.data[CONF_NAME]} White Channel"
if entry.unique_id: base_unique_id = entry.unique_id or entry.entry_id
self._attr_unique_id = f"{entry.unique_id}_white_channel" self._attr_unique_id = f"{base_unique_id}_white_channel"
@property @property
def current_option(self) -> str | None: def current_option(self) -> str | None:

View file

@ -25,7 +25,7 @@ async def async_setup_entry(
[ [
FluxPairedRemotes( FluxPairedRemotes(
coordinator, coordinator,
entry.unique_id, entry.unique_id or entry.entry_id,
f"{entry.data[CONF_NAME]} Paired Remotes", f"{entry.data[CONF_NAME]} Paired Remotes",
"paired_remotes", "paired_remotes",
) )

View file

@ -34,18 +34,18 @@ async def async_setup_entry(
"""Set up the Flux lights.""" """Set up the Flux lights."""
coordinator: FluxLedUpdateCoordinator = hass.data[DOMAIN][entry.entry_id] coordinator: FluxLedUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
entities: list[FluxSwitch | FluxRemoteAccessSwitch | FluxMusicSwitch] = [] entities: list[FluxSwitch | FluxRemoteAccessSwitch | FluxMusicSwitch] = []
unique_id = entry.unique_id base_unique_id = entry.unique_id or entry.entry_id
name = entry.data[CONF_NAME] name = entry.data[CONF_NAME]
if coordinator.device.device_type == DeviceType.Switch: if coordinator.device.device_type == DeviceType.Switch:
entities.append(FluxSwitch(coordinator, unique_id, name, None)) entities.append(FluxSwitch(coordinator, base_unique_id, name, None))
if entry.data.get(CONF_REMOTE_ACCESS_HOST): if entry.data.get(CONF_REMOTE_ACCESS_HOST):
entities.append(FluxRemoteAccessSwitch(coordinator.device, entry)) entities.append(FluxRemoteAccessSwitch(coordinator.device, entry))
if coordinator.device.microphone: if coordinator.device.microphone:
entities.append( entities.append(
FluxMusicSwitch(coordinator, unique_id, f"{name} Music", "music") FluxMusicSwitch(coordinator, base_unique_id, f"{name} Music", "music")
) )
if entities: if entities:
@ -74,8 +74,8 @@ class FluxRemoteAccessSwitch(FluxBaseEntity, SwitchEntity):
"""Initialize the light.""" """Initialize the light."""
super().__init__(device, entry) super().__init__(device, entry)
self._attr_name = f"{entry.data[CONF_NAME]} Remote Access" self._attr_name = f"{entry.data[CONF_NAME]} Remote Access"
if entry.unique_id: base_unique_id = entry.unique_id or entry.entry_id
self._attr_unique_id = f"{entry.unique_id}_remote_access" self._attr_unique_id = f"{base_unique_id}_remote_access"
async def async_turn_on(self, **kwargs: Any) -> None: async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the remote access on.""" """Turn the remote access on."""

View file

@ -7,10 +7,16 @@ from unittest.mock import patch
import pytest import pytest
from homeassistant.components import flux_led from homeassistant.components import flux_led
from homeassistant.components.flux_led.const import DOMAIN from homeassistant.components.flux_led.const import (
CONF_REMOTE_ACCESS_ENABLED,
CONF_REMOTE_ACCESS_HOST,
CONF_REMOTE_ACCESS_PORT,
DOMAIN,
)
from homeassistant.config_entries import ConfigEntryState from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_HOST, CONF_NAME, EVENT_HOMEASSISTANT_STARTED from homeassistant.const import CONF_HOST, CONF_NAME, EVENT_HOMEASSISTANT_STARTED
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from homeassistant.util.dt import utcnow from homeassistant.util.dt import utcnow
@ -156,3 +162,46 @@ async def test_time_sync_startup_and_next_day(hass: HomeAssistant) -> None:
async_fire_time_changed(hass, utcnow() + timedelta(hours=24)) async_fire_time_changed(hass, utcnow() + timedelta(hours=24))
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(bulb.async_set_time.mock_calls) == 2 assert len(bulb.async_set_time.mock_calls) == 2
async def test_unique_id_migrate_when_mac_discovered(hass: HomeAssistant) -> None:
"""Test unique id migrated when mac discovered."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_REMOTE_ACCESS_HOST: "any",
CONF_REMOTE_ACCESS_ENABLED: True,
CONF_REMOTE_ACCESS_PORT: 1234,
CONF_HOST: IP_ADDRESS,
CONF_NAME: DEFAULT_ENTRY_TITLE,
},
)
config_entry.add_to_hass(hass)
bulb = _mocked_bulb()
with _patch_discovery(no_device=True), _patch_wifibulb(device=bulb):
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done()
assert not config_entry.unique_id
entity_registry = er.async_get(hass)
assert (
entity_registry.async_get("light.bulb_rgbcw_ddeeff").unique_id
== config_entry.entry_id
)
assert (
entity_registry.async_get("switch.bulb_rgbcw_ddeeff_remote_access").unique_id
== f"{config_entry.entry_id}_remote_access"
)
with _patch_discovery(), _patch_wifibulb(device=bulb):
await hass.config_entries.async_reload(config_entry.entry_id)
await hass.async_block_till_done()
assert (
entity_registry.async_get("light.bulb_rgbcw_ddeeff").unique_id
== config_entry.unique_id
)
assert (
entity_registry.async_get("switch.bulb_rgbcw_ddeeff_remote_access").unique_id
== f"{config_entry.unique_id}_remote_access"
)

View file

@ -137,8 +137,8 @@ async def test_light_goes_unavailable_and_recovers(hass: HomeAssistant) -> None:
assert state.state == STATE_ON assert state.state == STATE_ON
async def test_light_no_unique_id(hass: HomeAssistant) -> None: async def test_light_mac_address_not_found(hass: HomeAssistant) -> None:
"""Test a light without a unique id.""" """Test a light when we cannot discover the mac address."""
config_entry = MockConfigEntry( config_entry = MockConfigEntry(
domain=DOMAIN, data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE} domain=DOMAIN, data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE}
) )
@ -150,7 +150,7 @@ async def test_light_no_unique_id(hass: HomeAssistant) -> None:
entity_id = "light.bulb_rgbcw_ddeeff" entity_id = "light.bulb_rgbcw_ddeeff"
entity_registry = er.async_get(hass) entity_registry = er.async_get(hass)
assert entity_registry.async_get(entity_id) is None assert entity_registry.async_get(entity_id).unique_id == config_entry.entry_id
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON

View file

@ -41,7 +41,7 @@ from . import (
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
async def test_number_unique_id(hass: HomeAssistant) -> None: async def test_effects_speed_unique_id(hass: HomeAssistant) -> None:
"""Test a number unique id.""" """Test a number unique id."""
config_entry = MockConfigEntry( config_entry = MockConfigEntry(
domain=DOMAIN, domain=DOMAIN,
@ -59,6 +59,23 @@ async def test_number_unique_id(hass: HomeAssistant) -> None:
assert entity_registry.async_get(entity_id).unique_id == MAC_ADDRESS assert entity_registry.async_get(entity_id).unique_id == MAC_ADDRESS
async def test_effects_speed_unique_id_no_discovery(hass: HomeAssistant) -> None:
"""Test a number unique id."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
)
config_entry.add_to_hass(hass)
bulb = _mocked_bulb()
with _patch_discovery(no_device=True), _patch_wifibulb(device=bulb):
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done()
entity_id = "number.bulb_rgbcw_ddeeff_effect_speed"
entity_registry = er.async_get(hass)
assert entity_registry.async_get(entity_id).unique_id == config_entry.entry_id
async def test_rgb_light_effect_speed(hass: HomeAssistant) -> None: async def test_rgb_light_effect_speed(hass: HomeAssistant) -> None:
"""Test an rgb light with an effect.""" """Test an rgb light with an effect."""
config_entry = MockConfigEntry( config_entry = MockConfigEntry(

View file

@ -14,6 +14,7 @@ from homeassistant.components.flux_led.const import CONF_WHITE_CHANNEL_TYPE, DOM
from homeassistant.components.select import DOMAIN as SELECT_DOMAIN from homeassistant.components.select import DOMAIN as SELECT_DOMAIN
from homeassistant.const import ATTR_ENTITY_ID, ATTR_OPTION, CONF_HOST, CONF_NAME from homeassistant.const import ATTR_ENTITY_ID, ATTR_OPTION, CONF_HOST, CONF_NAME
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from . import ( from . import (
@ -67,6 +68,47 @@ async def test_switch_power_restore_state(hass: HomeAssistant) -> None:
) )
async def test_power_restored_unique_id(hass: HomeAssistant) -> None:
"""Test a select unique id."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
unique_id=MAC_ADDRESS,
)
config_entry.add_to_hass(hass)
switch = _mocked_switch()
with _patch_discovery(), _patch_wifibulb(device=switch):
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done()
entity_id = "select.bulb_rgbcw_ddeeff_power_restored"
entity_registry = er.async_get(hass)
assert (
entity_registry.async_get(entity_id).unique_id
== f"{MAC_ADDRESS}_power_restored"
)
async def test_power_restored_unique_id_no_discovery(hass: HomeAssistant) -> None:
"""Test a select unique id."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
)
config_entry.add_to_hass(hass)
switch = _mocked_switch()
with _patch_discovery(no_device=True), _patch_wifibulb(device=switch):
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done()
entity_id = "select.bulb_rgbcw_ddeeff_power_restored"
entity_registry = er.async_get(hass)
assert (
entity_registry.async_get(entity_id).unique_id
== f"{config_entry.entry_id}_power_restored"
)
async def test_select_addressable_strip_config(hass: HomeAssistant) -> None: async def test_select_addressable_strip_config(hass: HomeAssistant) -> None:
"""Test selecting addressable strip configs.""" """Test selecting addressable strip configs."""
config_entry = MockConfigEntry( config_entry = MockConfigEntry(

View file

@ -2,7 +2,12 @@
from flux_led.const import MODE_MUSIC from flux_led.const import MODE_MUSIC
from homeassistant.components import flux_led from homeassistant.components import flux_led
from homeassistant.components.flux_led.const import CONF_REMOTE_ACCESS_ENABLED, DOMAIN from homeassistant.components.flux_led.const import (
CONF_REMOTE_ACCESS_ENABLED,
CONF_REMOTE_ACCESS_HOST,
CONF_REMOTE_ACCESS_PORT,
DOMAIN,
)
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import ( from homeassistant.const import (
ATTR_ENTITY_ID, ATTR_ENTITY_ID,
@ -12,6 +17,7 @@ from homeassistant.const import (
STATE_ON, STATE_ON,
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from . import ( from . import (
@ -65,11 +71,69 @@ async def test_switch_on_off(hass: HomeAssistant) -> None:
assert hass.states.get(entity_id).state == STATE_ON assert hass.states.get(entity_id).state == STATE_ON
async def test_remote_access_unique_id(hass: HomeAssistant) -> None:
"""Test a remote access switch unique id."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_REMOTE_ACCESS_HOST: "any",
CONF_REMOTE_ACCESS_ENABLED: True,
CONF_REMOTE_ACCESS_PORT: 1234,
CONF_HOST: IP_ADDRESS,
CONF_NAME: DEFAULT_ENTRY_TITLE,
},
unique_id=MAC_ADDRESS,
)
config_entry.add_to_hass(hass)
bulb = _mocked_bulb()
with _patch_discovery(), _patch_wifibulb(device=bulb):
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done()
entity_id = "switch.bulb_rgbcw_ddeeff_remote_access"
entity_registry = er.async_get(hass)
assert (
entity_registry.async_get(entity_id).unique_id == f"{MAC_ADDRESS}_remote_access"
)
async def test_effects_speed_unique_id_no_discovery(hass: HomeAssistant) -> None:
"""Test a remote access switch unique id when discovery fails."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_REMOTE_ACCESS_HOST: "any",
CONF_REMOTE_ACCESS_ENABLED: True,
CONF_REMOTE_ACCESS_PORT: 1234,
CONF_HOST: IP_ADDRESS,
CONF_NAME: DEFAULT_ENTRY_TITLE,
},
)
config_entry.add_to_hass(hass)
bulb = _mocked_bulb()
with _patch_discovery(no_device=True), _patch_wifibulb(device=bulb):
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done()
entity_id = "switch.bulb_rgbcw_ddeeff_remote_access"
entity_registry = er.async_get(hass)
assert (
entity_registry.async_get(entity_id).unique_id
== f"{config_entry.entry_id}_remote_access"
)
async def test_remote_access_on_off(hass: HomeAssistant) -> None: async def test_remote_access_on_off(hass: HomeAssistant) -> None:
"""Test enable/disable remote access.""" """Test enable/disable remote access."""
config_entry = MockConfigEntry( config_entry = MockConfigEntry(
domain=DOMAIN, domain=DOMAIN,
data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE}, data={
CONF_REMOTE_ACCESS_HOST: "any",
CONF_REMOTE_ACCESS_ENABLED: True,
CONF_REMOTE_ACCESS_PORT: 1234,
CONF_HOST: IP_ADDRESS,
CONF_NAME: DEFAULT_ENTRY_TITLE,
},
unique_id=MAC_ADDRESS, unique_id=MAC_ADDRESS,
) )
config_entry.add_to_hass(hass) config_entry.add_to_hass(hass)