arian/home-assistant-core
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

data grand lyon: list stops and lines in config flow (#173117)

Browse Source
This commit is contained in:
Crocmagnon
2026-06-08 17:25:01 +02:00
committed by GitHub
parent 05088bf991
commit d10ede2264
4 changed files with 299 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/data_grand_lyon/config_flow.py
+116 -24
View File
@@ -5,7 +5,7 @@ import logging
from typing import Any
from aiohttp import ClientError, ClientResponseError
from data_grand_lyon_ha import DataGrandLyonClient
from data_grand_lyon_ha import DataGrandLyonClient, TclStop, find_tcl_stop_by_id
import voluptuous as vol
from homeassistant.config_entries import (
@@ -18,6 +18,12 @@ from homeassistant.config_entries import (
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import callback
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.selector import (
SelectOptionDict,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
)
from .const import (
CONF_LINE,
@@ -43,13 +49,6 @@ STEP_RECONFIGURE_SCHEMA = vol.Schema(
}
)
STEP_STOP_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_LINE): str,
vol.Required(CONF_STOP_ID): vol.Coerce(int),
}
)
STEP_VELOV_STATION_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_STATION_ID): vol.Coerce(int),
@@ -179,33 +178,126 @@ class DataGrandLyonConfigFlow(ConfigFlow, domain=DOMAIN):
class StopSubentryFlowHandler(ConfigSubentryFlow):
"""Handle a subentry flow for adding a Data Grand Lyon stop."""
def __init__(self) -> None:
"""Initialize the flow."""
self._stops: list[TclStop] = []
self._selected_stop: TclStop | None = None
self._selected_stop_id: int | None = None
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Handle the user step to add a new stop."""
entry = self._get_entry()
"""Pick a stop from the list fetched from the API, or enter one manually."""
if not self._stops:
if error := await self._async_load_stops():
return self.async_abort(reason=error)
errors: dict[str, str] = {}
if user_input is not None:
line = user_input[CONF_LINE]
stop_id = user_input[CONF_STOP_ID]
unique_id = f"{line}_{stop_id}"
try:
stop_id = int(user_input[CONF_STOP_ID])
except ValueError:
errors[CONF_STOP_ID] = "invalid_stop_id"
else:
self._selected_stop_id = stop_id
self._selected_stop = find_tcl_stop_by_id(self._stops, stop_id)
return await self.async_step_pick_line()
for subentry in entry.subentries.values():
if subentry.unique_id == unique_id:
return self.async_abort(reason="already_configured")
name = f"{line} - Stop {stop_id}"
return self.async_create_entry(
title=name,
data={CONF_LINE: line, CONF_STOP_ID: stop_id},
unique_id=unique_id,
options = [
SelectOptionDict(value=str(stop.id), label=_stop_label(stop))
for stop in sorted(
self._stops, key=lambda s: (s.nom, s.commune or "", s.id or 0)
)
]
schema = vol.Schema(
{
vol.Required(CONF_STOP_ID): SelectSelector(
SelectSelectorConfig(
options=options,
mode=SelectSelectorMode.DROPDOWN,
sort=False,
custom_value=True,
)
)
}
)
return self.async_show_form(
step_id="user",
data_schema=STEP_STOP_DATA_SCHEMA,
data_schema=schema,
errors=errors,
)
async def async_step_pick_line(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Pick a line from the selected stop's desserte, or enter one manually."""
assert self._selected_stop_id is not None
if user_input is not None:
return self._create_stop(
line=user_input[CONF_LINE], stop_id=self._selected_stop_id
)
options = self._selected_stop.desserte if self._selected_stop else []
schema = vol.Schema(
{
vol.Required(CONF_LINE): SelectSelector(
SelectSelectorConfig(
options=options,
mode=SelectSelectorMode.DROPDOWN,
custom_value=True,
)
)
}
)
return self.async_show_form(step_id="pick_line", data_schema=schema)
async def _async_load_stops(self) -> str | None:
"""Fetch TCL stops from the API, returning an error key on failure."""
entry = self._get_entry()
session = async_get_clientsession(self.hass)
client = DataGrandLyonClient(
session=session,
username=entry.data[CONF_USERNAME],
password=entry.data[CONF_PASSWORD],
)
try:
self._stops = await client.get_tcl_stops()
except ClientResponseError as err:
if err.status in (401, 403):
return "invalid_auth"
return "cannot_connect"
except ClientError, TimeoutError:
return "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected error fetching Data Grand Lyon TCL stops")
return "unknown"
return None
def _create_stop(self, line: str, stop_id: int) -> SubentryFlowResult:
"""Create the stop subentry, aborting on duplicate."""
entry = self._get_entry()
unique_id = f"{line}_{stop_id}"
for subentry in entry.subentries.values():
if subentry.unique_id == unique_id:
return self.async_abort(reason="already_configured")
return self.async_create_entry(
title=f"{line} - Stop {stop_id}",
data={CONF_LINE: line, CONF_STOP_ID: stop_id},
unique_id=unique_id,
)
def _stop_label(stop: TclStop) -> str:
label = stop.nom
# variable extracted to please codespell.
address = stop.adresse # codespell:ignore adresse
if address or stop.commune:
label += " (" + ", ".join(filter(None, [address, stop.commune])) + ")"
label += f" - {stop.id}"
return label
class VelovStationSubentryFlowHandler(ConfigSubentryFlow):
"""Handle a subentry flow for adding a Vélo'v station."""
homeassistant/components/data_grand_lyon/strings.json
+16 -3
View File
@@ -46,17 +46,30 @@
"config_subentries": {
"stop": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]"
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"entry_type": "Transit stop",
"error": {
"invalid_stop_id": "Stop ID must be a number."
},
"initiate_flow": {
"user": "Add transit stop"
},
"step": {
"pick_line": {
"data": {
"line": "Line"
}
},
"user": {
"data": {
"line": "Line",
"stop_id": "Stop ID"
"stop_id": "Stop"
},
"data_description": {
"stop_id": "Search by stop name, address or city, or enter a stop ID directly."
}
}
}
tests/components/data_grand_lyon/conftest.py
+35
View File
@@ -7,6 +7,7 @@ from unittest.mock import AsyncMock, patch
from data_grand_lyon_ha import (
TclPassage,
TclPassageType,
TclStop,
VelovAvailabilityLevel,
VelovBikeStandAvailability,
VelovStation,
@@ -50,6 +51,39 @@ MOCK_DEPARTURES = [
),
]
MOCK_TCL_STOPS = [
TclStop(
id=100,
gid=1100,
adresse="Place Bellecour", # codespell:ignore adresse
ascenseur=False,
commune="Lyon 2",
desserte=["C3", "27"],
escalator=False,
insee="69382",
last_update=datetime(2026, 4, 10, 0, 0),
lat=45.757,
lon=4.832,
nom="Bellecour",
pmr=True,
),
TclStop(
id=200,
gid=1200,
adresse="Cours Lafayette", # codespell:ignore adresse
ascenseur=True,
commune="Lyon 3",
desserte=["C3", "T1"],
escalator=True,
insee="69383",
last_update=datetime(2026, 4, 10, 0, 0),
lat=45.763,
lon=4.846,
nom="Part-Dieu",
pmr=True,
),
]
MOCK_VELOV_STATION = VelovStation(
number=1001,
name="Place Bellecour",
@@ -147,5 +181,6 @@ def mock_tcl_client() -> Generator[AsyncMock]:
) as mock_cls:
client = mock_cls.return_value
client.get_tcl_passages.return_value = MOCK_DEPARTURES
client.get_tcl_stops.return_value = MOCK_TCL_STOPS
client.get_velov_stations.return_value = [MOCK_VELOV_STATION]
yield client
tests/components/data_grand_lyon/test_config_flow.py
+132 -9
View File
@@ -19,6 +19,8 @@ from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .conftest import MOCK_TCL_STOPS
from tests.common import MockConfigEntry
@@ -32,6 +34,16 @@ def mock_get_tcl_passages() -> Generator[AsyncMock]:
yield mock
@pytest.fixture
def mock_get_tcl_stops() -> Generator[AsyncMock]:
"""Mock get_tcl_stops in the stop subentry picker flow."""
with patch(
"homeassistant.components.data_grand_lyon.config_flow.DataGrandLyonClient.get_tcl_stops",
return_value=MOCK_TCL_STOPS,
) as mock:
yield mock
# Main config flow tests
@@ -274,11 +286,12 @@ async def test_reconfigure_flow_errors(
@pytest.mark.parametrize("mock_subentries", [[]])
async def test_stop_subentry_flow(
async def test_stop_subentry_picker_flow(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_get_tcl_stops: AsyncMock,
) -> None:
"""Test adding a stop subentry."""
"""Test adding a stop subentry by picking a stop and a line from the lists."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
@@ -289,23 +302,33 @@ async def test_stop_subentry_flow(
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert mock_get_tcl_stops.await_count == 1
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{CONF_LINE: "C3", CONF_STOP_ID: 456},
{CONF_STOP_ID: "200"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "pick_line"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{CONF_LINE: "T1"},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "C3 - Stop 456"
assert result["data"] == {CONF_LINE: "C3", CONF_STOP_ID: 456}
assert result["unique_id"] == "C3_456"
assert result["title"] == "T1 - Stop 200"
assert result["data"] == {CONF_LINE: "T1", CONF_STOP_ID: 200}
assert result["unique_id"] == "T1_200"
async def test_stop_subentry_already_configured(
@pytest.mark.parametrize("mock_subentries", [[]])
async def test_stop_subentry_custom_value_flow(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_get_tcl_stops: AsyncMock,
) -> None:
"""Test stop subentry aborts if same line+stop already exists."""
"""Test adding a stop subentry by typing a stop ID not present in the list."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
@@ -314,16 +337,116 @@ async def test_stop_subentry_already_configured(
(mock_config_entry.entry_id, SUBENTRY_TYPE_STOP),
context={"source": config_entries.SOURCE_USER},
)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{CONF_STOP_ID: "456"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "pick_line"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{CONF_LINE: "C3", CONF_STOP_ID: 100},
{CONF_LINE: "C3"},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "C3 - Stop 456"
assert result["data"] == {CONF_LINE: "C3", CONF_STOP_ID: 456}
assert result["unique_id"] == "C3_456"
@pytest.mark.parametrize("mock_subentries", [[]])
async def test_stop_subentry_invalid_stop_id(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_get_tcl_stops: AsyncMock,
) -> None:
"""Test typing a non-numeric stop ID re-renders the form with an error."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, SUBENTRY_TYPE_STOP),
context={"source": config_entries.SOURCE_USER},
)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{CONF_STOP_ID: "not-a-number"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {CONF_STOP_ID: "invalid_stop_id"}
async def test_stop_subentry_picker_already_configured(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_get_tcl_stops: AsyncMock,
) -> None:
"""Test picker stop subentry aborts if same line+stop already exists."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, SUBENTRY_TYPE_STOP),
context={"source": config_entries.SOURCE_USER},
)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{CONF_STOP_ID: "100"},
)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{CONF_LINE: "C3"},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
("side_effect", "reason"),
[
(
ClientResponseError(request_info=None, history=(), status=500),
"cannot_connect",
),
(
ClientResponseError(request_info=None, history=(), status=401),
"invalid_auth",
),
(ClientConnectionError("boom"), "cannot_connect"),
(TimeoutError("boom"), "cannot_connect"),
(RuntimeError("boom"), "unknown"),
],
)
@pytest.mark.parametrize("mock_subentries", [[]])
async def test_stop_subentry_picker_load_errors(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_get_tcl_stops: AsyncMock,
side_effect: Exception,
reason: str,
) -> None:
"""Test picker aborts with the right reason when loading stops fails."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
mock_get_tcl_stops.side_effect = side_effect
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, SUBENTRY_TYPE_STOP),
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == reason
# Vélo'v station subentry tests