evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add options flow to OpenSky (#98177)

Browse source
This commit is contained in:
Joost Lekkerkerker 2023-08-22 18:06:19 +02:00 • committed by GitHub
parent 59900a49e2
commit 19576e6c95
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 314 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

24
homeassistant/components/opensky/__init__.py
View file

@ -1,13 +1,17 @@
"""The opensky component."""
from __future__ import annotations
from aiohttp import BasicAuth
from python_opensky import OpenSky
from python_opensky.exceptions import OpenSkyUnauthenticatedError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN, PLATFORMS
from .const import CONF_CONTRIBUTING_USER, DOMAIN, PLATFORMS
from .coordinator import OpenSkyDataUpdateCoordinator
@ -15,11 +19,24 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up opensky from a config entry."""
client = OpenSky(session=async_get_clientsession(hass))
if CONF_USERNAME in entry.options and CONF_PASSWORD in entry.options:
try:
await client.authenticate(
BasicAuth(
login=entry.options[CONF_USERNAME],
password=entry.options[CONF_PASSWORD],
),
contributing_user=entry.options.get(CONF_CONTRIBUTING_USER, False),
)
except OpenSkyUnauthenticatedError as exc:
raise ConfigEntryNotReady from exc
coordinator = OpenSkyDataUpdateCoordinator(hass, client)
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(update_listener))
return True
@ -28,3 +45,8 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload opensky config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle options update."""
await hass.config_entries.async_reload(entry.entry_id)

84
homeassistant/components/opensky/config_flow.py
View file

@ -3,21 +3,45 @@ from __future__ import annotations
from typing import Any
from aiohttp import BasicAuth
from python_opensky import OpenSky
from python_opensky.exceptions import OpenSkyUnauthenticatedError
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME, CONF_RADIUS
from homeassistant.config_entries import (
ConfigEntry,
ConfigFlow,
OptionsFlowWithConfigEntry,
)
from homeassistant.const import (
CONF_LATITUDE,
CONF_LONGITUDE,
CONF_NAME,
CONF_PASSWORD,
CONF_RADIUS,
CONF_USERNAME,
)
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
from .const import DEFAULT_NAME, DOMAIN
from .const import CONF_CONTRIBUTING_USER, DEFAULT_NAME, DOMAIN
from .sensor import CONF_ALTITUDE, DEFAULT_ALTITUDE
class OpenSkyConfigFlowHandler(ConfigFlow, domain=DOMAIN):
"""Config flow handler for OpenSky."""
@staticmethod
@callback
def async_get_options_flow(
config_entry: ConfigEntry,
) -> OpenSkyOptionsFlowHandler:
"""Get the options flow for this handler."""
return OpenSkyOptionsFlowHandler(config_entry)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
@ -70,3 +94,57 @@ class OpenSkyConfigFlowHandler(ConfigFlow, domain=DOMAIN):
CONF_ALTITUDE: import_config.get(CONF_ALTITUDE, DEFAULT_ALTITUDE),
},
)
class OpenSkyOptionsFlowHandler(OptionsFlowWithConfigEntry):
"""OpenSky Options flow handler."""
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Initialize form."""
errors: dict[str, str] = {}
if user_input is not None:
authentication = CONF_USERNAME in user_input or CONF_PASSWORD in user_input
if authentication and CONF_USERNAME not in user_input:
errors["base"] = "username_missing"
if authentication and CONF_PASSWORD not in user_input:
errors["base"] = "password_missing"
if user_input[CONF_CONTRIBUTING_USER] and not authentication:
errors["base"] = "no_authentication"
if authentication and not errors:
async with OpenSky(
session=async_get_clientsession(self.hass)
) as opensky:
try:
await opensky.authenticate(
BasicAuth(
login=user_input[CONF_USERNAME],
password=user_input[CONF_PASSWORD],
),
contributing_user=user_input[CONF_CONTRIBUTING_USER],
)
except OpenSkyUnauthenticatedError:
errors["base"] = "invalid_auth"
if not errors:
return self.async_create_entry(
title=self.options.get(CONF_NAME, "OpenSky"),
data=user_input,
)
return self.async_show_form(
step_id="init",
errors=errors,
data_schema=self.add_suggested_values_to_schema(
vol.Schema(
{
vol.Required(CONF_RADIUS): vol.Coerce(float),
vol.Optional(CONF_ALTITUDE): vol.Coerce(float),
vol.Optional(CONF_USERNAME): str,
vol.Optional(CONF_PASSWORD): str,
vol.Optional(CONF_CONTRIBUTING_USER, default=False): bool,
}
),
user_input or self.options,
),
)

1
homeassistant/components/opensky/const.py
View file

@ -10,6 +10,7 @@ DEFAULT_NAME = "OpenSky"
DOMAIN = "opensky"
MANUFACTURER = "OpenSky Network"
CONF_ALTITUDE = "altitude"
CONF_CONTRIBUTING_USER = "contributing_user"
ATTR_ICAO24 = "icao24"
ATTR_CALLSIGN = "callsign"
ATTR_ALTITUDE = "altitude"

6
homeassistant/components/opensky/coordinator.py
View file

@ -41,8 +41,10 @@ class OpenSkyDataUpdateCoordinator(DataUpdateCoordinator[int]):
hass,
LOGGER,
name=DOMAIN,
# OpenSky free user has 400 credits, with 4 credits per API call. 100/24 = ~4 requests per hour
update_interval=timedelta(minutes=15),
update_interval={
True: timedelta(seconds=90),
False: timedelta(minutes=15),
}.get(opensky.is_authenticated),
)
self._opensky = opensky
self._previously_tracked: set[str] | None = None

20
homeassistant/components/opensky/strings.json
View file

@ -11,5 +11,25 @@
}
}
}
},
"options": {
"step": {
"init": {
"description": "You can login to your OpenSky account to increase the update frequency.",
"data": {
"radius": "[%key:component::opensky::config::step::user::data::radius%]",
"altitude": "[%key:component::opensky::config::step::user::data::altitude%]",
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]",
"contributing_user": "I'm contributing to OpenSky"
}
}
},
"error": {
"username_missing": "Username is missing",
"password_missing": "Password is missing",
"no_authentication": "You need to authenticate to be contributing",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
}
}
}

11
tests/components/opensky/__init__.py
View file

@ -1,9 +1,20 @@
"""Opensky tests."""
import json
from unittest.mock import patch
from python_opensky import StatesResponse
from tests.common import load_fixture
def patch_setup_entry() -> bool:
"""Patch interface."""
return patch(
"homeassistant.components.opensky.async_setup_entry", return_value=True
)
def get_states_response_fixture(fixture: str) -> StatesResponse:
"""Return the states response from json."""
json_fixture = load_fixture(fixture)
return StatesResponse.parse_obj(json.loads(json_fixture))

34
tests/components/opensky/conftest.py
View file

@ -6,8 +6,18 @@ from unittest.mock import patch
import pytest
from python_opensky import StatesResponse
from homeassistant.components.opensky.const import CONF_ALTITUDE, DOMAIN
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE, CONF_RADIUS
from homeassistant.components.opensky.const import (
CONF_ALTITUDE,
CONF_CONTRIBUTING_USER,
DOMAIN,
)
from homeassistant.const import (
CONF_LATITUDE,
CONF_LONGITUDE,
CONF_PASSWORD,
CONF_RADIUS,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
@ -50,6 +60,26 @@ def mock_config_entry_altitude() -> MockConfigEntry:
)
@pytest.fixture(name="config_entry_authenticated")
def mock_config_entry_authenticated() -> MockConfigEntry:
"""Create authenticated Opensky entry in Home Assistant."""
return MockConfigEntry(
domain=DOMAIN,
title="OpenSky",
data={
CONF_LATITUDE: 0.0,
CONF_LONGITUDE: 0.0,
},
options={
CONF_RADIUS: 10.0,
CONF_ALTITUDE: 12500.0,
CONF_USERNAME: "asd",
CONF_PASSWORD: "secret",
CONF_CONTRIBUTING_USER: True,
},
)
@pytest.fixture(name="setup_integration")
async def mock_setup_integration(
hass: HomeAssistant,

128
tests/components/opensky/test_config_flow.py
View file

@ -1,15 +1,31 @@
"""Test OpenSky config flow."""
from typing import Any
from unittest.mock import patch
import pytest
from python_opensky.exceptions import OpenSkyUnauthenticatedError
from homeassistant.components.opensky.const import CONF_ALTITUDE, DEFAULT_NAME, DOMAIN
from homeassistant import data_entry_flow
from homeassistant.components.opensky.const import (
CONF_ALTITUDE,
CONF_CONTRIBUTING_USER,
DEFAULT_NAME,
DOMAIN,
)
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME, CONF_RADIUS
from homeassistant.const import (
CONF_LATITUDE,
CONF_LONGITUDE,
CONF_NAME,
CONF_PASSWORD,
CONF_RADIUS,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from . import patch_setup_entry
from . import get_states_response_fixture, patch_setup_entry
from .conftest import ComponentSetup
from tests.common import MockConfigEntry
@ -149,3 +165,109 @@ async def test_importing_already_exists_flow(hass: HomeAssistant) -> None:
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
("user_input", "error"),
[
(
{CONF_USERNAME: "homeassistant", CONF_CONTRIBUTING_USER: False},
"password_missing",
),
({CONF_PASSWORD: "secret", CONF_CONTRIBUTING_USER: False}, "username_missing"),
({CONF_CONTRIBUTING_USER: True}, "no_authentication"),
(
{
CONF_USERNAME: "homeassistant",
CONF_PASSWORD: "secret",
CONF_CONTRIBUTING_USER: True,
},
"invalid_auth",
),
],
)
async def test_options_flow_failures(
hass: HomeAssistant,
setup_integration: ComponentSetup,
config_entry: MockConfigEntry,
user_input: dict[str, Any],
error: str,
) -> None:
"""Test load and unload entry."""
await setup_integration(config_entry)
entry = hass.config_entries.async_entries(DOMAIN)[0]
with patch(
"python_opensky.OpenSky.authenticate",
side_effect=OpenSkyUnauthenticatedError(),
):
result = await hass.config_entries.options.async_init(entry.entry_id)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_RADIUS: 10000, **user_input},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "init"
assert result["errors"]["base"] == error
with patch("python_opensky.OpenSky.authenticate"), patch(
"python_opensky.OpenSky.get_states",
return_value=get_states_response_fixture("opensky/states_1.json"),
):
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_RADIUS: 10000,
CONF_USERNAME: "homeassistant",
CONF_PASSWORD: "secret",
CONF_CONTRIBUTING_USER: True,
},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["data"] == {
CONF_RADIUS: 10000,
CONF_USERNAME: "homeassistant",
CONF_PASSWORD: "secret",
CONF_CONTRIBUTING_USER: True,
}
async def test_options_flow(
hass: HomeAssistant,
setup_integration: ComponentSetup,
config_entry: MockConfigEntry,
) -> None:
"""Test options flow."""
await setup_integration(config_entry)
entry = hass.config_entries.async_entries(DOMAIN)[0]
result = await hass.config_entries.options.async_init(entry.entry_id)
await hass.async_block_till_done()
with patch("python_opensky.OpenSky.authenticate"), patch(
"python_opensky.OpenSky.get_states",
return_value=get_states_response_fixture("opensky/states_1.json"),
):
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_RADIUS: 10000,
CONF_USERNAME: "homeassistant",
CONF_PASSWORD: "secret",
CONF_CONTRIBUTING_USER: True,
},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["data"] == {
CONF_RADIUS: 10000,
CONF_USERNAME: "homeassistant",
CONF_PASSWORD: "secret",
CONF_CONTRIBUTING_USER: True,
}

17
tests/components/opensky/test_init.py
View file

@ -4,6 +4,7 @@ from __future__ import annotations
from unittest.mock import patch
from python_opensky import OpenSkyError
from python_opensky.exceptions import OpenSkyUnauthenticatedError
from homeassistant.components.opensky.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
@ -48,3 +49,19 @@ async def test_load_entry_failure(
await hass.async_block_till_done()
entry = hass.config_entries.async_entries(DOMAIN)[0]
assert entry.state == ConfigEntryState.SETUP_RETRY
async def test_load_entry_authentication_failure(
hass: HomeAssistant,
config_entry_authenticated: MockConfigEntry,
) -> None:
"""Test auth failure while loading."""
config_entry_authenticated.add_to_hass(hass)
with patch(
"python_opensky.OpenSky.authenticate",
side_effect=OpenSkyUnauthenticatedError(),
):
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
entry = hass.config_entries.async_entries(DOMAIN)[0]
assert entry.state == ConfigEntryState.SETUP_RETRY