jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add UniqueID to AsusWRT config entry (#70478)

Browse Source
This commit is contained in:
ollo69 2022-05-03 22:56:57 +02:00 committed by GitHub
parent e30940ef2a
commit 93153b3790
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 201 additions and 100 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
homeassistant/components/asuswrt/config_flow.py
View File

@ -25,6 +25,7 @@ from homeassistant.const import (
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.device_registry import format_mac
from .const import (
CONF_DNSMASQ,
@ -41,11 +42,13 @@ from .const import (
PROTOCOL_SSH,
PROTOCOL_TELNET,
)
from .router import get_api
from .router import get_api, get_nvram_info
LABEL_MAC = "LABEL_MAC"
RESULT_CONN_ERROR = "cannot_connect"
RESULT_UNKNOWN = "unknown"
RESULT_SUCCESS = "success"
RESULT_UNKNOWN = "unknown"
_LOGGER = logging.getLogger(__name__)
@ -107,7 +110,9 @@ class AsusWrtFlowHandler(ConfigFlow, domain=DOMAIN):
)
@staticmethod
async def _async_check_connection(user_input: dict[str, Any]) -> str:
async def _async_check_connection(
user_input: dict[str, Any]
) -> tuple[str, str | None]:
"""Attempt to connect the AsusWrt router."""
host: str = user_input[CONF_HOST]
@ -117,29 +122,37 @@ class AsusWrtFlowHandler(ConfigFlow, domain=DOMAIN):
except OSError:
_LOGGER.error("Error connecting to the AsusWrt router at %s", host)
return RESULT_CONN_ERROR
return RESULT_CONN_ERROR, None
except Exception: # pylint: disable=broad-except
_LOGGER.exception(
"Unknown error connecting with AsusWrt router at %s", host
)
return RESULT_UNKNOWN
return RESULT_UNKNOWN, None
if not api.is_connected:
_LOGGER.error("Error connecting to the AsusWrt router at %s", host)
return RESULT_CONN_ERROR
return RESULT_CONN_ERROR, None
label_mac = await get_nvram_info(api, LABEL_MAC)
conf_protocol = user_input[CONF_PROTOCOL]
if conf_protocol == PROTOCOL_TELNET:
api.connection.disconnect()
return RESULT_SUCCESS
unique_id = None
if label_mac and "label_mac" in label_mac:
unique_id = format_mac(label_mac["label_mac"])
return RESULT_SUCCESS, unique_id
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initiated by the user."""
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
# if exist one entry without unique ID, we abort config flow
for unique_id in self._async_current_ids():
if unique_id is None:
return self.async_abort(reason="not_unique_id_exist")
if user_input is None:
return self._show_setup_form(user_input)
@ -166,17 +179,27 @@ class AsusWrtFlowHandler(ConfigFlow, domain=DOMAIN):
errors["base"] = "invalid_host"
if not errors:
result = await self._async_check_connection(user_input)
if result != RESULT_SUCCESS:
errors["base"] = result
result, unique_id = await self._async_check_connection(user_input)
if result == RESULT_SUCCESS:
if unique_id:
await self.async_set_unique_id(unique_id)
# we allow configure a single instance without unique id
elif self._async_current_entries():
return self.async_abort(reason="invalid_unique_id")
else:
_LOGGER.warning(
"This device do not provide a valid Unique ID."
" Configuration of multiple instance will not be possible"
)
if errors:
return self._show_setup_form(user_input, errors)
return self.async_create_entry(
title=host,
data=user_input,
)
return self.async_create_entry(
title=host,
data=user_input,
)
errors["base"] = result
return self._show_setup_form(user_input, errors)
@staticmethod
@callback

3
homeassistant/components/asuswrt/diagnostics.py
View File

@ -11,6 +11,7 @@ from homeassistant.const import (
ATTR_CONNECTIONS,
ATTR_IDENTIFIERS,
CONF_PASSWORD,
CONF_UNIQUE_ID,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
@ -19,7 +20,7 @@ from homeassistant.helpers import device_registry as dr, entity_registry as er
from .const import DATA_ASUSWRT, DOMAIN
from .router import AsusWrtRouter
TO_REDACT = {CONF_PASSWORD, CONF_USERNAME}
TO_REDACT = {CONF_PASSWORD, CONF_UNIQUE_ID, CONF_USERNAME}
TO_REDACT_DEV = {ATTR_CONNECTIONS, ATTR_IDENTIFIERS}

19
homeassistant/components/asuswrt/router.py
View File

@ -51,6 +51,7 @@ from .const import (
)
CONF_REQ_RELOAD = [CONF_DNSMASQ, CONF_INTERFACE, CONF_REQUIRE_IP]
DEFAULT_NAME = "Asuswrt"
KEY_COORDINATOR = "coordinator"
KEY_SENSORS = "sensors"
@ -260,10 +261,10 @@ class AsusWrtRouter:
raise ConfigEntryNotReady
# System
model = await _get_nvram_info(self._api, "MODEL")
model = await get_nvram_info(self._api, "MODEL")
if model and "model" in model:
self._model = model["model"]
firmware = await _get_nvram_info(self._api, "FIRMWARE")
firmware = await get_nvram_info(self._api, "FIRMWARE")
if firmware and "firmver" in firmware and "buildno" in firmware:
self._sw_v = f"{firmware['firmver']} (build {firmware['buildno']})"
@ -441,7 +442,7 @@ class AsusWrtRouter:
def device_info(self) -> DeviceInfo:
"""Return the device information."""
return DeviceInfo(
identifiers={(DOMAIN, "AsusWRT")},
identifiers={(DOMAIN, self.unique_id or "AsusWRT")},
name=self._host,
model=self._model,
manufacturer="Asus",
@ -464,6 +465,16 @@ class AsusWrtRouter:
"""Return router hostname."""
return self._host
@property
def unique_id(self) -> str | None:
"""Return router unique id."""
return self._entry.unique_id
@property
def name(self) -> str:
"""Return router name."""
return self._host if self.unique_id else DEFAULT_NAME
@property
def devices(self) -> dict[str, AsusWrtDevInfo]:
"""Return devices."""
@ -475,7 +486,7 @@ class AsusWrtRouter:
return self._sensors_coordinator
async def _get_nvram_info(api: AsusWrt, info_type: str) -> dict[str, Any]:
async def get_nvram_info(api: AsusWrt, info_type: str) -> dict[str, Any]:
"""Get AsusWrt router info from nvram."""
info = {}
try:

8
homeassistant/components/asuswrt/sensor.py
View File

@ -43,7 +43,6 @@ class AsusWrtSensorEntityDescription(SensorEntityDescription):
precision: int = 2
DEFAULT_PREFIX = "Asuswrt"
UNIT_DEVICES = "Devices"
CONNECTION_SENSORS: tuple[AsusWrtSensorEntityDescription, ...] = (
@ -190,8 +189,11 @@ class AsusWrtSensor(CoordinatorEntity, SensorEntity):
super().__init__(coordinator)
self.entity_description: AsusWrtSensorEntityDescription = description
self._attr_name = f"{DEFAULT_PREFIX} {description.name}"
self._attr_unique_id = f"{DOMAIN} {self.name}"
self._attr_name = f"{router.name} {description.name}"
if router.unique_id:
self._attr_unique_id = f"{DOMAIN} {router.unique_id} {description.name}"
else:
self._attr_unique_id = f"{DOMAIN} {self.name}"
self._attr_device_info = router.device_info
self._attr_extra_state_attributes = {"hostname": router.host}

3
homeassistant/components/asuswrt/strings.json
View File

@ -25,7 +25,8 @@
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]"
"invalid_unique_id": "Impossible to determine a valid unique id for the device",
"not_unique_id_exist": "A device without a valid UniqueID is already configured. Configuration of multiple instance is not possible"
}
},
"options": {

3
homeassistant/components/asuswrt/translations/en.json
View File

@ -1,7 +1,8 @@
{
"config": {
"abort": {
"single_instance_allowed": "Already configured. Only a single configuration possible."
"invalid_unique_id": "Impossible to determine a valid unique id for the device",
"not_unique_id_exist": "A device without a valid UniqueID is already configured. Configuration of multiple instance is not possible"
},
"error": {
"cannot_connect": "Failed to connect",

187
tests/components/asuswrt/test_config_flow.py
View File

@ -28,6 +28,7 @@ from tests.common import MockConfigEntry
HOST = "myrouter.asuswrt.com"
IP_ADDRESS = "192.168.1.1"
MAC_ADDR = "a1:b1:c1:d1:e1:f1"
SSH_KEY = "1234"
CONFIG_DATA = {
@ -39,19 +40,44 @@ CONFIG_DATA = {
CONF_MODE: "ap",
}
PATCH_GET_HOST = patch(
"homeassistant.components.asuswrt.config_flow.socket.gethostbyname",
return_value=IP_ADDRESS,
)
PATCH_SETUP_ENTRY = patch(
"homeassistant.components.asuswrt.async_setup_entry",
return_value=True,
)
@pytest.fixture(name="mock_unique_id")
def mock_unique_id_fixture():
"""Mock returned unique id."""
return {}
@pytest.fixture(name="connect")
def mock_controller_connect():
def mock_controller_connect(mock_unique_id):
"""Mock a successful connection."""
with patch("homeassistant.components.asuswrt.router.AsusWrt") as service_mock:
service_mock.return_value.connection.async_connect = AsyncMock()
service_mock.return_value.is_connected = True
service_mock.return_value.connection.disconnect = Mock()
service_mock.return_value.async_get_nvram = AsyncMock(
return_value=mock_unique_id
)
yield service_mock
async def test_user(hass, connect):
@pytest.mark.usefixtures("connect")
@pytest.mark.parametrize(
"unique_id",
[{}, {"label_mac": MAC_ADDR}],
)
async def test_user(hass, mock_unique_id, unique_id):
"""Test user config."""
mock_unique_id.update(unique_id)
flow_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER, "show_advanced_options": True}
)
@ -59,15 +85,10 @@ async def test_user(hass, connect):
assert flow_result["step_id"] == "user"
# test with all provided
with patch(
"homeassistant.components.asuswrt.async_setup_entry",
return_value=True,
) as mock_setup_entry, patch(
"homeassistant.components.asuswrt.config_flow.socket.gethostbyname",
return_value=IP_ADDRESS,
):
with PATCH_GET_HOST, PATCH_SETUP_ENTRY as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
flow_result["flow_id"], user_input=CONFIG_DATA
flow_result["flow_id"],
user_input=CONFIG_DATA,
)
await hass.async_block_till_done()
@ -78,10 +99,17 @@ async def test_user(hass, connect):
assert len(mock_setup_entry.mock_calls) == 1
async def test_error_no_password_ssh(hass):
"""Test we abort if component is already setup."""
@pytest.mark.parametrize(
["config", "error"],
[
({CONF_PASSWORD: None}, "pwd_or_ssh"),
({CONF_SSH_KEY: SSH_KEY}, "pwd_and_ssh"),
],
)
async def test_error_wrong_password_ssh(hass, config, error):
"""Test we abort for wrong password and ssh file combination."""
config_data = CONFIG_DATA.copy()
config_data.pop(CONF_PASSWORD)
config_data.update(config)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER, "show_advanced_options": True},
@ -89,36 +117,27 @@ async def test_error_no_password_ssh(hass):
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "pwd_or_ssh"}
async def test_error_both_password_ssh(hass):
"""Test we abort if component is already setup."""
config_data = CONFIG_DATA.copy()
config_data[CONF_SSH_KEY] = SSH_KEY
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER, "show_advanced_options": True},
data=config_data,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "pwd_and_ssh"}
assert result["errors"] == {"base": error}
async def test_error_invalid_ssh(hass):
"""Test we abort if component is already setup."""
"""Test we abort if invalid ssh file is provided."""
config_data = CONFIG_DATA.copy()
config_data.pop(CONF_PASSWORD)
config_data[CONF_SSH_KEY] = SSH_KEY
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER, "show_advanced_options": True},
data=config_data,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "ssh_not_file"}
with patch(
"homeassistant.components.asuswrt.config_flow.os.path.isfile",
return_value=False,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER, "show_advanced_options": True},
data=config_data,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "ssh_not_file"}
async def test_error_invalid_host(hass):
@ -137,60 +156,96 @@ async def test_error_invalid_host(hass):
assert result["errors"] == {"base": "invalid_host"}
async def test_abort_if_already_setup(hass):
"""Test we abort if component is already setup."""
async def test_abort_if_not_unique_id_setup(hass):
"""Test we abort if component without uniqueid is already setup."""
MockConfigEntry(
domain=DOMAIN,
data=CONFIG_DATA,
).add_to_hass(hass)
with patch(
"homeassistant.components.asuswrt.config_flow.socket.gethostbyname",
return_value=IP_ADDRESS,
):
# Should fail, same HOST (flow)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data=CONFIG_DATA,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "not_unique_id_exist"
@pytest.mark.usefixtures("connect")
async def test_update_uniqueid_exist(hass, mock_unique_id):
"""Test we update entry if uniqueid is already configured."""
mock_unique_id.update({"label_mac": MAC_ADDR})
existing_entry = MockConfigEntry(
domain=DOMAIN,
data={**CONFIG_DATA, CONF_HOST: "10.10.10.10"},
unique_id=MAC_ADDR,
)
existing_entry.add_to_hass(hass)
# test with all provided
with PATCH_GET_HOST, PATCH_SETUP_ENTRY:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER, "show_advanced_options": True},
data=CONFIG_DATA,
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == HOST
assert result["data"] == CONFIG_DATA
prev_entry = hass.config_entries.async_get_entry(existing_entry.entry_id)
assert not prev_entry
@pytest.mark.usefixtures("connect")
async def test_abort_invalid_unique_id(hass):
"""Test we abort if uniqueid not available."""
MockConfigEntry(
domain=DOMAIN,
data=CONFIG_DATA,
unique_id=MAC_ADDR,
).add_to_hass(hass)
with PATCH_GET_HOST:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data=CONFIG_DATA,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "single_instance_allowed"
assert result["reason"] == "invalid_unique_id"
async def test_on_connect_failed(hass):
@pytest.mark.parametrize(
["side_effect", "error"],
[
(OSError, "cannot_connect"),
(TypeError, "unknown"),
(None, "cannot_connect"),
],
)
async def test_on_connect_failed(hass, side_effect, error):
"""Test when we have errors connecting the router."""
flow_result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER, "show_advanced_options": True},
)
with patch("homeassistant.components.asuswrt.router.AsusWrt") as asus_wrt:
asus_wrt.return_value.connection.async_connect = AsyncMock()
asus_wrt.return_value.is_connected = False
result = await hass.config_entries.flow.async_configure(
flow_result["flow_id"], user_input=CONFIG_DATA
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "cannot_connect"}
with patch("homeassistant.components.asuswrt.router.AsusWrt") as asus_wrt:
asus_wrt.return_value.connection.async_connect = AsyncMock(side_effect=OSError)
result = await hass.config_entries.flow.async_configure(
flow_result["flow_id"], user_input=CONFIG_DATA
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "cannot_connect"}
with patch("homeassistant.components.asuswrt.router.AsusWrt") as asus_wrt:
with PATCH_GET_HOST, patch(
"homeassistant.components.asuswrt.router.AsusWrt"
) as asus_wrt:
asus_wrt.return_value.connection.async_connect = AsyncMock(
side_effect=TypeError
side_effect=side_effect
)
asus_wrt.return_value.is_connected = False
result = await hass.config_entries.flow.async_configure(
flow_result["flow_id"], user_input=CONFIG_DATA
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "unknown"}
assert result["errors"] == {"base": error}
async def test_options_flow(hass):
@ -202,7 +257,7 @@ async def test_options_flow(hass):
)
config_entry.add_to_hass(hass)
with patch("homeassistant.components.asuswrt.async_setup_entry", return_value=True):
with PATCH_SETUP_ENTRY:
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(config_entry.entry_id)

19
tests/components/asuswrt/test_sensor.py
View File

@ -7,7 +7,7 @@ import pytest
from homeassistant.components import device_tracker, sensor
from homeassistant.components.asuswrt.const import CONF_INTERFACE, DOMAIN
from homeassistant.components.asuswrt.sensor import DEFAULT_PREFIX
from homeassistant.components.asuswrt.router import DEFAULT_NAME
from homeassistant.components.device_tracker.const import CONF_CONSIDER_HOME
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import (
@ -39,6 +39,8 @@ CONFIG_DATA = {
CONF_MODE: "router",
}
MAC_ADDR = "a1:b2:c3:d4:e5:f6"
MOCK_BYTES_TOTAL = [60000000000, 50000000000]
MOCK_CURRENT_TRANSFER_RATES = [20000000, 10000000]
MOCK_LOAD_AVG = [1.1, 1.2, 1.3]
@ -157,7 +159,7 @@ def mock_controller_connect_sens_fail():
yield service_mock
def _setup_entry(hass):
def _setup_entry(hass, unique_id=None):
"""Create mock config entry."""
entity_reg = er.async_get(hass)
@ -166,11 +168,11 @@ def _setup_entry(hass):
domain=DOMAIN,
data=CONFIG_DATA,
options={CONF_CONSIDER_HOME: 60},
unique_id=unique_id,
)
# init variable
unique_id = DOMAIN
obj_prefix = slugify(DEFAULT_PREFIX)
obj_prefix = slugify(HOST if unique_id else DEFAULT_NAME)
sensor_prefix = f"{sensor.DOMAIN}.{obj_prefix}"
# Pre-enable the status sensor
@ -179,7 +181,7 @@ def _setup_entry(hass):
entity_reg.async_get_or_create(
sensor.DOMAIN,
DOMAIN,
f"{unique_id} {DEFAULT_PREFIX} {sensor_name}",
f"{DOMAIN} {unique_id or DEFAULT_NAME} {sensor_name}",
suggested_object_id=f"{obj_prefix}_{sensor_id}",
disabled_by=None,
)
@ -202,15 +204,20 @@ def _setup_entry(hass):
return config_entry, sensor_prefix
@pytest.mark.parametrize(
"entry_unique_id",
[None, MAC_ADDR],
)
async def test_sensors(
hass,
connect,
mock_devices,
mock_available_temps,
create_device_registry_devices,
entry_unique_id,
):
"""Test creating an AsusWRT sensor."""
config_entry, sensor_prefix = _setup_entry(hass)
config_entry, sensor_prefix = _setup_entry(hass, entry_unique_id)
config_entry.add_to_hass(hass)
# initial devices setup