github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Add config flow to linky (#26076) 

* Linky: setup ConfigFlow

* async_track_time_interval

* Review from @MartinHjelmare 1

* Review from @MartinHjelmare 2

* Review from @MartinHjelmare 3

* Review from @MartinHjelmare 4

* black --fast homeassistant tests

* Bump pylinky to 0.4.0 and add error user feedback

* Fix .coveragerc

* Linky platform moved to integration in config.yml and with multiple accounts

* Remove useless logs

* Review from @MartinHjelmare 5

* Add config flow tests

* Add config flow tests : login + fetch on failed
This commit is contained in:
Quentame 2019-09-04 07:04:26 +02:00 committed by Martin Hjelmare
parent ca97bba4b4
commit b4058b5c7f
15 changed files with 452 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -339,6 +339,7 @@ omit =
homeassistant/components/limitlessled/light.py
homeassistant/components/linksys_ap/device_tracker.py
homeassistant/components/linksys_smart/device_tracker.py
homeassistant/components/linky/__init__.py
homeassistant/components/linky/sensor.py
homeassistant/components/linode/*
homeassistant/components/linux_battery/sensor.py

2
CODEOWNERS
View File

@ -153,7 +153,7 @@ homeassistant/components/life360/* @pnbruckner
homeassistant/components/lifx/* @amelchio
homeassistant/components/lifx_cloud/* @amelchio
homeassistant/components/lifx_legacy/* @amelchio
homeassistant/components/linky/* @tiste @Quentame
homeassistant/components/linky/* @Quentame
homeassistant/components/linux_battery/* @fabaff
homeassistant/components/liveboxplaytv/* @pschmitt
homeassistant/components/logger/* @home-assistant/core

25
homeassistant/components/linky/.translations/en.json Normal file
View File

@ -0,0 +1,25 @@
{
"config": {
"abort": {
"username_exists": "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": {
"user": {
"data": {
"password": "Password",
"username": "Email"
},
"description": "Enter your credentials",
"title": "Linky"
}
},
"title": "Linky"
}
}

54
homeassistant/components/linky/__init__.py
View File

@ -1 +1,55 @@
"""The linky component."""
import logging
import voluptuous as vol
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_PASSWORD, CONF_TIMEOUT, CONF_USERNAME
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import HomeAssistantType
from .const import DEFAULT_TIMEOUT, DOMAIN
_LOGGER = logging.getLogger(__name__)
ACCOUNT_SCHEMA = vol.Schema(
{
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
}
)
CONFIG_SCHEMA = vol.Schema(
{DOMAIN: vol.Schema(vol.All(cv.ensure_list, [ACCOUNT_SCHEMA]))},
extra=vol.ALLOW_EXTRA,
)
async def async_setup(hass, config):
"""Set up Linky sensors from legacy config file."""
conf = config.get(DOMAIN)
if conf is None:
return True
for linky_account_conf in conf:
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=linky_account_conf.copy(),
)
)
return True
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry):
"""Set up Linky sensors."""
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, "sensor")
)
return True

118
homeassistant/components/linky/config_flow.py Normal file
View File

@ -0,0 +1,118 @@
"""Config flow to configure the Linky integration."""
import logging
import voluptuous as vol
from pylinky.client import LinkyClient
from pylinky.exceptions import (
PyLinkyAccessException,
PyLinkyEnedisException,
PyLinkyException,
PyLinkyWrongLoginException,
)
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
_LOGGER = logging.getLogger(__name__)
class LinkyFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow."""
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."""
if user_input is None:
user_input = {}
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(
CONF_USERNAME, default=user_input.get(CONF_USERNAME, "")
): str,
vol.Required(
CONF_PASSWORD, default=user_input.get(CONF_PASSWORD, "")
): str,
}
),
errors=errors or {},
)
async def async_step_user(self, user_input=None):
"""Handle a flow initiated by the user."""
errors = {}
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)
if self._configuration_exists(self._username):
errors[CONF_USERNAME] = "username_exists"
return self._show_setup_form(user_input, errors)
client = LinkyClient(self._username, self._password, None, self._timeout)
try:
await self.hass.async_add_executor_job(client.login)
await self.hass.async_add_executor_job(client.fetch_data)
except PyLinkyAccessException as exp:
_LOGGER.error(exp)
errors["base"] = "access"
return self._show_setup_form(user_input, errors)
except PyLinkyEnedisException as exp:
_LOGGER.error(exp)
errors["base"] = "enedis"
return self._show_setup_form(user_input, errors)
except PyLinkyWrongLoginException as exp:
_LOGGER.error(exp)
errors["base"] = "wrong_login"
return self._show_setup_form(user_input, errors)
except PyLinkyException as exp:
_LOGGER.error(exp)
errors["base"] = "unknown"
return self._show_setup_form(user_input, errors)
finally:
client.close_session()
return self.async_create_entry(
title=self._username,
data={
CONF_USERNAME: self._username,
CONF_PASSWORD: self._password,
CONF_TIMEOUT: self._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")
return await self.async_step_user(user_input)

5
homeassistant/components/linky/const.py Normal file
View File

@ -0,0 +1,5 @@
"""Linky component constants."""
DOMAIN = "linky"
DEFAULT_TIMEOUT = 10

4
homeassistant/components/linky/manifest.json
View File

@ -1,13 +1,13 @@
{
"domain": "linky",
"name": "Linky",
"config_flow": true,
"documentation": "https://www.home-assistant.io/components/linky",
"requirements": [
"pylinky==0.3.3"
"pylinky==0.4.0"
],
"dependencies": [],
"codeowners": [
"@tiste",
"@Quentame"
]
}

101
homeassistant/components/linky/sensor.py
View File

@ -1,12 +1,12 @@
"""Support for Linky."""
from datetime import timedelta
import json
import logging
from datetime import timedelta
from pylinky.client import DAILY, MONTHLY, YEARLY, LinkyClient, PyLinkyError
import voluptuous as vol
from pylinky.client import DAILY, MONTHLY, YEARLY, LinkyClient
from pylinky.client import PyLinkyException
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_ATTRIBUTION,
CONF_PASSWORD,
@ -14,10 +14,9 @@ from homeassistant.const import (
CONF_USERNAME,
ENERGY_KILO_WATT_HOUR,
)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.event import track_time_interval
import homeassistant.util.dt as dt_util
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.typing import HomeAssistantType
_LOGGER = logging.getLogger(__name__)
@ -29,7 +28,6 @@ INDEX_CURRENT = -1
INDEX_LAST = -2
ATTRIBUTION = "Data provided by Enedis"
DEFAULT_TIMEOUT = 10
SENSORS = {
"yesterday": ("Linky yesterday", DAILY, INDEX_LAST),
"current_month": ("Linky current month", MONTHLY, INDEX_CURRENT),
@ -41,59 +39,54 @@ SENSORS_INDEX_LABEL = 0
SENSORS_INDEX_SCALE = 1
SENSORS_INDEX_WHEN = 2
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
}
)
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Old way of setting up the Linky platform."""
pass
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Configure the platform and add the Linky sensor."""
username = config[CONF_USERNAME]
password = config[CONF_PASSWORD]
timeout = config[CONF_TIMEOUT]
async def async_setup_entry(
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities
) -> None:
"""Add Linky entries."""
account = LinkyAccount(
entry.data[CONF_USERNAME], entry.data[CONF_PASSWORD], entry.data[CONF_TIMEOUT]
)
account = LinkyAccount(hass, add_entities, username, password, timeout)
add_entities(account.sensors, True)
await hass.async_add_executor_job(account.update_linky_data)
sensors = [
LinkySensor("Linky yesterday", account, DAILY, INDEX_LAST),
LinkySensor("Linky current month", account, MONTHLY, INDEX_CURRENT),
LinkySensor("Linky last month", account, MONTHLY, INDEX_LAST),
LinkySensor("Linky current year", account, YEARLY, INDEX_CURRENT),
LinkySensor("Linky last year", account, YEARLY, INDEX_LAST),
]
async_track_time_interval(hass, account.update_linky_data, SCAN_INTERVAL)
async_add_entities(sensors, True)
class LinkyAccount:
"""Representation of a Linky account."""
def __init__(self, hass, add_entities, username, password, timeout):
def __init__(self, username, password, timeout):
"""Initialise the Linky account."""
self._username = username
self.__password = password
self._password = password
self._timeout = timeout
self._data = None
self.sensors = []
self.update_linky_data(dt_util.utcnow())
self.sensors.append(LinkySensor("Linky yesterday", self, DAILY, INDEX_LAST))
self.sensors.append(
LinkySensor("Linky current month", self, MONTHLY, INDEX_CURRENT)
)
self.sensors.append(LinkySensor("Linky last month", self, MONTHLY, INDEX_LAST))
self.sensors.append(
LinkySensor("Linky current year", self, YEARLY, INDEX_CURRENT)
)
self.sensors.append(LinkySensor("Linky last year", self, YEARLY, INDEX_LAST))
track_time_interval(hass, self.update_linky_data, SCAN_INTERVAL)
def update_linky_data(self, event_time):
def update_linky_data(self, event_time=None):
"""Fetch new state data for the sensor."""
client = LinkyClient(self._username, self.__password, None, self._timeout)
client = LinkyClient(self._username, self._password, None, self._timeout)
try:
client.login()
client.fetch_data()
self._data = client.get_data()
_LOGGER.debug(json.dumps(self._data, indent=2))
except PyLinkyError as exp:
except PyLinkyException as exp:
_LOGGER.error(exp)
finally:
client.close_session()
@ -115,12 +108,12 @@ class LinkySensor(Entity):
def __init__(self, name, account: LinkyAccount, scale, when):
"""Initialize the sensor."""
self._name = name
self.__account = account
self._account = account
self._scale = scale
self.__when = when
self._when = when
self._username = account.username
self.__time = None
self.__consumption = None
self._time = None
self._consumption = None
@property
def name(self):
@ -130,7 +123,7 @@ class LinkySensor(Entity):
@property
def state(self):
"""Return the state of the sensor."""
return self.__consumption
return self._consumption
@property
def unit_of_measurement(self):
@ -147,18 +140,18 @@ class LinkySensor(Entity):
"""Return the state attributes of the sensor."""
return {
ATTR_ATTRIBUTION: ATTRIBUTION,
"time": self.__time,
"time": self._time,
CONF_USERNAME: self._username,
}
def update(self):
async def async_update(self) -> None:
"""Retrieve the new data for the sensor."""
data = self.__account.data[self._scale][self.__when]
self.__consumption = data[CONSUMPTION]
self.__time = data[TIME]
data = self._account.data[self._scale][self._when]
self._consumption = data[CONSUMPTION]
self._time = data[TIME]
if self._scale is not YEARLY:
year_index = INDEX_CURRENT
if self.__time.endswith("Dec"):
if self._time.endswith("Dec"):
year_index = INDEX_LAST
self.__time += " " + self.__account.data[YEARLY][year_index][TIME]
self._time += " " + self._account.data[YEARLY][year_index][TIME]

25
homeassistant/components/linky/strings.json Normal file
View File

@ -0,0 +1,25 @@
{
"config": {
"title": "Linky",
"step": {
"user": {
"title": "Linky",
"description": "Enter your credentials",
"data": {
"username": "Email",
"password": "Password"
}
}
},
"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"
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -30,6 +30,7 @@ FLOWS = [
"iqvia",
"life360",
"lifx",
"linky",
"locative",
"logi_circle",
"luftdaten",

2
requirements_all.txt
View File

@ -1259,7 +1259,7 @@ pylgnetcast-homeassistant==0.2.0.dev0
pylgtv==0.1.9
# homeassistant.components.linky
pylinky==0.3.3
pylinky==0.4.0
# homeassistant.components.litejet
pylitejet==0.1

3
requirements_test_all.txt
View File

@ -291,6 +291,9 @@ pyhomematic==0.1.60
# homeassistant.components.iqvia
pyiqvia==0.2.1
# homeassistant.components.linky
pylinky==0.4.0
# homeassistant.components.litejet
pylitejet==0.1

1
script/gen_requirements_all.py
View File

@ -120,6 +120,7 @@ TEST_REQUIREMENTS = (
"pyheos",
"pyhomematic",
"pyiqvia",
"pylinky",
"pylitejet",
"pymfy",
"pymonoprice",

1
tests/components/linky/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the Linky component."""

167
tests/components/linky/test_config_flow.py Normal file
View File

@ -0,0 +1,167 @@
"""Tests for the Linky config flow."""
import pytest
from unittest.mock import patch
from pylinky.exceptions import (
PyLinkyAccessException,
PyLinkyEnedisException,
PyLinkyException,
PyLinkyWrongLoginException,
)
from homeassistant import data_entry_flow
from homeassistant.components.linky import config_flow
from homeassistant.components.linky.const import DOMAIN, DEFAULT_TIMEOUT
from homeassistant.const import CONF_PASSWORD, CONF_TIMEOUT, CONF_USERNAME
from tests.common import MockConfigEntry
USERNAME = "username"
PASSWORD = "password"
TIMEOUT = 20
@pytest.fixture(name="login")
def mock_controller_login():
"""Mock a successful login."""
with patch("pylinky.client.LinkyClient.login", return_value=True):
yield
@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
@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):
"""Test user config."""
flow = init_config_flow(hass)
result = await flow.async_step_user()
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}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
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):
"""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}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
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}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == USERNAME
assert result["data"][CONF_USERNAME] == USERNAME
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):
"""Test we abort if Linky is already setup."""
flow = init_config_flow(hass)
MockConfigEntry(
domain=DOMAIN, data={CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD}
).add_to_hass(hass)
# Should fail, same USERNAME (import)
result = await flow.async_step_import(
{CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "username_exists"
# Should fail, same USERNAME (flow)
result = await flow.async_step_user(
{CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {CONF_USERNAME: "username_exists"}
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):
"""Test when we have errors during fetch."""
flow = init_config_flow(hass)
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"}
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"}