jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-04-27 04:46:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add config flow to NS (#151567)

Browse Source 
Signed-off-by: Heindrich Paul <heindrich.paul@gmail.com>
Co-authored-by: Norbert Rittel <norbert@rittel.de>
Co-authored-by: Franck Nijhof <frenck@frenck.nl>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
Heindrich Paul
2025-09-15 15:13:43 +02:00
committed by GitHub
parent 410c3df6dd
commit b503f792b5
16 changed files with 1134 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
CODEOWNERS generated
View File

@@ -1017,7 +1017,8 @@ build.json @home-assistant/supervisor
/tests/components/nanoleaf/ @milanmeu @joostlek
/homeassistant/components/nasweb/ @nasWebio
/tests/components/nasweb/ @nasWebio
/homeassistant/components/nederlandse_spoorwegen/ @YarmoM
/homeassistant/components/nederlandse_spoorwegen/ @YarmoM @heindrichpaul
/tests/components/nederlandse_spoorwegen/ @YarmoM @heindrichpaul
/homeassistant/components/ness_alarm/ @nickw444
/tests/components/ness_alarm/ @nickw444
/homeassistant/components/nest/ @allenporter

57
homeassistant/components/nederlandse_spoorwegen/__init__.py
View File

@@ -1 +1,56 @@
"""The nederlandse_spoorwegen component."""
"""The Nederlandse Spoorwegen integration."""
from __future__ import annotations
import logging
from ns_api import NSAPI, RequestParametersError
import requests
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
_LOGGER = logging.getLogger(__name__)
type NSConfigEntry = ConfigEntry[NSAPI]
PLATFORMS = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: NSConfigEntry) -> bool:
"""Set up Nederlandse Spoorwegen from a config entry."""
api_key = entry.data[CONF_API_KEY]
client = NSAPI(api_key)
try:
await hass.async_add_executor_job(client.get_stations)
except (
requests.exceptions.ConnectionError,
requests.exceptions.HTTPError,
) as error:
_LOGGER.error("Could not connect to the internet: %s", error)
raise ConfigEntryNotReady from error
except RequestParametersError as error:
_LOGGER.error("Could not fetch stations, please check configuration: %s", error)
raise ConfigEntryNotReady from error
entry.runtime_data = client
entry.async_on_unload(entry.add_update_listener(async_reload_entry))
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_reload_entry(hass: HomeAssistant, entry: NSConfigEntry) -> None:
"""Reload NS integration when options are updated."""
await hass.config_entries.async_reload(entry.entry_id)
async def async_unload_entry(hass: HomeAssistant, entry: NSConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

176
homeassistant/components/nederlandse_spoorwegen/config_flow.py Normal file
View File

@@ -0,0 +1,176 @@
"""Config flow for Nederlandse Spoorwegen integration."""
from __future__ import annotations
import logging
from typing import Any
from ns_api import NSAPI, Station
from requests.exceptions import (
ConnectionError as RequestsConnectionError,
HTTPError,
Timeout,
)
import voluptuous as vol
from homeassistant.config_entries import (
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
ConfigSubentryData,
ConfigSubentryFlow,
SubentryFlowResult,
)
from homeassistant.const import CONF_API_KEY
from homeassistant.core import callback
from homeassistant.helpers.selector import (
SelectOptionDict,
SelectSelector,
SelectSelectorConfig,
TimeSelector,
)
from .const import (
CONF_FROM,
CONF_NAME,
CONF_ROUTES,
CONF_TIME,
CONF_TO,
CONF_VIA,
DOMAIN,
)
_LOGGER = logging.getLogger(__name__)
class NSConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Nederlandse Spoorwegen."""
VERSION = 1
MINOR_VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step of the config flow (API key)."""
errors: dict[str, str] = {}
if user_input is not None:
self._async_abort_entries_match(user_input)
client = NSAPI(user_input[CONF_API_KEY])
try:
await self.hass.async_add_executor_job(client.get_stations)
except HTTPError:
errors["base"] = "invalid_auth"
except (RequestsConnectionError, Timeout):
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected exception validating API key")
errors["base"] = "unknown"
if not errors:
return self.async_create_entry(
title="Nederlandse Spoorwegen",
data={CONF_API_KEY: user_input[CONF_API_KEY]},
)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema({vol.Required(CONF_API_KEY): str}),
errors=errors,
)
async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
"""Handle import from YAML configuration."""
self._async_abort_entries_match({CONF_API_KEY: import_data[CONF_API_KEY]})
client = NSAPI(import_data[CONF_API_KEY])
try:
stations = await self.hass.async_add_executor_job(client.get_stations)
except HTTPError:
return self.async_abort(reason="invalid_auth")
except (RequestsConnectionError, Timeout):
return self.async_abort(reason="cannot_connect")
except Exception:
_LOGGER.exception("Unexpected exception validating API key")
return self.async_abort(reason="unknown")
station_codes = {station.code for station in stations}
subentries: list[ConfigSubentryData] = []
for route in import_data.get(CONF_ROUTES, []):
# Convert station codes to uppercase for consistency with UI routes
for key in (CONF_FROM, CONF_TO, CONF_VIA):
if key in route:
route[key] = route[key].upper()
if route[key] not in station_codes:
return self.async_abort(reason="invalid_station")
subentries.append(
ConfigSubentryData(
title=route[CONF_NAME],
subentry_type="route",
data=route,
unique_id=None,
)
)
return self.async_create_entry(
title="Nederlandse Spoorwegen",
data={CONF_API_KEY: import_data[CONF_API_KEY]},
subentries=subentries,
)
@classmethod
@callback
def async_get_supported_subentry_types(
cls, config_entry: ConfigEntry
) -> dict[str, type[ConfigSubentryFlow]]:
"""Return subentries supported by this integration."""
return {"route": RouteSubentryFlowHandler}
class RouteSubentryFlowHandler(ConfigSubentryFlow):
"""Handle subentry flow for adding and modifying routes."""
def __init__(self) -> None:
"""Initialize route subentry flow."""
self.stations: dict[str, Station] = {}
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Add a new route subentry."""
if user_input is not None:
return self.async_create_entry(title=user_input[CONF_NAME], data=user_input)
client = NSAPI(self._get_entry().data[CONF_API_KEY])
if not self.stations:
try:
self.stations = {
station.code: station
for station in await self.hass.async_add_executor_job(
client.get_stations
)
}
except (RequestsConnectionError, Timeout, HTTPError, ValueError):
return self.async_abort(reason="cannot_connect")
options = [
SelectOptionDict(label=station.names["long"], value=code)
for code, station in self.stations.items()
]
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_NAME): str,
vol.Required(CONF_FROM): SelectSelector(
SelectSelectorConfig(options=options, sort=True),
),
vol.Required(CONF_TO): SelectSelector(
SelectSelectorConfig(options=options, sort=True),
),
vol.Optional(CONF_VIA): SelectSelector(
SelectSelectorConfig(options=options, sort=True),
),
vol.Optional(CONF_TIME): TimeSelector(),
}
),
)

17
homeassistant/components/nederlandse_spoorwegen/const.py Normal file
View File

@@ -0,0 +1,17 @@
"""Constants for the Nederlandse Spoorwegen integration."""
DOMAIN = "nederlandse_spoorwegen"
CONF_ROUTES = "routes"
CONF_FROM = "from"
CONF_TO = "to"
CONF_VIA = "via"
CONF_TIME = "time"
CONF_NAME = "name"
# Attribute and schema keys
ATTR_ROUTE = "route"
ATTR_TRIPS = "trips"
ATTR_FIRST_TRIP = "first_trip"
ATTR_NEXT_TRIP = "next_trip"
ATTR_ROUTES = "routes"

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

@@ -1,8 +1,10 @@
{
"domain": "nederlandse_spoorwegen",
"name": "Nederlandse Spoorwegen (NS)",
"codeowners": ["@YarmoM"],
"codeowners": ["@YarmoM", "@heindrichpaul"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/nederlandse_spoorwegen",
"integration_type": "service",
"iot_class": "cloud_polling",
"quality_scale": "legacy",
"requirements": ["nsapi==3.1.2"]

127
homeassistant/components/nederlandse_spoorwegen/sensor.py
View File

@@ -5,8 +5,6 @@ from __future__ import annotations
from datetime import datetime, timedelta
import logging
import ns_api
from ns_api import RequestParametersError
import requests
import voluptuous as vol
@@ -14,13 +12,21 @@ from homeassistant.components.sensor import (
PLATFORM_SCHEMA as SENSOR_PLATFORM_SCHEMA,
SensorEntity,
)
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import CONF_API_KEY, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import config_validation as cv, issue_registry as ir
from homeassistant.helpers.entity_platform import (
AddConfigEntryEntitiesCallback,
AddEntitiesCallback,
)
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import Throttle, dt as dt_util
from homeassistant.util.dt import parse_time
from . import NSConfigEntry
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
@@ -50,57 +56,84 @@ PLATFORM_SCHEMA = SENSOR_PLATFORM_SCHEMA.extend(
)
def setup_platform(
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the departure sensor."""
nsapi = ns_api.NSAPI(config[CONF_API_KEY])
try:
stations = nsapi.get_stations()
except (
requests.exceptions.ConnectionError,
requests.exceptions.HTTPError,
) as error:
_LOGGER.error("Could not connect to the internet: %s", error)
raise PlatformNotReady from error
except RequestParametersError as error:
_LOGGER.error("Could not fetch stations, please check configuration: %s", error)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=config,
)
if (
result.get("type") is FlowResultType.ABORT
and result.get("reason") != "already_configured"
):
ir.async_create_issue(
hass,
DOMAIN,
f"deprecated_yaml_import_issue_{result.get('reason')}",
breaks_in_ha_version="2026.4.0",
is_fixable=False,
issue_domain=DOMAIN,
severity=ir.IssueSeverity.WARNING,
translation_key=f"deprecated_yaml_import_issue_{result.get('reason')}",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "Nederlandse Spoorwegen",
},
)
return
sensors = []
for departure in config.get(CONF_ROUTES, {}):
if not valid_stations(
stations,
[departure.get(CONF_FROM), departure.get(CONF_VIA), departure.get(CONF_TO)],
):
ir.async_create_issue(
hass,
HOMEASSISTANT_DOMAIN,
"deprecated_yaml",
breaks_in_ha_version="2026.4.0",
is_fixable=False,
issue_domain=DOMAIN,
severity=ir.IssueSeverity.WARNING,
translation_key="deprecated_yaml",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "Nederlandse Spoorwegen",
},
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: NSConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the departure sensor from a config entry."""
client = config_entry.runtime_data
for subentry in config_entry.subentries.values():
if subentry.subentry_type != "route":
continue
sensors.append(
NSDepartureSensor(
nsapi,
departure.get(CONF_NAME),
departure.get(CONF_FROM),
departure.get(CONF_TO),
departure.get(CONF_VIA),
departure.get(CONF_TIME),
)
async_add_entities(
[
NSDepartureSensor(
client,
subentry.data[CONF_NAME],
subentry.data[CONF_FROM],
subentry.data[CONF_TO],
subentry.data.get(CONF_VIA),
parse_time(subentry.data[CONF_TIME])
if CONF_TIME in subentry.data
else None,
)
],
config_subentry_id=subentry.subentry_id,
update_before_add=True,
)
add_entities(sensors, True)
def valid_stations(stations, given_stations):
"""Verify the existence of the given station codes."""
for station in given_stations:
if station is None:
continue
if not any(s.code == station.upper() for s in stations):
_LOGGER.warning("Station '%s' is not a valid station", station)
return False
return True
class NSDepartureSensor(SensorEntity):

74
homeassistant/components/nederlandse_spoorwegen/strings.json Normal file
View File

@@ -0,0 +1,74 @@
{
"config": {
"step": {
"user": {
"description": "Set up your Nederlandse Spoorwegen integration.",
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]"
},
"data_description": {
"api_key": "Your NS API key."
}
}
},
"error": {
"cannot_connect": "Could not connect to NS API. Check your API key.",
"invalid_auth": "[%key:common::config_flow::error::invalid_api_key%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]"
}
},
"config_subentries": {
"route": {
"step": {
"user": {
"description": "Select your departure and destination stations from the dropdown lists.",
"data": {
"name": "Route name",
"from": "Departure station",
"to": "Destination station",
"via": "Via station",
"time": "Departure time"
},
"data_description": {
"name": "A name for this route",
"from": "The station to depart from",
"to": "The station to arrive at",
"via": "An optional intermediate station",
"time": "Optional planned departure time"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"abort": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"initiate_flow": {
"user": "Add route"
},
"entry_type": "Route"
}
},
"issues": {
"deprecated_yaml_import_issue_invalid_auth": {
"title": "Nederlandse Spoorwegen YAML configuration deprecated",
"description": "Configuring Nederlandse Spoorwegen using YAML sensor platform is deprecated.\n\nWhile importing your configuration, an invalid API key was found. Please update your YAML configuration, or remove the existing YAML configuration and set the integration up via the UI."
},
"deprecated_yaml_import_issue_cannot_connect": {
"title": "[%key:component::nederlandse_spoorwegen::issues::deprecated_yaml_import_issue_invalid_auth::title%]",
"description": "Configuring Nederlandse Spoorwegen using YAML sensor platform is deprecated.\n\nWhile importing your configuration, Home Assistant could not connect to the NS API. Please check your internet connection and the status of the NS API, then restart Home Assistant to try again, or remove the existing YAML configuration and set the integration up via the UI."
},
"deprecated_yaml_import_issue_unknown": {
"title": "[%key:component::nederlandse_spoorwegen::issues::deprecated_yaml_import_issue_invalid_auth::title%]",
"description": "Configuring Nederlandse Spoorwegen using YAML sensor platform is deprecated.\n\nWhile importing your configuration, an unknown error occurred. Please restart Home Assistant to try again, or remove the existing YAML configuration and set the integration up via the UI."
},
"deprecated_yaml_import_issue_invalid_station": {
"title": "[%key:component::nederlandse_spoorwegen::issues::deprecated_yaml_import_issue_invalid_auth::title%]",
"description": "Configuring Nederlandse Spoorwegen using YAML sensor platform is deprecated.\n\nWhile importing your configuration an invalid station was found. Please update your YAML configuration, or remove the existing YAML configuration and set the integration up via the UI."
}
}
}

1
homeassistant/generated/config_flows.py generated
View File

@@ -418,6 +418,7 @@ FLOWS = {
"nanoleaf",
"nasweb",
"neato",
"nederlandse_spoorwegen",
"nest",
"netatmo",
"netgear",

4
homeassistant/generated/integrations.json
View File

@@ -4280,8 +4280,8 @@
},
"nederlandse_spoorwegen": {
"name": "Nederlandse Spoorwegen (NS)",
"integration_type": "hub",
"config_flow": false,
"integration_type": "service",
"config_flow": true,
"iot_class": "cloud_polling"
},
"neff": {

3
requirements_test_all.txt generated
View File

@@ -1324,6 +1324,9 @@ notifications-android-tv==0.1.5
# homeassistant.components.notify_events
notify-events==1.0.4
# homeassistant.components.nederlandse_spoorwegen
nsapi==3.1.2
# homeassistant.components.nsw_fuel_station
nsw-fuel-api-client==1.1.0

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

@@ -0,0 +1 @@
"""Tests for the Nederlandse Spoorwegen integration."""

68
tests/components/nederlandse_spoorwegen/conftest.py Normal file
View File

@@ -0,0 +1,68 @@
"""Fixtures for Nederlandse Spoorwegen tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
from ns_api import Station
import pytest
from homeassistant.components.nederlandse_spoorwegen.const import (
CONF_FROM,
CONF_TO,
CONF_VIA,
DOMAIN,
)
from homeassistant.config_entries import ConfigSubentryData
from homeassistant.const import CONF_API_KEY, CONF_NAME
from .const import API_KEY
from tests.common import MockConfigEntry, load_json_object_fixture
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.nederlandse_spoorwegen.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
def mock_nsapi() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.nederlandse_spoorwegen.config_flow.NSAPI",
autospec=True,
) as mock_nsapi:
client = mock_nsapi.return_value
stations = load_json_object_fixture("stations.json", DOMAIN)
client.get_stations.return_value = [
Station(station) for station in stations["payload"]
]
yield client
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Mock config entry."""
return MockConfigEntry(
title="Nederlandse Spoorwegen",
data={CONF_API_KEY: API_KEY},
domain=DOMAIN,
subentries_data=[
ConfigSubentryData(
data={
CONF_NAME: "To work",
CONF_FROM: "Ams",
CONF_TO: "Rot",
CONF_VIA: "Ht",
},
subentry_type="route",
title="Test Route",
unique_id=None,
),
],
)

3
tests/components/nederlandse_spoorwegen/const.py Normal file
View File

@@ -0,0 +1,3 @@
"""Constants for the Nederlandse Spoorwegen integration tests."""
API_KEY = "abc1234567"

262
tests/components/nederlandse_spoorwegen/fixtures/stations.json Normal file
View File

@@ -0,0 +1,262 @@
{
"payload": [
{
"EVACode": "8400058",
"UICCode": "8400058",
"UICCdCode": "118400058",
"cdCode": 58,
"code": "ASD",
"ingangsDatum": "2025-06-03",
"heeftFaciliteiten": true,
"heeftReisassistentie": true,
"heeftVertrektijden": true,
"land": "NL",
"lat": 52.3788871765137,
"lng": 4.90027761459351,
"radius": 525,
"naderenRadius": 1200,
"namen": {
"lang": "Amsterdam Centraal",
"middel": "Amsterdam C.",
"kort": "Amsterdm C"
},
"synoniemen": ["Amsterdam CS", "Amsterdam"],
"nearbyMeLocationId": {
"value": "ASD",
"type": "stationV2"
},
"sporen": [
{
"spoorNummer": "1"
},
{
"spoorNummer": "2"
},
{
"spoorNummer": "2a"
},
{
"spoorNummer": "2b"
},
{
"spoorNummer": "4"
},
{
"spoorNummer": "4a"
},
{
"spoorNummer": "4b"
},
{
"spoorNummer": "5"
},
{
"spoorNummer": "5a"
},
{
"spoorNummer": "5b"
},
{
"spoorNummer": "7"
},
{
"spoorNummer": "7a"
},
{
"spoorNummer": "7b"
},
{
"spoorNummer": "8"
},
{
"spoorNummer": "8a"
},
{
"spoorNummer": "8b"
},
{
"spoorNummer": "10"
},
{
"spoorNummer": "10a"
},
{
"spoorNummer": "10b"
},
{
"spoorNummer": "11"
},
{
"spoorNummer": "11a"
},
{
"spoorNummer": "11b"
},
{
"spoorNummer": "13"
},
{
"spoorNummer": "13a"
},
{
"spoorNummer": "13b"
},
{
"spoorNummer": "14"
},
{
"spoorNummer": "14a"
},
{
"spoorNummer": "14b"
},
{
"spoorNummer": "15"
},
{
"spoorNummer": "15a"
},
{
"spoorNummer": "15b"
}
],
"stationType": "MEGA_STATION"
},
{
"EVACode": "8400319",
"UICCode": "8400319",
"UICCdCode": "118400319",
"cdCode": 319,
"code": "HT",
"ingangsDatum": "2025-06-03",
"heeftFaciliteiten": true,
"heeftReisassistentie": true,
"heeftVertrektijden": true,
"land": "NL",
"lat": 51.69048,
"lng": 5.29362,
"radius": 525,
"naderenRadius": 1200,
"namen": {
"lang": "'s-Hertogenbosch",
"middel": "'s-Hertogenbosch",
"kort": "Den Bosch"
},
"synoniemen": ["Den Bosch", "Hertogenbosch ('s)"],
"nearbyMeLocationId": {
"value": "HT",
"type": "stationV2"
},
"sporen": [
{
"spoorNummer": "1"
},
{
"spoorNummer": "3"
},
{
"spoorNummer": "3a"
},
{
"spoorNummer": "3b"
},
{
"spoorNummer": "4"
},
{
"spoorNummer": "4a"
},
{
"spoorNummer": "4b"
},
{
"spoorNummer": "6"
},
{
"spoorNummer": "6a"
},
{
"spoorNummer": "6b"
},
{
"spoorNummer": "7"
},
{
"spoorNummer": "7a"
},
{
"spoorNummer": "7b"
}
],
"stationType": "KNOOPPUNT_INTERCITY_STATION"
},
{
"EVACode": "8400530",
"UICCode": "8400530",
"UICCdCode": "118400530",
"cdCode": 530,
"code": "RTD",
"ingangsDatum": "2017-02-01",
"heeftFaciliteiten": true,
"heeftReisassistentie": true,
"heeftVertrektijden": true,
"land": "NL",
"lat": 51.9249992370605,
"lng": 4.46888875961304,
"radius": 525,
"naderenRadius": 1000,
"namen": {
"lang": "Rotterdam Centraal",
"middel": "Rotterdam C.",
"kort": "Rotterdm C"
},
"synoniemen": ["Rotterdam CS", "Rotterdam"],
"nearbyMeLocationId": {
"value": "RTD",
"type": "stationV2"
},
"sporen": [
{
"spoorNummer": "2"
},
{
"spoorNummer": "3"
},
{
"spoorNummer": "4"
},
{
"spoorNummer": "6"
},
{
"spoorNummer": "7"
},
{
"spoorNummer": "8"
},
{
"spoorNummer": "9"
},
{
"spoorNummer": "11"
},
{
"spoorNummer": "12"
},
{
"spoorNummer": "13"
},
{
"spoorNummer": "14"
},
{
"spoorNummer": "15"
},
{
"spoorNummer": "16"
}
],
"stationType": "MEGA_STATION"
}
]
}

333
tests/components/nederlandse_spoorwegen/test_config_flow.py Normal file
View File

@@ -0,0 +1,333 @@
"""Test config flow for Nederlandse Spoorwegen integration."""
from datetime import time
from typing import Any
from unittest.mock import AsyncMock
import pytest
from requests import ConnectionError as RequestsConnectionError, HTTPError, Timeout
from homeassistant.components.nederlandse_spoorwegen.const import (
CONF_FROM,
CONF_ROUTES,
CONF_TIME,
CONF_TO,
CONF_VIA,
DOMAIN,
)
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
from homeassistant.const import CONF_API_KEY, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .const import API_KEY
from tests.common import MockConfigEntry
async def test_full_flow(
hass: HomeAssistant, mock_nsapi: AsyncMock, mock_setup_entry: AsyncMock
) -> None:
"""Test successful user config flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert not result["errors"]
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_API_KEY: API_KEY}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Nederlandse Spoorwegen"
assert result["data"] == {CONF_API_KEY: API_KEY}
assert len(mock_setup_entry.mock_calls) == 1
async def test_creating_route(
hass: HomeAssistant,
mock_nsapi: AsyncMock,
mock_setup_entry: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test creating a route after setting up the main config entry."""
mock_config_entry.add_to_hass(hass)
assert len(mock_config_entry.subentries) == 1
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "route"), context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert not result["errors"]
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
user_input={
CONF_FROM: "ASD",
CONF_TO: "RTD",
CONF_VIA: "HT",
CONF_NAME: "Home to Work",
CONF_TIME: "08:30",
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Home to Work"
assert result["data"] == {
CONF_FROM: "ASD",
CONF_TO: "RTD",
CONF_VIA: "HT",
CONF_NAME: "Home to Work",
CONF_TIME: "08:30",
}
assert len(mock_config_entry.subentries) == 2
@pytest.mark.parametrize(
("exception", "expected_error"),
[
(HTTPError("Invalid API key"), "invalid_auth"),
(Timeout("Cannot connect"), "cannot_connect"),
(RequestsConnectionError("Cannot connect"), "cannot_connect"),
(Exception("Unexpected error"), "unknown"),
],
)
async def test_flow_exceptions(
hass: HomeAssistant,
mock_nsapi: AsyncMock,
mock_setup_entry: AsyncMock,
exception: Exception,
expected_error: str,
) -> None:
"""Test config flow handling different exceptions."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert not result["errors"]
mock_nsapi.get_stations.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_API_KEY: API_KEY}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": expected_error}
mock_nsapi.get_stations.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_API_KEY: API_KEY}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Nederlandse Spoorwegen"
assert result["data"] == {CONF_API_KEY: API_KEY}
assert len(mock_setup_entry.mock_calls) == 1
async def test_fetching_stations_failed(
hass: HomeAssistant,
mock_nsapi: AsyncMock,
mock_setup_entry: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test creating a route after setting up the main config entry."""
mock_config_entry.add_to_hass(hass)
assert len(mock_config_entry.subentries) == 1
mock_nsapi.get_stations.side_effect = RequestsConnectionError("Unexpected error")
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "route"), context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
async def test_already_configured(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test config flow aborts if already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert not result["errors"]
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_API_KEY: API_KEY}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_config_flow_import_success(
hass: HomeAssistant, mock_nsapi: AsyncMock, mock_setup_entry: AsyncMock
) -> None:
"""Test successful import flow from YAML configuration."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={CONF_API_KEY: API_KEY},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Nederlandse Spoorwegen"
assert result["data"] == {CONF_API_KEY: API_KEY}
assert not result["result"].subentries
@pytest.mark.parametrize(
("routes_data", "expected_routes_data"),
[
(
# Test with uppercase station codes (UI behavior)
[
{
CONF_NAME: "Home to Work",
CONF_FROM: "ASD",
CONF_TO: "RTD",
CONF_VIA: "HT",
CONF_TIME: time(hour=8, minute=30),
}
],
[
{
CONF_NAME: "Home to Work",
CONF_FROM: "ASD",
CONF_TO: "RTD",
CONF_VIA: "HT",
CONF_TIME: time(hour=8, minute=30),
}
],
),
(
# Test with lowercase station codes (converted to uppercase)
[
{
CONF_NAME: "Rotterdam-Amsterdam",
CONF_FROM: "rtd", # lowercase input
CONF_TO: "asd", # lowercase input
},
{
CONF_NAME: "Amsterdam-Haarlem",
CONF_FROM: "asd", # lowercase input
CONF_TO: "ht", # lowercase input
CONF_VIA: "rtd", # lowercase input
},
],
[
{
CONF_NAME: "Rotterdam-Amsterdam",
CONF_FROM: "RTD", # converted to uppercase
CONF_TO: "ASD", # converted to uppercase
},
{
CONF_NAME: "Amsterdam-Haarlem",
CONF_FROM: "ASD", # converted to uppercase
CONF_TO: "HT", # converted to uppercase
CONF_VIA: "RTD", # converted to uppercase
},
],
),
],
)
async def test_config_flow_import_with_routes(
hass: HomeAssistant,
mock_nsapi: AsyncMock,
mock_setup_entry: AsyncMock,
routes_data: list[dict[str, Any]],
expected_routes_data: list[dict[str, Any]],
) -> None:
"""Test import flow with routes from YAML configuration."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_API_KEY: API_KEY,
CONF_ROUTES: routes_data,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Nederlandse Spoorwegen"
assert result["data"] == {CONF_API_KEY: API_KEY}
assert len(result["result"].subentries) == len(expected_routes_data)
subentries = list(result["result"].subentries.values())
for expected_route in expected_routes_data:
route_entry = next(
entry for entry in subentries if entry.title == expected_route[CONF_NAME]
)
assert route_entry.data == expected_route
assert route_entry.subentry_type == "route"
async def test_config_flow_import_with_unknown_station(
hass: HomeAssistant, mock_nsapi: AsyncMock, mock_setup_entry: AsyncMock
) -> None:
"""Test import flow aborts with unknown station in routes."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_API_KEY: API_KEY,
CONF_ROUTES: [
{
CONF_NAME: "Home to Work",
CONF_FROM: "HRM",
CONF_TO: "RTD",
CONF_VIA: "HT",
CONF_TIME: time(hour=8, minute=30),
}
],
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "invalid_station"
async def test_config_flow_import_already_configured(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test import flow when integration is already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={CONF_API_KEY: API_KEY},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
("exception", "expected_error"),
[
(HTTPError("Invalid API key"), "invalid_auth"),
(Timeout("Cannot connect"), "cannot_connect"),
(RequestsConnectionError("Cannot connect"), "cannot_connect"),
(Exception("Unexpected error"), "unknown"),
],
)
async def test_import_flow_exceptions(
hass: HomeAssistant,
mock_nsapi: AsyncMock,
exception: Exception,
expected_error: str,
) -> None:
"""Test config flow handling different exceptions."""
mock_nsapi.get_stations.side_effect = exception
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data={CONF_API_KEY: API_KEY}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == expected_error

53
tests/components/nederlandse_spoorwegen/test_sensor.py Normal file
View File

@@ -0,0 +1,53 @@
"""Test the Nederlandse Spoorwegen sensor."""
from unittest.mock import AsyncMock
from homeassistant.components.nederlandse_spoorwegen.const import (
CONF_FROM,
CONF_ROUTES,
CONF_TO,
CONF_VIA,
DOMAIN,
)
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.const import CONF_API_KEY, CONF_NAME, CONF_PLATFORM
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
import homeassistant.helpers.issue_registry as ir
from homeassistant.setup import async_setup_component
from .const import API_KEY
async def test_config_import(
hass: HomeAssistant,
mock_nsapi,
mock_setup_entry: AsyncMock,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test sensor initialization."""
await async_setup_component(
hass,
SENSOR_DOMAIN,
{
SENSOR_DOMAIN: [
{
CONF_PLATFORM: DOMAIN,
CONF_API_KEY: API_KEY,
CONF_ROUTES: [
{
CONF_NAME: "Spoorwegen Nederlande Station",
CONF_FROM: "ASD",
CONF_TO: "RTD",
CONF_VIA: "HT",
}
],
}
]
},
)
await hass.async_block_till_done()
assert len(issue_registry.issues) == 1
assert (HOMEASSISTANT_DOMAIN, "deprecated_yaml") in issue_registry.issues
assert len(hass.config_entries.async_entries(DOMAIN)) == 1