Add config-flow to NextBus (#92149)

This commit is contained in:
Ian 2023-09-19 08:10:29 -07:00 committed by GitHub
parent d9227a7e3d
commit c3f74ae022
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
12 changed files with 710 additions and 137 deletions

View file

@ -1 +1,18 @@
"""NextBus sensor."""
"""NextBus platform."""
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
PLATFORMS = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up platforms for NextBus."""
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

View file

@ -0,0 +1,236 @@
"""Config flow to configure the Nextbus integration."""
from collections import Counter
import logging
from py_nextbus import NextBusClient
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.selector import (
SelectOptionDict,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
)
from .const import CONF_AGENCY, CONF_ROUTE, CONF_STOP, DOMAIN
_LOGGER = logging.getLogger(__name__)
def _dict_to_select_selector(options: dict[str, str]) -> SelectSelector:
return SelectSelector(
SelectSelectorConfig(
options=sorted(
(
SelectOptionDict(value=key, label=value)
for key, value in options.items()
),
key=lambda o: o["label"],
),
mode=SelectSelectorMode.DROPDOWN,
)
)
def _get_agency_tags(client: NextBusClient) -> dict[str, str]:
return {a["tag"]: a["title"] for a in client.get_agency_list()["agency"]}
def _get_route_tags(client: NextBusClient, agency_tag: str) -> dict[str, str]:
return {a["tag"]: a["title"] for a in client.get_route_list(agency_tag)["route"]}
def _get_stop_tags(
client: NextBusClient, agency_tag: str, route_tag: str
) -> dict[str, str]:
route_config = client.get_route_config(route_tag, agency_tag)
tags = {a["tag"]: a["title"] for a in route_config["route"]["stop"]}
title_counts = Counter(tags.values())
stop_directions: dict[str, str] = {}
for direction in route_config["route"]["direction"]:
for stop in direction["stop"]:
stop_directions[stop["tag"]] = direction["name"]
# Append directions for stops with shared titles
for tag, title in tags.items():
if title_counts[title] > 1:
tags[tag] = f"{title} ({stop_directions[tag]})"
return tags
def _validate_import(
client: NextBusClient, agency_tag: str, route_tag: str, stop_tag: str
) -> str | tuple[str, str, str]:
agency_tags = _get_agency_tags(client)
agency = agency_tags.get(agency_tag)
if not agency:
return "invalid_agency"
route_tags = _get_route_tags(client, agency_tag)
route = route_tags.get(route_tag)
if not route:
return "invalid_route"
stop_tags = _get_stop_tags(client, agency_tag, route_tag)
stop = stop_tags.get(stop_tag)
if not stop:
return "invalid_stop"
return agency, route, stop
def _unique_id_from_data(data: dict[str, str]) -> str:
return f"{data[CONF_AGENCY]}_{data[CONF_ROUTE]}_{data[CONF_STOP]}"
class NextBusFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle Nextbus configuration."""
VERSION = 1
_agency_tags: dict[str, str]
_route_tags: dict[str, str]
_stop_tags: dict[str, str]
def __init__(self):
"""Initialize NextBus config flow."""
self.data: dict[str, str] = {}
self._client = NextBusClient(output_format="json")
_LOGGER.info("Init new config flow")
async def async_step_import(self, config_input: dict[str, str]) -> FlowResult:
"""Handle import of config."""
agency_tag = config_input[CONF_AGENCY]
route_tag = config_input[CONF_ROUTE]
stop_tag = config_input[CONF_STOP]
validation_result = await self.hass.async_add_executor_job(
_validate_import,
self._client,
agency_tag,
route_tag,
stop_tag,
)
if isinstance(validation_result, str):
return self.async_abort(reason=validation_result)
data = {
CONF_AGENCY: agency_tag,
CONF_ROUTE: route_tag,
CONF_STOP: stop_tag,
CONF_NAME: config_input.get(
CONF_NAME,
f"{config_input[CONF_AGENCY]} {config_input[CONF_ROUTE]}",
),
}
await self.async_set_unique_id(_unique_id_from_data(data))
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=" ".join(validation_result),
data=data,
)
async def async_step_user(
self,
user_input: dict[str, str] | None = None,
) -> FlowResult:
"""Handle a flow initiated by the user."""
return await self.async_step_agency(user_input)
async def async_step_agency(
self,
user_input: dict[str, str] | None = None,
) -> FlowResult:
"""Select agency."""
if user_input is not None:
self.data[CONF_AGENCY] = user_input[CONF_AGENCY]
return await self.async_step_route()
self._agency_tags = await self.hass.async_add_executor_job(
_get_agency_tags, self._client
)
return self.async_show_form(
step_id="agency",
data_schema=vol.Schema(
{
vol.Required(CONF_AGENCY): _dict_to_select_selector(
self._agency_tags
),
}
),
)
async def async_step_route(
self,
user_input: dict[str, str] | None = None,
) -> FlowResult:
"""Select route."""
if user_input is not None:
self.data[CONF_ROUTE] = user_input[CONF_ROUTE]
return await self.async_step_stop()
self._route_tags = await self.hass.async_add_executor_job(
_get_route_tags, self._client, self.data[CONF_AGENCY]
)
return self.async_show_form(
step_id="route",
data_schema=vol.Schema(
{
vol.Required(CONF_ROUTE): _dict_to_select_selector(
self._route_tags
),
}
),
)
async def async_step_stop(
self,
user_input: dict[str, str] | None = None,
) -> FlowResult:
"""Select stop."""
if user_input is not None:
self.data[CONF_STOP] = user_input[CONF_STOP]
await self.async_set_unique_id(_unique_id_from_data(self.data))
self._abort_if_unique_id_configured()
agency_tag = self.data[CONF_AGENCY]
route_tag = self.data[CONF_ROUTE]
stop_tag = self.data[CONF_STOP]
agency_name = self._agency_tags[agency_tag]
route_name = self._route_tags[route_tag]
stop_name = self._stop_tags[stop_tag]
return self.async_create_entry(
title=f"{agency_name} {route_name} {stop_name}",
data=self.data,
)
self._stop_tags = await self.hass.async_add_executor_job(
_get_stop_tags,
self._client,
self.data[CONF_AGENCY],
self.data[CONF_ROUTE],
)
return self.async_show_form(
step_id="stop",
data_schema=vol.Schema(
{
vol.Required(CONF_STOP): _dict_to_select_selector(self._stop_tags),
}
),
)

View file

@ -2,6 +2,7 @@
"domain": "nextbus",
"name": "NextBus",
"codeowners": ["@vividboarder"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/nextbus",
"iot_class": "cloud_polling",
"loggers": ["py_nextbus"],

View file

@ -12,14 +12,16 @@ from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
)
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
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
from homeassistant.util.dt import utc_from_timestamp
from .const import CONF_AGENCY, CONF_ROUTE, CONF_STOP
from .const import CONF_AGENCY, CONF_ROUTE, CONF_STOP, DOMAIN
from .util import listify, maybe_first
_LOGGER = logging.getLogger(__name__)
@ -34,59 +36,54 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
)
def validate_value(value_name, value, value_list):
"""Validate tag value is in the list of items and logs error if not."""
valid_values = {v["tag"]: v["title"] for v in value_list}
if value not in valid_values:
_LOGGER.error(
"Invalid %s tag `%s`. Please use one of the following: %s",
value_name,
value,
", ".join(f"{title}: {tag}" for tag, title in valid_values.items()),
)
return False
return True
def validate_tags(client, agency, route, stop):
"""Validate provided tags."""
# Validate agencies
if not validate_value("agency", agency, client.get_agency_list()["agency"]):
return False
# Validate the route
if not validate_value("route", route, client.get_route_list(agency)["route"]):
return False
# Validate the stop
route_config = client.get_route_config(route, agency)["route"]
if not validate_value("stop", stop, route_config["stop"]):
return False
return True
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:
"""Load values from configuration and initialize the platform."""
agency = config[CONF_AGENCY]
route = config[CONF_ROUTE]
stop = config[CONF_STOP]
name = config.get(CONF_NAME)
"""Initialize nextbus import from config."""
async_create_issue(
hass,
HOMEASSISTANT_DOMAIN,
f"deprecated_yaml_{DOMAIN}",
is_fixable=False,
breaks_in_ha_version="2024.4.0",
severity=IssueSeverity.WARNING,
translation_key="deprecated_yaml",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "NextBus",
},
)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=config
)
)
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Load values from configuration and initialize the platform."""
client = NextBusClient(output_format="json")
# Ensures that the tags provided are valid, also logs out valid values
if not validate_tags(client, agency, route, stop):
_LOGGER.error("Invalid config value(s)")
return
_LOGGER.debug(config.data)
add_entities([NextBusDepartureSensor(client, agency, route, stop, name)], True)
sensor = NextBusDepartureSensor(
client,
config.unique_id,
config.data[CONF_AGENCY],
config.data[CONF_ROUTE],
config.data[CONF_STOP],
config.data.get(CONF_NAME) or config.title,
)
async_add_entities((sensor,), True)
class NextBusDepartureSensor(SensorEntity):
@ -103,17 +100,14 @@ class NextBusDepartureSensor(SensorEntity):
_attr_device_class = SensorDeviceClass.TIMESTAMP
_attr_icon = "mdi:bus"
def __init__(self, client, agency, route, stop, name=None):
def __init__(self, client, unique_id, agency, route, stop, name):
"""Initialize sensor with all required config."""
self.agency = agency
self.route = route
self.stop = stop
self._attr_extra_state_attributes = {}
# Maybe pull a more user friendly name from the API here
self._attr_name = f"{agency} {route}"
if name:
self._attr_name = name
self._attr_unique_id = unique_id
self._attr_name = name
self._client = client

View file

@ -0,0 +1,33 @@
{
"title": "NextBus predictions",
"config": {
"step": {
"agency": {
"title": "Select metro agency",
"data": {
"agency": "Metro agency"
}
},
"route": {
"title": "Select route",
"data": {
"route": "Route"
}
},
"stop": {
"title": "Select stop",
"data": {
"stop": "Stop"
}
}
},
"error": {
"invalid_agency": "The agency value selected is not valid",
"invalid_route": "The route value selected is not valid",
"invalid_stop": "The stop value selected is not valid"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]"
}
}
}

View file

@ -17,7 +17,7 @@ def listify(maybe_list: Any) -> list[Any]:
return [maybe_list]
def maybe_first(maybe_list: list[Any]) -> Any:
def maybe_first(maybe_list: list[Any] | None) -> Any:
"""Return the first item out of a list or returns back the input."""
if isinstance(maybe_list, list) and maybe_list:
return maybe_list[0]

View file

@ -301,6 +301,7 @@ FLOWS = {
"netatmo",
"netgear",
"nexia",
"nextbus",
"nextcloud",
"nextdns",
"nfandroidtv",

View file

@ -3712,9 +3712,8 @@
"supported_by": "overkiz"
},
"nextbus": {
"name": "NextBus",
"integration_type": "hub",
"config_flow": false,
"config_flow": true,
"iot_class": "cloud_polling"
},
"nextcloud": {
@ -6798,6 +6797,7 @@
"mobile_app",
"moehlenhoff_alpha2",
"moon",
"nextbus",
"nmap_tracker",
"plant",
"proximity",

View file

@ -0,0 +1,36 @@
"""Test helpers for NextBus tests."""
from unittest.mock import MagicMock
import pytest
@pytest.fixture
def mock_nextbus_lists(mock_nextbus: MagicMock) -> MagicMock:
"""Mock all list functions in nextbus to test validate logic."""
instance = mock_nextbus.return_value
instance.get_agency_list.return_value = {
"agency": [{"tag": "sf-muni", "title": "San Francisco Muni"}]
}
instance.get_route_list.return_value = {
"route": [{"tag": "F", "title": "F - Market & Wharves"}]
}
instance.get_route_config.return_value = {
"route": {
"stop": [
{"tag": "5650", "title": "Market St & 7th St"},
{"tag": "5651", "title": "Market St & 7th St"},
],
"direction": [
{
"name": "Outbound",
"stop": [{"tag": "5650"}],
},
{
"name": "Inbound",
"stop": [{"tag": "5651"}],
},
],
}
}
return instance

View file

@ -0,0 +1,162 @@
"""Test the NextBus config flow."""
from collections.abc import Generator
from unittest.mock import MagicMock, patch
import pytest
from homeassistant import config_entries, setup
from homeassistant.components.nextbus.const import (
CONF_AGENCY,
CONF_ROUTE,
CONF_STOP,
DOMAIN,
)
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
@pytest.fixture
def mock_setup_entry() -> Generator[MagicMock, None, None]:
"""Create a mock for the nextbus component setup."""
with patch(
"homeassistant.components.nextbus.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
def mock_nextbus() -> Generator[MagicMock, None, None]:
"""Create a mock py_nextbus module."""
with patch("homeassistant.components.nextbus.config_flow.NextBusClient") as client:
yield client
async def test_import_config(
hass: HomeAssistant, mock_setup_entry: MagicMock, mock_nextbus_lists: MagicMock
) -> None:
"""Test config is imported and component set up."""
await setup.async_setup_component(hass, "persistent_notification", {})
data = {
CONF_AGENCY: "sf-muni",
CONF_ROUTE: "F",
CONF_STOP: "5650",
}
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=data,
)
await hass.async_block_till_done()
assert result.get("type") == FlowResultType.CREATE_ENTRY
assert (
result.get("title")
== "San Francisco Muni F - Market & Wharves Market St & 7th St (Outbound)"
)
assert result.get("data") == {CONF_NAME: "sf-muni F", **data}
assert len(mock_setup_entry.mock_calls) == 1
# Check duplicate entries are aborted
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=data,
)
await hass.async_block_till_done()
assert result.get("type") == FlowResultType.ABORT
assert result.get("reason") == "already_configured"
@pytest.mark.parametrize(
("override", "expected_reason"),
(
({CONF_AGENCY: "not muni"}, "invalid_agency"),
({CONF_ROUTE: "not F"}, "invalid_route"),
({CONF_STOP: "not 5650"}, "invalid_stop"),
),
)
async def test_import_config_invalid(
hass: HomeAssistant,
mock_setup_entry: MagicMock,
mock_nextbus_lists: MagicMock,
override: dict[str, str],
expected_reason: str,
) -> None:
"""Test user is redirected to user setup flow because they have invalid config."""
await setup.async_setup_component(hass, "persistent_notification", {})
data = {
CONF_AGENCY: "sf-muni",
CONF_ROUTE: "F",
CONF_STOP: "5650",
**override,
}
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=data,
)
await hass.async_block_till_done()
assert result.get("type") == FlowResultType.ABORT
assert result.get("reason") == expected_reason
async def test_user_config(
hass: HomeAssistant, mock_setup_entry: MagicMock, mock_nextbus_lists: MagicMock
) -> None:
"""Test we get the form."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result.get("type") == FlowResultType.FORM
assert result.get("step_id") == "agency"
# Select agency
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_AGENCY: "sf-muni",
},
)
await hass.async_block_till_done()
assert result.get("type") == "form"
assert result.get("step_id") == "route"
# Select route
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_ROUTE: "F",
},
)
await hass.async_block_till_done()
assert result.get("type") == FlowResultType.FORM
assert result.get("step_id") == "stop"
# Select stop
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_STOP: "5650",
},
)
await hass.async_block_till_done()
assert result.get("type") == FlowResultType.CREATE_ENTRY
assert result.get("data") == {
"agency": "sf-muni",
"route": "F",
"stop": "5650",
}
assert len(mock_setup_entry.mock_calls) == 1

View file

@ -1,15 +1,24 @@
"""The tests for the nexbus sensor component."""
from collections.abc import Generator
from copy import deepcopy
from unittest.mock import patch
from unittest.mock import MagicMock, patch
import pytest
import homeassistant.components.nextbus.sensor as nextbus
import homeassistant.components.sensor as sensor
from homeassistant.core import HomeAssistant
from homeassistant.components import sensor
from homeassistant.components.nextbus.const import (
CONF_AGENCY,
CONF_ROUTE,
CONF_STOP,
DOMAIN,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_NAME
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
from homeassistant.helpers import issue_registry as ir
from homeassistant.setup import async_setup_component
from tests.common import assert_setup_component
from tests.common import MockConfigEntry
VALID_AGENCY = "sf-muni"
VALID_ROUTE = "F"
@ -17,24 +26,34 @@ VALID_STOP = "5650"
VALID_AGENCY_TITLE = "San Francisco Muni"
VALID_ROUTE_TITLE = "F-Market & Wharves"
VALID_STOP_TITLE = "Market St & 7th St"
SENSOR_ID_SHORT = "sensor.sf_muni_f"
SENSOR_ID = "sensor.san_francisco_muni_f_market_wharves_market_st_7th_st"
CONFIG_BASIC = {
"sensor": {
"platform": "nextbus",
"agency": VALID_AGENCY,
"route": VALID_ROUTE,
"stop": VALID_STOP,
}
PLATFORM_CONFIG = {
sensor.DOMAIN: {
"platform": DOMAIN,
CONF_AGENCY: VALID_AGENCY,
CONF_ROUTE: VALID_ROUTE,
CONF_STOP: VALID_STOP,
},
}
CONFIG_INVALID_MISSING = {"sensor": {"platform": "nextbus"}}
CONFIG_BASIC = {
DOMAIN: {
CONF_AGENCY: VALID_AGENCY,
CONF_ROUTE: VALID_ROUTE,
CONF_STOP: VALID_STOP,
}
}
BASIC_RESULTS = {
"predictions": {
"agencyTitle": VALID_AGENCY_TITLE,
"agencyTag": VALID_AGENCY,
"routeTitle": VALID_ROUTE_TITLE,
"routeTag": VALID_ROUTE,
"stopTitle": VALID_STOP_TITLE,
"stopTag": VALID_STOP,
"direction": {
"title": "Outbound",
"prediction": [
@ -48,24 +67,19 @@ BASIC_RESULTS = {
}
async def assert_setup_sensor(hass, config, count=1):
"""Set up the sensor and assert it's been created."""
with assert_setup_component(count):
assert await async_setup_component(hass, sensor.DOMAIN, config)
await hass.async_block_till_done()
@pytest.fixture
def mock_nextbus():
def mock_nextbus() -> Generator[MagicMock, None, None]:
"""Create a mock py_nextbus module."""
with patch(
"homeassistant.components.nextbus.sensor.NextBusClient"
) as NextBusClient:
yield NextBusClient
"homeassistant.components.nextbus.sensor.NextBusClient",
) as client:
yield client
@pytest.fixture
def mock_nextbus_predictions(mock_nextbus):
def mock_nextbus_predictions(
mock_nextbus: MagicMock,
) -> Generator[MagicMock, None, None]:
"""Create a mock of NextBusClient predictions."""
instance = mock_nextbus.return_value
instance.get_predictions_for_multi_stops.return_value = BASIC_RESULTS
@ -73,63 +87,69 @@ def mock_nextbus_predictions(mock_nextbus):
return instance.get_predictions_for_multi_stops
@pytest.fixture
def mock_nextbus_lists(mock_nextbus):
"""Mock all list functions in nextbus to test validate logic."""
instance = mock_nextbus.return_value
instance.get_agency_list.return_value = {
"agency": [{"tag": "sf-muni", "title": "San Francisco Muni"}]
}
instance.get_route_list.return_value = {
"route": [{"tag": "F", "title": "F - Market & Wharves"}]
}
instance.get_route_config.return_value = {
"route": {"stop": [{"tag": "5650", "title": "Market St & 7th St"}]}
}
async def assert_setup_sensor(
hass: HomeAssistant,
config: dict[str, str],
expected_state=ConfigEntryState.LOADED,
) -> MockConfigEntry:
"""Set up the sensor and assert it's been created."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data=config[DOMAIN],
title=f"{VALID_AGENCY_TITLE} {VALID_ROUTE_TITLE} {VALID_STOP_TITLE}",
unique_id=f"{VALID_AGENCY}_{VALID_ROUTE}_{VALID_STOP}",
)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is expected_state
return config_entry
async def test_legacy_yaml_setup(
hass: HomeAssistant,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test config setup and yaml deprecation."""
with patch(
"homeassistant.components.nextbus.config_flow.NextBusClient",
) as NextBusClient:
NextBusClient.return_value.get_predictions_for_multi_stops.return_value = (
BASIC_RESULTS
)
await async_setup_component(hass, sensor.DOMAIN, PLATFORM_CONFIG)
await hass.async_block_till_done()
issue = issue_registry.async_get_issue(
HOMEASSISTANT_DOMAIN, f"deprecated_yaml_{DOMAIN}"
)
assert issue
async def test_valid_config(
hass: HomeAssistant, mock_nextbus, mock_nextbus_lists
hass: HomeAssistant, mock_nextbus: MagicMock, mock_nextbus_lists: MagicMock
) -> None:
"""Test that sensor is set up properly with valid config."""
await assert_setup_sensor(hass, CONFIG_BASIC)
async def test_invalid_config(
hass: HomeAssistant, mock_nextbus, mock_nextbus_lists
) -> None:
"""Checks that component is not setup when missing information."""
await assert_setup_sensor(hass, CONFIG_INVALID_MISSING, count=0)
async def test_validate_tags(
hass: HomeAssistant, mock_nextbus, mock_nextbus_lists
) -> None:
"""Test that additional validation against the API is successful."""
# with self.subTest('Valid everything'):
assert nextbus.validate_tags(mock_nextbus(), VALID_AGENCY, VALID_ROUTE, VALID_STOP)
# with self.subTest('Invalid agency'):
assert not nextbus.validate_tags(
mock_nextbus(), "not-valid", VALID_ROUTE, VALID_STOP
)
# with self.subTest('Invalid route'):
assert not nextbus.validate_tags(mock_nextbus(), VALID_AGENCY, "0", VALID_STOP)
# with self.subTest('Invalid stop'):
assert not nextbus.validate_tags(mock_nextbus(), VALID_AGENCY, VALID_ROUTE, 0)
async def test_verify_valid_state(
hass: HomeAssistant, mock_nextbus, mock_nextbus_lists, mock_nextbus_predictions
hass: HomeAssistant,
mock_nextbus: MagicMock,
mock_nextbus_lists: MagicMock,
mock_nextbus_predictions: MagicMock,
) -> None:
"""Verify all attributes are set from a valid response."""
await assert_setup_sensor(hass, CONFIG_BASIC)
mock_nextbus_predictions.assert_called_once_with(
[{"stop_tag": VALID_STOP, "route_tag": VALID_ROUTE}], VALID_AGENCY
)
state = hass.states.get(SENSOR_ID_SHORT)
state = hass.states.get(SENSOR_ID)
assert state is not None
assert state.state == "2019-03-28T21:09:31+00:00"
assert state.attributes["agency"] == VALID_AGENCY_TITLE
@ -140,14 +160,20 @@ async def test_verify_valid_state(
async def test_message_dict(
hass: HomeAssistant, mock_nextbus, mock_nextbus_lists, mock_nextbus_predictions
hass: HomeAssistant,
mock_nextbus: MagicMock,
mock_nextbus_lists: MagicMock,
mock_nextbus_predictions: MagicMock,
) -> None:
"""Verify that a single dict message is rendered correctly."""
mock_nextbus_predictions.return_value = {
"predictions": {
"agencyTitle": VALID_AGENCY_TITLE,
"agencyTag": VALID_AGENCY,
"routeTitle": VALID_ROUTE_TITLE,
"routeTag": VALID_ROUTE,
"stopTitle": VALID_STOP_TITLE,
"stopTag": VALID_STOP,
"message": {"text": "Message"},
"direction": {
"title": "Outbound",
@ -162,20 +188,26 @@ async def test_message_dict(
await assert_setup_sensor(hass, CONFIG_BASIC)
state = hass.states.get(SENSOR_ID_SHORT)
state = hass.states.get(SENSOR_ID)
assert state is not None
assert state.attributes["message"] == "Message"
async def test_message_list(
hass: HomeAssistant, mock_nextbus, mock_nextbus_lists, mock_nextbus_predictions
hass: HomeAssistant,
mock_nextbus: MagicMock,
mock_nextbus_lists: MagicMock,
mock_nextbus_predictions: MagicMock,
) -> None:
"""Verify that a list of messages are rendered correctly."""
mock_nextbus_predictions.return_value = {
"predictions": {
"agencyTitle": VALID_AGENCY_TITLE,
"agencyTag": VALID_AGENCY,
"routeTitle": VALID_ROUTE_TITLE,
"routeTag": VALID_ROUTE,
"stopTitle": VALID_STOP_TITLE,
"stopTag": VALID_STOP,
"message": [{"text": "Message 1"}, {"text": "Message 2"}],
"direction": {
"title": "Outbound",
@ -190,20 +222,26 @@ async def test_message_list(
await assert_setup_sensor(hass, CONFIG_BASIC)
state = hass.states.get(SENSOR_ID_SHORT)
state = hass.states.get(SENSOR_ID)
assert state is not None
assert state.attributes["message"] == "Message 1 -- Message 2"
async def test_direction_list(
hass: HomeAssistant, mock_nextbus, mock_nextbus_lists, mock_nextbus_predictions
hass: HomeAssistant,
mock_nextbus: MagicMock,
mock_nextbus_lists: MagicMock,
mock_nextbus_predictions: MagicMock,
) -> None:
"""Verify that a list of messages are rendered correctly."""
mock_nextbus_predictions.return_value = {
"predictions": {
"agencyTitle": VALID_AGENCY_TITLE,
"agencyTag": VALID_AGENCY,
"routeTitle": VALID_ROUTE_TITLE,
"routeTag": VALID_ROUTE,
"stopTitle": VALID_STOP_TITLE,
"stopTag": VALID_STOP,
"message": [{"text": "Message 1"}, {"text": "Message 2"}],
"direction": [
{
@ -224,7 +262,7 @@ async def test_direction_list(
await assert_setup_sensor(hass, CONFIG_BASIC)
state = hass.states.get(SENSOR_ID_SHORT)
state = hass.states.get(SENSOR_ID)
assert state is not None
assert state.state == "2019-03-28T21:09:31+00:00"
assert state.attributes["agency"] == VALID_AGENCY_TITLE
@ -235,46 +273,67 @@ async def test_direction_list(
async def test_custom_name(
hass: HomeAssistant, mock_nextbus, mock_nextbus_lists, mock_nextbus_predictions
hass: HomeAssistant,
mock_nextbus: MagicMock,
mock_nextbus_lists: MagicMock,
mock_nextbus_predictions: MagicMock,
) -> None:
"""Verify that a custom name can be set via config."""
config = deepcopy(CONFIG_BASIC)
config["sensor"]["name"] = "Custom Name"
config[DOMAIN][CONF_NAME] = "Custom Name"
await assert_setup_sensor(hass, config)
state = hass.states.get("sensor.custom_name")
assert state is not None
assert state.name == "Custom Name"
@pytest.mark.parametrize(
"prediction_results",
(
{},
{"Error": "Failed"},
),
)
async def test_no_predictions(
hass: HomeAssistant, mock_nextbus, mock_nextbus_predictions, mock_nextbus_lists
hass: HomeAssistant,
mock_nextbus: MagicMock,
mock_nextbus_predictions: MagicMock,
mock_nextbus_lists: MagicMock,
prediction_results: dict[str, str],
) -> None:
"""Verify there are no exceptions when no predictions are returned."""
mock_nextbus_predictions.return_value = {}
mock_nextbus_predictions.return_value = prediction_results
await assert_setup_sensor(hass, CONFIG_BASIC)
state = hass.states.get(SENSOR_ID_SHORT)
state = hass.states.get(SENSOR_ID)
assert state is not None
assert state.state == "unknown"
async def test_verify_no_upcoming(
hass: HomeAssistant, mock_nextbus, mock_nextbus_lists, mock_nextbus_predictions
hass: HomeAssistant,
mock_nextbus: MagicMock,
mock_nextbus_lists: MagicMock,
mock_nextbus_predictions: MagicMock,
) -> None:
"""Verify attributes are set despite no upcoming times."""
mock_nextbus_predictions.return_value = {
"predictions": {
"agencyTitle": VALID_AGENCY_TITLE,
"agencyTag": VALID_AGENCY,
"routeTitle": VALID_ROUTE_TITLE,
"routeTag": VALID_ROUTE,
"stopTitle": VALID_STOP_TITLE,
"stopTag": VALID_STOP,
"direction": {"title": "Outbound", "prediction": []},
}
}
await assert_setup_sensor(hass, CONFIG_BASIC)
state = hass.states.get(SENSOR_ID_SHORT)
state = hass.states.get(SENSOR_ID)
assert state is not None
assert state.state == "unknown"
assert state.attributes["upcoming"] == "No upcoming predictions"

View file

@ -0,0 +1,34 @@
"""Test NextBus util functions."""
from typing import Any
import pytest
from homeassistant.components.nextbus.util import listify, maybe_first
@pytest.mark.parametrize(
("input", "expected"),
(
("foo", ["foo"]),
(["foo"], ["foo"]),
(None, []),
),
)
def test_listify(input: Any, expected: list[Any]) -> None:
"""Test input listification."""
assert listify(input) == expected
@pytest.mark.parametrize(
("input", "expected"),
(
([], []),
(None, None),
("test", "test"),
(["test"], "test"),
(["test", "second"], "test"),
),
)
def test_maybe_first(input: list[Any] | None, expected: Any) -> None:
"""Test maybe getting the first thing from a list."""
assert maybe_first(input) == expected