Cleanup code config flow smhi ()

pull/64219/head
G Johansson 2022-01-17 02:32:16 +01:00 committed by GitHub
parent 57ab296d8a
commit 3cfa2bb6fe
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 189 additions and 374 deletions
homeassistant/components/smhi
tests/components/smhi

View File

@ -8,22 +8,26 @@ import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import aiohttp_client
import homeassistant.helpers.config_validation as cv
from homeassistant.util import slugify
from .const import DOMAIN, HOME_LOCATION_NAME
from .const import DEFAULT_NAME, DOMAIN, HOME_LOCATION_NAME
@callback
def smhi_locations(hass: HomeAssistant) -> set[str]:
"""Return configurations of SMHI component."""
return {
slugify(entry.data[CONF_NAME])
for entry in hass.config_entries.async_entries(DOMAIN)
}
async def async_check_location(
hass: HomeAssistant, longitude: float, latitude: float
) -> bool:
"""Return true if location is ok."""
session = aiohttp_client.async_get_clientsession(hass)
smhi_api = Smhi(longitude, latitude, session=session)
try:
await smhi_api.async_get_forecast()
except SmhiForecastException:
return False
return True
class SmhiFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
@ -31,94 +35,52 @@ class SmhiFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
def __init__(self) -> None:
"""Initialize SMHI forecast configuration flow."""
self._errors: dict[str, str] = {}
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initialized by the user."""
self._errors = {}
errors: dict[str, str] = {}
if user_input is not None:
is_ok = await self._check_location(
user_input[CONF_LONGITUDE], user_input[CONF_LATITUDE]
)
if is_ok:
name = slugify(user_input[CONF_NAME])
if not self._name_in_configuration_exists(name):
latitude = user_input[CONF_LATITUDE]
longitude = user_input[CONF_LONGITUDE]
await self.async_set_unique_id(f"{latitude}-{longitude}")
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=user_input[CONF_NAME], data=user_input
)
lat: float = user_input[CONF_LATITUDE]
lon: float = user_input[CONF_LONGITUDE]
if await async_check_location(self.hass, lon, lat):
name = f"{DEFAULT_NAME} {round(lat, 6)} {round(lon, 6)}"
if (
lat == self.hass.config.latitude
and lon == self.hass.config.longitude
):
name = HOME_LOCATION_NAME
self._errors[CONF_NAME] = "name_exists"
else:
self._errors["base"] = "wrong_location"
user_input[CONF_NAME] = (
HOME_LOCATION_NAME if name == HOME_LOCATION_NAME else DEFAULT_NAME
)
# If hass config has the location set and is a valid coordinate the
# default location is set as default values in the form
if (
not smhi_locations(self.hass)
and await self._homeassistant_location_exists()
):
return await self._show_config_form(
name=HOME_LOCATION_NAME,
latitude=self.hass.config.latitude,
longitude=self.hass.config.longitude,
)
await self.async_set_unique_id(f"{lat}-{lon}")
self._abort_if_unique_id_configured()
return self.async_create_entry(title=name, data=user_input)
return await self._show_config_form()
errors["base"] = "wrong_location"
async def _homeassistant_location_exists(self) -> bool:
"""Return true if default location is set and is valid."""
# Return true if valid location
return (
self.hass.config.latitude != 0.0
and self.hass.config.longitude != 0.0
and await self._check_location(
self.hass.config.longitude, self.hass.config.latitude
)
)
default_lat: float = self.hass.config.latitude
default_lon: float = self.hass.config.longitude
def _name_in_configuration_exists(self, name: str) -> bool:
"""Return True if name exists in configuration."""
return name in smhi_locations(self.hass)
for entry in self.hass.config_entries.async_entries(DOMAIN):
if (
entry.data[CONF_LATITUDE] == self.hass.config.latitude
and entry.data[CONF_LONGITUDE] == self.hass.config.longitude
):
default_lat = 0
default_lon = 0
async def _show_config_form(
self,
name: str | None = None,
latitude: float | None = None,
longitude: float | None = None,
) -> FlowResult:
"""Show the configuration form to edit location data."""
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_NAME, default=name): str,
vol.Required(CONF_LATITUDE, default=latitude): cv.latitude,
vol.Required(CONF_LONGITUDE, default=longitude): cv.longitude,
vol.Required(CONF_LATITUDE, default=default_lat): cv.latitude,
vol.Required(CONF_LONGITUDE, default=default_lon): cv.longitude,
}
),
errors=self._errors,
errors=errors,
)
async def _check_location(self, longitude: float, latitude: float) -> bool:
"""Return true if location is ok."""
try:
session = aiohttp_client.async_get_clientsession(self.hass)
smhi_api = Smhi(longitude, latitude, session=session)
await smhi_api.async_get_forecast()
return True
except SmhiForecastException:
# The API will throw an exception if faulty location
pass
return False

View File

@ -10,5 +10,6 @@ ATTR_SMHI_THUNDER_PROBABILITY: Final = "thunder_probability"
DOMAIN = "smhi"
HOME_LOCATION_NAME = "Home"
DEFAULT_NAME = "Weather"
ENTITY_ID_SENSOR_FORMAT = WEATHER_DOMAIN + ".smhi_{}"

View File

@ -7,14 +7,12 @@
"user": {
"title": "Location in Sweden",
"data": {
"name": "[%key:common::config_flow::data::name%]",
"latitude": "[%key:common::config_flow::data::latitude%]",
"longitude": "[%key:common::config_flow::data::longitude%]"
}
}
},
"error": {
"name_exists": "Name already exists",
"wrong_location": "Location Sweden only"
}
}

View File

@ -4,15 +4,13 @@
"already_configured": "Account is already configured"
},
"error": {
"name_exists": "Name already exists",
"wrong_location": "Location Sweden only"
},
"step": {
"user": {
"data": {
"latitude": "Latitude",
"longitude": "Longitude",
"name": "Name"
"longitude": "Longitude"
},
"title": "Location in Sweden"
}

View File

@ -1,309 +1,165 @@
"""Tests for SMHI config flow."""
from unittest.mock import Mock, patch
"""Test the Smhi config flow."""
from __future__ import annotations
from smhi.smhi_lib import Smhi as SmhiApi, SmhiForecastException
from unittest.mock import patch
from homeassistant.components.smhi import config_flow
from smhi.smhi_lib import SmhiForecastException
from homeassistant import config_entries
from homeassistant.components.smhi.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import RESULT_TYPE_ABORT
from homeassistant.data_entry_flow import (
RESULT_TYPE_ABORT,
RESULT_TYPE_CREATE_ENTRY,
RESULT_TYPE_FORM,
)
from tests.common import MockConfigEntry
# pylint: disable=protected-access
async def test_homeassistant_location_exists() -> None:
"""Test if Home Assistant location exists it should return True."""
hass = Mock()
flow = config_flow.SmhiFlowHandler()
flow.hass = hass
with patch.object(flow, "_check_location", return_value=True):
# Test exists
hass.config.location_name = "Home"
hass.config.latitude = 17.8419
hass.config.longitude = 59.3262
async def test_form(hass: HomeAssistant) -> None:
"""Test we get the form and create an entry."""
assert await flow._homeassistant_location_exists() is True
# Test not exists
hass.config.location_name = None
hass.config.latitude = 0
hass.config.longitude = 0
assert await flow._homeassistant_location_exists() is False
async def test_name_in_configuration_exists() -> None:
"""Test if home location exists in configuration."""
hass = Mock()
flow = config_flow.SmhiFlowHandler()
flow.hass = hass
# Test exists
hass.config.location_name = "Home"
hass.config.latitude = 17.8419
hass.config.longitude = 59.3262
# Check not exists
with patch.object(
config_flow,
"smhi_locations",
return_value={"test": "something", "test2": "something else"},
):
assert flow._name_in_configuration_exists("no_exist_name") is False
# Check exists
with patch.object(
config_flow,
"smhi_locations",
return_value={"test": "something", "name_exist": "config"},
):
assert flow._name_in_configuration_exists("name_exist") is True
def test_smhi_locations(hass) -> None:
"""Test return empty set."""
locations = config_flow.smhi_locations(hass)
assert not locations
async def test_show_config_form() -> None:
"""Test show configuration form."""
hass = Mock()
flow = config_flow.SmhiFlowHandler()
flow.hass = hass
result = await flow._show_config_form()
assert result["type"] == "form"
assert result["step_id"] == "user"
async def test_show_config_form_default_values() -> None:
"""Test show configuration form."""
hass = Mock()
flow = config_flow.SmhiFlowHandler()
flow.hass = hass
result = await flow._show_config_form(name="test", latitude="65", longitude="17")
assert result["type"] == "form"
assert result["step_id"] == "user"
async def test_flow_with_home_location(hass) -> None:
"""Test config flow .
Tests the flow when a default location is configured
then it should return a form with default values
"""
flow = config_flow.SmhiFlowHandler()
flow.hass = hass
with patch.object(flow, "_check_location", return_value=True):
hass.config.location_name = "Home"
hass.config.latitude = 17.8419
hass.config.longitude = 59.3262
result = await flow.async_step_user()
assert result["type"] == "form"
assert result["step_id"] == "user"
async def test_flow_show_form() -> None:
"""Test show form scenarios first time.
Test when the form should show when no configurations exists
"""
hass = Mock()
flow = config_flow.SmhiFlowHandler()
flow.hass = hass
# Test show form when Home Assistant config exists and
# home is already configured, then new config is allowed
with patch.object(
flow, "_show_config_form", return_value=None
) as config_form, patch.object(
flow, "_homeassistant_location_exists", return_value=True
), patch.object(
config_flow,
"smhi_locations",
return_value={"test": "something", "name_exist": "config"},
):
await flow.async_step_user()
assert len(config_form.mock_calls) == 1
# Test show form when Home Assistant config not and
# home is not configured
with patch.object(
flow, "_show_config_form", return_value=None
) as config_form, patch.object(
flow, "_homeassistant_location_exists", return_value=False
), patch.object(
config_flow,
"smhi_locations",
return_value={"test": "something", "name_exist": "config"},
):
await flow.async_step_user()
assert len(config_form.mock_calls) == 1
async def test_flow_show_form_name_exists() -> None:
"""Test show form if name already exists.
Test when the form should show when no configurations exists
"""
hass = Mock()
flow = config_flow.SmhiFlowHandler()
flow.hass = hass
test_data = {"name": "home", CONF_LONGITUDE: "0", CONF_LATITUDE: "0"}
# Test show form when Home Assistant config exists and
# home is already configured, then new config is allowed
with patch.object(
flow, "_show_config_form", return_value=None
) as config_form, patch.object(
flow, "_name_in_configuration_exists", return_value=True
), patch.object(
config_flow,
"smhi_locations",
return_value={"test": "something", "name_exist": "config"},
), patch.object(
flow, "_check_location", return_value=True
):
await flow.async_step_user(user_input=test_data)
assert len(config_form.mock_calls) == 1
assert len(flow._errors) == 1
async def test_flow_entry_created_from_user_input() -> None:
"""Test that create data from user input.
Test when the form should show when no configurations exists
"""
hass = Mock()
flow = config_flow.SmhiFlowHandler()
flow.hass = hass
test_data = {"name": "home", CONF_LONGITUDE: "0", CONF_LATITUDE: "0"}
# Test that entry created when user_input name not exists
with patch.object(
flow, "_show_config_form", return_value=None
) as config_form, patch.object(
flow, "_name_in_configuration_exists", return_value=False
), patch.object(
flow, "_homeassistant_location_exists", return_value=False
), patch.object(
config_flow,
"smhi_locations",
return_value={"test": "something", "name_exist": "config"},
), patch.object(
flow, "_check_location", return_value=True
), patch.object(
flow, "async_set_unique_id", return_value=None
):
result = await flow.async_step_user(user_input=test_data)
assert result["type"] == "create_entry"
assert result["data"] == test_data
assert not config_form.mock_calls
async def test_flow_entry_created_user_input_faulty() -> None:
"""Test that create data from user input and are faulty.
Test when the form should show when user puts faulty location
in the config gui. Then the form should show with error
"""
hass = Mock()
flow = config_flow.SmhiFlowHandler()
flow.hass = hass
test_data = {"name": "home", CONF_LONGITUDE: "0", CONF_LATITUDE: "0"}
# Test that entry created when user_input name not exists
with patch.object(flow, "_check_location", return_value=True), patch.object(
flow, "_show_config_form", return_value=None
) as config_form, patch.object(
flow, "_name_in_configuration_exists", return_value=False
), patch.object(
flow, "_homeassistant_location_exists", return_value=False
), patch.object(
config_flow,
"smhi_locations",
return_value={"test": "something", "name_exist": "config"},
), patch.object(
flow, "_check_location", return_value=False
):
await flow.async_step_user(user_input=test_data)
assert len(config_form.mock_calls) == 1
assert len(flow._errors) == 1
async def test_check_location_correct() -> None:
"""Test check location when correct input."""
hass = Mock()
flow = config_flow.SmhiFlowHandler()
flow.hass = hass
with patch.object(
config_flow.aiohttp_client, "async_get_clientsession"
), patch.object(SmhiApi, "async_get_forecast", return_value=None):
assert await flow._check_location("58", "17") is True
async def test_check_location_faulty() -> None:
"""Test check location when faulty input."""
hass = Mock()
flow = config_flow.SmhiFlowHandler()
flow.hass = hass
with patch.object(
config_flow.aiohttp_client, "async_get_clientsession"
), patch.object(SmhiApi, "async_get_forecast", side_effect=SmhiForecastException()):
assert await flow._check_location("58", "17") is False
async def test_flow_unique_id_not_unique(hass: HomeAssistant) -> None:
"""Test that unique id is unique."""
test_data = {"name": "home", CONF_LONGITUDE: "0", CONF_LATITUDE: "0"}
MockConfigEntry(
domain=DOMAIN,
data={"name": "away", CONF_LONGITUDE: "0", CONF_LATITUDE: "0"},
unique_id="0.0-0.0",
).add_to_hass(hass)
hass.config.latitude = 0.0
hass.config.longitude = 0.0
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {}
with patch(
"homeassistant.components.smhi.config_flow.Smhi.async_get_forecast",
return_value={"test": "something", "test2": "something else"},
), patch(
"homeassistant.components.smhi.async_setup_entry",
return_value=True,
), patch(
"homeassistant.components.smhi.config_flow.Smhi.async_get_forecast",
return_value={"kalle": "anka"},
):
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
test_data,
{
CONF_LATITUDE: 0.0,
CONF_LONGITUDE: 0.0,
},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_ABORT
assert result2["reason"] == "already_configured"
assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == "Home"
assert result2["data"] == {
"latitude": 0.0,
"longitude": 0.0,
"name": "Home",
}
assert len(mock_setup_entry.mock_calls) == 1
# Check title is "Weather" when not home coordinates
result3 = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.smhi.config_flow.Smhi.async_get_forecast",
return_value={"test": "something", "test2": "something else"},
), patch(
"homeassistant.components.smhi.async_setup_entry",
return_value=True,
):
result4 = await hass.config_entries.flow.async_configure(
result3["flow_id"],
{
CONF_LATITUDE: 1.0,
CONF_LONGITUDE: 1.0,
},
)
await hass.async_block_till_done()
assert result4["type"] == RESULT_TYPE_CREATE_ENTRY
assert result4["title"] == "Weather 1.0 1.0"
assert result4["data"] == {
"latitude": 1.0,
"longitude": 1.0,
"name": "Weather",
}
async def test_form_invalid_coordinates(hass: HomeAssistant) -> None:
"""Test we handle invalid coordinates."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.smhi.config_flow.Smhi.async_get_forecast",
side_effect=SmhiForecastException,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_LATITUDE: 0.0,
CONF_LONGITUDE: 0.0,
},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_FORM
assert result2["errors"] == {"base": "wrong_location"}
# Continue flow with new coordinates
with patch(
"homeassistant.components.smhi.config_flow.Smhi.async_get_forecast",
return_value={"test": "something", "test2": "something else"},
), patch(
"homeassistant.components.smhi.async_setup_entry",
return_value=True,
):
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_LATITUDE: 2.0,
CONF_LONGITUDE: 2.0,
},
)
await hass.async_block_till_done()
assert result3["type"] == RESULT_TYPE_CREATE_ENTRY
assert result3["title"] == "Weather 2.0 2.0"
assert result3["data"] == {
"latitude": 2.0,
"longitude": 2.0,
"name": "Weather",
}
async def test_form_unique_id_exist(hass: HomeAssistant) -> None:
"""Test we handle unique id already exist."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="1.0-1.0",
data={
"latitude": 1.0,
"longitude": 1.0,
"name": "Weather",
},
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.smhi.config_flow.Smhi.async_get_forecast",
return_value={"test": "something", "test2": "something else"},
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_LATITUDE: 1.0,
CONF_LONGITUDE: 1.0,
},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_ABORT
assert result2["reason"] == "already_configured"