Handle WLED devices with CCT channels (#64581)

This commit is contained in:
Franck Nijhof 2022-01-21 09:19:50 +01:00 committed by GitHub
parent d634a08361
commit 1c94431efe
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 79 additions and 3 deletions

View File

@ -5,7 +5,7 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform from homeassistant.const import Platform
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from .const import DOMAIN from .const import DOMAIN, LOGGER
from .coordinator import WLEDDataUpdateCoordinator from .coordinator import WLEDDataUpdateCoordinator
PLATFORMS = ( PLATFORMS = (
@ -23,6 +23,15 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up WLED from a config entry.""" """Set up WLED from a config entry."""
coordinator = WLEDDataUpdateCoordinator(hass, entry=entry) coordinator = WLEDDataUpdateCoordinator(hass, entry=entry)
await coordinator.async_config_entry_first_refresh() await coordinator.async_config_entry_first_refresh()
if coordinator.data.info.leds.cct:
LOGGER.error(
"WLED device '%s' has a CCT channel, which is not supported by "
"this integration",
entry.title,
)
return False
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
# Set up all platforms for this device/entry. # Set up all platforms for this device/entry.

View File

@ -41,6 +41,8 @@ class WLEDFlowHandler(ConfigFlow, domain=DOMAIN):
except WLEDConnectionError: except WLEDConnectionError:
errors["base"] = "cannot_connect" errors["base"] = "cannot_connect"
else: else:
if device.info.leds.cct:
return self.async_abort(reason="cct_unsupported")
await self.async_set_unique_id(device.info.mac_address) await self.async_set_unique_id(device.info.mac_address)
self._abort_if_unique_id_configured( self._abort_if_unique_id_configured(
updates={CONF_HOST: user_input[CONF_HOST]} updates={CONF_HOST: user_input[CONF_HOST]}
@ -77,6 +79,9 @@ class WLEDFlowHandler(ConfigFlow, domain=DOMAIN):
except WLEDConnectionError: except WLEDConnectionError:
return self.async_abort(reason="cannot_connect") return self.async_abort(reason="cannot_connect")
if self.discovered_device.info.leds.cct:
return self.async_abort(reason="cct_unsupported")
await self.async_set_unique_id(self.discovered_device.info.mac_address) await self.async_set_unique_id(self.discovered_device.info.mac_address)
self._abort_if_unique_id_configured(updates={CONF_HOST: discovery_info.host}) self._abort_if_unique_id_configured(updates={CONF_HOST: discovery_info.host})

View File

@ -113,6 +113,7 @@ class WLEDDataUpdateCoordinator(DataUpdateCoordinator[WLEDDevice]):
# If the device supports a WebSocket, try activating it. # If the device supports a WebSocket, try activating it.
if ( if (
device.info.websocket is not None device.info.websocket is not None
and device.info.leds.cct is not True
and not self.wled.connected and not self.wled.connected
and not self.unsub and not self.unsub
): ):

View File

@ -18,7 +18,8 @@
}, },
"abort": { "abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]", "already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]" "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"cct_unsupported": "This WLED device uses CCT channels, which is not supported by this integration"
} }
}, },
"options": { "options": {

View File

@ -2,7 +2,8 @@
"config": { "config": {
"abort": { "abort": {
"already_configured": "Device is already configured", "already_configured": "Device is already configured",
"cannot_connect": "Failed to connect" "cannot_connect": "Failed to connect",
"cct_unsupported": "This WLED device uses CCT channels, which is not supported by this integration"
}, },
"error": { "error": {
"cannot_connect": "Failed to connect" "cannot_connect": "Failed to connect"

View File

@ -139,6 +139,23 @@ async def test_user_device_exists_abort(
assert result.get("reason") == "already_configured" assert result.get("reason") == "already_configured"
async def test_user_with_cct_channel_abort(
hass: HomeAssistant,
mock_wled_config_flow: MagicMock,
) -> None:
"""Test we abort user flow if WLED device uses a CCT channel."""
mock_wled_config_flow.update.return_value.info.leds.cct = True
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_HOST: "192.168.1.123"},
)
assert result.get("type") == RESULT_TYPE_ABORT
assert result.get("reason") == "cct_unsupported"
async def test_zeroconf_without_mac_device_exists_abort( async def test_zeroconf_without_mac_device_exists_abort(
hass: HomeAssistant, hass: HomeAssistant,
mock_config_entry: MockConfigEntry, mock_config_entry: MockConfigEntry,
@ -187,6 +204,30 @@ async def test_zeroconf_with_mac_device_exists_abort(
assert result.get("reason") == "already_configured" assert result.get("reason") == "already_configured"
async def test_zeroconf_with_cct_channel_abort(
hass: HomeAssistant,
mock_wled_config_flow: MagicMock,
) -> None:
"""Test we abort zeroconf flow if WLED device uses a CCT channel."""
mock_wled_config_flow.update.return_value.info.leds.cct = True
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=zeroconf.ZeroconfServiceInfo(
host="192.168.1.123",
hostname="example.local.",
name="mock_name",
port=None,
properties={CONF_MAC: "aabbccddeeff"},
type="mock_type",
),
)
assert result.get("type") == RESULT_TYPE_ABORT
assert result.get("reason") == "cct_unsupported"
async def test_options_flow( async def test_options_flow(
hass: HomeAssistant, mock_config_entry: MockConfigEntry hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None: ) -> None:

View File

@ -68,3 +68,21 @@ async def test_setting_unique_id(
"""Test we set unique ID if not set yet.""" """Test we set unique ID if not set yet."""
assert hass.data[DOMAIN] assert hass.data[DOMAIN]
assert init_integration.unique_id == "aabbccddeeff" assert init_integration.unique_id == "aabbccddeeff"
async def test_error_config_entry_with_cct_channel(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_wled: AsyncMock,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test the WLED fails entry setup with a CCT channel."""
mock_wled.update.return_value.info.leds.cct = True
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
# Ensure config entry is errored and are connected and disconnected
assert mock_config_entry.state is ConfigEntryState.SETUP_ERROR
assert "has a CCT channel, which is not supported" in caplog.text