evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add support for enabling/disabling cloud access in flux_led (#61138)

Browse source 
Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
This commit is contained in:
J. Nick Koston 2021-12-19 00:59:16 -06:00 • committed by GitHub
parent d7c5e41802
commit a6b680cd32
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
13 changed files with 428 additions and 111 deletions
Download patch file Download diff file Expand all files Collapse all files

74
homeassistant/components/flux_led/__init__.py
View file

@ -3,21 +3,14 @@ from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any, Final
from typing import Any, Final, cast
from flux_led import DeviceType
from flux_led.aio import AIOWifiLedBulb
from flux_led.const import ATTR_ID
from flux_led.scanner import FluxLEDDiscovery
from homeassistant import config_entries
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
EVENT_HOMEASSISTANT_STARTED,
Platform,
)
from homeassistant.const import CONF_HOST, EVENT_HOMEASSISTANT_STARTED, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
@ -36,16 +29,18 @@ from .const import (
STARTUP_SCAN_TIMEOUT,
)
from .discovery import (
async_clear_discovery_cache,
async_discover_device,
async_discover_devices,
async_name_from_discovery,
async_get_discovery,
async_trigger_discovery,
async_update_entry_from_discovery,
)
_LOGGER = logging.getLogger(__name__)
PLATFORMS_BY_TYPE: Final = {
DeviceType.Bulb: [Platform.LIGHT, Platform.NUMBER],
DeviceType.Bulb: [Platform.LIGHT, Platform.NUMBER, Platform.SWITCH],
DeviceType.Switch: [Platform.SWITCH],
}
DISCOVERY_INTERVAL: Final = timedelta(minutes=15)
@ -58,22 +53,6 @@ def async_wifi_bulb_for_host(host: str) -> AIOWifiLedBulb:
return AIOWifiLedBulb(host)
@callback
def async_update_entry_from_discovery(
hass: HomeAssistant, entry: config_entries.ConfigEntry, device: FluxLEDDiscovery
) -> None:
"""Update a config entry from a flux_led discovery."""
name = async_name_from_discovery(device)
mac_address = device[ATTR_ID]
assert mac_address is not None
hass.config_entries.async_update_entry(
entry,
data={**entry.data, CONF_NAME: name},
title=name,
unique_id=dr.format_mac(mac_address),
)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the flux_led component."""
domain_data = hass.data.setdefault(DOMAIN, {})
@ -92,18 +71,9 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
return True
async def async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Update listener."""
await hass.config_entries.async_reload(entry.entry_id)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Flux LED/MagicLight from a config entry."""
host = entry.data[CONF_HOST]
if not entry.unique_id:
if discovery := await async_discover_device(hass, host):
async_update_entry_from_discovery(hass, entry, discovery)
device: AIOWifiLedBulb = async_wifi_bulb_for_host(host)
signal = SIGNAL_STATE_UPDATED.format(device.ipaddr)
@ -119,11 +89,32 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
str(ex) or f"Timed out trying to connect to {device.ipaddr}"
) from ex
coordinator = FluxLedUpdateCoordinator(hass, device)
# UDP probe after successful connect only
directed_discovery = None
if discovery := async_get_discovery(hass, host):
directed_discovery = False
elif discovery := await async_discover_device(hass, host):
directed_discovery = True
if discovery:
if entry.unique_id:
assert discovery[ATTR_ID] is not None
mac = dr.format_mac(cast(str, discovery[ATTR_ID]))
if mac != entry.unique_id:
# The device is offline and another flux_led device is now using the ip address
raise ConfigEntryNotReady(
f"Unexpected device found at {host}; Expected {entry.unique_id}, found {mac}"
)
if directed_discovery:
# Only update the entry once we have verified the unique id
# is either missing or we have verified it matches
async_update_entry_from_discovery(hass, entry, discovery)
device.discovery = discovery
coordinator = FluxLedUpdateCoordinator(hass, device, entry)
hass.data[DOMAIN][entry.entry_id] = coordinator
platforms = PLATFORMS_BY_TYPE[device.device_type]
hass.config_entries.async_setup_platforms(entry, platforms)
entry.async_on_unload(entry.add_update_listener(async_update_listener))
return True
@ -133,6 +124,8 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
device: AIOWifiLedBulb = hass.data[DOMAIN][entry.entry_id].device
platforms = PLATFORMS_BY_TYPE[device.device_type]
if unload_ok := await hass.config_entries.async_unload_platforms(entry, platforms):
# Make sure we probe the device again in case something has changed externally
async_clear_discovery_cache(hass, entry.data[CONF_HOST])
del hass.data[DOMAIN][entry.entry_id]
await device.async_stop()
return unload_ok
@ -142,12 +135,11 @@ class FluxLedUpdateCoordinator(DataUpdateCoordinator):
"""DataUpdateCoordinator to gather data for a specific flux_led device."""
def __init__(
self,
hass: HomeAssistant,
device: AIOWifiLedBulb,
self, hass: HomeAssistant, device: AIOWifiLedBulb, entry: ConfigEntry
) -> None:
"""Initialize DataUpdateCoordinator to gather data for specific device."""
self.device = device
self.entry = entry
super().__init__(
hass,
_LOGGER,

30
homeassistant/components/flux_led/config_flow.py
View file

@ -10,13 +10,13 @@ import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import dhcp
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_MODE, CONF_NAME, CONF_PROTOCOL
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_NAME, CONF_PROTOCOL
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.typing import DiscoveryInfoType
from . import async_update_entry_from_discovery, async_wifi_bulb_for_host
from . import async_wifi_bulb_for_host
from .const import (
CONF_CUSTOM_EFFECT_COLORS,
CONF_CUSTOM_EFFECT_SPEED_PCT,
@ -33,6 +33,8 @@ from .discovery import (
async_discover_device,
async_discover_devices,
async_name_from_discovery,
async_populate_data_from_discovery,
async_update_entry_from_discovery,
)
CONF_DEVICE: Final = "device"
@ -73,7 +75,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
CONF_PROTOCOL: user_input.get(CONF_PROTOCOL),
},
options={
CONF_MODE: user_input[CONF_MODE],
CONF_CUSTOM_EFFECT_COLORS: user_input[CONF_CUSTOM_EFFECT_COLORS],
CONF_CUSTOM_EFFECT_SPEED_PCT: user_input[CONF_CUSTOM_EFFECT_SPEED_PCT],
CONF_CUSTOM_EFFECT_TRANSITION: user_input[
@ -86,7 +87,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle discovery via dhcp."""
self._discovered_device = FluxLEDDiscovery(
ipaddr=discovery_info.ip,
model=discovery_info.hostname,
model=None,
id=discovery_info.macaddress.replace(":", ""),
model_num=None,
version_num=None,
@ -115,11 +116,12 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
mac = dr.format_mac(mac_address)
host = device[ATTR_IPADDR]
await self.async_set_unique_id(mac)
self._abort_if_unique_id_configured(updates={CONF_HOST: host})
for entry in self._async_current_entries(include_ignore=False):
if entry.data[CONF_HOST] == host:
if not entry.unique_id:
async_update_entry_from_discovery(self.hass, entry, device)
if entry.unique_id == mac or entry.data[CONF_HOST] == host:
if async_update_entry_from_discovery(self.hass, entry, device):
self.hass.async_create_task(
self.hass.config_entries.async_reload(entry.entry_id)
)
return self.async_abort(reason="already_configured")
self.context[CONF_HOST] = host
for progress in self._async_in_progress():
@ -164,12 +166,14 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Create a config entry from a device."""
self._async_abort_entries_match({CONF_HOST: device[ATTR_IPADDR]})
name = async_name_from_discovery(device)
return self.async_create_entry(
title=name,
data={
data: dict[str, Any] = {
CONF_HOST: device[ATTR_IPADDR],
CONF_NAME: name,
},
}
async_populate_data_from_discovery(data, data, device)
return self.async_create_entry(
title=name,
data=data,
)
async def async_step_user(
@ -259,7 +263,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
model=model,
id=mac_address,
model_num=bulb.model_num,
version_num=bulb.version_num,
version_num=None, # This is the minor version number
firmware_date=None,
model_info=None,
model_description=bulb.model_data.description,

4
homeassistant/components/flux_led/const.py
View file

@ -53,6 +53,10 @@ DISCOVER_SCAN_TIMEOUT: Final = 10
CONF_DEVICES: Final = "devices"
CONF_CUSTOM_EFFECT: Final = "custom_effect"
CONF_MODEL: Final = "model"
CONF_MINOR_VERSION: Final = "minor_version"
CONF_REMOTE_ACCESS_ENABLED: Final = "remote_access_enabled"
CONF_REMOTE_ACCESS_HOST: Final = "remote_access_host"
CONF_REMOTE_ACCESS_PORT: Final = "remote_access_port"
MODE_AUTO: Final = "auto"
MODE_RGB: Final = "rgb"

93
homeassistant/components/flux_led/discovery.py
View file

@ -2,21 +2,54 @@
from __future__ import annotations
import asyncio
from collections.abc import Mapping
import logging
from typing import Any, Final
from flux_led.aioscanner import AIOBulbScanner
from flux_led.const import ATTR_ID, ATTR_IPADDR, ATTR_MODEL, ATTR_MODEL_DESCRIPTION
from flux_led.const import (
ATTR_ID,
ATTR_IPADDR,
ATTR_MODEL,
ATTR_MODEL_DESCRIPTION,
ATTR_REMOTE_ACCESS_ENABLED,
ATTR_REMOTE_ACCESS_HOST,
ATTR_REMOTE_ACCESS_PORT,
ATTR_VERSION_NUM,
)
from flux_led.scanner import FluxLEDDiscovery
from homeassistant import config_entries
from homeassistant.components import network
from homeassistant.const import CONF_HOST, CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.util.network import is_ip_address
from .const import DISCOVER_SCAN_TIMEOUT, DOMAIN
from .const import (
CONF_MINOR_VERSION,
CONF_MODEL,
CONF_REMOTE_ACCESS_ENABLED,
CONF_REMOTE_ACCESS_HOST,
CONF_REMOTE_ACCESS_PORT,
DISCOVER_SCAN_TIMEOUT,
DOMAIN,
FLUX_LED_DISCOVERY,
)
_LOGGER = logging.getLogger(__name__)
CONF_TO_DISCOVERY: Final = {
CONF_HOST: ATTR_IPADDR,
CONF_REMOTE_ACCESS_ENABLED: ATTR_REMOTE_ACCESS_ENABLED,
CONF_REMOTE_ACCESS_HOST: ATTR_REMOTE_ACCESS_HOST,
CONF_REMOTE_ACCESS_PORT: ATTR_REMOTE_ACCESS_PORT,
CONF_MINOR_VERSION: ATTR_VERSION_NUM,
CONF_MODEL: ATTR_MODEL,
}
@callback
def async_name_from_discovery(device: FluxLEDDiscovery) -> str:
"""Convert a flux_led discovery to a human readable name."""
@ -29,6 +62,62 @@ def async_name_from_discovery(device: FluxLEDDiscovery) -> str:
return f"{device[ATTR_MODEL]} {short_mac}"
@callback
def async_populate_data_from_discovery(
current_data: Mapping[str, Any],
data_updates: dict[str, Any],
device: FluxLEDDiscovery,
) -> None:
"""Copy discovery data into config entry data."""
for conf_key, discovery_key in CONF_TO_DISCOVERY.items():
if (
device.get(discovery_key) is not None
and current_data.get(conf_key) != device[discovery_key] # type: ignore[misc]
):
data_updates[conf_key] = device[discovery_key] # type: ignore[misc]
@callback
def async_update_entry_from_discovery(
hass: HomeAssistant, entry: config_entries.ConfigEntry, device: FluxLEDDiscovery
) -> bool:
"""Update a config entry from a flux_led discovery."""
data_updates: dict[str, Any] = {}
mac_address = device[ATTR_ID]
assert mac_address is not None
updates: dict[str, Any] = {}
if not entry.unique_id:
updates["unique_id"] = dr.format_mac(mac_address)
async_populate_data_from_discovery(entry.data, data_updates, device)
if not entry.data.get(CONF_NAME) or is_ip_address(entry.data[CONF_NAME]):
updates["title"] = data_updates[CONF_NAME] = async_name_from_discovery(device)
if data_updates:
updates["data"] = {**entry.data, **data_updates}
if updates:
return hass.config_entries.async_update_entry(entry, **updates)
return False
@callback
def async_get_discovery(hass: HomeAssistant, host: str) -> FluxLEDDiscovery | None:
"""Check if a device was already discovered via a broadcast discovery."""
discoveries: list[FluxLEDDiscovery] = hass.data[DOMAIN][FLUX_LED_DISCOVERY]
for discovery in discoveries:
if discovery[ATTR_IPADDR] == host:
return discovery
return None
@callback
def async_clear_discovery_cache(hass: HomeAssistant, host: str) -> None:
"""Clear the host from the discovery cache."""
domain_data = hass.data[DOMAIN]
discoveries: list[FluxLEDDiscovery] = domain_data[FLUX_LED_DISCOVERY]
domain_data[FLUX_LED_DISCOVERY] = [
discovery for discovery in discoveries if discovery[ATTR_IPADDR] != host
]
async def async_discover_devices(
hass: HomeAssistant, timeout: int, address: str | None = None
) -> list[FluxLEDDiscovery]:

54
homeassistant/components/flux_led/entity.py
View file

@ -6,18 +6,56 @@ from typing import Any
from flux_led.aiodevice import AIOWifiLedBulb
from homeassistant import config_entries
from homeassistant.const import CONF_NAME
from homeassistant.core import callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity import DeviceInfo, Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import FluxLedUpdateCoordinator
from .const import SIGNAL_STATE_UPDATED
from .const import CONF_MINOR_VERSION, CONF_MODEL, SIGNAL_STATE_UPDATED
def _async_device_info(
unique_id: str, device: AIOWifiLedBulb, entry: config_entries.ConfigEntry
) -> DeviceInfo:
version_num = device.version_num
if minor_version := entry.data.get(CONF_MINOR_VERSION):
sw_version = version_num + int(hex(minor_version)[2:]) / 100
sw_version_str = f"{sw_version:0.3f}"
else:
sw_version_str = str(device.version_num)
return DeviceInfo(
connections={(dr.CONNECTION_NETWORK_MAC, unique_id)},
manufacturer="Zengge",
model=device.model,
name=entry.data[CONF_NAME],
sw_version=sw_version_str,
hw_version=entry.data.get(CONF_MODEL),
)
class FluxBaseEntity(Entity):
"""Representation of a Flux entity without a coordinator."""
def __init__(
self,
device: AIOWifiLedBulb,
entry: config_entries.ConfigEntry,
) -> None:
"""Initialize the light."""
self._device: AIOWifiLedBulb = device
self.entry = entry
if entry.unique_id:
self._attr_device_info = _async_device_info(
entry.unique_id, self._device, entry
)
class FluxEntity(CoordinatorEntity):
"""Representation of a Flux entity."""
"""Representation of a Flux entity with a coordinator."""
coordinator: FluxLedUpdateCoordinator
@ -33,13 +71,9 @@ class FluxEntity(CoordinatorEntity):
self._responding = True
self._attr_name = name
self._attr_unique_id = unique_id
if self.unique_id:
self._attr_device_info = DeviceInfo(
connections={(dr.CONNECTION_NETWORK_MAC, self.unique_id)},
manufacturer="Magic Home (Zengge)",
model=self._device.model,
name=self.name,
sw_version=str(self._device.version_num),
if unique_id:
self._attr_device_info = _async_device_info(
unique_id, self._device, coordinator.entry
)
@property

2
homeassistant/components/flux_led/light.py
View file

@ -34,7 +34,6 @@ from homeassistant.const import (
CONF_DEVICES,
CONF_HOST,
CONF_MAC,
CONF_MODE,
CONF_NAME,
CONF_PROTOCOL,
)
@ -158,7 +157,6 @@ async def async_setup_platform(
CONF_MAC: discovered_mac_by_host.get(host),
CONF_NAME: device_config[CONF_NAME],
CONF_PROTOCOL: device_config.get(CONF_PROTOCOL),
CONF_MODE: device_config.get(ATTR_MODE, MODE_AUTO),
CONF_CUSTOM_EFFECT_COLORS: custom_effect_colors,
CONF_CUSTOM_EFFECT_SPEED_PCT: custom_effects.get(
CONF_SPEED_PCT, DEFAULT_EFFECT_SPEED

77
homeassistant/components/flux_led/switch.py
View file

@ -3,16 +3,26 @@ from __future__ import annotations
from typing import Any
from flux_led import DeviceType
from flux_led.aio import AIOWifiLedBulb
from homeassistant import config_entries
from homeassistant.components.switch import SwitchEntity
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import FluxLedUpdateCoordinator
from .const import DOMAIN
from .entity import FluxOnOffEntity
from .const import (
CONF_REMOTE_ACCESS_ENABLED,
CONF_REMOTE_ACCESS_HOST,
CONF_REMOTE_ACCESS_PORT,
DOMAIN,
)
from .discovery import async_clear_discovery_cache
from .entity import FluxBaseEntity, FluxOnOffEntity
async def async_setup_entry(
@ -22,16 +32,23 @@ async def async_setup_entry(
) -> None:
"""Set up the Flux lights."""
coordinator: FluxLedUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
[
entities: list[FluxSwitch | FluxRemoteAccessSwitch] = []
if coordinator.device.device_type == DeviceType.Switch:
entities.append(
FluxSwitch(
coordinator,
entry.unique_id,
entry.data[CONF_NAME],
)
]
)
if entry.data.get(CONF_REMOTE_ACCESS_HOST):
entities.append(FluxRemoteAccessSwitch(coordinator.device, entry))
if entities:
async_add_entities(entities)
class FluxSwitch(FluxOnOffEntity, CoordinatorEntity, SwitchEntity):
"""Representation of a Flux switch."""
@ -40,3 +57,53 @@ class FluxSwitch(FluxOnOffEntity, CoordinatorEntity, SwitchEntity):
"""Turn the device on."""
if not self.is_on:
await self._device.async_turn_on()
class FluxRemoteAccessSwitch(FluxBaseEntity, SwitchEntity):
"""Representation of a Flux remote access switch."""
_attr_should_poll = False
_attr_entity_category = EntityCategory.CONFIG
def __init__(
self,
device: AIOWifiLedBulb,
entry: config_entries.ConfigEntry,
) -> None:
"""Initialize the light."""
super().__init__(device, entry)
self._attr_name = f"{entry.data[CONF_NAME]} Remote Access"
if entry.unique_id:
self._attr_unique_id = f"{entry.unique_id}_remote_access"
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the remote access on."""
await self._device.async_enable_remote_access(
self.entry.data[CONF_REMOTE_ACCESS_HOST],
self.entry.data[CONF_REMOTE_ACCESS_PORT],
)
await self._async_update_entry(True)
async def _async_update_entry(self, new_state: bool) -> None:
"""Update the entry with the new state on success."""
async_clear_discovery_cache(self.hass, self._device.ipaddr)
self.hass.config_entries.async_update_entry(
self.entry,
data={**self.entry.data, CONF_REMOTE_ACCESS_ENABLED: new_state},
)
self.async_write_ha_state()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the remote access off."""
await self._device.async_disable_remote_access()
await self._async_update_entry(False)
@property
def is_on(self) -> bool:
"""Return true if remote access is enabled."""
return bool(self.entry.data[CONF_REMOTE_ACCESS_ENABLED])
@property
def icon(self) -> str:
"""Return icon based on state."""
return "mdi:cloud-outline" if self.is_on else "mdi:cloud-off-outline"

3
homeassistant/components/flux_led/translations/en.json
View file

@ -27,8 +27,7 @@
"data": {
"custom_effect_colors": "Custom Effect: List of 1 to 16 [R,G,B] colors. Example: [255,0,255],[60,128,0]",
"custom_effect_speed_pct": "Custom Effect: Speed in percents for the effect that switch colors.",
"custom_effect_transition": "Custom Effect: Type of transition between the colors.",
"mode": "The chosen brightness mode."
"custom_effect_transition": "Custom Effect: Type of transition between the colors."
}
}
}

6
tests/components/flux_led/__init__.py
View file

@ -57,6 +57,9 @@ FLUX_DISCOVERY = FluxLEDDiscovery(
firmware_date=datetime.date(2021, 5, 5),
model_info=MODEL,
model_description=MODEL_DESCRIPTION,
remote_access_enabled=True,
remote_access_host="the.cloud",
remote_access_port=8816,
)
@ -80,6 +83,9 @@ def _mocked_bulb() -> AIOWifiLedBulb:
bulb.async_turn_off = AsyncMock()
bulb.async_turn_on = AsyncMock()
bulb.async_set_levels = AsyncMock()
bulb.async_set_zones = AsyncMock()
bulb.async_disable_remote_access = AsyncMock()
bulb.async_enable_remote_access = AsyncMock()
bulb.min_temp = 2700
bulb.max_temp = 6500
bulb.getRgb = MagicMock(return_value=[255, 0, 0])

62
tests/components/flux_led/test_config_flow.py
View file

@ -11,6 +11,11 @@ from homeassistant.components.flux_led.const import (
CONF_CUSTOM_EFFECT_COLORS,
CONF_CUSTOM_EFFECT_SPEED_PCT,
CONF_CUSTOM_EFFECT_TRANSITION,
CONF_MINOR_VERSION,
CONF_MODEL,
CONF_REMOTE_ACCESS_ENABLED,
CONF_REMOTE_ACCESS_HOST,
CONF_REMOTE_ACCESS_PORT,
DOMAIN,
MODE_RGB,
TRANSITION_JUMP,
@ -20,7 +25,6 @@ from homeassistant.const import (
CONF_DEVICE,
CONF_HOST,
CONF_MAC,
CONF_MODE,
CONF_NAME,
CONF_PROTOCOL,
)
@ -34,6 +38,7 @@ from . import (
FLUX_DISCOVERY_PARTIAL,
IP_ADDRESS,
MAC_ADDRESS,
MODEL,
MODULE,
_patch_discovery,
_patch_wifibulb,
@ -88,7 +93,16 @@ async def test_discovery(hass: HomeAssistant):
assert result3["type"] == "create_entry"
assert result3["title"] == DEFAULT_ENTRY_TITLE
assert result3["data"] == {CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE}
assert result3["data"] == {
CONF_MINOR_VERSION: 4,
CONF_HOST: IP_ADDRESS,
CONF_NAME: DEFAULT_ENTRY_TITLE,
CONF_MODEL: MODEL,
CONF_REMOTE_ACCESS_ENABLED: True,
CONF_REMOTE_ACCESS_HOST: "the.cloud",
CONF_REMOTE_ACCESS_PORT: 8816,
CONF_MINOR_VERSION: 0x04,
}
mock_setup.assert_called_once()
mock_setup_entry.assert_called_once()
@ -160,8 +174,14 @@ async def test_discovery_with_existing_device_present(hass: HomeAssistant):
assert result3["type"] == "create_entry"
assert result3["title"] == DEFAULT_ENTRY_TITLE
assert result3["data"] == {
CONF_MINOR_VERSION: 4,
CONF_HOST: IP_ADDRESS,
CONF_NAME: DEFAULT_ENTRY_TITLE,
CONF_MODEL: MODEL,
CONF_REMOTE_ACCESS_ENABLED: True,
CONF_REMOTE_ACCESS_HOST: "the.cloud",
CONF_REMOTE_ACCESS_PORT: 8816,
CONF_MINOR_VERSION: 0x04,
}
await hass.async_block_till_done()
@ -204,7 +224,7 @@ async def test_import(hass: HomeAssistant):
CONF_MAC: MAC_ADDRESS,
CONF_NAME: "floor lamp",
CONF_PROTOCOL: "ledenet",
CONF_MODE: MODE_RGB,
CONF_MODEL: MODE_RGB,
CONF_CUSTOM_EFFECT_COLORS: "[255,0,0], [0,0,255]",
CONF_CUSTOM_EFFECT_SPEED_PCT: 30,
CONF_CUSTOM_EFFECT_TRANSITION: TRANSITION_STROBE,
@ -229,7 +249,6 @@ async def test_import(hass: HomeAssistant):
CONF_PROTOCOL: "ledenet",
}
assert result["options"] == {
CONF_MODE: MODE_RGB,
CONF_CUSTOM_EFFECT_COLORS: "[255,0,0], [0,0,255]",
CONF_CUSTOM_EFFECT_SPEED_PCT: 30,
CONF_CUSTOM_EFFECT_TRANSITION: TRANSITION_STROBE,
@ -278,7 +297,16 @@ async def test_manual_working_discovery(hass: HomeAssistant):
await hass.async_block_till_done()
assert result4["type"] == "create_entry"
assert result4["title"] == DEFAULT_ENTRY_TITLE
assert result4["data"] == {CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE}
assert result4["data"] == {
CONF_MINOR_VERSION: 4,
CONF_HOST: IP_ADDRESS,
CONF_NAME: DEFAULT_ENTRY_TITLE,
CONF_MODEL: MODEL,
CONF_REMOTE_ACCESS_ENABLED: True,
CONF_REMOTE_ACCESS_HOST: "the.cloud",
CONF_REMOTE_ACCESS_PORT: 8816,
CONF_MINOR_VERSION: 0x04,
}
# Duplicate
result = await hass.config_entries.flow.async_init(
@ -376,7 +404,16 @@ async def test_discovered_by_discovery(hass):
await hass.async_block_till_done()
assert result2["type"] == "create_entry"
assert result2["data"] == {CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE}
assert result2["data"] == {
CONF_MINOR_VERSION: 4,
CONF_HOST: IP_ADDRESS,
CONF_NAME: DEFAULT_ENTRY_TITLE,
CONF_MODEL: MODEL,
CONF_REMOTE_ACCESS_ENABLED: True,
CONF_REMOTE_ACCESS_HOST: "the.cloud",
CONF_REMOTE_ACCESS_PORT: 8816,
CONF_MINOR_VERSION: 0x04,
}
assert mock_async_setup.called
assert mock_async_setup_entry.called
@ -402,7 +439,16 @@ async def test_discovered_by_dhcp_udp_responds(hass):
await hass.async_block_till_done()
assert result2["type"] == "create_entry"
assert result2["data"] == {CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE}
assert result2["data"] == {
CONF_MINOR_VERSION: 4,
CONF_HOST: IP_ADDRESS,
CONF_NAME: DEFAULT_ENTRY_TITLE,
CONF_MODEL: MODEL,
CONF_REMOTE_ACCESS_ENABLED: True,
CONF_REMOTE_ACCESS_HOST: "the.cloud",
CONF_REMOTE_ACCESS_PORT: 8816,
CONF_MINOR_VERSION: 0x04,
}
assert mock_async_setup.called
assert mock_async_setup_entry.called
@ -538,7 +584,7 @@ async def test_options(hass: HomeAssistant):
domain=DOMAIN,
data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
options={
CONF_MODE: MODE_RGB,
CONF_MODEL: MODE_RGB,
CONF_CUSTOM_EFFECT_COLORS: "[255,0,0], [0,0,255]",
CONF_CUSTOM_EFFECT_SPEED_PCT: 30,
CONF_CUSTOM_EFFECT_TRANSITION: TRANSITION_STROBE,

28
tests/components/flux_led/test_init.py
View file

@ -3,8 +3,6 @@ from __future__ import annotations
from unittest.mock import patch
from flux_led.aioscanner import AIOBulbScanner
from flux_led.scanner import FluxLEDDiscovery
import pytest
from homeassistant.components import flux_led
@ -107,32 +105,24 @@ async def test_config_entry_retry(hass: HomeAssistant) -> None:
],
)
async def test_config_entry_fills_unique_id_with_directed_discovery(
hass: HomeAssistant, discovery: FluxLEDDiscovery, title: str
hass: HomeAssistant, discovery: dict[str, str], title: str
) -> None:
"""Test that the unique id is added if its missing via directed (not broadcast) discovery."""
config_entry = MockConfigEntry(
domain=DOMAIN, data={CONF_NAME: "bogus", CONF_HOST: IP_ADDRESS}, unique_id=None
domain=DOMAIN, data={CONF_HOST: IP_ADDRESS}, unique_id=None
)
config_entry.add_to_hass(hass)
assert config_entry.unique_id is None
class MockBulbScanner(AIOBulbScanner):
def __init__(self) -> None:
self._last_address: str | None = None
super().__init__()
async def async_scan(
self, timeout: int = 10, address: str | None = None
) -> list[FluxLEDDiscovery]:
self._last_address = address
async def _discovery(self, *args, address=None, **kwargs):
# Only return discovery results when doing directed discovery
return [discovery] if address == IP_ADDRESS else []
def getBulbInfo(self) -> FluxLEDDiscovery:
return [discovery] if self._last_address == IP_ADDRESS else []
with patch(
"homeassistant.components.flux_led.discovery.AIOBulbScanner",
return_value=MockBulbScanner(),
"homeassistant.components.flux_led.discovery.AIOBulbScanner.async_scan",
new=_discovery,
), patch(
"homeassistant.components.flux_led.discovery.AIOBulbScanner.getBulbInfo",
return_value=[discovery],
), _patch_wifibulb():
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done()

63
tests/components/flux_led/test_light.py
View file

@ -1,6 +1,6 @@
"""Tests for light platform."""
from datetime import timedelta
from unittest.mock import AsyncMock, Mock
from unittest.mock import AsyncMock, Mock, patch
from flux_led.const import (
COLOR_MODE_ADDRESSABLE as FLUX_COLOR_MODE_ADDRESSABLE,
@ -21,6 +21,11 @@ from homeassistant.components.flux_led.const import (
CONF_CUSTOM_EFFECT_SPEED_PCT,
CONF_CUSTOM_EFFECT_TRANSITION,
CONF_DEVICES,
CONF_MINOR_VERSION,
CONF_MODEL,
CONF_REMOTE_ACCESS_ENABLED,
CONF_REMOTE_ACCESS_HOST,
CONF_REMOTE_ACCESS_PORT,
CONF_SPEED_PCT,
CONF_TRANSITION,
DOMAIN,
@ -59,8 +64,10 @@ from homeassistant.util.dt import utcnow
from . import (
DEFAULT_ENTRY_TITLE,
FLUX_DISCOVERY,
IP_ADDRESS,
MAC_ADDRESS,
MODEL,
_mocked_bulb,
_patch_discovery,
_patch_wifibulb,
@ -1145,7 +1152,26 @@ async def test_migrate_from_yaml_with_custom_effect(hass: HomeAssistant) -> None
}
],
}
with _patch_discovery(), _patch_wifibulb():
last_address = None
async def _discovery(self, *args, address=None, **kwargs):
# Only return discovery results when doing directed discovery
nonlocal last_address
last_address = address
return [FLUX_DISCOVERY] if address == IP_ADDRESS else []
def _mock_getBulbInfo(*args, **kwargs):
nonlocal last_address
return [FLUX_DISCOVERY] if last_address == IP_ADDRESS else []
with patch(
"homeassistant.components.flux_led.discovery.AIOBulbScanner.async_scan",
new=_discovery,
), patch(
"homeassistant.components.flux_led.discovery.AIOBulbScanner.getBulbInfo",
new=_mock_getBulbInfo,
), _patch_wifibulb():
await async_setup_component(hass, LIGHT_DOMAIN, config)
await hass.async_block_till_done()
await hass.async_block_till_done()
@ -1165,9 +1191,13 @@ async def test_migrate_from_yaml_with_custom_effect(hass: HomeAssistant) -> None
CONF_HOST: IP_ADDRESS,
CONF_NAME: "flux_lamppost",
CONF_PROTOCOL: "ledenet",
CONF_MODEL: MODEL,
CONF_REMOTE_ACCESS_ENABLED: True,
CONF_REMOTE_ACCESS_HOST: "the.cloud",
CONF_REMOTE_ACCESS_PORT: 8816,
CONF_MINOR_VERSION: 0x04,
}
assert migrated_entry.options == {
CONF_MODE: "auto",
CONF_CUSTOM_EFFECT_COLORS: "[(255, 0, 0), (255, 255, 0), (0, 255, 0)]",
CONF_CUSTOM_EFFECT_SPEED_PCT: 30,
CONF_CUSTOM_EFFECT_TRANSITION: "strobe",
@ -1189,7 +1219,26 @@ async def test_migrate_from_yaml_no_custom_effect(hass: HomeAssistant) -> None:
}
],
}
with _patch_discovery(), _patch_wifibulb():
last_address = None
async def _discovery(self, *args, address=None, **kwargs):
# Only return discovery results when doing directed discovery
nonlocal last_address
last_address = address
return [FLUX_DISCOVERY] if address == IP_ADDRESS else []
def _mock_getBulbInfo(*args, **kwargs):
nonlocal last_address
return [FLUX_DISCOVERY] if last_address == IP_ADDRESS else []
with patch(
"homeassistant.components.flux_led.discovery.AIOBulbScanner.async_scan",
new=_discovery,
), patch(
"homeassistant.components.flux_led.discovery.AIOBulbScanner.getBulbInfo",
new=_mock_getBulbInfo,
), _patch_wifibulb():
await async_setup_component(hass, LIGHT_DOMAIN, config)
await hass.async_block_till_done()
await hass.async_block_till_done()
@ -1209,9 +1258,13 @@ async def test_migrate_from_yaml_no_custom_effect(hass: HomeAssistant) -> None:
CONF_HOST: IP_ADDRESS,
CONF_NAME: "flux_lamppost",
CONF_PROTOCOL: "ledenet",
CONF_MODEL: MODEL,
CONF_REMOTE_ACCESS_ENABLED: True,
CONF_REMOTE_ACCESS_HOST: "the.cloud",
CONF_REMOTE_ACCESS_PORT: 8816,
CONF_MINOR_VERSION: 0x04,
}
assert migrated_entry.options == {
CONF_MODE: "auto",
CONF_CUSTOM_EFFECT_COLORS: None,
CONF_CUSTOM_EFFECT_SPEED_PCT: 50,
CONF_CUSTOM_EFFECT_TRANSITION: "gradual",

39
tests/components/flux_led/test_switch.py
View file

@ -1,6 +1,6 @@
"""Tests for switch platform."""
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, DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
@ -16,6 +16,7 @@ from . import (
DEFAULT_ENTRY_TITLE,
IP_ADDRESS,
MAC_ADDRESS,
_mocked_bulb,
_mocked_switch,
_patch_discovery,
_patch_wifibulb,
@ -27,7 +28,7 @@ from tests.common import MockConfigEntry
async def test_switch_on_off(hass: HomeAssistant) -> None:
"""Test a switch light."""
"""Test a smart plug."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
@ -60,3 +61,37 @@ async def test_switch_on_off(hass: HomeAssistant) -> None:
await async_mock_device_turn_on(hass, switch)
assert hass.states.get(entity_id).state == STATE_ON
async def test_remote_access_on_off(hass: HomeAssistant) -> None:
"""Test enable/disable remote access."""
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)
bulb = _mocked_bulb()
with _patch_discovery(), _patch_wifibulb(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"
state = hass.states.get(entity_id)
assert state.state == STATE_ON
await hass.services.async_call(
SWITCH_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
bulb.async_disable_remote_access.assert_called_once()
assert hass.states.get(entity_id).state == STATE_OFF
assert config_entry.data[CONF_REMOTE_ACCESS_ENABLED] is False
await hass.services.async_call(
SWITCH_DOMAIN, "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
bulb.async_enable_remote_access.assert_called_once()
assert hass.states.get(entity_id).state == STATE_ON
assert config_entry.data[CONF_REMOTE_ACCESS_ENABLED] is True