Add set_client_name action to TP-Link Omada (#181450)

This commit is contained in:
Mattheinrichs
2026-09-08 22:23:49 +02:00
committed by GitHub
parent f134bdc394
commit 728380ba21
6 changed files with 645 additions and 23 deletions
@@ -31,6 +31,9 @@
"services": {
"reconnect_client": {
"service": "mdi:sync"
},
"set_client_name": {
"service": "mdi:rename-box"
}
}
}
+132 -22
View File
@@ -2,20 +2,24 @@
from typing import cast
from tplink_omada_client.exceptions import OmadaClientException
from tplink_omada_client import OmadaClientSettings
from tplink_omada_client.exceptions import OmadaClientException, RequestFailed
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
from homeassistant.const import ATTR_CONFIG_ENTRY_ID
from homeassistant.const import ATTR_CONFIG_ENTRY_ID, ATTR_DEVICE_ID, ATTR_NAME
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import config_validation as cv, selector
from homeassistant.helpers import (
config_validation as cv,
device_registry as dr,
selector,
)
from homeassistant.helpers.service import async_register_admin_service
from .const import DOMAIN
from .controller import OmadaSiteController
SERVICE_RECONNECT_CLIENT = "reconnect_client"
ATTR_MAC = "mac"
@@ -50,18 +54,6 @@ def _get_controller(call: ServiceCall) -> OmadaSiteController:
return entry.runtime_data
SCHEMA_RECONNECT_CLIENT = vol.Schema(
{
vol.Optional(ATTR_CONFIG_ENTRY_ID): selector.ConfigEntrySelector(
{
"integration": DOMAIN,
}
),
vol.Required(ATTR_MAC): cv.string,
}
)
async def _handle_reconnect_client(call: ServiceCall) -> None:
"""Handle the service action to force reconnection of a network client."""
controller = _get_controller(call)
@@ -78,14 +70,132 @@ async def _handle_reconnect_client(call: ServiceCall) -> None:
) from ex
SERVICES = [
(SERVICE_RECONNECT_CLIENT, SCHEMA_RECONNECT_CLIENT, _handle_reconnect_client)
]
def _controller_mac(mac: str) -> str:
"""Normalize a registry MAC to the controller's canonical format."""
return mac.upper().replace(":", "-")
async def _resolve_client_controller(
call: ServiceCall,
) -> tuple[OmadaSiteController, str]:
"""Resolve the controller and MAC of the client referenced by the call."""
hass = call.hass
entry = hass.config_entries.async_get_entry(call.data[ATTR_CONFIG_ENTRY_ID])
if not entry or entry.domain != DOMAIN:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="controller_not_found",
)
if entry.state is not ConfigEntryState.LOADED:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="controller_unavailable",
)
controller = cast(ConfigEntry[OmadaSiteController], entry).runtime_data
device = dr.async_get(hass).async_get(call.data[ATTR_DEVICE_ID])
if device is None or not isinstance(device, dr.DeviceEntry):
# Child devices carry no connections to resolve a MAC from.
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="client_device_not_found",
)
macs = [
connection_id
for connection_type, connection_id in device.connections
if connection_type == dr.CONNECTION_NETWORK_MAC
]
if not macs:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="client_device_no_mac",
)
known_macs: list[str] = []
for mac in macs:
controller_mac = _controller_mac(mac)
try:
await controller.omada_client.get_client(controller_mac)
except RequestFailed as ex:
# -41011 is the controller's "client not found" code (no public accessor).
if getattr(ex, "_error_code", None) == -41011:
continue
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="client_query_failed",
translation_placeholders={"mac": controller_mac},
) from ex
except OmadaClientException as ex:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="client_query_failed",
translation_placeholders={"mac": controller_mac},
) from ex
known_macs.append(controller_mac)
if len(known_macs) == 1:
return controller, known_macs[0]
if not known_macs:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="client_device_not_tracked",
)
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="client_mac_ambiguous",
)
async def _handle_set_client_name(call: ServiceCall) -> None:
"""Handle the service action to set the name of a network client."""
controller, mac = await _resolve_client_controller(call)
name: str = call.data[ATTR_NAME]
try:
await controller.omada_client.update_client(mac, OmadaClientSettings(name=name))
except OmadaClientException as ex:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="set_client_name_failed",
translation_placeholders={"mac": mac},
) from ex
@callback
def async_setup_services(hass: HomeAssistant) -> None:
"""Set up the services for the TP-Link Omada integration."""
for service_name, schema, handler in SERVICES:
hass.services.async_register(DOMAIN, service_name, handler, schema=schema)
async_register_admin_service(
hass,
DOMAIN,
"set_client_name",
_handle_set_client_name,
schema=vol.Schema(
{
vol.Required(ATTR_CONFIG_ENTRY_ID): selector.ConfigEntrySelector(
{
"integration": DOMAIN,
}
),
vol.Required(ATTR_DEVICE_ID): selector.DeviceSelector(),
vol.Required(ATTR_NAME): vol.All(cv.string, vol.Length(min=1)),
}
),
)
hass.services.async_register(
DOMAIN,
"reconnect_client",
_handle_reconnect_client,
schema=vol.Schema(
{
vol.Optional(ATTR_CONFIG_ENTRY_ID): selector.ConfigEntrySelector(
{
"integration": DOMAIN,
}
),
vol.Required(ATTR_MAC): cv.string,
}
),
)
@@ -9,3 +9,18 @@ reconnect_client:
example: "01-23-45-67-89-AB"
selector:
text:
set_client_name:
fields:
config_entry_id:
required: true
selector:
config_entry:
integration: tplink_omada
device_id:
required: true
selector:
device:
name:
required: true
selector:
text:
@@ -111,6 +111,21 @@
"cannot_connect": {
"message": "Omada controller could not be reached."
},
"client_device_no_mac": {
"message": "The selected device has no network MAC address associated with it."
},
"client_device_not_found": {
"message": "The selected device was not found."
},
"client_device_not_tracked": {
"message": "The selected device is not a network client of the Omada controller."
},
"client_mac_ambiguous": {
"message": "The selected device has multiple network addresses on the Omada controller. Select a different device."
},
"client_query_failed": {
"message": "Failed to query client with MAC {mac} from the Omada controller."
},
"controller_not_found": {
"message": "Specified TP-Link Omada controller not found"
},
@@ -129,6 +144,9 @@
"reconnect_failed": {
"message": "Failed to reconnect client with MAC {mac}"
},
"set_client_name_failed": {
"message": "Failed to set client name for device with MAC {mac}"
},
"switch_action_failed": {
"message": "Failed to update the switch."
},
@@ -153,6 +171,24 @@
}
},
"name": "Reconnect wireless client"
},
"set_client_name": {
"description": "Sets the name of a client on the Omada network.",
"fields": {
"config_entry_id": {
"description": "The instance of the Omada integration to which the client is connected.",
"name": "Omada controller"
},
"device_id": {
"description": "The device whose client name will be set.",
"name": "Device"
},
"name": {
"description": "New name for the client.",
"name": "Name"
}
},
"name": "Set client name"
}
}
}
@@ -108,7 +108,9 @@ async def mock_omada_site_client(hass: HomeAssistant) -> AsyncGenerator[AsyncMoc
site_client.get_known_clients.return_value = async_empty()
site_client.get_connected_clients.return_value = async_empty()
site_client.get_client = AsyncMock()
site_client.reconnect_client = AsyncMock()
site_client.update_client = AsyncMock()
return site_client
+457 -1
View File
@@ -3,12 +3,15 @@
from unittest.mock import MagicMock
import pytest
from tplink_omada_client.exceptions import OmadaClientException
from tplink_omada_client import OmadaClientSettings
from tplink_omada_client.exceptions import OmadaClientException, RequestFailed
import voluptuous as vol
from homeassistant.components.tplink_omada.const import DOMAIN
from homeassistant.components.tplink_omada.services import async_setup_services
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import device_registry as dr
from tests.common import MockConfigEntry
@@ -160,3 +163,456 @@ async def test_service_reconnect_failed_raises_homeassistanterror(
assert err.value.translation_placeholders == {"mac": mac}
mock_omada_site_client.reconnect_client.assert_awaited_once_with(mac)
def _add_client_device(
hass: HomeAssistant, config_entry: MockConfigEntry, mac: str
) -> str:
"""Register a device with a network MAC connection."""
device = dr.async_get(hass).async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, mac)},
)
return device.id
def _add_foreign_device(hass: HomeAssistant, mac: str) -> str:
"""Register a device owned by another integration with a network MAC."""
foreign_entry = MockConfigEntry(domain="sonos", unique_id="foreign")
foreign_entry.add_to_hass(hass)
device = dr.async_get(hass).async_get_or_create(
config_entry_id=foreign_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, mac)},
)
return device.id
def _add_multi_mac_device(
hass: HomeAssistant, config_entry: MockConfigEntry, macs: list[str]
) -> str:
"""Register a device with multiple network MAC connections."""
device = dr.async_get(hass).async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, mac) for mac in macs},
)
return device.id
def _add_device_without_mac(hass: HomeAssistant, config_entry: MockConfigEntry) -> str:
"""Register a device without a network MAC connection."""
device = dr.async_get(hass).async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, "no-mac-device")},
)
return device.id
async def test_service_set_client_name(
hass: HomeAssistant,
mock_omada_site_client: MagicMock,
mock_omada_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test setting the name of a client."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
mac = "aa:bb:cc:dd:ee:ff"
device_id = _add_client_device(hass, mock_config_entry, mac)
await hass.services.async_call(
DOMAIN,
"set_client_name",
{
"config_entry_id": mock_config_entry.entry_id,
"device_id": device_id,
"name": "Ting sensor",
},
blocking=True,
)
mock_omada_site_client.update_client.assert_awaited_once_with(
mac.upper().replace(":", "-"),
OmadaClientSettings(name="Ting sensor"),
)
async def test_service_set_client_name_invalid_config_entry_id(
hass: HomeAssistant,
mock_omada_site_client: MagicMock,
mock_omada_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test set client name with invalid config entry raises ServiceValidationError."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
with pytest.raises(ServiceValidationError) as err:
await hass.services.async_call(
DOMAIN,
"set_client_name",
{
"config_entry_id": "invalid_entry_id",
"device_id": "device1",
"name": "Ting sensor",
},
blocking=True,
)
assert err.value.translation_key == "controller_not_found"
assert err.value.translation_domain == DOMAIN
async def test_service_set_client_name_entry_not_loaded(
hass: HomeAssistant,
mock_omada_site_client: MagicMock,
mock_omada_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test set client name raises error when entry is not loaded."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
unloaded_entry = MockConfigEntry(
title="Unloaded Omada Controller",
domain=DOMAIN,
unique_id="67890",
)
unloaded_entry.add_to_hass(hass)
with pytest.raises(ServiceValidationError) as err:
await hass.services.async_call(
DOMAIN,
"set_client_name",
{
"config_entry_id": unloaded_entry.entry_id,
"device_id": "device1",
"name": "Ting sensor",
},
blocking=True,
)
assert err.value.translation_key == "controller_unavailable"
assert err.value.translation_domain == DOMAIN
async def test_service_set_client_name_unknown_device(
hass: HomeAssistant,
mock_omada_site_client: MagicMock,
mock_omada_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test set client name with an unknown device raises ServiceValidationError."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
with pytest.raises(ServiceValidationError) as err:
await hass.services.async_call(
DOMAIN,
"set_client_name",
{
"config_entry_id": mock_config_entry.entry_id,
"device_id": "nonexistent_device",
"name": "Ting sensor",
},
blocking=True,
)
assert err.value.translation_key == "client_device_not_found"
assert err.value.translation_domain == DOMAIN
async def test_service_set_client_name_device_without_mac(
hass: HomeAssistant,
mock_omada_site_client: MagicMock,
mock_omada_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test set client name with a device without a MAC raises ServiceValidationError."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
device_id = _add_device_without_mac(hass, mock_config_entry)
with pytest.raises(ServiceValidationError) as err:
await hass.services.async_call(
DOMAIN,
"set_client_name",
{
"config_entry_id": mock_config_entry.entry_id,
"device_id": device_id,
"name": "Ting sensor",
},
blocking=True,
)
assert err.value.translation_key == "client_device_no_mac"
assert err.value.translation_domain == DOMAIN
async def test_service_set_client_name_failed_raises_homeassistanterror(
hass: HomeAssistant,
mock_omada_site_client: MagicMock,
mock_omada_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test set client name raises correct exception on controller failure."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
mac = "aa:bb:cc:dd:ee:ff"
device_id = _add_client_device(hass, mock_config_entry, mac)
mock_omada_site_client.update_client.side_effect = OmadaClientException
with pytest.raises(HomeAssistantError) as err:
await hass.services.async_call(
DOMAIN,
"set_client_name",
{
"config_entry_id": mock_config_entry.entry_id,
"device_id": device_id,
"name": "Ting sensor",
},
blocking=True,
)
assert err.value.translation_key == "set_client_name_failed"
assert err.value.translation_domain == DOMAIN
assert err.value.translation_placeholders == {"mac": mac.upper().replace(":", "-")}
mock_omada_site_client.update_client.assert_awaited_once_with(
mac.upper().replace(":", "-"),
OmadaClientSettings(name="Ting sensor"),
)
async def test_service_set_client_name_empty_name_rejected(
hass: HomeAssistant,
mock_omada_site_client: MagicMock,
mock_omada_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test set client name with an empty name is rejected by the schema."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
mac = "aa:bb:cc:dd:ee:ff"
device_id = _add_client_device(hass, mock_config_entry, mac)
with pytest.raises(vol.Invalid):
await hass.services.async_call(
DOMAIN,
"set_client_name",
{
"config_entry_id": mock_config_entry.entry_id,
"device_id": device_id,
"name": "",
},
blocking=True,
)
mock_omada_site_client.update_client.assert_not_awaited()
async def test_service_set_client_name_non_client_device(
hass: HomeAssistant,
mock_omada_site_client: MagicMock,
mock_omada_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test set client name with a non-client device raises ServiceValidationError."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
mac = "aa:bb:cc:dd:ee:ff"
device_id = _add_client_device(hass, mock_config_entry, mac)
mock_omada_site_client.get_client.side_effect = RequestFailed(-41011, "not found")
with pytest.raises(ServiceValidationError) as err:
await hass.services.async_call(
DOMAIN,
"set_client_name",
{
"config_entry_id": mock_config_entry.entry_id,
"device_id": device_id,
"name": "Ting sensor",
},
blocking=True,
)
assert err.value.translation_key == "client_device_not_tracked"
assert err.value.translation_domain == DOMAIN
mock_omada_site_client.update_client.assert_not_awaited()
async def test_service_set_client_name_foreign_device(
hass: HomeAssistant,
mock_omada_site_client: MagicMock,
mock_omada_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test set client name with a device owned by another integration."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
mac = "aa:bb:cc:dd:ee:ff"
device_id = _add_foreign_device(hass, mac)
await hass.services.async_call(
DOMAIN,
"set_client_name",
{
"config_entry_id": mock_config_entry.entry_id,
"device_id": device_id,
"name": "Ting sensor",
},
blocking=True,
)
mock_omada_site_client.get_client.assert_awaited_once_with(
mac.upper().replace(":", "-")
)
mock_omada_site_client.update_client.assert_awaited_once_with(
mac.upper().replace(":", "-"),
OmadaClientSettings(name="Ting sensor"),
)
async def test_service_set_client_name_foreign_controller_entry(
hass: HomeAssistant,
mock_omada_site_client: MagicMock,
mock_omada_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test set client name with a non-Omada config entry raises an error."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
foreign_entry = MockConfigEntry(domain="sonos", unique_id="foreign_entry")
foreign_entry.add_to_hass(hass)
with pytest.raises(ServiceValidationError) as err:
await hass.services.async_call(
DOMAIN,
"set_client_name",
{
"config_entry_id": foreign_entry.entry_id,
"device_id": "device1",
"name": "Ting sensor",
},
blocking=True,
)
assert err.value.translation_key == "controller_not_found"
assert err.value.translation_domain == DOMAIN
mock_omada_site_client.update_client.assert_not_awaited()
async def test_service_set_client_name_single_known_mac(
hass: HomeAssistant,
mock_omada_site_client: MagicMock,
mock_omada_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test set client name resolves a device with multiple MACs to the tracked one."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
client_mac = "aa:bb:cc:dd:ee:ff"
other_mac = "11:22:33:44:55:66"
device_id = _add_multi_mac_device(hass, mock_config_entry, [client_mac, other_mac])
async def get_client(mac: str) -> MagicMock:
if mac == other_mac.upper().replace(":", "-"):
raise RequestFailed(-41011, "not found")
return MagicMock()
mock_omada_site_client.get_client.side_effect = get_client
await hass.services.async_call(
DOMAIN,
"set_client_name",
{
"config_entry_id": mock_config_entry.entry_id,
"device_id": device_id,
"name": "Ting sensor",
},
blocking=True,
)
mock_omada_site_client.update_client.assert_awaited_once_with(
client_mac.upper().replace(":", "-"),
OmadaClientSettings(name="Ting sensor"),
)
async def test_service_set_client_name_multiple_known_macs(
hass: HomeAssistant,
mock_omada_site_client: MagicMock,
mock_omada_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test set client name with multiple tracked MACs raises an ambiguity error."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
device_id = _add_multi_mac_device(
hass, mock_config_entry, ["aa:bb:cc:dd:ee:ff", "11:22:33:44:55:66"]
)
with pytest.raises(ServiceValidationError) as err:
await hass.services.async_call(
DOMAIN,
"set_client_name",
{
"config_entry_id": mock_config_entry.entry_id,
"device_id": device_id,
"name": "Ting sensor",
},
blocking=True,
)
assert err.value.translation_key == "client_mac_ambiguous"
assert err.value.translation_domain == DOMAIN
mock_omada_site_client.update_client.assert_not_awaited()
async def test_service_set_client_name_query_failed_raises_homeassistanterror(
hass: HomeAssistant,
mock_omada_site_client: MagicMock,
mock_omada_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test set client name surfaces a controller failure querying the client."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
mac = "aa:bb:cc:dd:ee:ff"
device_id = _add_client_device(hass, mock_config_entry, mac)
mock_omada_site_client.get_client.side_effect = RequestFailed(-30109, "boom")
with pytest.raises(HomeAssistantError) as err:
await hass.services.async_call(
DOMAIN,
"set_client_name",
{
"config_entry_id": mock_config_entry.entry_id,
"device_id": device_id,
"name": "Ting sensor",
},
blocking=True,
)
assert err.value.translation_key == "client_query_failed"
assert err.value.translation_domain == DOMAIN
assert err.value.translation_placeholders == {"mac": mac.upper().replace(":", "-")}
mock_omada_site_client.update_client.assert_not_awaited()