Include model name in flux_led discovery when available (#60105)

This commit is contained in:
J. Nick Koston 2021-11-22 20:21:46 -06:00 committed by GitHub
parent cb3b19b000
commit 4af5cde738
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 184 additions and 77 deletions

View File

@ -3,11 +3,12 @@ from __future__ import annotations
from datetime import timedelta from datetime import timedelta
import logging import logging
from typing import Any, Final from typing import Any, Final, cast
from flux_led import DeviceType from flux_led import DeviceType
from flux_led.aio import AIOWifiLedBulb from flux_led.aio import AIOWifiLedBulb
from flux_led.aioscanner import AIOBulbScanner from flux_led.aioscanner import AIOBulbScanner
from flux_led.const import ATTR_ID, ATTR_IPADDR, ATTR_MODEL, ATTR_MODEL_DESCRIPTION
from homeassistant import config_entries from homeassistant import config_entries
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
@ -24,11 +25,8 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda
from .const import ( from .const import (
DISCOVER_SCAN_TIMEOUT, DISCOVER_SCAN_TIMEOUT,
DOMAIN, DOMAIN,
FLUX_HOST,
FLUX_LED_DISCOVERY, FLUX_LED_DISCOVERY,
FLUX_LED_EXCEPTIONS, FLUX_LED_EXCEPTIONS,
FLUX_MAC,
FLUX_MODEL,
SIGNAL_STATE_UPDATED, SIGNAL_STATE_UPDATED,
STARTUP_SCAN_TIMEOUT, STARTUP_SCAN_TIMEOUT,
) )
@ -49,17 +47,28 @@ def async_wifi_bulb_for_host(host: str) -> AIOWifiLedBulb:
return AIOWifiLedBulb(host) return AIOWifiLedBulb(host)
@callback
def async_name_from_discovery(device: dict[str, Any]) -> str:
"""Convert a flux_led discovery to a human readable name."""
if (mac := device.get(ATTR_ID)) is None:
return cast(str, device[ATTR_IPADDR])
short_mac = mac[-6:]
if device.get(ATTR_MODEL_DESCRIPTION):
return f"{device[ATTR_MODEL_DESCRIPTION]} {short_mac}"
return f"{device[ATTR_MODEL]} {short_mac}"
@callback @callback
def async_update_entry_from_discovery( def async_update_entry_from_discovery(
hass: HomeAssistant, entry: config_entries.ConfigEntry, device: dict[str, Any] hass: HomeAssistant, entry: config_entries.ConfigEntry, device: dict[str, Any]
) -> None: ) -> None:
"""Update a config entry from a flux_led discovery.""" """Update a config entry from a flux_led discovery."""
name = f"{device[FLUX_MODEL]} {device[FLUX_MAC]}" name = async_name_from_discovery(device)
hass.config_entries.async_update_entry( hass.config_entries.async_update_entry(
entry, entry,
data={**entry.data, CONF_NAME: name}, data={**entry.data, CONF_NAME: name},
title=name, title=name,
unique_id=dr.format_mac(device[FLUX_MAC]), unique_id=dr.format_mac(device[ATTR_ID]),
) )
@ -86,7 +95,7 @@ async def async_discover_device(
# If we are missing the unique_id we should be able to fetch it # If we are missing the unique_id we should be able to fetch it
# from the device by doing a directed discovery at the host only # from the device by doing a directed discovery at the host only
for device in await async_discover_devices(hass, DISCOVER_SCAN_TIMEOUT, host): for device in await async_discover_devices(hass, DISCOVER_SCAN_TIMEOUT, host):
if device[FLUX_HOST] == host: if device[ATTR_IPADDR] == host:
return device return device
return None return None

View File

@ -4,6 +4,7 @@ from __future__ import annotations
import logging import logging
from typing import Any, Final from typing import Any, Final
from flux_led.const import ATTR_ID, ATTR_IPADDR, ATTR_MODEL, ATTR_MODEL_DESCRIPTION
import voluptuous as vol import voluptuous as vol
from homeassistant import config_entries from homeassistant import config_entries
@ -17,6 +18,7 @@ from homeassistant.helpers.typing import DiscoveryInfoType
from . import ( from . import (
async_discover_device, async_discover_device,
async_discover_devices, async_discover_devices,
async_name_from_discovery,
async_update_entry_from_discovery, async_update_entry_from_discovery,
async_wifi_bulb_for_host, async_wifi_bulb_for_host,
) )
@ -27,10 +29,7 @@ from .const import (
DEFAULT_EFFECT_SPEED, DEFAULT_EFFECT_SPEED,
DISCOVER_SCAN_TIMEOUT, DISCOVER_SCAN_TIMEOUT,
DOMAIN, DOMAIN,
FLUX_HOST,
FLUX_LED_EXCEPTIONS, FLUX_LED_EXCEPTIONS,
FLUX_MAC,
FLUX_MODEL,
TRANSITION_GRADUAL, TRANSITION_GRADUAL,
TRANSITION_JUMP, TRANSITION_JUMP,
TRANSITION_STROBE, TRANSITION_STROBE,
@ -38,6 +37,7 @@ from .const import (
CONF_DEVICE: Final = "device" CONF_DEVICE: Final = "device"
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -85,9 +85,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult: async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult:
"""Handle discovery via dhcp.""" """Handle discovery via dhcp."""
self._discovered_device = { self._discovered_device = {
FLUX_HOST: discovery_info[dhcp.IP_ADDRESS], ATTR_IPADDR: discovery_info[dhcp.IP_ADDRESS],
FLUX_MODEL: discovery_info[dhcp.HOSTNAME], ATTR_MODEL: discovery_info[dhcp.HOSTNAME],
FLUX_MAC: discovery_info[dhcp.MAC_ADDRESS].replace(":", ""), ATTR_ID: discovery_info[dhcp.MAC_ADDRESS].replace(":", ""),
} }
return await self._async_handle_discovery() return await self._async_handle_discovery()
@ -101,8 +101,8 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def _async_handle_discovery(self) -> FlowResult: async def _async_handle_discovery(self) -> FlowResult:
"""Handle any discovery.""" """Handle any discovery."""
device = self._discovered_device device = self._discovered_device
mac = dr.format_mac(device[FLUX_MAC]) mac = dr.format_mac(device[ATTR_ID])
host = device[FLUX_HOST] host = device[ATTR_IPADDR]
await self.async_set_unique_id(mac) await self.async_set_unique_id(mac)
self._abort_if_unique_id_configured(updates={CONF_HOST: host}) self._abort_if_unique_id_configured(updates={CONF_HOST: host})
for entry in self._async_current_entries(include_ignore=False): for entry in self._async_current_entries(include_ignore=False):
@ -113,6 +113,14 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
for progress in self._async_in_progress(): for progress in self._async_in_progress():
if progress.get("context", {}).get(CONF_HOST) == host: if progress.get("context", {}).get(CONF_HOST) == host:
return self.async_abort(reason="already_in_progress") return self.async_abort(reason="already_in_progress")
if not device.get(ATTR_MODEL_DESCRIPTION):
try:
device = await self._async_try_connect(host)
except FLUX_LED_EXCEPTIONS:
return self.async_abort(reason="cannot_connect")
else:
if device.get(ATTR_MODEL_DESCRIPTION):
self._discovered_device = device
return await self.async_step_discovery_confirm() return await self.async_step_discovery_confirm()
async def async_step_discovery_confirm( async def async_step_discovery_confirm(
@ -123,7 +131,12 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self._async_create_entry_from_device(self._discovered_device) return self._async_create_entry_from_device(self._discovered_device)
self._set_confirm_only() self._set_confirm_only()
placeholders = self._discovered_device device = self._discovered_device
placeholders = {
"model": device.get(ATTR_MODEL_DESCRIPTION, device[ATTR_MODEL]),
"id": device[ATTR_ID][-6:],
"ipaddr": device[ATTR_IPADDR],
}
self.context["title_placeholders"] = placeholders self.context["title_placeholders"] = placeholders
return self.async_show_form( return self.async_show_form(
step_id="discovery_confirm", description_placeholders=placeholders step_id="discovery_confirm", description_placeholders=placeholders
@ -132,15 +145,12 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
@callback @callback
def _async_create_entry_from_device(self, device: dict[str, Any]) -> FlowResult: def _async_create_entry_from_device(self, device: dict[str, Any]) -> FlowResult:
"""Create a config entry from a device.""" """Create a config entry from a device."""
self._async_abort_entries_match({CONF_HOST: device[FLUX_HOST]}) self._async_abort_entries_match({CONF_HOST: device[ATTR_IPADDR]})
if device.get(FLUX_MAC): name = async_name_from_discovery(device)
name = f"{device[FLUX_MODEL]} {device[FLUX_MAC]}"
else:
name = device[FLUX_HOST]
return self.async_create_entry( return self.async_create_entry(
title=name, title=name,
data={ data={
CONF_HOST: device[FLUX_HOST], CONF_HOST: device[ATTR_IPADDR],
CONF_NAME: name, CONF_NAME: name,
}, },
) )
@ -158,9 +168,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
except FLUX_LED_EXCEPTIONS: except FLUX_LED_EXCEPTIONS:
errors["base"] = "cannot_connect" errors["base"] = "cannot_connect"
else: else:
if device[FLUX_MAC]: if device[ATTR_ID]:
await self.async_set_unique_id( await self.async_set_unique_id(
dr.format_mac(device[FLUX_MAC]), raise_on_progress=False dr.format_mac(device[ATTR_ID]), raise_on_progress=False
) )
self._abort_if_unique_id_configured(updates={CONF_HOST: host}) self._abort_if_unique_id_configured(updates={CONF_HOST: host})
return self._async_create_entry_from_device(device) return self._async_create_entry_from_device(device)
@ -189,12 +199,13 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self.hass, DISCOVER_SCAN_TIMEOUT self.hass, DISCOVER_SCAN_TIMEOUT
) )
self._discovered_devices = { self._discovered_devices = {
dr.format_mac(device[FLUX_MAC]): device for device in discovered_devices dr.format_mac(device[ATTR_ID]): device for device in discovered_devices
} }
devices_name = { devices_name = {
mac: f"{device[FLUX_MODEL]} {mac} ({device[FLUX_HOST]})" mac: f"{async_name_from_discovery(device)} ({device[ATTR_IPADDR]})"
for mac, device in self._discovered_devices.items() for mac, device in self._discovered_devices.items()
if mac not in current_unique_ids and device[FLUX_HOST] not in current_hosts if mac not in current_unique_ids
and device[ATTR_IPADDR] not in current_hosts
} }
# Check if there is at least one device # Check if there is at least one device
if not devices_name: if not devices_name:
@ -214,7 +225,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
await bulb.async_setup(lambda: None) await bulb.async_setup(lambda: None)
finally: finally:
await bulb.async_stop() await bulb.async_stop()
return {FLUX_MAC: None, FLUX_MODEL: None, FLUX_HOST: host} return {ATTR_ID: None, ATTR_MODEL: None, ATTR_IPADDR: host}
class OptionsFlow(config_entries.OptionsFlow): class OptionsFlow(config_entries.OptionsFlow):

View File

@ -77,7 +77,3 @@ EFFECT_SUPPORT_MODES = {COLOR_MODE_RGB, COLOR_MODE_RGBW, COLOR_MODE_RGBWW}
CONF_CUSTOM_EFFECT_COLORS: Final = "custom_effect_colors" CONF_CUSTOM_EFFECT_COLORS: Final = "custom_effect_colors"
CONF_CUSTOM_EFFECT_SPEED_PCT: Final = "custom_effect_speed_pct" CONF_CUSTOM_EFFECT_SPEED_PCT: Final = "custom_effect_speed_pct"
CONF_CUSTOM_EFFECT_TRANSITION: Final = "custom_effect_transition" CONF_CUSTOM_EFFECT_TRANSITION: Final = "custom_effect_transition"
FLUX_HOST: Final = "ipaddr"
FLUX_MAC: Final = "id"
FLUX_MODEL: Final = "model"

View File

@ -6,6 +6,7 @@ import logging
import random import random
from typing import Any, Final, cast from typing import Any, Final, cast
from flux_led.const import ATTR_ID, ATTR_IPADDR
from flux_led.utils import ( from flux_led.utils import (
color_temp_to_white_levels, color_temp_to_white_levels,
rgbcw_brightness, rgbcw_brightness,
@ -71,9 +72,7 @@ from .const import (
DEFAULT_EFFECT_SPEED, DEFAULT_EFFECT_SPEED,
DOMAIN, DOMAIN,
EFFECT_SUPPORT_MODES, EFFECT_SUPPORT_MODES,
FLUX_HOST,
FLUX_LED_DISCOVERY, FLUX_LED_DISCOVERY,
FLUX_MAC,
MODE_AUTO, MODE_AUTO,
MODE_RGB, MODE_RGB,
MODE_RGBW, MODE_RGBW,
@ -141,7 +140,7 @@ async def async_setup_platform(
"""Set up the flux led platform.""" """Set up the flux led platform."""
domain_data = hass.data[DOMAIN] domain_data = hass.data[DOMAIN]
discovered_mac_by_host = { discovered_mac_by_host = {
device[FLUX_HOST]: device[FLUX_MAC] device[ATTR_IPADDR]: device[ATTR_ID]
for device in domain_data[FLUX_LED_DISCOVERY] for device in domain_data[FLUX_LED_DISCOVERY]
} }
for host, device_config in config.get(CONF_DEVICES, {}).items(): for host, device_config in config.get(CONF_DEVICES, {}).items():

View File

@ -3,7 +3,7 @@
"name": "Flux LED/MagicHome", "name": "Flux LED/MagicHome",
"config_flow": true, "config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/flux_led", "documentation": "https://www.home-assistant.io/integrations/flux_led",
"requirements": ["flux_led==0.24.27"], "requirements": ["flux_led==0.24.28"],
"quality_scale": "platinum", "quality_scale": "platinum",
"codeowners": ["@icemanch"], "codeowners": ["@icemanch"],
"iot_class": "local_push", "iot_class": "local_push",

View File

@ -658,7 +658,7 @@ fjaraskupan==1.0.2
flipr-api==1.4.1 flipr-api==1.4.1
# homeassistant.components.flux_led # homeassistant.components.flux_led
flux_led==0.24.27 flux_led==0.24.28
# homeassistant.components.homekit # homeassistant.components.homekit
fnvhash==0.1.0 fnvhash==0.1.0

View File

@ -399,7 +399,7 @@ fjaraskupan==1.0.2
flipr-api==1.4.1 flipr-api==1.4.1
# homeassistant.components.flux_led # homeassistant.components.flux_led
flux_led==0.24.27 flux_led==0.24.28
# homeassistant.components.homekit # homeassistant.components.homekit
fnvhash==0.1.0 fnvhash==0.1.0

View File

@ -8,30 +8,47 @@ from unittest.mock import AsyncMock, MagicMock, patch
from flux_led import DeviceType from flux_led import DeviceType
from flux_led.aio import AIOWifiLedBulb from flux_led.aio import AIOWifiLedBulb
from flux_led.const import ( from flux_led.const import (
ATTR_ID,
ATTR_IPADDR,
ATTR_MODEL,
ATTR_MODEL_DESCRIPTION,
COLOR_MODE_CCT as FLUX_COLOR_MODE_CCT, COLOR_MODE_CCT as FLUX_COLOR_MODE_CCT,
COLOR_MODE_RGB as FLUX_COLOR_MODE_RGB, COLOR_MODE_RGB as FLUX_COLOR_MODE_RGB,
) )
from flux_led.protocol import LEDENETRawState from flux_led.protocol import LEDENETRawState
from homeassistant.components import dhcp from homeassistant.components import dhcp
from homeassistant.components.flux_led.const import FLUX_HOST, FLUX_MAC, FLUX_MODEL
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
MODULE = "homeassistant.components.flux_led" MODULE = "homeassistant.components.flux_led"
MODULE_CONFIG_FLOW = "homeassistant.components.flux_led.config_flow" MODULE_CONFIG_FLOW = "homeassistant.components.flux_led.config_flow"
IP_ADDRESS = "127.0.0.1" IP_ADDRESS = "127.0.0.1"
MODEL = "AZ120444" MODEL = "AZ120444"
MODEL_DESCRIPTION = "RGBW Controller"
MAC_ADDRESS = "aa:bb:cc:dd:ee:ff" MAC_ADDRESS = "aa:bb:cc:dd:ee:ff"
FLUX_MAC_ADDRESS = "aabbccddeeff" FLUX_MAC_ADDRESS = "aabbccddeeff"
SHORT_MAC_ADDRESS = "ddeeff"
DEFAULT_ENTRY_TITLE = f"{MODEL_DESCRIPTION} {SHORT_MAC_ADDRESS}"
DEFAULT_ENTRY_TITLE_PARTIAL = f"{MODEL} {SHORT_MAC_ADDRESS}"
DEFAULT_ENTRY_TITLE = f"{MODEL} {FLUX_MAC_ADDRESS}"
DHCP_DISCOVERY = dhcp.DhcpServiceInfo( DHCP_DISCOVERY = dhcp.DhcpServiceInfo(
hostname=MODEL, hostname=MODEL,
ip=IP_ADDRESS, ip=IP_ADDRESS,
macaddress=MAC_ADDRESS, macaddress=MAC_ADDRESS,
) )
FLUX_DISCOVERY = {FLUX_HOST: IP_ADDRESS, FLUX_MODEL: MODEL, FLUX_MAC: FLUX_MAC_ADDRESS} FLUX_DISCOVERY_PARTIAL = {
ATTR_IPADDR: IP_ADDRESS,
ATTR_MODEL: MODEL,
ATTR_ID: FLUX_MAC_ADDRESS,
}
FLUX_DISCOVERY = {
ATTR_IPADDR: IP_ADDRESS,
ATTR_MODEL: MODEL,
ATTR_ID: FLUX_MAC_ADDRESS,
ATTR_MODEL_DESCRIPTION: MODEL_DESCRIPTION,
}
def _mocked_bulb() -> AIOWifiLedBulb: def _mocked_bulb() -> AIOWifiLedBulb:

View File

@ -29,6 +29,7 @@ from homeassistant.data_entry_flow import RESULT_TYPE_ABORT, RESULT_TYPE_FORM
from . import ( from . import (
DEFAULT_ENTRY_TITLE, DEFAULT_ENTRY_TITLE,
DEFAULT_ENTRY_TITLE_PARTIAL,
DHCP_DISCOVERY, DHCP_DISCOVERY,
FLUX_DISCOVERY, FLUX_DISCOVERY,
IP_ADDRESS, IP_ADDRESS,
@ -350,19 +351,14 @@ async def test_discovered_by_discovery_and_dhcp(hass):
assert result3["reason"] == "already_in_progress" assert result3["reason"] == "already_in_progress"
@pytest.mark.parametrize( async def test_discovered_by_discovery(hass):
"source, data", """Test we can setup when discovered from discovery."""
[
(config_entries.SOURCE_DHCP, DHCP_DISCOVERY),
(config_entries.SOURCE_DISCOVERY, FLUX_DISCOVERY),
],
)
async def test_discovered_by_dhcp_or_discovery(hass, source, data):
"""Test we can setup when discovered from dhcp or discovery."""
with _patch_discovery(), _patch_wifibulb(): with _patch_discovery(), _patch_wifibulb():
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": source}, data=data DOMAIN,
context={"source": config_entries.SOURCE_DISCOVERY},
data=FLUX_DISCOVERY,
) )
await hass.async_block_till_done() await hass.async_block_till_done()
@ -383,6 +379,74 @@ async def test_discovered_by_dhcp_or_discovery(hass, source, data):
assert mock_async_setup_entry.called assert mock_async_setup_entry.called
async def test_discovered_by_dhcp_udp_responds(hass):
"""Test we can setup when discovered from dhcp but with udp response."""
with _patch_discovery(), _patch_wifibulb():
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_DHCP}, data=DHCP_DISCOVERY
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] is None
with _patch_discovery(), _patch_wifibulb(), patch(
f"{MODULE}.async_setup", return_value=True
) as mock_async_setup, patch(
f"{MODULE}.async_setup_entry", return_value=True
) as mock_async_setup_entry:
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
await hass.async_block_till_done()
assert result2["type"] == "create_entry"
assert result2["data"] == {CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE}
assert mock_async_setup.called
assert mock_async_setup_entry.called
async def test_discovered_by_dhcp_no_udp_response(hass):
"""Test we can setup when discovered from dhcp but no udp response."""
with _patch_discovery(no_device=True), _patch_wifibulb():
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_DHCP}, data=DHCP_DISCOVERY
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] is None
with _patch_discovery(no_device=True), _patch_wifibulb(), patch(
f"{MODULE}.async_setup", return_value=True
) as mock_async_setup, patch(
f"{MODULE}.async_setup_entry", return_value=True
) as mock_async_setup_entry:
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
await hass.async_block_till_done()
assert result2["type"] == "create_entry"
assert result2["data"] == {
CONF_HOST: IP_ADDRESS,
CONF_NAME: DEFAULT_ENTRY_TITLE_PARTIAL,
}
assert mock_async_setup.called
assert mock_async_setup_entry.called
async def test_discovered_by_dhcp_no_udp_response_or_tcp_response(hass):
"""Test we can setup when discovered from dhcp but no udp response or tcp response."""
with _patch_discovery(no_device=True), _patch_wifibulb(no_device=True):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_DHCP}, data=DHCP_DISCOVERY
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "cannot_connect"
@pytest.mark.parametrize( @pytest.mark.parametrize(
"source, data", "source, data",
[ [
@ -445,4 +509,4 @@ async def test_options(hass: HomeAssistant):
assert result2["type"] == "create_entry" assert result2["type"] == "create_entry"
assert result2["data"] == user_input assert result2["data"] == user_input
assert result2["data"] == config_entry.options assert result2["data"] == config_entry.options
assert hass.states.get("light.az120444_aabbccddeeff") is not None assert hass.states.get("light.rgbw_controller_ddeeff") is not None

View File

@ -3,6 +3,8 @@ from __future__ import annotations
from unittest.mock import patch from unittest.mock import patch
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 DOMAIN
from homeassistant.config_entries import ConfigEntryState from homeassistant.config_entries import ConfigEntryState
@ -13,7 +15,9 @@ from homeassistant.util.dt import utcnow
from . import ( from . import (
DEFAULT_ENTRY_TITLE, DEFAULT_ENTRY_TITLE,
DEFAULT_ENTRY_TITLE_PARTIAL,
FLUX_DISCOVERY, FLUX_DISCOVERY,
FLUX_DISCOVERY_PARTIAL,
IP_ADDRESS, IP_ADDRESS,
MAC_ADDRESS, MAC_ADDRESS,
_patch_discovery, _patch_discovery,
@ -67,8 +71,15 @@ async def test_config_entry_retry(hass: HomeAssistant) -> None:
assert config_entry.state == ConfigEntryState.SETUP_RETRY assert config_entry.state == ConfigEntryState.SETUP_RETRY
@pytest.mark.parametrize(
"discovery,title",
[
(FLUX_DISCOVERY, DEFAULT_ENTRY_TITLE),
(FLUX_DISCOVERY_PARTIAL, DEFAULT_ENTRY_TITLE_PARTIAL),
],
)
async def test_config_entry_fills_unique_id_with_directed_discovery( async def test_config_entry_fills_unique_id_with_directed_discovery(
hass: HomeAssistant, hass: HomeAssistant, discovery: dict[str, str], title: str
) -> None: ) -> None:
"""Test that the unique id is added if its missing via directed (not broadcast) discovery.""" """Test that the unique id is added if its missing via directed (not broadcast) discovery."""
config_entry = MockConfigEntry( config_entry = MockConfigEntry(
@ -78,7 +89,7 @@ async def test_config_entry_fills_unique_id_with_directed_discovery(
async def _discovery(self, *args, address=None, **kwargs): async def _discovery(self, *args, address=None, **kwargs):
# Only return discovery results when doing directed discovery # Only return discovery results when doing directed discovery
return [FLUX_DISCOVERY] if address == IP_ADDRESS else [] return [discovery] if address == IP_ADDRESS else []
with patch( with patch(
"homeassistant.components.flux_led.AIOBulbScanner.async_scan", new=_discovery "homeassistant.components.flux_led.AIOBulbScanner.async_scan", new=_discovery
@ -88,5 +99,5 @@ async def test_config_entry_fills_unique_id_with_directed_discovery(
assert config_entry.state == ConfigEntryState.LOADED assert config_entry.state == ConfigEntryState.LOADED
assert config_entry.unique_id == MAC_ADDRESS assert config_entry.unique_id == MAC_ADDRESS
assert config_entry.data[CONF_NAME] == DEFAULT_ENTRY_TITLE assert config_entry.data[CONF_NAME] == title
assert config_entry.title == DEFAULT_ENTRY_TITLE assert config_entry.title == title

View File

@ -85,7 +85,7 @@ async def test_light_unique_id(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.az120444_aabbccddeeff" entity_id = "light.rgbw_controller_ddeeff"
entity_registry = er.async_get(hass) entity_registry = er.async_get(hass)
assert entity_registry.async_get(entity_id).unique_id == MAC_ADDRESS assert entity_registry.async_get(entity_id).unique_id == MAC_ADDRESS
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
@ -105,7 +105,7 @@ async def test_light_goes_unavailable_and_recovers(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.az120444_aabbccddeeff" entity_id = "light.rgbw_controller_ddeeff"
entity_registry = er.async_get(hass) entity_registry = er.async_get(hass)
assert entity_registry.async_get(entity_id).unique_id == MAC_ADDRESS assert entity_registry.async_get(entity_id).unique_id == MAC_ADDRESS
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
@ -137,7 +137,7 @@ async def test_light_no_unique_id(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.az120444_aabbccddeeff" entity_id = "light.rgbw_controller_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) is None
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
@ -195,7 +195,7 @@ async def test_rgb_light(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.az120444_aabbccddeeff" entity_id = "light.rgbw_controller_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -301,7 +301,7 @@ async def test_rgb_cct_light(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.az120444_aabbccddeeff" entity_id = "light.rgbw_controller_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -420,7 +420,7 @@ async def test_rgbw_light(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.az120444_aabbccddeeff" entity_id = "light.rgbw_controller_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -522,7 +522,7 @@ async def test_rgb_or_w_light(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.az120444_aabbccddeeff" entity_id = "light.rgbw_controller_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -633,7 +633,7 @@ async def test_rgbcw_light(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.az120444_aabbccddeeff" entity_id = "light.rgbw_controller_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -762,7 +762,7 @@ async def test_white_light(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.az120444_aabbccddeeff" entity_id = "light.rgbw_controller_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -817,7 +817,7 @@ async def test_rgb_light_custom_effects(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.az120444_aabbccddeeff" entity_id = "light.rgbw_controller_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -899,7 +899,7 @@ async def test_rgb_light_custom_effects_invalid_colors(
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.az120444_aabbccddeeff" entity_id = "light.rgbw_controller_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -928,7 +928,7 @@ async def test_rgb_light_custom_effect_via_service(
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.az120444_aabbccddeeff" entity_id = "light.rgbw_controller_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -1073,7 +1073,7 @@ async def test_addressable_light(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "light.az120444_aabbccddeeff" entity_id = "light.rgbw_controller_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON

View File

@ -45,7 +45,7 @@ async def test_number_unique_id(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "number.az120444_aabbccddeeff_effect_speed" entity_id = "number.rgbw_controller_ddeeff_effect_speed"
entity_registry = er.async_get(hass) entity_registry = er.async_get(hass)
assert entity_registry.async_get(entity_id).unique_id == MAC_ADDRESS assert entity_registry.async_get(entity_id).unique_id == MAC_ADDRESS
@ -70,8 +70,8 @@ async def test_rgb_light_effect_speed(hass: HomeAssistant) -> None:
await async_mock_device_turn_on(hass, bulb) await async_mock_device_turn_on(hass, bulb)
light_entity_id = "light.az120444_aabbccddeeff" light_entity_id = "light.rgbw_controller_ddeeff"
number_entity_id = "number.az120444_aabbccddeeff_effect_speed" number_entity_id = "number.rgbw_controller_ddeeff_effect_speed"
with pytest.raises(HomeAssistantError): with pytest.raises(HomeAssistantError):
await hass.services.async_call( await hass.services.async_call(
NUMBER_DOMAIN, NUMBER_DOMAIN,
@ -135,8 +135,8 @@ async def test_original_addressable_light_effect_speed(hass: HomeAssistant) -> N
await async_mock_device_turn_on(hass, bulb) await async_mock_device_turn_on(hass, bulb)
light_entity_id = "light.az120444_aabbccddeeff" light_entity_id = "light.rgbw_controller_ddeeff"
number_entity_id = "number.az120444_aabbccddeeff_effect_speed" number_entity_id = "number.rgbw_controller_ddeeff_effect_speed"
state = hass.states.get(light_entity_id) state = hass.states.get(light_entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON
@ -192,8 +192,8 @@ async def test_addressable_light_effect_speed(hass: HomeAssistant) -> None:
await async_mock_device_turn_on(hass, bulb) await async_mock_device_turn_on(hass, bulb)
light_entity_id = "light.az120444_aabbccddeeff" light_entity_id = "light.rgbw_controller_ddeeff"
number_entity_id = "number.az120444_aabbccddeeff_effect_speed" number_entity_id = "number.rgbw_controller_ddeeff_effect_speed"
state = hass.states.get(light_entity_id) state = hass.states.get(light_entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON

View File

@ -39,7 +39,7 @@ async def test_switch_on_off(hass: HomeAssistant) -> None:
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}}) await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done() await hass.async_block_till_done()
entity_id = "switch.az120444_aabbccddeeff" entity_id = "switch.rgbw_controller_ddeeff"
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state.state == STATE_ON assert state.state == STATE_ON