mirror of
https://github.com/home-assistant/core.git
synced 2025-07-23 13:17:32 +00:00
Use config_entry.unique_id in Linky (#31051)
* Use config_entry.unique_id in Linky * Reviews * _show_setup_form not async
This commit is contained in:
parent
d3ac3e48a3
commit
a73a1a4489
@ -1,13 +1,12 @@
|
||||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"username_exists": "Account already configured"
|
||||
"already_configured": "Account already configured"
|
||||
},
|
||||
"error": {
|
||||
"access": "Could not access to Enedis.fr, please check your internet connection",
|
||||
"enedis": "Enedis.fr answered with an error: please retry later (usually not between 11PM and 2AM)",
|
||||
"unknown": "Unknown error: please retry later (usually not between 11PM and 2AM)",
|
||||
"username_exists": "Account already configured",
|
||||
"wrong_login": "Login error: please check your email & password"
|
||||
},
|
||||
"step": {
|
||||
|
@ -47,6 +47,12 @@ async def async_setup(hass, config):
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry):
|
||||
"""Set up Linky sensors."""
|
||||
# For backwards compat
|
||||
if entry.unique_id is None:
|
||||
hass.config_entries.async_update_entry(
|
||||
entry, unique_id=entry.data[CONF_USERNAME]
|
||||
)
|
||||
|
||||
hass.async_create_task(
|
||||
hass.config_entries.async_forward_entry_setup(entry, "sensor")
|
||||
)
|
||||
|
@ -12,9 +12,9 @@ import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.const import CONF_PASSWORD, CONF_TIMEOUT, CONF_USERNAME
|
||||
from homeassistant.core import callback
|
||||
|
||||
from .const import DEFAULT_TIMEOUT, DOMAIN
|
||||
from .const import DEFAULT_TIMEOUT
|
||||
from .const import DOMAIN # pylint: disable=unused-import
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
@ -25,20 +25,6 @@ class LinkyFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
VERSION = 1
|
||||
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize Linky config flow."""
|
||||
self._username = None
|
||||
self._password = None
|
||||
self._timeout = None
|
||||
|
||||
def _configuration_exists(self, username: str) -> bool:
|
||||
"""Return True if username exists in configuration."""
|
||||
for entry in self.hass.config_entries.async_entries(DOMAIN):
|
||||
if entry.data[CONF_USERNAME] == username:
|
||||
return True
|
||||
return False
|
||||
|
||||
@callback
|
||||
def _show_setup_form(self, user_input=None, errors=None):
|
||||
"""Show the setup form to the user."""
|
||||
|
||||
@ -67,15 +53,16 @@ class LinkyFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
if user_input is None:
|
||||
return self._show_setup_form(user_input, None)
|
||||
|
||||
self._username = user_input[CONF_USERNAME]
|
||||
self._password = user_input[CONF_PASSWORD]
|
||||
self._timeout = user_input.get(CONF_TIMEOUT, DEFAULT_TIMEOUT)
|
||||
username = user_input[CONF_USERNAME]
|
||||
password = user_input[CONF_PASSWORD]
|
||||
timeout = user_input.get(CONF_TIMEOUT, DEFAULT_TIMEOUT)
|
||||
|
||||
if self._configuration_exists(self._username):
|
||||
errors[CONF_USERNAME] = "username_exists"
|
||||
return self._show_setup_form(user_input, errors)
|
||||
# Check if already configured
|
||||
if self.unique_id is None:
|
||||
await self.async_set_unique_id(username)
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
client = LinkyClient(self._username, self._password, None, self._timeout)
|
||||
client = LinkyClient(username, password, None, timeout)
|
||||
try:
|
||||
await self.hass.async_add_executor_job(client.login)
|
||||
await self.hass.async_add_executor_job(client.fetch_data)
|
||||
@ -99,20 +86,14 @@ class LinkyFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
client.close_session()
|
||||
|
||||
return self.async_create_entry(
|
||||
title=self._username,
|
||||
title=username,
|
||||
data={
|
||||
CONF_USERNAME: self._username,
|
||||
CONF_PASSWORD: self._password,
|
||||
CONF_TIMEOUT: self._timeout,
|
||||
CONF_USERNAME: username,
|
||||
CONF_PASSWORD: password,
|
||||
CONF_TIMEOUT: timeout,
|
||||
},
|
||||
)
|
||||
|
||||
async def async_step_import(self, user_input=None):
|
||||
"""Import a config entry.
|
||||
|
||||
Only host was required in the yaml file all other fields are optional
|
||||
"""
|
||||
if self._configuration_exists(user_input[CONF_USERNAME]):
|
||||
return self.async_abort(reason="username_exists")
|
||||
|
||||
"""Import a config entry."""
|
||||
return await self.async_step_user(user_input)
|
||||
|
@ -12,14 +12,13 @@
|
||||
}
|
||||
},
|
||||
"error":{
|
||||
"username_exists": "Account already configured",
|
||||
"access": "Could not access to Enedis.fr, please check your internet connection",
|
||||
"enedis": "Enedis.fr answered with an error: please retry later (usually not between 11PM and 2AM)",
|
||||
"wrong_login": "Login error: please check your email & password",
|
||||
"unknown": "Unknown error: please retry later (usually not between 11PM and 2AM)"
|
||||
},
|
||||
"abort":{
|
||||
"username_exists": "Account already configured"
|
||||
"already_configured": "Account already configured"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -1,5 +1,5 @@
|
||||
"""Tests for the Linky config flow."""
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from pylinky.exceptions import (
|
||||
PyLinkyAccessException,
|
||||
@ -10,13 +10,15 @@ from pylinky.exceptions import (
|
||||
import pytest
|
||||
|
||||
from homeassistant import data_entry_flow
|
||||
from homeassistant.components.linky import config_flow
|
||||
from homeassistant.components.linky.const import DEFAULT_TIMEOUT, DOMAIN
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
|
||||
from homeassistant.const import CONF_PASSWORD, CONF_TIMEOUT, CONF_USERNAME
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
USERNAME = "username"
|
||||
USERNAME = "username@hotmail.fr"
|
||||
USERNAME_2 = "username@free.fr"
|
||||
PASSWORD = "password"
|
||||
TIMEOUT = 20
|
||||
|
||||
@ -24,145 +26,158 @@ TIMEOUT = 20
|
||||
@pytest.fixture(name="login")
|
||||
def mock_controller_login():
|
||||
"""Mock a successful login."""
|
||||
with patch("pylinky.client.LinkyClient.login", return_value=True):
|
||||
yield
|
||||
with patch(
|
||||
"homeassistant.components.linky.config_flow.LinkyClient"
|
||||
) as service_mock:
|
||||
service_mock.return_value.login = Mock(return_value=True)
|
||||
service_mock.return_value.close_session = Mock(return_value=None)
|
||||
yield service_mock
|
||||
|
||||
|
||||
@pytest.fixture(name="fetch_data")
|
||||
def mock_controller_fetch_data():
|
||||
"""Mock a successful get data."""
|
||||
with patch("pylinky.client.LinkyClient.fetch_data", return_value={}):
|
||||
yield
|
||||
with patch(
|
||||
"homeassistant.components.linky.config_flow.LinkyClient"
|
||||
) as service_mock:
|
||||
service_mock.return_value.fetch_data = Mock(return_value={})
|
||||
service_mock.return_value.close_session = Mock(return_value=None)
|
||||
yield service_mock
|
||||
|
||||
|
||||
@pytest.fixture(name="close_session")
|
||||
def mock_controller_close_session():
|
||||
"""Mock a successful closing session."""
|
||||
with patch("pylinky.client.LinkyClient.close_session", return_value=None):
|
||||
yield
|
||||
|
||||
|
||||
def init_config_flow(hass):
|
||||
"""Init a configuration flow."""
|
||||
flow = config_flow.LinkyFlowHandler()
|
||||
flow.hass = hass
|
||||
return flow
|
||||
|
||||
|
||||
async def test_user(hass, login, fetch_data, close_session):
|
||||
async def test_user(hass: HomeAssistantType, login, fetch_data):
|
||||
"""Test user config."""
|
||||
flow = init_config_flow(hass)
|
||||
|
||||
result = await flow.async_step_user()
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}, data=None
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
# test with all provided
|
||||
result = await flow.async_step_user(
|
||||
{CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD}
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
data={CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["result"].unique_id == USERNAME
|
||||
assert result["title"] == USERNAME
|
||||
assert result["data"][CONF_USERNAME] == USERNAME
|
||||
assert result["data"][CONF_PASSWORD] == PASSWORD
|
||||
assert result["data"][CONF_TIMEOUT] == DEFAULT_TIMEOUT
|
||||
|
||||
|
||||
async def test_import(hass, login, fetch_data, close_session):
|
||||
async def test_import(hass: HomeAssistantType, login, fetch_data):
|
||||
"""Test import step."""
|
||||
flow = init_config_flow(hass)
|
||||
|
||||
# import with username and password
|
||||
result = await flow.async_step_import(
|
||||
{CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD}
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data={CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["result"].unique_id == USERNAME
|
||||
assert result["title"] == USERNAME
|
||||
assert result["data"][CONF_USERNAME] == USERNAME
|
||||
assert result["data"][CONF_PASSWORD] == PASSWORD
|
||||
assert result["data"][CONF_TIMEOUT] == DEFAULT_TIMEOUT
|
||||
|
||||
# import with all
|
||||
result = await flow.async_step_import(
|
||||
{CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD, CONF_TIMEOUT: TIMEOUT}
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data={
|
||||
CONF_USERNAME: USERNAME_2,
|
||||
CONF_PASSWORD: PASSWORD,
|
||||
CONF_TIMEOUT: TIMEOUT,
|
||||
},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["title"] == USERNAME
|
||||
assert result["data"][CONF_USERNAME] == USERNAME
|
||||
assert result["result"].unique_id == USERNAME_2
|
||||
assert result["title"] == USERNAME_2
|
||||
assert result["data"][CONF_USERNAME] == USERNAME_2
|
||||
assert result["data"][CONF_PASSWORD] == PASSWORD
|
||||
assert result["data"][CONF_TIMEOUT] == TIMEOUT
|
||||
|
||||
|
||||
async def test_abort_if_already_setup(hass, login, fetch_data, close_session):
|
||||
async def test_abort_if_already_setup(hass: HomeAssistantType, login, fetch_data):
|
||||
"""Test we abort if Linky is already setup."""
|
||||
flow = init_config_flow(hass)
|
||||
MockConfigEntry(
|
||||
domain=DOMAIN, data={CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD}
|
||||
domain=DOMAIN,
|
||||
data={CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD},
|
||||
unique_id=USERNAME,
|
||||
).add_to_hass(hass)
|
||||
|
||||
# Should fail, same USERNAME (import)
|
||||
result = await flow.async_step_import(
|
||||
{CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD}
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data={CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "username_exists"
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
# Should fail, same USERNAME (flow)
|
||||
result = await flow.async_step_user(
|
||||
{CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD}
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
data={CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_login_failed(hass: HomeAssistantType, login):
|
||||
"""Test when we have errors during login."""
|
||||
login.return_value.login.side_effect = PyLinkyAccessException()
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data={CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {CONF_USERNAME: "username_exists"}
|
||||
assert result["errors"] == {"base": "access"}
|
||||
hass.config_entries.flow.async_abort(result["flow_id"])
|
||||
|
||||
login.return_value.login.side_effect = PyLinkyWrongLoginException()
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
data={CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "wrong_login"}
|
||||
hass.config_entries.flow.async_abort(result["flow_id"])
|
||||
|
||||
|
||||
async def test_abort_on_login_failed(hass, close_session):
|
||||
"""Test when we have errors during login."""
|
||||
flow = init_config_flow(hass)
|
||||
|
||||
with patch(
|
||||
"pylinky.client.LinkyClient.login", side_effect=PyLinkyAccessException()
|
||||
):
|
||||
result = await flow.async_step_user(
|
||||
{CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "access"}
|
||||
|
||||
with patch(
|
||||
"pylinky.client.LinkyClient.login", side_effect=PyLinkyWrongLoginException()
|
||||
):
|
||||
result = await flow.async_step_user(
|
||||
{CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "wrong_login"}
|
||||
|
||||
|
||||
async def test_abort_on_fetch_failed(hass, login, close_session):
|
||||
async def test_fetch_failed(hass: HomeAssistantType, login):
|
||||
"""Test when we have errors during fetch."""
|
||||
flow = init_config_flow(hass)
|
||||
login.return_value.fetch_data.side_effect = PyLinkyAccessException()
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
data={CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "access"}
|
||||
hass.config_entries.flow.async_abort(result["flow_id"])
|
||||
|
||||
with patch(
|
||||
"pylinky.client.LinkyClient.fetch_data", side_effect=PyLinkyAccessException()
|
||||
):
|
||||
result = await flow.async_step_user(
|
||||
{CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "access"}
|
||||
login.return_value.fetch_data.side_effect = PyLinkyEnedisException()
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
data={CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "enedis"}
|
||||
hass.config_entries.flow.async_abort(result["flow_id"])
|
||||
|
||||
with patch(
|
||||
"pylinky.client.LinkyClient.fetch_data", side_effect=PyLinkyEnedisException()
|
||||
):
|
||||
result = await flow.async_step_user(
|
||||
{CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "enedis"}
|
||||
|
||||
with patch("pylinky.client.LinkyClient.fetch_data", side_effect=PyLinkyException()):
|
||||
result = await flow.async_step_user(
|
||||
{CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "unknown"}
|
||||
login.return_value.fetch_data.side_effect = PyLinkyException()
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
data={CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "unknown"}
|
||||
hass.config_entries.flow.async_abort(result["flow_id"])
|
||||
|
Loading…
x
Reference in New Issue
Block a user