Swiss public transport config flow (#105648)
* add config flow * unit tests * yaml config import flow * change deprecation period and simply code * keep name for legacy yaml - removing the name now would break current implementations - it will be removed together with the deprectation of yaml config flow * improve error handling, simpler unique_id, cleanup * simplify issues for yaml import flow * improve typing and clean name handling * streamline unit tests - happy path + errors - mock opendata instead of aiohttp * parametrize unit tests * improve strings * add missing aborts * update coverage ignore * remove redundant test * minor clean up of constants
This commit is contained in:
Cyrill Raccaud
GitHub
parent
54f460b7c9
commit
7e685f2bc7
14 changed files with 501 additions and 34 deletions
|
|
@ -1243,6 +1243,7 @@ omit =
|
|||
homeassistant/components/surepetcare/entity.py
|
||||
homeassistant/components/surepetcare/sensor.py
|
||||
homeassistant/components/swiss_hydrological_data/sensor.py
|
||||
homeassistant/components/swiss_public_transport/__init__.py
|
||||
homeassistant/components/swiss_public_transport/sensor.py
|
||||
homeassistant/components/swisscom/device_tracker.py
|
||||
homeassistant/components/switchbee/__init__.py
|
||||
|
|
|
|||
|
|
@ -1262,7 +1262,8 @@ build.json @home-assistant/supervisor
|
|||
/homeassistant/components/surepetcare/ @benleb @danielhiversen
|
||||
/tests/components/surepetcare/ @benleb @danielhiversen
|
||||
/homeassistant/components/swiss_hydrological_data/ @fabaff
|
||||
/homeassistant/components/swiss_public_transport/ @fabaff
|
||||
/homeassistant/components/swiss_public_transport/ @fabaff @miaucl
|
||||
/tests/components/swiss_public_transport/ @fabaff @miaucl
|
||||
/homeassistant/components/switch/ @home-assistant/core
|
||||
/tests/components/switch/ @home-assistant/core
|
||||
/homeassistant/components/switch_as_x/ @home-assistant/core
|
||||
|
|
|
|||
|
|
@ -1 +1,64 @@
|
|||
"""The swiss_public_transport component."""
|
||||
import logging
|
||||
|
||||
from opendata_transport import OpendataTransport
|
||||
from opendata_transport.exceptions import (
|
||||
OpendataTransportConnectionError,
|
||||
OpendataTransportError,
|
||||
)
|
||||
|
||||
from homeassistant import config_entries, core
|
||||
from homeassistant.const import Platform
|
||||
from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
|
||||
from .const import CONF_DESTINATION, CONF_START, DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
PLATFORMS: list[Platform] = [Platform.SENSOR]
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: core.HomeAssistant, entry: config_entries.ConfigEntry
|
||||
) -> bool:
|
||||
"""Set up Swiss public transport from a config entry."""
|
||||
config = entry.data
|
||||
|
||||
start = config[CONF_START]
|
||||
destination = config[CONF_DESTINATION]
|
||||
|
||||
session = async_get_clientsession(hass)
|
||||
opendata = OpendataTransport(start, destination, session)
|
||||
|
||||
try:
|
||||
await opendata.async_get_data()
|
||||
except OpendataTransportConnectionError as e:
|
||||
raise ConfigEntryNotReady(
|
||||
f"Timeout while connecting for entry '{start} {destination}'"
|
||||
) from e
|
||||
except OpendataTransportError as e:
|
||||
_LOGGER.error(
|
||||
"Setup failed for entry '%s %s', check at http://transport.opendata.ch/examples/stationboard.html if your station names are valid",
|
||||
start,
|
||||
destination,
|
||||
)
|
||||
raise ConfigEntryError(
|
||||
f"Setup failed for entry '{start} {destination}' with invalid data"
|
||||
) from e
|
||||
|
||||
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = opendata
|
||||
|
||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(
|
||||
hass: core.HomeAssistant, entry: config_entries.ConfigEntry
|
||||
) -> bool:
|
||||
"""Unload a config entry."""
|
||||
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
|
||||
hass.data[DOMAIN].pop(entry.entry_id)
|
||||
|
||||
return unload_ok
|
||||
|
|
|
|||
101
homeassistant/components/swiss_public_transport/config_flow.py
Normal file
101
homeassistant/components/swiss_public_transport/config_flow.py
Normal file
|
|
@ -0,0 +1,101 @@
|
|||
"""Config flow for swiss_public_transport."""
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from opendata_transport import OpendataTransport
|
||||
from opendata_transport.exceptions import (
|
||||
OpendataTransportConnectionError,
|
||||
OpendataTransportError,
|
||||
)
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.const import CONF_NAME
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
|
||||
from .const import CONF_DESTINATION, CONF_START, DOMAIN
|
||||
|
||||
DATA_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_START): cv.string,
|
||||
vol.Required(CONF_DESTINATION): cv.string,
|
||||
}
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class SwissPublicTransportConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Swiss public transport config flow."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Async user step to set up the connection."""
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is not None:
|
||||
self._async_abort_entries_match(
|
||||
{
|
||||
CONF_START: user_input[CONF_START],
|
||||
CONF_DESTINATION: user_input[CONF_DESTINATION],
|
||||
}
|
||||
)
|
||||
|
||||
session = async_get_clientsession(self.hass)
|
||||
opendata = OpendataTransport(
|
||||
user_input[CONF_START], user_input[CONF_DESTINATION], session
|
||||
)
|
||||
try:
|
||||
await opendata.async_get_data()
|
||||
except OpendataTransportConnectionError:
|
||||
errors["base"] = "cannot_connect"
|
||||
except OpendataTransportError:
|
||||
errors["base"] = "bad_config"
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Unknown error")
|
||||
errors["base"] = "unknown"
|
||||
else:
|
||||
return self.async_create_entry(
|
||||
title=f"{user_input[CONF_START]} {user_input[CONF_DESTINATION]}",
|
||||
data=user_input,
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user", data_schema=DATA_SCHEMA, errors=errors
|
||||
)
|
||||
|
||||
async def async_step_import(self, import_input: dict[str, Any]) -> FlowResult:
|
||||
"""Async import step to set up the connection."""
|
||||
self._async_abort_entries_match(
|
||||
{
|
||||
CONF_START: import_input[CONF_START],
|
||||
CONF_DESTINATION: import_input[CONF_DESTINATION],
|
||||
}
|
||||
)
|
||||
|
||||
session = async_get_clientsession(self.hass)
|
||||
opendata = OpendataTransport(
|
||||
import_input[CONF_START], import_input[CONF_DESTINATION], session
|
||||
)
|
||||
try:
|
||||
await opendata.async_get_data()
|
||||
except OpendataTransportConnectionError:
|
||||
return self.async_abort(reason="cannot_connect")
|
||||
except OpendataTransportError:
|
||||
return self.async_abort(reason="bad_config")
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.error(
|
||||
"Unknown error raised by python-opendata-transport for '%s %s', check at http://transport.opendata.ch/examples/stationboard.html if your station names and your parameters are valid",
|
||||
import_input[CONF_START],
|
||||
import_input[CONF_DESTINATION],
|
||||
)
|
||||
return self.async_abort(reason="unknown")
|
||||
|
||||
return self.async_create_entry(
|
||||
title=import_input[CONF_NAME],
|
||||
data=import_input,
|
||||
)
|
||||
9
homeassistant/components/swiss_public_transport/const.py
Normal file
9
homeassistant/components/swiss_public_transport/const.py
Normal file
|
|
@ -0,0 +1,9 @@
|
|||
"""Constants for the swiss_public_transport integration."""
|
||||
|
||||
DOMAIN = "swiss_public_transport"
|
||||
|
||||
|
||||
CONF_DESTINATION = "to"
|
||||
CONF_START = "from"
|
||||
|
||||
DEFAULT_NAME = "Next Destination"
|
||||
|
|
@ -1,7 +1,8 @@
|
|||
{
|
||||
"domain": "swiss_public_transport",
|
||||
"name": "Swiss public transport",
|
||||
"codeowners": ["@fabaff"],
|
||||
"codeowners": ["@fabaff", "@miaucl"],
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/swiss_public_transport",
|
||||
"iot_class": "cloud_polling",
|
||||
"loggers": ["opendata_transport"],
|
||||
|
|
|
|||
|
|
@ -4,21 +4,27 @@ from __future__ import annotations
|
|||
from datetime import timedelta
|
||||
import logging
|
||||
|
||||
from opendata_transport import OpendataTransport
|
||||
from opendata_transport.exceptions import OpendataTransportError
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries, core
|
||||
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
|
||||
from homeassistant.config_entries import SOURCE_IMPORT
|
||||
from homeassistant.const import CONF_NAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
import homeassistant.util.dt as dt_util
|
||||
|
||||
from .const import CONF_DESTINATION, CONF_START, DEFAULT_NAME, DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
SCAN_INTERVAL = timedelta(seconds=90)
|
||||
|
||||
ATTR_DEPARTURE_TIME1 = "next_departure"
|
||||
ATTR_DEPARTURE_TIME2 = "next_on_departure"
|
||||
ATTR_DURATION = "duration"
|
||||
|
|
@ -30,14 +36,6 @@ ATTR_TRAIN_NUMBER = "train_number"
|
|||
ATTR_TRANSFERS = "transfers"
|
||||
ATTR_DELAY = "delay"
|
||||
|
||||
CONF_DESTINATION = "to"
|
||||
CONF_START = "from"
|
||||
|
||||
DEFAULT_NAME = "Next Departure"
|
||||
|
||||
|
||||
SCAN_INTERVAL = timedelta(seconds=90)
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
vol.Required(CONF_DESTINATION): cv.string,
|
||||
|
|
@ -47,31 +45,65 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
|||
)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: core.HomeAssistant,
|
||||
config_entry: config_entries.ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the sensor from a config entry created in the integrations UI."""
|
||||
opendata = hass.data[DOMAIN][config_entry.entry_id]
|
||||
|
||||
start = config_entry.data[CONF_START]
|
||||
destination = config_entry.data[CONF_DESTINATION]
|
||||
name = config_entry.title
|
||||
|
||||
async_add_entities(
|
||||
[SwissPublicTransportSensor(opendata, start, destination, name)],
|
||||
update_before_add=True,
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_platform(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> None:
|
||||
"""Set up the Swiss public transport sensor."""
|
||||
|
||||
name = config.get(CONF_NAME)
|
||||
start = config.get(CONF_START)
|
||||
destination = config.get(CONF_DESTINATION)
|
||||
|
||||
session = async_get_clientsession(hass)
|
||||
opendata = OpendataTransport(start, destination, session)
|
||||
|
||||
try:
|
||||
await opendata.async_get_data()
|
||||
except OpendataTransportError:
|
||||
_LOGGER.error(
|
||||
"Check at http://transport.opendata.ch/examples/stationboard.html "
|
||||
"if your station names are valid"
|
||||
"""Set up the sensor platform."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data=config,
|
||||
)
|
||||
if (
|
||||
result["type"] == FlowResultType.CREATE_ENTRY
|
||||
or result["reason"] == "already_configured"
|
||||
):
|
||||
async_create_issue(
|
||||
hass,
|
||||
HOMEASSISTANT_DOMAIN,
|
||||
f"deprecated_yaml_{DOMAIN}",
|
||||
breaks_in_ha_version="2024.7.0",
|
||||
is_fixable=False,
|
||||
issue_domain=DOMAIN,
|
||||
severity=IssueSeverity.WARNING,
|
||||
translation_key="deprecated_yaml",
|
||||
translation_placeholders={
|
||||
"domain": DOMAIN,
|
||||
"integration_title": "Swiss public transport",
|
||||
},
|
||||
)
|
||||
else:
|
||||
async_create_issue(
|
||||
hass,
|
||||
DOMAIN,
|
||||
f"deprecated_yaml_import_issue_${result['reason']}",
|
||||
breaks_in_ha_version="2024.7.0",
|
||||
is_fixable=False,
|
||||
issue_domain=DOMAIN,
|
||||
severity=IssueSeverity.WARNING,
|
||||
translation_key=f"deprecated_yaml_import_issue_${result['reason']}",
|
||||
)
|
||||
return
|
||||
|
||||
async_add_entities([SwissPublicTransportSensor(opendata, start, destination, name)])
|
||||
|
||||
|
||||
class SwissPublicTransportSensor(SensorEntity):
|
||||
|
|
@ -86,7 +118,7 @@ class SwissPublicTransportSensor(SensorEntity):
|
|||
self._name = name
|
||||
self._from = start
|
||||
self._to = destination
|
||||
self._remaining_time = ""
|
||||
self._remaining_time = None
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
|
|
@ -129,7 +161,7 @@ class SwissPublicTransportSensor(SensorEntity):
|
|||
"""Get the latest data from opendata.ch and update the states."""
|
||||
|
||||
try:
|
||||
if self._remaining_time.total_seconds() < 0:
|
||||
if not self._remaining_time or self._remaining_time.total_seconds() < 0:
|
||||
await self._opendata.async_get_data()
|
||||
except OpendataTransportError:
|
||||
_LOGGER.error("Unable to retrieve data from transport.opendata.ch")
|
||||
|
|
|
|||
39
homeassistant/components/swiss_public_transport/strings.json
Normal file
39
homeassistant/components/swiss_public_transport/strings.json
Normal file
|
|
@ -0,0 +1,39 @@
|
|||
{
|
||||
"config": {
|
||||
"error": {
|
||||
"cannot_connect": "Cannot connect to server",
|
||||
"bad_config": "Request failed due to bad config: Check at [stationboard](http://transport.opendata.ch/examples/stationboard.html) if your station names are valid",
|
||||
"unknown": "An unknown error was raised by python-opendata-transport"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]",
|
||||
"cannot_connect": "Cannot connect to server",
|
||||
"bad_config": "Request failed due to bad config: Check at [stationboard](http://transport.opendata.ch/examples/stationboard.html) if your station names are valid",
|
||||
"unknown": "An unknown error was raised by python-opendata-transport"
|
||||
},
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"from": "Start station",
|
||||
"to": "End station"
|
||||
},
|
||||
"description": "Provide start and end station for your connection\n\nCheck here for valid stations: [stationboard](http://transport.opendata.ch/examples/stationboard.html)",
|
||||
"title": "Swiss Public Transport"
|
||||
}
|
||||
}
|
||||
},
|
||||
"issues": {
|
||||
"deprecated_yaml_import_issue_cannot_connect": {
|
||||
"title": "The swiss public transport YAML configuration import cannot connect to server",
|
||||
"description": "Configuring swiss public transport using YAML is being removed but there was an connection error importing your YAML configuration.\n\nMake sure your home assistant can reach the [opendata server](http://transport.opendata.ch). In case the server is down, try again later."
|
||||
},
|
||||
"deprecated_yaml_import_issue_bad_config": {
|
||||
"title": "The swiss public transport YAML configuration import request failed due to bad config",
|
||||
"description": "Configuring swiss public transport using YAML is being removed but there was bad config imported in your YAML configuration..\n\nCheck here for valid stations: [stationboard](http://transport.opendata.ch/examples/stationboard.html)"
|
||||
},
|
||||
"deprecated_yaml_import_issue_unknown": {
|
||||
"title": "The swiss public transport YAML configuration import failed with unknown error raised by python-opendata-transport",
|
||||
"description": "Configuring swiss public transport using YAML is being removed but there was an unknown error importing your YAML configuration.\n\nCheck your configuration or have a look at the documentation of the integration."
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -475,6 +475,7 @@ FLOWS = {
|
|||
"sun",
|
||||
"sunweg",
|
||||
"surepetcare",
|
||||
"swiss_public_transport",
|
||||
"switchbee",
|
||||
"switchbot",
|
||||
"switchbot_cloud",
|
||||
|
|
|
|||
|
|
@ -5613,7 +5613,7 @@
|
|||
"swiss_public_transport": {
|
||||
"name": "Swiss public transport",
|
||||
"integration_type": "hub",
|
||||
"config_flow": false,
|
||||
"config_flow": true,
|
||||
"iot_class": "cloud_polling"
|
||||
},
|
||||
"swisscom": {
|
||||
|
|
|
|||
|
|
@ -1655,6 +1655,9 @@ python-miio==0.5.12
|
|||
# homeassistant.components.mystrom
|
||||
python-mystrom==2.2.0
|
||||
|
||||
# homeassistant.components.swiss_public_transport
|
||||
python-opendata-transport==0.3.0
|
||||
|
||||
# homeassistant.components.opensky
|
||||
python-opensky==1.0.0
|
||||
|
||||
|
|
|
|||
1
tests/components/swiss_public_transport/__init__.py
Normal file
1
tests/components/swiss_public_transport/__init__.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
"""Tests for the swiss_public_transport integration."""
|
||||
15
tests/components/swiss_public_transport/conftest.py
Normal file
15
tests/components/swiss_public_transport/conftest.py
Normal file
|
|
@ -0,0 +1,15 @@
|
|||
"""Common fixtures for the swiss_public_transport tests."""
|
||||
from collections.abc import Generator
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
|
||||
"""Override async_setup_entry."""
|
||||
with patch(
|
||||
"homeassistant.components.swiss_public_transport.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry:
|
||||
yield mock_setup_entry
|
||||
200
tests/components/swiss_public_transport/test_config_flow.py
Normal file
200
tests/components/swiss_public_transport/test_config_flow.py
Normal file
|
|
@ -0,0 +1,200 @@
|
|||
"""Test the swiss_public_transport config flow."""
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from opendata_transport.exceptions import (
|
||||
OpendataTransportConnectionError,
|
||||
OpendataTransportError,
|
||||
)
|
||||
import pytest
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.swiss_public_transport import config_flow
|
||||
from homeassistant.components.swiss_public_transport.const import (
|
||||
CONF_DESTINATION,
|
||||
CONF_START,
|
||||
)
|
||||
from homeassistant.const import CONF_NAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
pytestmark = pytest.mark.usefixtures("mock_setup_entry")
|
||||
|
||||
MOCK_DATA_STEP = {
|
||||
CONF_START: "test_start",
|
||||
CONF_DESTINATION: "test_destination",
|
||||
}
|
||||
|
||||
|
||||
async def test_flow_user_init_data_success(hass: HomeAssistant) -> None:
|
||||
"""Test success response."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": "user"}
|
||||
)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "user"
|
||||
assert result["handler"] == "swiss_public_transport"
|
||||
assert result["data_schema"] == config_flow.DATA_SCHEMA
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.swiss_public_transport.config_flow.OpendataTransport.async_get_data",
|
||||
autospec=True,
|
||||
return_value=True,
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": "user"}
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input=MOCK_DATA_STEP,
|
||||
)
|
||||
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["result"].title == "test_start test_destination"
|
||||
|
||||
assert result["data"] == MOCK_DATA_STEP
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("raise_error", "text_error"),
|
||||
[
|
||||
(OpendataTransportConnectionError(), "cannot_connect"),
|
||||
(OpendataTransportError(), "bad_config"),
|
||||
(IndexError(), "unknown"),
|
||||
],
|
||||
)
|
||||
async def test_flow_user_init_data_unknown_error_and_recover(
|
||||
hass: HomeAssistant, raise_error, text_error
|
||||
) -> None:
|
||||
"""Test unknown errors."""
|
||||
with patch(
|
||||
"homeassistant.components.swiss_public_transport.config_flow.OpendataTransport.async_get_data",
|
||||
autospec=True,
|
||||
side_effect=raise_error,
|
||||
) as mock_OpendataTransport:
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": "user"}
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input=MOCK_DATA_STEP,
|
||||
)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["errors"]["base"] == text_error
|
||||
|
||||
# Recover
|
||||
mock_OpendataTransport.side_effect = None
|
||||
mock_OpendataTransport.return_value = True
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": "user"}
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input=MOCK_DATA_STEP,
|
||||
)
|
||||
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["result"].title == "test_start test_destination"
|
||||
|
||||
assert result["data"] == MOCK_DATA_STEP
|
||||
|
||||
|
||||
async def test_flow_user_init_data_already_configured(hass: HomeAssistant) -> None:
|
||||
"""Test we abort user data set when entry is already configured."""
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=config_flow.DOMAIN,
|
||||
data=MOCK_DATA_STEP,
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": "user"}
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input=MOCK_DATA_STEP,
|
||||
)
|
||||
|
||||
assert result["type"] == FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
MOCK_DATA_IMPORT = {
|
||||
CONF_START: "test_start",
|
||||
CONF_DESTINATION: "test_destination",
|
||||
CONF_NAME: "test_name",
|
||||
}
|
||||
|
||||
|
||||
async def test_import(
|
||||
hass: HomeAssistant,
|
||||
mock_setup_entry: AsyncMock,
|
||||
) -> None:
|
||||
"""Test import flow."""
|
||||
with patch(
|
||||
"homeassistant.components.swiss_public_transport.config_flow.OpendataTransport.async_get_data",
|
||||
autospec=True,
|
||||
return_value=True,
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN,
|
||||
context={"source": config_entries.SOURCE_IMPORT},
|
||||
data=MOCK_DATA_IMPORT,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert result["data"] == MOCK_DATA_IMPORT
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("raise_error", "text_error"),
|
||||
[
|
||||
(OpendataTransportConnectionError(), "cannot_connect"),
|
||||
(OpendataTransportError(), "bad_config"),
|
||||
(IndexError(), "unknown"),
|
||||
],
|
||||
)
|
||||
async def test_import_cannot_connect_error(
|
||||
hass: HomeAssistant, raise_error, text_error
|
||||
) -> None:
|
||||
"""Test import flow cannot_connect error."""
|
||||
with patch(
|
||||
"homeassistant.components.swiss_public_transport.config_flow.OpendataTransport.async_get_data",
|
||||
autospec=True,
|
||||
side_effect=raise_error,
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN,
|
||||
context={"source": config_entries.SOURCE_IMPORT},
|
||||
data=MOCK_DATA_IMPORT,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == FlowResultType.ABORT
|
||||
assert result["reason"] == text_error
|
||||
|
||||
|
||||
async def test_import_already_configured(hass: HomeAssistant) -> None:
|
||||
"""Test we abort import when entry is already configured."""
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=config_flow.DOMAIN,
|
||||
data=MOCK_DATA_IMPORT,
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN,
|
||||
context={"source": config_entries.SOURCE_IMPORT},
|
||||
data=MOCK_DATA_IMPORT,
|
||||
)
|
||||
|
||||
assert result["type"] == FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
Loading…
Add table
Add a link
Reference in a new issue