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

Add authentication to tplink integration for newer devices (#105143)

Browse source 
* Add authentication flows to tplink integration to enable newer device protocol support

* Add websession passing to tplink integration discover methods

* Use SmartDevice.connect()

* Update to use DeviceConfig

* Use credential hashes

* Bump python-kasa to 0.6.0.dev0

* Fix tests and address review comments

* Add autodetection for L530, P110, and L900

This adds mac address prefixes for the devices I have.
The wildcards are left quite lax assuming different series may share the same prefix.

* Bump tplink to 0.6.0.dev1

* Add config flow tests

* Use short_mac if alias is None and try legacy connect on discovery timeout

* Add config_flow tests

* Add init tests

* Migrate to aiohttp

* add some more ouis

* final

* ip change fix

* add fixmes

* fix O(n) searching

* fix O(n) searching

* move code that cannot fail outside of try block

* fix missing reauth_successful string

* add doc strings, cleanups

* error message by password

* dry

* adjust discovery timeout

* integration discovery already formats mac

* tweaks

* cleanups

* cleanups

* Update post review and fix broken tests

* Fix TODOs and FIXMEs in test_config_flow

* Add pragma no cover

* bump, apply suggestions

* remove no cover

* use iden check

* Apply suggestions from code review

* Fix branched test and update integration title

* legacy typing

* Update homeassistant/components/tplink/__init__.py

* lint

* Remove more unused consts

* Update test docstrings

* Add sdb9696 to tplink codeowners

* Update docstring on test for invalid DeviceConfig

* Update test stored credentials test

---------

Co-authored-by: Teemu Rytilahti <tpr@iki.fi>
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Steven B 2024-01-21 15:25:12 +00:00 • committed by GitHub
parent c3da51db4e
commit 9b3d3b3b2d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
18 changed files with 1661 additions and 161 deletions
Download patch file Download diff file Expand all files Collapse all files

4
CODEOWNERS
View file

@ -1371,8 +1371,8 @@ build.json @home-assistant/supervisor
/tests/components/tomorrowio/ @raman325 @lymanepp
/homeassistant/components/totalconnect/ @austinmroczek
/tests/components/totalconnect/ @austinmroczek
/homeassistant/components/tplink/ @rytilahti @thegardenmonkey @bdraco
/tests/components/tplink/ @rytilahti @thegardenmonkey @bdraco
/homeassistant/components/tplink/ @rytilahti @thegardenmonkey @bdraco @sdb9696
/tests/components/tplink/ @rytilahti @thegardenmonkey @bdraco @sdb9696
/homeassistant/components/tplink_omada/ @MarkGodwin
/tests/components/tplink_omada/ @MarkGodwin
/homeassistant/components/traccar/ @ludeeus

125
homeassistant/components/tplink/__init__.py
View file

@ -3,37 +3,66 @@ from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
from typing import Any
from kasa import SmartDevice, SmartDeviceException
from kasa.discover import Discover
from aiohttp import ClientSession
from kasa import (
AuthenticationException,
Credentials,
DeviceConfig,
Discover,
SmartDevice,
SmartDeviceException,
)
from kasa.httpclient import get_cookie_jar
from homeassistant import config_entries
from homeassistant.components import network
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_ALIAS,
CONF_AUTHENTICATION,
CONF_HOST,
CONF_MAC,
CONF_NAME,
CONF_MODEL,
CONF_PASSWORD,
CONF_USERNAME,
EVENT_HOMEASSISTANT_STARTED,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import (
config_validation as cv,
device_registry as dr,
discovery_flow,
)
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN, PLATFORMS
from .const import (
CONF_DEVICE_CONFIG,
CONNECT_TIMEOUT,
DISCOVERY_TIMEOUT,
DOMAIN,
PLATFORMS,
)
from .coordinator import TPLinkDataUpdateCoordinator
from .models import TPLinkData
DISCOVERY_INTERVAL = timedelta(minutes=15)
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
_LOGGER = logging.getLogger(__name__)
def create_async_tplink_clientsession(hass: HomeAssistant) -> ClientSession:
"""Return aiohttp clientsession with cookie jar configured."""
return async_create_clientsession(
hass, verify_ssl=False, cookie_jar=get_cookie_jar()
)
@callback
def async_trigger_discovery(
@ -47,17 +76,31 @@ def async_trigger_discovery(
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
data={
CONF_NAME: device.alias,
CONF_ALIAS: device.alias or mac_alias(device.mac),
CONF_HOST: device.host,
CONF_MAC: formatted_mac,
CONF_DEVICE_CONFIG: device.config.to_dict(
credentials_hash=device.credentials_hash,
exclude_credentials=True,
),
},
)
async def async_discover_devices(hass: HomeAssistant) -> dict[str, SmartDevice]:
"""Discover TPLink devices on configured network interfaces."""
credentials = await get_credentials(hass)
broadcast_addresses = await network.async_get_ipv4_broadcast_addresses(hass)
tasks = [Discover.discover(target=str(address)) for address in broadcast_addresses]
tasks = [
Discover.discover(
target=str(address),
discovery_timeout=DISCOVERY_TIMEOUT,
timeout=CONNECT_TIMEOUT,
credentials=credentials,
)
for address in broadcast_addresses
]
discovered_devices: dict[str, SmartDevice] = {}
for device_list in await asyncio.gather(*tasks):
for device in device_list.values():
@ -67,7 +110,7 @@ async def async_discover_devices(hass: HomeAssistant) -> dict[str, SmartDevice]:
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the TP-Link component."""
hass.data[DOMAIN] = {}
hass.data.setdefault(DOMAIN, {})
if discovered_devices := await async_discover_devices(hass):
async_trigger_discovery(hass, discovered_devices)
@ -86,12 +129,51 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up TPLink from a config entry."""
host = entry.data[CONF_HOST]
host: str = entry.data[CONF_HOST]
credentials = await get_credentials(hass)
config: DeviceConfig | None = None
if config_dict := entry.data.get(CONF_DEVICE_CONFIG):
try:
config = DeviceConfig.from_dict(config_dict)
except SmartDeviceException:
_LOGGER.warning(
"Invalid connection type dict for %s: %s", host, config_dict
)
if not config:
config = DeviceConfig(host)
config.timeout = CONNECT_TIMEOUT
if config.uses_http is True:
config.http_client = create_async_tplink_clientsession(hass)
if credentials:
config.credentials = credentials
try:
device: SmartDevice = await Discover.discover_single(host, timeout=10)
device: SmartDevice = await SmartDevice.connect(config=config)
except AuthenticationException as ex:
raise ConfigEntryAuthFailed from ex
except SmartDeviceException as ex:
raise ConfigEntryNotReady from ex
device_config_dict = device.config.to_dict(
credentials_hash=device.credentials_hash, exclude_credentials=True
)
updates: dict[str, Any] = {}
if device_config_dict != config_dict:
updates[CONF_DEVICE_CONFIG] = device_config_dict
if entry.data.get(CONF_ALIAS) != device.alias:
updates[CONF_ALIAS] = device.alias
if entry.data.get(CONF_MODEL) != device.model:
updates[CONF_MODEL] = device.model
if updates:
hass.config_entries.async_update_entry(
entry,
data={
**entry.data,
**updates,
},
)
found_mac = dr.format_mac(device.mac)
if found_mac != entry.unique_id:
# If the mac address of the device does not match the unique_id
@ -130,6 +212,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass_data.pop(entry.entry_id)
await device.protocol.close()
return unload_ok
@ -141,3 +224,25 @@ def legacy_device_id(device: SmartDevice) -> str:
if "_" not in device_id:
return device_id
return device_id.split("_")[1]
async def get_credentials(hass: HomeAssistant) -> Credentials | None:
"""Retrieve the credentials from hass data."""
if DOMAIN in hass.data and CONF_AUTHENTICATION in hass.data[DOMAIN]:
auth = hass.data[DOMAIN][CONF_AUTHENTICATION]
return Credentials(auth[CONF_USERNAME], auth[CONF_PASSWORD])
return None
async def set_credentials(hass: HomeAssistant, username: str, password: str) -> None:
"""Save the credentials to HASS data."""
hass.data.setdefault(DOMAIN, {})[CONF_AUTHENTICATION] = {
CONF_USERNAME: username,
CONF_PASSWORD: password,
}
def mac_alias(mac: str) -> str:
"""Convert a MAC address to a short address for the UI."""
return mac.replace(":", "")[-4:].upper()

330
homeassistant/components/tplink/config_flow.py
View file

@ -1,28 +1,57 @@
"""Config flow for TP-Link."""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from kasa import SmartDevice, SmartDeviceException
from kasa.discover import Discover
from kasa import (
AuthenticationException,
Credentials,
DeviceConfig,
Discover,
SmartDevice,
SmartDeviceException,
TimeoutException,
)
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import dhcp
from homeassistant.const import CONF_DEVICE, CONF_HOST, CONF_MAC
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntry, ConfigEntryState
from homeassistant.const import (
CONF_ALIAS,
CONF_DEVICE,
CONF_HOST,
CONF_MAC,
CONF_MODEL,
CONF_PASSWORD,
CONF_USERNAME,
)
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.data_entry_flow import AbortFlow, FlowResult
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.typing import DiscoveryInfoType
from . import async_discover_devices
from .const import DOMAIN
from . import (
async_discover_devices,
create_async_tplink_clientsession,
get_credentials,
mac_alias,
set_credentials,
)
from .const import CONF_DEVICE_CONFIG, CONNECT_TIMEOUT, DOMAIN
STEP_AUTH_DATA_SCHEMA = vol.Schema(
{vol.Required(CONF_USERNAME): str, vol.Required(CONF_PASSWORD): str}
)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for tplink."""
VERSION = 1
MINOR_VERSION = 2
reauth_entry: ConfigEntry | None = None
def __init__(self) -> None:
"""Initialize the config flow."""
@ -40,27 +69,114 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
) -> FlowResult:
"""Handle integration discovery."""
return await self._async_handle_discovery(
discovery_info[CONF_HOST], discovery_info[CONF_MAC]
discovery_info[CONF_HOST],
discovery_info[CONF_MAC],
discovery_info[CONF_DEVICE_CONFIG],
)
async def _async_handle_discovery(self, host: str, mac: str) -> FlowResult:
@callback
def _update_config_if_entry_in_setup_error(
self, entry: ConfigEntry, host: str, config: dict
) -> None:
"""If discovery encounters a device that is in SETUP_ERROR update the device config."""
if entry.state is not ConfigEntryState.SETUP_ERROR:
return
entry_data = entry.data
entry_config_dict = entry_data.get(CONF_DEVICE_CONFIG)
if entry_config_dict == config and entry_data[CONF_HOST] == host:
return
self.hass.config_entries.async_update_entry(
entry, data={**entry.data, CONF_DEVICE_CONFIG: config, CONF_HOST: host}
)
self.hass.async_create_task(
self.hass.config_entries.async_reload(entry.entry_id),
f"config entry reload {entry.title} {entry.domain} {entry.entry_id}",
)
raise AbortFlow("already_configured")
async def _async_handle_discovery(
self, host: str, formatted_mac: str, config: dict | None = None
) -> FlowResult:
"""Handle any discovery."""
await self.async_set_unique_id(dr.format_mac(mac))
current_entry = await self.async_set_unique_id(
formatted_mac, raise_on_progress=False
)
if config and current_entry:
self._update_config_if_entry_in_setup_error(current_entry, host, config)
self._abort_if_unique_id_configured(updates={CONF_HOST: host})
self._async_abort_entries_match({CONF_HOST: host})
self.context[CONF_HOST] = host
for progress in self._async_in_progress():
if progress.get("context", {}).get(CONF_HOST) == host:
return self.async_abort(reason="already_in_progress")
credentials = await get_credentials(self.hass)
try:
self._discovered_device = await self._async_try_connect(
host, raise_on_progress=True
await self._async_try_discover_and_update(
host, credentials, raise_on_progress=True
)
except AuthenticationException:
return await self.async_step_discovery_auth_confirm()
except SmartDeviceException:
return self.async_abort(reason="cannot_connect")
return await self.async_step_discovery_confirm()
async def async_step_discovery_auth_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Dialog that informs the user that auth is required."""
assert self._discovered_device is not None
errors = {}
credentials = await get_credentials(self.hass)
if credentials and credentials != self._discovered_device.config.credentials:
try:
device = await self._async_try_connect(
self._discovered_device, credentials
)
except AuthenticationException:
pass # Authentication exceptions should continue to the rest of the step
else:
self._discovered_device = device
return await self.async_step_discovery_confirm()
if user_input:
username = user_input[CONF_USERNAME]
password = user_input[CONF_PASSWORD]
credentials = Credentials(username, password)
try:
device = await self._async_try_connect(
self._discovered_device, credentials
)
except AuthenticationException:
errors[CONF_PASSWORD] = "invalid_auth"
except SmartDeviceException:
errors["base"] = "cannot_connect"
else:
self._discovered_device = device
await set_credentials(self.hass, username, password)
self.hass.async_create_task(self._async_reload_requires_auth_entries())
return await self.async_step_discovery_confirm()
placeholders = self._async_make_placeholders_from_discovery()
self.context["title_placeholders"] = placeholders
return self.async_show_form(
step_id="discovery_auth_confirm",
data_schema=STEP_AUTH_DATA_SCHEMA,
errors=errors,
description_placeholders=placeholders,
)
def _async_make_placeholders_from_discovery(self) -> dict[str, str]:
"""Make placeholders for the discovery steps."""
discovered_device = self._discovered_device
assert discovered_device is not None
return {
"name": discovered_device.alias or mac_alias(discovered_device.mac),
"model": discovered_device.model,
"host": discovered_device.host,
}
async def async_step_discovery_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
@ -70,11 +186,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self._async_create_entry_from_device(self._discovered_device)
self._set_confirm_only()
placeholders = {
"name": self._discovered_device.alias,
"model": self._discovered_device.model,
"host": self._discovered_device.host,
}
placeholders = self._async_make_placeholders_from_discovery()
self.context["title_placeholders"] = placeholders
return self.async_show_form(
step_id="discovery_confirm", description_placeholders=placeholders
@ -88,8 +200,15 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if user_input is not None:
if not (host := user_input[CONF_HOST]):
return await self.async_step_pick_device()
self._async_abort_entries_match({CONF_HOST: host})
self.context[CONF_HOST] = host
credentials = await get_credentials(self.hass)
try:
device = await self._async_try_connect(host, raise_on_progress=False)
device = await self._async_try_discover_and_update(
host, credentials, raise_on_progress=False
)
except AuthenticationException:
return await self.async_step_user_auth_confirm()
except SmartDeviceException:
errors["base"] = "cannot_connect"
else:
@ -101,6 +220,37 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_user_auth_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Dialog that informs the user that auth is required."""
errors = {}
host = self.context[CONF_HOST]
assert self._discovered_device is not None
if user_input:
username = user_input[CONF_USERNAME]
password = user_input[CONF_PASSWORD]
credentials = Credentials(username, password)
try:
device = await self._async_try_connect(
self._discovered_device, credentials
)
except AuthenticationException:
errors[CONF_PASSWORD] = "invalid_auth"
except SmartDeviceException:
errors["base"] = "cannot_connect"
else:
await set_credentials(self.hass, username, password)
self.hass.async_create_task(self._async_reload_requires_auth_entries())
return self._async_create_entry_from_device(device)
return self.async_show_form(
step_id="user_auth_confirm",
data_schema=STEP_AUTH_DATA_SCHEMA,
errors=errors,
description_placeholders={CONF_HOST: host},
)
async def async_step_pick_device(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
@ -108,7 +258,21 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if user_input is not None:
mac = user_input[CONF_DEVICE]
await self.async_set_unique_id(mac, raise_on_progress=False)
return self._async_create_entry_from_device(self._discovered_devices[mac])
self._discovered_device = self._discovered_devices[mac]
host = self._discovered_device.host
self.context[CONF_HOST] = host
credentials = await get_credentials(self.hass)
try:
device = await self._async_try_connect(
self._discovered_device, credentials
)
except AuthenticationException:
return await self.async_step_user_auth_confirm()
except SmartDeviceException:
return self.async_abort(reason="cannot_connect")
return self._async_create_entry_from_device(device)
configured_devices = {
entry.unique_id for entry in self._async_current_entries()
@ -116,7 +280,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._discovered_devices = await async_discover_devices(self.hass)
devices_name = {
formatted_mac: (
f"{device.alias} {device.model} ({device.host}) {formatted_mac}"
f"{device.alias or mac_alias(device.mac)} {device.model} ({device.host}) {formatted_mac}"
)
for formatted_mac, device in self._discovered_devices.items()
if formatted_mac not in configured_devices
@ -129,6 +293,25 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
data_schema=vol.Schema({vol.Required(CONF_DEVICE): vol.In(devices_name)}),
)
async def _async_reload_requires_auth_entries(self) -> None:
"""Reload any in progress config flow that now have credentials."""
_config_entries = self.hass.config_entries
if reauth_entry := self.reauth_entry:
await _config_entries.async_reload(reauth_entry.entry_id)
for flow in _config_entries.flow.async_progress_by_handler(
DOMAIN, include_uninitialized=True
):
context: dict[str, Any] = flow["context"]
if context.get("source") != SOURCE_REAUTH:
continue
entry_id: str = context["entry_id"]
if entry := _config_entries.async_get_entry(entry_id):
await _config_entries.async_reload(entry.entry_id)
if entry.state is ConfigEntryState.LOADED:
_config_entries.flow.async_abort(flow["flow_id"])
@callback
def _async_create_entry_from_device(self, device: SmartDevice) -> FlowResult:
"""Create a config entry from a smart device."""
@ -137,16 +320,113 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
title=f"{device.alias} {device.model}",
data={
CONF_HOST: device.host,
CONF_ALIAS: device.alias,
CONF_MODEL: device.model,
CONF_DEVICE_CONFIG: device.config.to_dict(
credentials_hash=device.credentials_hash,
exclude_credentials=True,
),
},
)
async def _async_try_discover_and_update(
self,
host: str,
credentials: Credentials | None,
raise_on_progress: bool,
) -> SmartDevice:
"""Try to discover the device and call update.
Will try to connect to legacy devices if discovery fails.
"""
try:
self._discovered_device = await Discover.discover_single(
host, credentials=credentials
)
except TimeoutException:
# Try connect() to legacy devices if discovery fails
self._discovered_device = await SmartDevice.connect(
config=DeviceConfig(host)
)
else:
if self._discovered_device.config.uses_http:
self._discovered_device.config.http_client = (
create_async_tplink_clientsession(self.hass)
)
await self._discovered_device.update()
await self.async_set_unique_id(
dr.format_mac(self._discovered_device.mac),
raise_on_progress=raise_on_progress,
)
return self._discovered_device
async def _async_try_connect(
self, host: str, raise_on_progress: bool = True
self,
discovered_device: SmartDevice,
credentials: Credentials | None,
) -> SmartDevice:
"""Try to connect."""
self._async_abort_entries_match({CONF_HOST: host})
device: SmartDevice = await Discover.discover_single(host)
self._async_abort_entries_match({CONF_HOST: discovered_device.host})
config = discovered_device.config
if credentials:
config.credentials = credentials
config.timeout = CONNECT_TIMEOUT
if config.uses_http:
config.http_client = create_async_tplink_clientsession(self.hass)
self._discovered_device = await SmartDevice.connect(config=config)
await self.async_set_unique_id(
dr.format_mac(device.mac), raise_on_progress=raise_on_progress
dr.format_mac(self._discovered_device.mac),
raise_on_progress=False,
)
return self._discovered_device
async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
"""Start the reauthentication flow if the device needs updated credentials."""
self.reauth_entry = self.hass.config_entries.async_get_entry(
self.context["entry_id"]
)
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Dialog that informs the user that reauth is required."""
errors: dict[str, str] = {}
reauth_entry = self.reauth_entry
assert reauth_entry is not None
entry_data = reauth_entry.data
host = entry_data[CONF_HOST]
if user_input:
username = user_input[CONF_USERNAME]
password = user_input[CONF_PASSWORD]
credentials = Credentials(username, password)
try:
await self._async_try_discover_and_update(
host,
credentials=credentials,
raise_on_progress=True,
)
except AuthenticationException:
errors[CONF_PASSWORD] = "invalid_auth"
except SmartDeviceException:
errors["base"] = "cannot_connect"
else:
await set_credentials(self.hass, username, password)
self.hass.async_create_task(self._async_reload_requires_auth_entries())
return self.async_abort(reason="reauth_successful")
# Old config entries will not have these values.
alias = entry_data.get(CONF_ALIAS) or "unknown"
model = entry_data.get(CONF_MODEL) or "unknown"
placeholders = {"name": alias, "model": model, "host": host}
self.context["title_placeholders"] = placeholders
return self.async_show_form(
step_id="reauth_confirm",
data_schema=STEP_AUTH_DATA_SCHEMA,
errors=errors,
description_placeholders=placeholders,
)
return device

9
homeassistant/components/tplink/const.py
View file

@ -7,15 +7,14 @@ from homeassistant.const import Platform
DOMAIN = "tplink"
DISCOVERY_TIMEOUT = 5 # Home Assistant will complain if startup takes > 10s
CONNECT_TIMEOUT = 5
ATTR_CURRENT_A: Final = "current_a"
ATTR_CURRENT_POWER_W: Final = "current_power_w"
ATTR_TODAY_ENERGY_KWH: Final = "today_energy_kwh"
ATTR_TOTAL_ENERGY_KWH: Final = "total_energy_kwh"
CONF_DIMMER: Final = "dimmer"
CONF_LIGHT: Final = "light"
CONF_STRIP: Final = "strip"
CONF_SWITCH: Final = "switch"
CONF_SENSOR: Final = "sensor"
CONF_DEVICE_CONFIG: Final = "device_config"
PLATFORMS: Final = [Platform.LIGHT, Platform.SENSOR, Platform.SWITCH]

34
homeassistant/components/tplink/manifest.json
View file

@ -1,13 +1,17 @@
{
"domain": "tplink",
"name": "TP-Link Kasa Smart",
"codeowners": ["@rytilahti", "@thegardenmonkey", "@bdraco"],
"name": "TP-Link Smart Home",
"codeowners": ["@rytilahti", "@thegardenmonkey", "@bdraco", "@sdb9696"],
"config_flow": true,
"dependencies": ["network"],
"dhcp": [
{
"registered_devices": true
},
{
"hostname": "e[sp]*",
"macaddress": "3C52A1*"
},
{
"hostname": "e[sp]*",
"macaddress": "54AF97*"
@ -32,6 +36,10 @@
"hostname": "hs*",
"macaddress": "9C5322*"
},
{
"hostname": "k[lps]*",
"macaddress": "5091E3*"
},
{
"hostname": "k[lps]*",
"macaddress": "9C5322*"
@ -163,11 +171,31 @@
{
"hostname": "k[lps]*",
"macaddress": "1C61B4*"
},
{
"hostname": "l5*",
"macaddress": "5CE931*"
},
{
"hostname": "p1*",
"macaddress": "482254*"
},
{
"hostname": "p1*",
"macaddress": "30DE4B*"
},
{
"hostname": "l9*",
"macaddress": "A842A1*"
},
{
"hostname": "l9*",
"macaddress": "3460F9*"
}
],
"documentation": "https://www.home-assistant.io/integrations/tplink",
"iot_class": "local_polling",
"loggers": ["kasa"],
"quality_scale": "platinum",
"requirements": ["python-kasa[speedups]==0.5.4"]
"requirements": ["python-kasa[speedups]==0.6.0.1"]
}

31
homeassistant/components/tplink/strings.json
View file

@ -18,6 +18,34 @@
},
"discovery_confirm": {
"description": "Do you want to set up {name} {model} ({host})?"
},
"user_auth_confirm": {
"title": "Authenticate",
"description": "The device requires authentication, please input your credentials below.",
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
}
},
"discovery_auth_confirm": {
"title": "Authenticate",
"description": "The device requires authentication, please input your credentials below.",
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
}
},
"reauth": {
"title": "[%key:common::config_flow::title::reauth%]",
"description": "The device needs updated credentials, please input your credentials below."
},
"reauth_confirm": {
"title": "[%key:common::config_flow::title::reauth%]",
"description": "The device needs updated credentials, please input your credentials below.",
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
}
}
},
"error": {
@ -25,7 +53,8 @@
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]"
"no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
}
},
"entity": {

6
homeassistant/components/tplink/switch.py
View file

@ -41,7 +41,9 @@ async def async_setup_entry(
elif device.is_plug:
entities.append(SmartPlugSwitch(device, parent_coordinator))
entities.append(SmartPlugLedSwitch(device, parent_coordinator))
# this will be removed on the led is implemented
if hasattr(device, "led"):
entities.append(SmartPlugLedSwitch(device, parent_coordinator))
async_add_entities(entities)
@ -86,7 +88,7 @@ class SmartPlugLedSwitch(CoordinatedTPLinkEntity, SwitchEntity):
class SmartPlugSwitch(CoordinatedTPLinkEntity, SwitchEntity):
"""Representation of a TPLink Smart Plug switch."""
_attr_name = None
_attr_name: str | None = None
def __init__(
self,

35
homeassistant/generated/dhcp.py
View file

@ -603,6 +603,11 @@ DHCP: list[dict[str, str | bool]] = [
"domain": "tplink",
"registered_devices": True,
},
{
"domain": "tplink",
"hostname": "e[sp]*",
"macaddress": "3C52A1*",
},
{
"domain": "tplink",
"hostname": "e[sp]*",
@ -633,6 +638,11 @@ DHCP: list[dict[str, str | bool]] = [
"hostname": "hs*",
"macaddress": "9C5322*",
},
{
"domain": "tplink",
"hostname": "k[lps]*",
"macaddress": "5091E3*",
},
{
"domain": "tplink",
"hostname": "k[lps]*",
@ -798,6 +808,31 @@ DHCP: list[dict[str, str | bool]] = [
"hostname": "k[lps]*",
"macaddress": "1C61B4*",
},
{
"domain": "tplink",
"hostname": "l5*",
"macaddress": "5CE931*",
},
{
"domain": "tplink",
"hostname": "p1*",
"macaddress": "482254*",
},
{
"domain": "tplink",
"hostname": "p1*",
"macaddress": "30DE4B*",
},
{
"domain": "tplink",
"hostname": "l9*",
"macaddress": "A842A1*",
},
{
"domain": "tplink",
"hostname": "l9*",
"macaddress": "3460F9*",
},
{
"domain": "tuya",
"macaddress": "105A17*",

2
homeassistant/generated/integrations.json
View file

@ -6086,7 +6086,7 @@
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_polling",
"name": "TP-Link Kasa Smart"
"name": "TP-Link Smart Home"
},
"tplink_omada": {
"integration_type": "hub",

2
requirements_all.txt
View file

@ -2213,7 +2213,7 @@ python-join-api==0.0.9
python-juicenet==1.1.0
# homeassistant.components.tplink
python-kasa[speedups]==0.5.4
python-kasa[speedups]==0.6.0.1
# homeassistant.components.lirc
# python-lirc==1.2.3

2
requirements_test_all.txt
View file

@ -1683,7 +1683,7 @@ python-izone==1.2.9
python-juicenet==1.1.0
# homeassistant.components.tplink
python-kasa[speedups]==0.5.4
python-kasa[speedups]==0.6.0.1
# homeassistant.components.matter
python-matter-server==5.1.1

99
tests/components/tplink/__init__.py
View file

@ -3,6 +3,10 @@
from unittest.mock import AsyncMock, MagicMock, patch
from kasa import (
ConnectionType,
DeviceConfig,
DeviceFamilyType,
EncryptType,
SmartBulb,
SmartDevice,
SmartDimmer,
@ -13,7 +17,13 @@ from kasa import (
from kasa.exceptions import SmartDeviceException
from kasa.protocol import TPLinkSmartHomeProtocol
from homeassistant.components.tplink import CONF_HOST
from homeassistant.components.tplink import (
CONF_ALIAS,
CONF_DEVICE_CONFIG,
CONF_HOST,
CONF_MODEL,
Credentials,
)
from homeassistant.components.tplink.const import DOMAIN
from homeassistant.core import HomeAssistant
@ -22,10 +32,61 @@ from tests.common import MockConfigEntry
MODULE = "homeassistant.components.tplink"
MODULE_CONFIG_FLOW = "homeassistant.components.tplink.config_flow"
IP_ADDRESS = "127.0.0.1"
IP_ADDRESS2 = "127.0.0.2"
ALIAS = "My Bulb"
MODEL = "HS100"
MAC_ADDRESS = "aa:bb:cc:dd:ee:ff"
MAC_ADDRESS2 = "11:22:33:44:55:66"
DEFAULT_ENTRY_TITLE = f"{ALIAS} {MODEL}"
CREDENTIALS_HASH_LEGACY = ""
DEVICE_CONFIG_LEGACY = DeviceConfig(IP_ADDRESS)
DEVICE_CONFIG_DICT_LEGACY = DEVICE_CONFIG_LEGACY.to_dict(
credentials_hash=CREDENTIALS_HASH_LEGACY, exclude_credentials=True
)
CREDENTIALS = Credentials("foo", "bar")
CREDENTIALS_HASH_AUTH = "abcdefghijklmnopqrstuv=="
DEVICE_CONFIG_AUTH = DeviceConfig(
IP_ADDRESS,
credentials=CREDENTIALS,
connection_type=ConnectionType(
DeviceFamilyType.IotSmartPlugSwitch, EncryptType.Klap
),
uses_http=True,
)
DEVICE_CONFIG_AUTH2 = DeviceConfig(
IP_ADDRESS2,
credentials=CREDENTIALS,
connection_type=ConnectionType(
DeviceFamilyType.IotSmartPlugSwitch, EncryptType.Klap
),
uses_http=True,
)
DEVICE_CONFIG_DICT_AUTH = DEVICE_CONFIG_AUTH.to_dict(
credentials_hash=CREDENTIALS_HASH_AUTH, exclude_credentials=True
)
DEVICE_CONFIG_DICT_AUTH2 = DEVICE_CONFIG_AUTH2.to_dict(
credentials_hash=CREDENTIALS_HASH_AUTH, exclude_credentials=True
)
CREATE_ENTRY_DATA_LEGACY = {
CONF_HOST: IP_ADDRESS,
CONF_ALIAS: ALIAS,
CONF_MODEL: MODEL,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_LEGACY,
}
CREATE_ENTRY_DATA_AUTH = {
CONF_HOST: IP_ADDRESS,
CONF_ALIAS: ALIAS,
CONF_MODEL: MODEL,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_AUTH,
}
CREATE_ENTRY_DATA_AUTH2 = {
CONF_HOST: IP_ADDRESS2,
CONF_ALIAS: ALIAS,
CONF_MODEL: MODEL,
CONF_DEVICE_CONFIG: DEVICE_CONFIG_DICT_AUTH2,
}
def _mock_protocol() -> TPLinkSmartHomeProtocol:
@ -34,11 +95,16 @@ def _mock_protocol() -> TPLinkSmartHomeProtocol:
return protocol
def _mocked_bulb() -> SmartBulb:
def _mocked_bulb(
device_config=DEVICE_CONFIG_LEGACY,
credentials_hash=CREDENTIALS_HASH_LEGACY,
mac=MAC_ADDRESS,
alias=ALIAS,
) -> SmartBulb:
bulb = MagicMock(auto_spec=SmartBulb, name="Mocked bulb")
bulb.update = AsyncMock()
bulb.mac = MAC_ADDRESS
bulb.alias = ALIAS
bulb.mac = mac
bulb.alias = alias
bulb.model = MODEL
bulb.host = IP_ADDRESS
bulb.brightness = 50
@ -52,7 +118,7 @@ def _mocked_bulb() -> SmartBulb:
bulb.effect = None
bulb.effect_list = None
bulb.hsv = (10, 30, 5)
bulb.device_id = MAC_ADDRESS
bulb.device_id = mac
bulb.valid_temperature_range.min = 4000
bulb.valid_temperature_range.max = 9000
bulb.hw_info = {"sw_ver": "1.0.0", "hw_ver": "1.0.0"}
@ -62,6 +128,8 @@ def _mocked_bulb() -> SmartBulb:
bulb.set_hsv = AsyncMock()
bulb.set_color_temp = AsyncMock()
bulb.protocol = _mock_protocol()
bulb.config = device_config
bulb.credentials_hash = credentials_hash
return bulb
@ -103,6 +171,8 @@ def _mocked_smart_light_strip() -> SmartLightStrip:
strip.set_effect = AsyncMock()
strip.set_custom_effect = AsyncMock()
strip.protocol = _mock_protocol()
strip.config = DEVICE_CONFIG_LEGACY
strip.credentials_hash = CREDENTIALS_HASH_LEGACY
return strip
@ -134,6 +204,8 @@ def _mocked_dimmer() -> SmartDimmer:
dimmer.set_color_temp = AsyncMock()
dimmer.set_led = AsyncMock()
dimmer.protocol = _mock_protocol()
dimmer.config = DEVICE_CONFIG_LEGACY
dimmer.credentials_hash = CREDENTIALS_HASH_LEGACY
return dimmer
@ -155,6 +227,8 @@ def _mocked_plug() -> SmartPlug:
plug.turn_on = AsyncMock()
plug.set_led = AsyncMock()
plug.protocol = _mock_protocol()
plug.config = DEVICE_CONFIG_LEGACY
plug.credentials_hash = CREDENTIALS_HASH_LEGACY
return plug
@ -176,6 +250,8 @@ def _mocked_strip() -> SmartStrip:
strip.turn_on = AsyncMock()
strip.set_led = AsyncMock()
strip.protocol = _mock_protocol()
strip.config = DEVICE_CONFIG_LEGACY
strip.credentials_hash = CREDENTIALS_HASH_LEGACY
plug0 = _mocked_plug()
plug0.alias = "Plug0"
plug0.device_id = "bb:bb:cc:dd:ee:ff_PLUG0DEVICEID"
@ -212,6 +288,15 @@ def _patch_single_discovery(device=None, no_device=False):
)
def _patch_connect(device=None, no_device=False):
async def _connect(*args, **kwargs):
if no_device:
raise SmartDeviceException
return device if device else _mocked_bulb()
return patch("homeassistant.components.tplink.SmartDevice.connect", new=_connect)
async def initialize_config_entry_for_device(
hass: HomeAssistant, dev: SmartDevice
) -> MockConfigEntry:
@ -225,7 +310,9 @@ async def initialize_config_entry_for_device(
)
config_entry.add_to_hass(hass)
with _patch_discovery(device=dev), _patch_single_discovery(device=dev):
with _patch_discovery(device=dev), _patch_single_discovery(
device=dev
), _patch_connect(device=dev):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()

119
tests/components/tplink/conftest.py
View file

@ -1,18 +1,75 @@
"""tplink conftest."""
from collections.abc import Generator
import copy
from unittest.mock import DEFAULT, AsyncMock, patch
import pytest
from . import _patch_discovery
from homeassistant.components.tplink import DOMAIN
from homeassistant.core import HomeAssistant
from tests.common import mock_device_registry, mock_registry
from . import (
CREATE_ENTRY_DATA_LEGACY,
CREDENTIALS_HASH_AUTH,
DEVICE_CONFIG_AUTH,
IP_ADDRESS,
IP_ADDRESS2,
MAC_ADDRESS,
MAC_ADDRESS2,
_mocked_bulb,
)
from tests.common import MockConfigEntry, mock_device_registry, mock_registry
@pytest.fixture
def mock_discovery():
"""Mock python-kasa discovery."""
with _patch_discovery() as mock_discover:
mock_discover.return_value = {}
yield mock_discover
with patch.multiple(
"homeassistant.components.tplink.Discover",
discover=DEFAULT,
discover_single=DEFAULT,
) as mock_discovery:
device = _mocked_bulb(
device_config=copy.deepcopy(DEVICE_CONFIG_AUTH),
credentials_hash=CREDENTIALS_HASH_AUTH,
alias=None,
)
devices = {
"127.0.0.1": _mocked_bulb(
device_config=copy.deepcopy(DEVICE_CONFIG_AUTH),
credentials_hash=CREDENTIALS_HASH_AUTH,
alias=None,
)
}
mock_discovery["discover"].return_value = devices
mock_discovery["discover_single"].return_value = device
mock_discovery["mock_device"] = device
yield mock_discovery
@pytest.fixture
def mock_connect():
"""Mock python-kasa connect."""
with patch("homeassistant.components.tplink.SmartDevice.connect") as mock_connect:
devices = {
IP_ADDRESS: _mocked_bulb(
device_config=DEVICE_CONFIG_AUTH, credentials_hash=CREDENTIALS_HASH_AUTH
),
IP_ADDRESS2: _mocked_bulb(
device_config=DEVICE_CONFIG_AUTH,
credentials_hash=CREDENTIALS_HASH_AUTH,
mac=MAC_ADDRESS2,
),
}
def get_device(config):
nonlocal devices
return devices[config.host]
mock_connect.side_effect = get_device
yield {"connect": mock_connect, "mock_devices": devices}
@pytest.fixture(name="device_reg")
@ -30,3 +87,55 @@ def entity_reg_fixture(hass):
@pytest.fixture(autouse=True)
def tplink_mock_get_source_ip(mock_get_source_ip):
"""Mock network util's async_get_source_ip."""
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
"""Override async_setup_entry."""
with patch.multiple(
async_setup=DEFAULT,
async_setup_entry=DEFAULT,
) as mock_setup_entry:
mock_setup_entry["async_setup"].return_value = True
mock_setup_entry["async_setup_entry"].return_value = True
yield mock_setup_entry
@pytest.fixture
def mock_init() -> Generator[AsyncMock, None, None]:
"""Override async_setup_entry."""
with patch.multiple(
"homeassistant.components.tplink",
async_setup=DEFAULT,
async_setup_entry=DEFAULT,
async_unload_entry=DEFAULT,
) as mock_init:
mock_init["async_setup"].return_value = True
mock_init["async_setup_entry"].return_value = True
mock_init["async_unload_entry"].return_value = True
yield mock_init
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Mock ConfigEntry."""
return MockConfigEntry(
title="TPLink",
domain=DOMAIN,
data={**CREATE_ENTRY_DATA_LEGACY},
unique_id=MAC_ADDRESS,
)
@pytest.fixture
async def mock_added_config_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_init,
) -> MockConfigEntry:
"""Mock ConfigEntry that's been added to HA."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert DOMAIN in hass.config_entries.async_domains()
return mock_config_entry

830
tests/components/tplink/test_config_flow.py
View file

File diff suppressed because it is too large Load diff

137
tests/components/tplink/test_init.py
View file

@ -1,25 +1,35 @@
"""Tests for the TP-Link component."""
from __future__ import annotations
import copy
from datetime import timedelta
from unittest.mock import MagicMock, patch
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from homeassistant import setup
from homeassistant.components import tplink
from homeassistant.components.tplink.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_HOST, EVENT_HOMEASSISTANT_STARTED
from homeassistant.components.tplink.const import CONF_DEVICE_CONFIG, DOMAIN
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
from homeassistant.const import (
CONF_AUTHENTICATION,
CONF_HOST,
CONF_PASSWORD,
CONF_USERNAME,
EVENT_HOMEASSISTANT_STARTED,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_registry import EntityRegistry
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from . import (
CREATE_ENTRY_DATA_AUTH,
DEVICE_CONFIG_AUTH,
IP_ADDRESS,
MAC_ADDRESS,
_mocked_dimmer,
_patch_connect,
_patch_discovery,
_patch_single_discovery,
)
@ -57,7 +67,7 @@ async def test_config_entry_reload(hass: HomeAssistant) -> None:
domain=DOMAIN, data={}, unique_id=MAC_ADDRESS
)
already_migrated_config_entry.add_to_hass(hass)
with _patch_discovery(), _patch_single_discovery():
with _patch_discovery(), _patch_single_discovery(), _patch_connect():
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
assert already_migrated_config_entry.state == ConfigEntryState.LOADED
@ -72,7 +82,9 @@ async def test_config_entry_retry(hass: HomeAssistant) -> None:
domain=DOMAIN, data={CONF_HOST: IP_ADDRESS}, unique_id=MAC_ADDRESS
)
already_migrated_config_entry.add_to_hass(hass)
with _patch_discovery(no_device=True), _patch_single_discovery(no_device=True):
with _patch_discovery(no_device=True), _patch_single_discovery(
no_device=True
), _patch_connect(no_device=True):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
assert already_migrated_config_entry.state == ConfigEntryState.SETUP_RETRY
@ -102,7 +114,9 @@ async def test_dimmer_switch_unique_id_fix_original_entity_still_exists(
original_name="Rollout dimmer",
)
with _patch_discovery(device=dimmer), _patch_single_discovery(device=dimmer):
with _patch_discovery(device=dimmer), _patch_single_discovery(
device=dimmer
), _patch_connect(device=dimmer):
await setup.async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
@ -126,7 +140,7 @@ async def test_config_entry_wrong_mac_Address(
domain=DOMAIN, data={CONF_HOST: "127.0.0.1"}, unique_id=mismatched_mac
)
already_migrated_config_entry.add_to_hass(hass)
with _patch_discovery(), _patch_single_discovery():
with _patch_discovery(), _patch_single_discovery(), _patch_connect():
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
assert already_migrated_config_entry.state == ConfigEntryState.SETUP_RETRY
@ -135,3 +149,110 @@ async def test_config_entry_wrong_mac_Address(
"Unexpected device found at 127.0.0.1; expected aa:bb:cc:dd:ee:f0, found aa:bb:cc:dd:ee:ff"
in caplog.text
)
async def test_config_entry_device_config(
hass: HomeAssistant,
mock_discovery: AsyncMock,
mock_connect: AsyncMock,
) -> None:
"""Test that a config entry can be loaded with DeviceConfig."""
mock_config_entry = MockConfigEntry(
title="TPLink",
domain=DOMAIN,
data={**CREATE_ENTRY_DATA_AUTH},
unique_id=MAC_ADDRESS,
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
async def test_config_entry_with_stored_credentials(
hass: HomeAssistant,
mock_discovery: AsyncMock,
mock_connect: AsyncMock,
) -> None:
"""Test that a config entry can be loaded when stored credentials are set."""
stored_credentials = tplink.Credentials("fake_username1", "fake_password1")
mock_config_entry = MockConfigEntry(
title="TPLink",
domain=DOMAIN,
data={**CREATE_ENTRY_DATA_AUTH},
unique_id=MAC_ADDRESS,
)
auth = {
CONF_USERNAME: stored_credentials.username,
CONF_PASSWORD: stored_credentials.password,
}
hass.data.setdefault(DOMAIN, {})[CONF_AUTHENTICATION] = auth
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
config = DEVICE_CONFIG_AUTH
assert config.credentials != stored_credentials
config.credentials = stored_credentials
mock_connect["connect"].assert_called_once_with(config=config)
async def test_config_entry_device_config_invalid(
hass: HomeAssistant,
mock_discovery: AsyncMock,
mock_connect: AsyncMock,
caplog,
) -> None:
"""Test that an invalid device config logs an error and loads the config entry."""
entry_data = copy.deepcopy(CREATE_ENTRY_DATA_AUTH)
entry_data[CONF_DEVICE_CONFIG] = {"foo": "bar"}
mock_config_entry = MockConfigEntry(
title="TPLink",
domain=DOMAIN,
data={**entry_data},
unique_id=MAC_ADDRESS,
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
assert (
f"Invalid connection type dict for {IP_ADDRESS}: {entry_data.get(CONF_DEVICE_CONFIG)}"
in caplog.text
)
@pytest.mark.parametrize(
("error_type", "entry_state", "reauth_flows"),
[
(tplink.AuthenticationException, ConfigEntryState.SETUP_ERROR, True),
(tplink.SmartDeviceException, ConfigEntryState.SETUP_RETRY, False),
],
ids=["invalid-auth", "unknown-error"],
)
async def test_config_entry_errors(
hass: HomeAssistant,
mock_discovery: AsyncMock,
mock_connect: AsyncMock,
error_type,
entry_state,
reauth_flows,
) -> None:
"""Test that device exceptions are handled correctly during init."""
mock_connect["connect"].side_effect = error_type
mock_config_entry = MockConfigEntry(
title="TPLink",
domain=DOMAIN,
data={**CREATE_ENTRY_DATA_AUTH},
unique_id=MAC_ADDRESS,
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is entry_state
assert (
any(mock_config_entry.async_get_active_flows(hass, {SOURCE_REAUTH}))
== reauth_flows
)

27
tests/components/tplink/test_light.py
View file

@ -33,6 +33,7 @@ from . import (
MAC_ADDRESS,
_mocked_bulb,
_mocked_smart_light_strip,
_patch_connect,
_patch_discovery,
_patch_single_discovery,
)
@ -48,7 +49,7 @@ async def test_light_unique_id(hass: HomeAssistant) -> None:
already_migrated_config_entry.add_to_hass(hass)
bulb = _mocked_bulb()
bulb.color_temp = None
with _patch_discovery(device=bulb), _patch_single_discovery(device=bulb):
with _patch_discovery(device=bulb), _patch_connect(device=bulb):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
@ -69,7 +70,7 @@ async def test_color_light(
)
already_migrated_config_entry.add_to_hass(hass)
bulb.color_temp = None
with _patch_discovery(device=bulb), _patch_single_discovery(device=bulb):
with _patch_discovery(device=bulb), _patch_connect(device=bulb):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
@ -151,7 +152,7 @@ async def test_color_light_no_temp(hass: HomeAssistant) -> None:
bulb = _mocked_bulb()
bulb.is_variable_color_temp = False
type(bulb).color_temp = PropertyMock(side_effect=Exception)
with _patch_discovery(device=bulb), _patch_single_discovery(device=bulb):
with _patch_discovery(device=bulb), _patch_connect(device=bulb):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
@ -212,7 +213,7 @@ async def test_color_temp_light(
bulb.color_temp = 4000
bulb.is_variable_color_temp = True
with _patch_discovery(device=bulb), _patch_single_discovery(device=bulb):
with _patch_discovery(device=bulb), _patch_connect(device=bulb):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
@ -295,7 +296,7 @@ async def test_brightness_only_light(hass: HomeAssistant) -> None:
bulb.is_color = False
bulb.is_variable_color_temp = False
with _patch_discovery(device=bulb), _patch_single_discovery(device=bulb):
with _patch_discovery(device=bulb), _patch_connect(device=bulb):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
@ -340,7 +341,7 @@ async def test_on_off_light(hass: HomeAssistant) -> None:
bulb.is_variable_color_temp = False
bulb.is_dimmable = False
with _patch_discovery(device=bulb), _patch_single_discovery(device=bulb):
with _patch_discovery(device=bulb), _patch_connect(device=bulb):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
@ -375,7 +376,7 @@ async def test_off_at_start_light(hass: HomeAssistant) -> None:
bulb.is_dimmable = False
bulb.is_on = False
with _patch_discovery(device=bulb), _patch_single_discovery(device=bulb):
with _patch_discovery(device=bulb), _patch_connect(device=bulb):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
@ -397,7 +398,7 @@ async def test_dimmer_turn_on_fix(hass: HomeAssistant) -> None:
bulb.is_dimmer = True
bulb.is_on = False
with _patch_discovery(device=bulb), _patch_single_discovery(device=bulb):
with _patch_discovery(device=bulb), _patch_connect(device=bulb):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
@ -421,7 +422,9 @@ async def test_smart_strip_effects(hass: HomeAssistant) -> None:
already_migrated_config_entry.add_to_hass(hass)
strip = _mocked_smart_light_strip()
with _patch_discovery(device=strip), _patch_single_discovery(device=strip):
with _patch_discovery(device=strip), _patch_single_discovery(
device=strip
), _patch_connect(device=strip):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
@ -501,7 +504,7 @@ async def test_smart_strip_custom_random_effect(hass: HomeAssistant) -> None:
already_migrated_config_entry.add_to_hass(hass)
strip = _mocked_smart_light_strip()
with _patch_discovery(device=strip), _patch_single_discovery(device=strip):
with _patch_discovery(device=strip), _patch_connect(device=strip):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
@ -664,7 +667,7 @@ async def test_smart_strip_custom_random_effect_at_start(hass: HomeAssistant) ->
"name": "Custom",
"enable": 0,
}
with _patch_discovery(device=strip), _patch_single_discovery(device=strip):
with _patch_discovery(device=strip), _patch_connect(device=strip):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
@ -691,7 +694,7 @@ async def test_smart_strip_custom_sequence_effect(hass: HomeAssistant) -> None:
already_migrated_config_entry.add_to_hass(hass)
strip = _mocked_smart_light_strip()
with _patch_discovery(device=strip), _patch_single_discovery(device=strip):
with _patch_discovery(device=strip), _patch_connect(device=strip):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()

16
tests/components/tplink/test_sensor.py
View file

@ -8,13 +8,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
from . import (
MAC_ADDRESS,
_mocked_bulb,
_mocked_plug,
_patch_discovery,
_patch_single_discovery,
)
from . import MAC_ADDRESS, _mocked_bulb, _mocked_plug, _patch_connect, _patch_discovery
from tests.common import MockConfigEntry
@ -35,7 +29,7 @@ async def test_color_light_with_an_emeter(hass: HomeAssistant) -> None:
current=5,
)
bulb.emeter_today = 5000.0036
with _patch_discovery(device=bulb), _patch_single_discovery(device=bulb):
with _patch_discovery(device=bulb), _patch_connect(device=bulb):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
await hass.async_block_till_done()
@ -75,7 +69,7 @@ async def test_plug_with_an_emeter(hass: HomeAssistant) -> None:
current=5.035,
)
plug.emeter_today = None
with _patch_discovery(device=plug), _patch_single_discovery(device=plug):
with _patch_discovery(device=plug), _patch_connect(device=plug):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
await hass.async_block_till_done()
@ -103,7 +97,7 @@ async def test_color_light_no_emeter(hass: HomeAssistant) -> None:
bulb = _mocked_bulb()
bulb.color_temp = None
bulb.has_emeter = False
with _patch_discovery(device=bulb), _patch_single_discovery(device=bulb):
with _patch_discovery(device=bulb), _patch_connect(device=bulb):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
await hass.async_block_till_done()
@ -139,7 +133,7 @@ async def test_sensor_unique_id(hass: HomeAssistant) -> None:
current=5,
)
plug.emeter_today = None
with _patch_discovery(device=plug), _patch_single_discovery(device=plug):
with _patch_discovery(device=plug), _patch_connect(device=plug):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()

14
tests/components/tplink/test_switch.py
View file

@ -20,8 +20,8 @@ from . import (
_mocked_dimmer,
_mocked_plug,
_mocked_strip,
_patch_connect,
_patch_discovery,
_patch_single_discovery,
)
from tests.common import MockConfigEntry, async_fire_time_changed
@ -34,7 +34,7 @@ async def test_plug(hass: HomeAssistant) -> None:
)
already_migrated_config_entry.add_to_hass(hass)
plug = _mocked_plug()
with _patch_discovery(device=plug), _patch_single_discovery(device=plug):
with _patch_discovery(device=plug), _patch_connect(device=plug):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
@ -69,7 +69,7 @@ async def test_led_switch(hass: HomeAssistant, dev, domain: str) -> None:
domain=DOMAIN, data={}, unique_id=MAC_ADDRESS
)
already_migrated_config_entry.add_to_hass(hass)
with _patch_discovery(device=dev), _patch_single_discovery(device=dev):
with _patch_discovery(device=dev), _patch_connect(device=dev):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
@ -100,7 +100,7 @@ async def test_plug_unique_id(hass: HomeAssistant) -> None:
)
already_migrated_config_entry.add_to_hass(hass)
plug = _mocked_plug()
with _patch_discovery(device=plug), _patch_single_discovery(device=plug):
with _patch_discovery(device=plug), _patch_connect(device=plug):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
@ -116,7 +116,7 @@ async def test_plug_update_fails(hass: HomeAssistant) -> None:
)
already_migrated_config_entry.add_to_hass(hass)
plug = _mocked_plug()
with _patch_discovery(device=plug), _patch_single_discovery(device=plug):
with _patch_discovery(device=plug), _patch_connect(device=plug):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
@ -138,7 +138,7 @@ async def test_strip(hass: HomeAssistant) -> None:
)
already_migrated_config_entry.add_to_hass(hass)
strip = _mocked_strip()
with _patch_discovery(device=strip), _patch_single_discovery(device=strip):
with _patch_discovery(device=strip), _patch_connect(device=strip):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
@ -186,7 +186,7 @@ async def test_strip_unique_ids(hass: HomeAssistant) -> None:
)
already_migrated_config_entry.add_to_hass(hass)
strip = _mocked_strip()
with _patch_discovery(device=strip), _patch_single_discovery(device=strip):
with _patch_discovery(device=strip), _patch_connect(device=strip):
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()