Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add trafikverket_ferry integration (#70443)

pull/70802/head
G Johansson 2022-04-27 07:40:53 +02:00 committed by GitHub
parent 5e49457c65
commit 6adcf500b3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 775 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.coveragerc
View File

@ -1278,6 +1278,9 @@ omit =
homeassistant/components/tradfri/light.py
homeassistant/components/tradfri/sensor.py
homeassistant/components/tradfri/switch.py
homeassistant/components/trafikverket_ferry/__init__.py
homeassistant/components/trafikverket_ferry/coordinator.py
homeassistant/components/trafikverket_ferry/sensor.py
homeassistant/components/trafikverket_train/__init__.py
homeassistant/components/trafikverket_train/sensor.py
homeassistant/components/trafikverket_weatherstation/__init__.py

2
CODEOWNERS
View File

@ -1048,6 +1048,8 @@ build.json @home-assistant/supervisor
/tests/components/trace/ @home-assistant/core
/homeassistant/components/tractive/ @Danielhiversen @zhulik @bieniu
/tests/components/tractive/ @Danielhiversen @zhulik @bieniu
/homeassistant/components/trafikverket_ferry/ @gjohansson-ST
/tests/components/trafikverket_ferry/ @gjohansson-ST
/homeassistant/components/trafikverket_train/ @endor-force @gjohansson-ST
/tests/components/trafikverket_train/ @endor-force @gjohansson-ST
/homeassistant/components/trafikverket_weatherstation/ @endor-force @gjohansson-ST

26
homeassistant/components/trafikverket_ferry/__init__.py Normal file
View File

@ -0,0 +1,26 @@
"""The trafikverket_ferry component."""
from __future__ import annotations
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from .const import DOMAIN, PLATFORMS
from .coordinator import TVDataUpdateCoordinator
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Trafikverket Ferry from a config entry."""
coordinator = TVDataUpdateCoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload Trafikverket Ferry config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

162
homeassistant/components/trafikverket_ferry/config_flow.py Normal file
View File

@ -0,0 +1,162 @@
"""Adds config flow for Trafikverket Ferry integration."""
from __future__ import annotations
from typing import Any
from pytrafikverket import TrafikverketFerry
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_API_KEY, CONF_NAME, CONF_WEEKDAY, WEEKDAYS
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import selector
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import CONF_FROM, CONF_TIME, CONF_TO, DOMAIN
from .util import create_unique_id
ERROR_INVALID_AUTH = "Source: Security, message: Invalid authentication"
ERROR_INVALID_ROUTE = "No FerryAnnouncement found"
DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_API_KEY): selector.TextSelector(
selector.TextSelectorConfig()
),
vol.Required(CONF_FROM): selector.TextSelector(selector.TextSelectorConfig()),
vol.Optional(CONF_TO): selector.TextSelector(selector.TextSelectorConfig()),
vol.Optional(CONF_TIME): selector.TimeSelector(selector.TimeSelectorConfig()),
vol.Required(CONF_WEEKDAY, default=WEEKDAYS): selector.SelectSelector(
selector.SelectSelectorConfig(
options=WEEKDAYS,
multiple=True,
mode=selector.SelectSelectorMode.DROPDOWN,
)
),
}
)
DATA_SCHEMA_REAUTH = vol.Schema(
{
vol.Required(CONF_API_KEY): selector.TextSelector(
selector.TextSelectorConfig()
),
}
)
class TVFerryConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Trafikverket Ferry integration."""
VERSION = 1
entry: config_entries.ConfigEntry | None
async def validate_input(
self, api_key: str, ferry_from: str, ferry_to: str
) -> None:
"""Validate input from user input."""
web_session = async_get_clientsession(self.hass)
ferry_api = TrafikverketFerry(web_session, api_key)
await ferry_api.async_get_next_ferry_stop(ferry_from, ferry_to)
async def async_step_reauth(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle re-authentication with Trafikverket."""
self.entry = self.hass.config_entries.async_get_entry(self.context["entry_id"])
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Confirm re-authentication with Trafikverket."""
errors: dict[str, str] = {}
if user_input:
api_key = user_input[CONF_API_KEY]
assert self.entry is not None
try:
await self.validate_input(
api_key, self.entry.data[CONF_FROM], self.entry.data[CONF_TO]
)
except ValueError as err:
if str(err) == ERROR_INVALID_AUTH:
errors["base"] = "invalid_auth"
elif str(err) == ERROR_INVALID_ROUTE:
errors["base"] = "invalid_route"
else:
errors["base"] = "cannot_connect"
else:
self.hass.config_entries.async_update_entry(
self.entry,
data={
**self.entry.data,
CONF_API_KEY: api_key,
},
)
await self.hass.config_entries.async_reload(self.entry.entry_id)
return self.async_abort(reason="reauth_successful")
return self.async_show_form(
step_id="reauth_confirm",
data_schema=DATA_SCHEMA_REAUTH,
errors=errors,
)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the user step."""
errors: dict[str, str] = {}
if user_input is not None:
api_key: str = user_input[CONF_API_KEY]
ferry_from: str = user_input[CONF_FROM]
ferry_to: str = user_input.get(CONF_TO, "")
ferry_time: str = user_input[CONF_TIME]
weekdays: list[str] = user_input[CONF_WEEKDAY]
name = f"{ferry_from}"
if ferry_to:
name = name + f" to {ferry_to}"
if ferry_time != "00:00:00":
name = name + f" at {str(ferry_time)}"
try:
await self.validate_input(api_key, ferry_from, ferry_to)
except ValueError as err:
if str(err) == ERROR_INVALID_AUTH:
errors["base"] = "invalid_auth"
elif str(err) == ERROR_INVALID_ROUTE:
errors["base"] = "invalid_route"
else:
errors["base"] = "cannot_connect"
else:
if not errors:
unique_id = create_unique_id(
ferry_from,
ferry_to,
ferry_time,
weekdays,
)
await self.async_set_unique_id(unique_id)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=name,
data={
CONF_API_KEY: api_key,
CONF_NAME: name,
CONF_FROM: ferry_from,
CONF_TO: ferry_to,
CONF_TIME: ferry_time,
CONF_WEEKDAY: weekdays,
},
)
return self.async_show_form(
step_id="user",
data_schema=DATA_SCHEMA,
errors=errors,
)

11
homeassistant/components/trafikverket_ferry/const.py Normal file
View File

@ -0,0 +1,11 @@
"""Adds constants for Trafikverket Ferry integration."""
from homeassistant.const import Platform
DOMAIN = "trafikverket_ferry"
PLATFORMS = [Platform.SENSOR]
ATTRIBUTION = "Data provided by Trafikverket"
CONF_TRAINS = "trains"
CONF_FROM = "from"
CONF_TO = "to"
CONF_TIME = "time"

93
homeassistant/components/trafikverket_ferry/coordinator.py Normal file
View File

@ -0,0 +1,93 @@
"""DataUpdateCoordinator for the Trafikverket Ferry integration."""
from __future__ import annotations
from datetime import date, datetime, time, timedelta
import logging
from typing import Any
from pytrafikverket import TrafikverketFerry
from pytrafikverket.trafikverket_ferry import FerryStop
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY, CONF_WEEKDAY, WEEKDAYS
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util.dt import UTC, as_utc, parse_time
from .const import CONF_FROM, CONF_TIME, CONF_TO, DOMAIN
_LOGGER = logging.getLogger(__name__)
TIME_BETWEEN_UPDATES = timedelta(minutes=5)
def next_weekday(fromdate: date, weekday: int) -> date:
"""Return the date of the next time a specific weekday happen."""
days_ahead = weekday - fromdate.weekday()
if days_ahead <= 0:
days_ahead += 7
return fromdate + timedelta(days_ahead)
def next_departuredate(departure: list[str]) -> date:
"""Calculate the next departuredate from an array input of short days."""
today_date = date.today()
today_weekday = date.weekday(today_date)
if WEEKDAYS[today_weekday] in departure:
return today_date
for day in departure:
next_departure = WEEKDAYS.index(day)
if next_departure > today_weekday:
return next_weekday(today_date, next_departure)
return next_weekday(today_date, WEEKDAYS.index(departure[0]))
class TVDataUpdateCoordinator(DataUpdateCoordinator):
"""A Trafikverket Data Update Coordinator."""
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Initialize the Trafikverket coordinator."""
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=TIME_BETWEEN_UPDATES,
)
self._ferry_api = TrafikverketFerry(
async_get_clientsession(hass), entry.data[CONF_API_KEY]
)
self._from: str = entry.data[CONF_FROM]
self._to: str = entry.data[CONF_TO]
self._time: time | None = parse_time(entry.data[CONF_TIME])
self._weekdays: list[str] = entry.data[CONF_WEEKDAY]
async def _async_update_data(self) -> dict[str, Any]:
"""Fetch data from Trafikverket."""
departure_day = next_departuredate(self._weekdays)
currenttime = datetime.now()
when = (
datetime.combine(departure_day, self._time)
if self._time
else datetime.now()
)
if currenttime > when:
when = currenttime
try:
routedata: FerryStop = await self._ferry_api.async_get_next_ferry_stop(
self._from, self._to, when
)
except ValueError as error:
raise UpdateFailed(
f"Departure {when} encountered a problem: {error}"
) from error
states = {
"departure_time": routedata.departure_time.replace(tzinfo=UTC),
"departure_from": routedata.from_harbor_name,
"departure_to": routedata.to_harbor_name,
"departure_modified": as_utc(routedata.modified_time.replace(tzinfo=UTC)),
"departure_information": routedata.other_information,
}
return states

10
homeassistant/components/trafikverket_ferry/manifest.json Normal file
View File

@ -0,0 +1,10 @@
{
"domain": "trafikverket_ferry",
"name": "Trafikverket Ferry",
"documentation": "https://www.home-assistant.io/integrations/trafikverket_ferry",
"requirements": ["pytrafikverket==0.1.6.2"],
"codeowners": ["@gjohansson-ST"],
"config_flow": true,
"iot_class": "cloud_polling",
"loggers": ["pytrafikverket"]
}

143
homeassistant/components/trafikverket_ferry/sensor.py Normal file
View File

@ -0,0 +1,143 @@
"""Ferry information for departures, provided by Trafikverket."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import timedelta
import logging
from typing import Any
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceEntryType
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import ATTRIBUTION, DOMAIN
from .coordinator import TVDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
ATTR_FROM = "from_harbour"
ATTR_TO = "to_harbour"
ATTR_MODIFIED_TIME = "modified_time"
ATTR_OTHER_INFO = "other_info"
ICON = "mdi:ferry"
SCAN_INTERVAL = timedelta(minutes=5)
@dataclass
class TrafikverketRequiredKeysMixin:
"""Mixin for required keys."""
value_fn: Callable[[dict[str, Any]], StateType]
info_fn: Callable[[dict[str, Any]], StateType | list]
@dataclass
class TrafikverketSensorEntityDescription(
SensorEntityDescription, TrafikverketRequiredKeysMixin
):
"""Describes Trafikverket sensor entity."""
SENSOR_TYPES: tuple[TrafikverketSensorEntityDescription, ...] = (
TrafikverketSensorEntityDescription(
key="departure_time",
name="Departure Time",
icon="mdi:clock",
device_class=SensorDeviceClass.TIMESTAMP,
value_fn=lambda data: data["departure_time"],
info_fn=lambda data: data["departure_information"],
),
TrafikverketSensorEntityDescription(
key="departure_from",
name="Departure From",
icon="mdi:ferry",
value_fn=lambda data: data["departure_from"],
info_fn=lambda data: data["departure_information"],
),
TrafikverketSensorEntityDescription(
key="departure_to",
name="Departure To",
icon="mdi:ferry",
value_fn=lambda data: data["departure_to"],
info_fn=lambda data: data["departure_information"],
),
TrafikverketSensorEntityDescription(
key="departure_modified",
name="Departure Modified",
icon="mdi:clock",
device_class=SensorDeviceClass.TIMESTAMP,
value_fn=lambda data: data["departure_modified"],
info_fn=lambda data: data["departure_information"],
entity_registry_enabled_default=False,
),
)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the Trafikverket sensor entry."""
coordinator: TVDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
[
FerrySensor(coordinator, entry.data[CONF_NAME], entry.entry_id, description)
for description in SENSOR_TYPES
]
)
class FerrySensor(CoordinatorEntity[TVDataUpdateCoordinator], SensorEntity):
"""Contains data about a ferry departure."""
entity_description: TrafikverketSensorEntityDescription
_attr_attribution = ATTRIBUTION
def __init__(
self,
coordinator: TVDataUpdateCoordinator,
name: str,
entry_id: str,
entity_description: TrafikverketSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self._attr_name = f"{name} {entity_description.name}"
self._attr_unique_id = f"{entry_id}-{entity_description.key}"
self.entity_description = entity_description
self._attr_device_info = DeviceInfo(
entry_type=DeviceEntryType.SERVICE,
identifiers={(DOMAIN, entry_id)},
manufacturer="Trafikverket",
model="v1.2",
name=name,
configuration_url="https://api.trafikinfo.trafikverket.se/",
)
self._update_attr()
def _update_attr(self) -> None:
"""Update _attr."""
self._attr_native_value = self.entity_description.value_fn(
self.coordinator.data
)
self._attr_extra_state_attributes = {
"other_information": self.entity_description.info_fn(self.coordinator.data),
}
@callback
def _handle_coordinator_update(self) -> None:
self._update_attr()
return super()._handle_coordinator_update()

30
homeassistant/components/trafikverket_ferry/strings.json Normal file
View File

@ -0,0 +1,30 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"invalid_route": "Could not find route with provided information",
"incorrect_api_key": "Invalid API key for selected account"
},
"step": {
"user": {
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]",
"to": "To Harbour",
"from": "From Harbour",
"time": "Time",
"weekday": "Weekdays"
}
},
"reauth_confirm": {
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]"
}
}
}
}
}

30
homeassistant/components/trafikverket_ferry/translations/en.json Normal file
View File

@ -0,0 +1,30 @@
{
"config": {
"abort": {
"already_configured": "Account is already configured",
"reauth_successful": "Re-authentication was successful"
},
"error": {
"cannot_connect": "Failed to connect",
"incorrect_api_key": "Invalid API key for selected account",
"invalid_auth": "Invalid authentication",
"invalid_route": "Could not find route with provided information"
},
"step": {
"reauth_confirm": {
"data": {
"api_key": "API Key"
}
},
"user": {
"data": {
"api_key": "API Key",
"from": "From Harbour",
"time": "Time",
"to": "To Harbour",
"weekday": "Weekdays"
}
}
}
}
}

14
homeassistant/components/trafikverket_ferry/util.py Normal file
View File

@ -0,0 +1,14 @@
"""Utils for trafikverket_ferry."""
from __future__ import annotations
from datetime import time
def create_unique_id(
ferry_from: str, ferry_to: str, ferry_time: time | str | None, weekdays: list[str]
) -> str:
"""Create unique id."""
return (
f"{ferry_from.casefold().replace(' ', '')}-{ferry_to.casefold().replace(' ', '')}"
f"-{str(ferry_time)}-{str(weekdays)}"
)

1
homeassistant/generated/config_flows.py
View File

@ -355,6 +355,7 @@ FLOWS = {
"traccar",
"tractive",
"tradfri",
"trafikverket_ferry",
"trafikverket_train",
"trafikverket_weatherstation",
"transmission",

1
requirements_all.txt
View File

@ -1966,6 +1966,7 @@ pytraccar==0.10.0
# homeassistant.components.tradfri
pytradfri[async]==9.0.0
# homeassistant.components.trafikverket_ferry
# homeassistant.components.trafikverket_train
# homeassistant.components.trafikverket_weatherstation
pytrafikverket==0.1.6.2

1
requirements_test_all.txt
View File

@ -1286,6 +1286,7 @@ pytraccar==0.10.0
# homeassistant.components.tradfri
pytradfri[async]==9.0.0
# homeassistant.components.trafikverket_ferry
# homeassistant.components.trafikverket_train
# homeassistant.components.trafikverket_weatherstation
pytrafikverket==0.1.6.2

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

@ -0,0 +1 @@
"""Tests for the Trafikverket Ferry integration."""

247
tests/components/trafikverket_ferry/test_config_flow.py Normal file
View File

@ -0,0 +1,247 @@
"""Test the Trafikverket Ferry config flow."""
from __future__ import annotations
from unittest.mock import patch
import pytest
from homeassistant import config_entries
from homeassistant.components.trafikverket_ferry.const import (
CONF_FROM,
CONF_TIME,
CONF_TO,
DOMAIN,
)
from homeassistant.const import CONF_API_KEY, CONF_NAME, CONF_WEEKDAY, WEEKDAYS
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import (
RESULT_TYPE_ABORT,
RESULT_TYPE_CREATE_ENTRY,
RESULT_TYPE_FORM,
)
from tests.common import MockConfigEntry
async def test_form(hass: HomeAssistant) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {}
with patch(
"homeassistant.components.trafikverket_ferry.config_flow.TrafikverketFerry.async_get_next_ferry_stop",
), patch(
"homeassistant.components.trafikverket_ferry.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_API_KEY: "1234567890",
CONF_FROM: "Ekerö",
CONF_TO: "Slagsta",
CONF_TIME: "10:00",
CONF_WEEKDAY: ["mon", "fri"],
},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == "Ekerö to Slagsta at 10:00"
assert result2["data"] == {
"api_key": "1234567890",
"name": "Ekerö to Slagsta at 10:00",
"from": "Ekerö",
"to": "Slagsta",
"time": "10:00",
"weekday": ["mon", "fri"],
}
assert len(mock_setup_entry.mock_calls) == 1
assert result2["result"].unique_id == "{}-{}-{}-{}".format(
"eker\u00f6", "slagsta", "10:00", "['mon', 'fri']"
)
@pytest.mark.parametrize(
"error_message,base_error",
[
(
"Source: Security, message: Invalid authentication",
"invalid_auth",
),
(
"No FerryAnnouncement found",
"invalid_route",
),
(
"Unknown",
"cannot_connect",
),
],
)
async def test_flow_fails(
hass: HomeAssistant, error_message: str, base_error: str
) -> None:
"""Test config flow errors."""
result4 = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result4["type"] == RESULT_TYPE_FORM
assert result4["step_id"] == config_entries.SOURCE_USER
with patch(
"homeassistant.components.trafikverket_ferry.config_flow.TrafikverketFerry.async_get_next_ferry_stop",
side_effect=ValueError(error_message),
):
result4 = await hass.config_entries.flow.async_configure(
result4["flow_id"],
user_input={
CONF_API_KEY: "1234567890",
CONF_FROM: "Ekerö",
CONF_TO: "Slagsta",
CONF_TIME: "00:00",
},
)
assert result4["errors"] == {"base": base_error}
async def test_reauth_flow(hass: HomeAssistant) -> None:
"""Test a reauthentication flow."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_API_KEY: "1234567890",
CONF_NAME: "Ekerö to Slagsta at 10:00",
CONF_FROM: "Ekerö",
CONF_TO: "Slagsta",
CONF_TIME: "10:00",
CONF_WEEKDAY: WEEKDAYS,
},
unique_id=f"eker\u00f6-slagsta-10:00-{WEEKDAYS}",
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": config_entries.SOURCE_REAUTH,
"unique_id": entry.unique_id,
"entry_id": entry.entry_id,
},
data=entry.data,
)
assert result["step_id"] == "reauth_confirm"
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {}
with patch(
"homeassistant.components.trafikverket_ferry.config_flow.TrafikverketFerry.async_get_next_ferry_stop",
), patch(
"homeassistant.components.trafikverket_ferry.async_setup_entry",
return_value=True,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_API_KEY: "1234567891"},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_ABORT
assert result2["reason"] == "reauth_successful"
assert entry.data == {
"api_key": "1234567891",
"name": "Ekerö to Slagsta at 10:00",
"from": "Ekerö",
"to": "Slagsta",
"time": "10:00",
"weekday": ["mon", "tue", "wed", "thu", "fri", "sat", "sun"],
}
@pytest.mark.parametrize(
"sideeffect,p_error",
[
(
ValueError("Source: Security, message: Invalid authentication"),
"invalid_auth",
),
(
ValueError("No FerryAnnouncement found"),
"invalid_route",
),
(
ValueError("Unknown"),
"cannot_connect",
),
],
)
async def test_reauth_flow_error(
hass: HomeAssistant, sideeffect: Exception, p_error: str
) -> None:
"""Test a reauthentication flow with error."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_API_KEY: "1234567890",
CONF_NAME: "Ekerö to Slagsta at 10:00",
CONF_FROM: "Ekerö",
CONF_TO: "Slagsta",
CONF_TIME: "10:00",
CONF_WEEKDAY: WEEKDAYS,
},
unique_id=f"eker\u00f6-slagsta-10:00-{WEEKDAYS}",
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": config_entries.SOURCE_REAUTH,
"unique_id": entry.unique_id,
"entry_id": entry.entry_id,
},
data=entry.data,
)
with patch(
"homeassistant.components.trafikverket_ferry.config_flow.TrafikverketFerry.async_get_next_ferry_stop",
side_effect=sideeffect,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_API_KEY: "1234567890"},
)
await hass.async_block_till_done()
assert result2["step_id"] == "reauth_confirm"
assert result2["type"] == RESULT_TYPE_FORM
assert result2["errors"] == {"base": p_error}
with patch(
"homeassistant.components.trafikverket_ferry.config_flow.TrafikverketFerry.async_get_next_ferry_stop",
), patch(
"homeassistant.components.trafikverket_ferry.async_setup_entry",
return_value=True,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_API_KEY: "1234567891"},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_ABORT
assert result2["reason"] == "reauth_successful"
assert entry.data == {
"api_key": "1234567891",
"name": "Ekerö to Slagsta at 10:00",
"from": "Ekerö",
"to": "Slagsta",
"time": "10:00",
"weekday": ["mon", "tue", "wed", "thu", "fri", "sat", "sun"],
}