evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add ConfigFlow for here_travel_time (#69212)

Browse source 
* Add ConfigFlow for here_travel_time

* Use Selectors and Menu

* Use separate config flow steps for menus

* Move time options together

* Update homeassistant/components/here_travel_time/config_flow.py

Co-authored-by: Allen Porter <allen.porter@gmail.com>

* Blacken config_flow

* Initialize _config

* Only catch HERE errors

* Fix unknown error test

* Implement async_step_import

* Only catch errors for validate_api_key

* Split lat/lon

* Add additional test coverage

* Use TimeSelector in option flow

* Assert config entry data/option

Co-authored-by: Allen Porter <allen.porter@gmail.com>
This commit is contained in:
Kevin Stillhammer 2022-05-21 16:36:02 +02:00 • committed by GitHub
parent dd0f9350ac
commit f1ac9f8cca
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
13 changed files with 1686 additions and 394 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

146
homeassistant/components/here_travel_time/__init__.py
View file

@ -8,7 +8,15 @@ import async_timeout
from herepy import NoRouteFoundError, RouteMode, RoutingApi, RoutingResponse
import voluptuous as vol
from homeassistant.const import ATTR_ATTRIBUTION, CONF_UNIT_SYSTEM_IMPERIAL, Platform
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_ATTRIBUTION,
CONF_API_KEY,
CONF_MODE,
CONF_UNIT_SYSTEM,
CONF_UNIT_SYSTEM_IMPERIAL,
Platform,
)
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.location import find_coordinates
@ -24,9 +32,20 @@ from .const import (
ATTR_ORIGIN,
ATTR_ORIGIN_NAME,
ATTR_ROUTE,
CONF_ARRIVAL_TIME,
CONF_DEPARTURE_TIME,
CONF_DESTINATION_ENTITY_ID,
CONF_DESTINATION_LATITUDE,
CONF_DESTINATION_LONGITUDE,
CONF_ORIGIN_ENTITY_ID,
CONF_ORIGIN_LATITUDE,
CONF_ORIGIN_LONGITUDE,
CONF_ROUTE_MODE,
CONF_TRAFFIC_MODE,
DEFAULT_SCAN_INTERVAL,
DOMAIN,
NO_ROUTE_ERROR_MESSAGE,
ROUTE_MODE_FASTEST,
TRAFFIC_MODE_ENABLED,
TRAVEL_MODES_VEHICLE,
)
@ -37,6 +56,74 @@ PLATFORMS = [Platform.SENSOR]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Set up HERE Travel Time from a config entry."""
api_key = config_entry.data[CONF_API_KEY]
here_client = RoutingApi(api_key)
setup_options(hass, config_entry)
arrival = (
dt.parse_time(config_entry.options[CONF_ARRIVAL_TIME])
if config_entry.options[CONF_ARRIVAL_TIME] is not None
else None
)
departure = (
dt.parse_time(config_entry.options[CONF_DEPARTURE_TIME])
if config_entry.options[CONF_DEPARTURE_TIME] is not None
else None
)
here_travel_time_config = HERETravelTimeConfig(
destination_latitude=config_entry.data.get(CONF_DESTINATION_LATITUDE),
destination_longitude=config_entry.data.get(CONF_DESTINATION_LONGITUDE),
destination_entity_id=config_entry.data.get(CONF_DESTINATION_ENTITY_ID),
origin_latitude=config_entry.data.get(CONF_ORIGIN_LATITUDE),
origin_longitude=config_entry.data.get(CONF_ORIGIN_LONGITUDE),
origin_entity_id=config_entry.data.get(CONF_ORIGIN_ENTITY_ID),
travel_mode=config_entry.data[CONF_MODE],
route_mode=config_entry.options[CONF_ROUTE_MODE],
units=config_entry.options[CONF_UNIT_SYSTEM],
arrival=arrival,
departure=departure,
)
coordinator = HereTravelTimeDataUpdateCoordinator(
hass,
here_client,
here_travel_time_config,
)
hass.data.setdefault(DOMAIN, {})[config_entry.entry_id] = coordinator
hass.config_entries.async_setup_platforms(config_entry, PLATFORMS)
return True
def setup_options(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
"""Set up options for a config entry if not set."""
if not config_entry.options:
hass.config_entries.async_update_entry(
config_entry,
options={
CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED,
CONF_ROUTE_MODE: ROUTE_MODE_FASTEST,
CONF_ARRIVAL_TIME: None,
CONF_DEPARTURE_TIME: None,
CONF_UNIT_SYSTEM: hass.config.units.name,
},
)
async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(
config_entry, PLATFORMS
)
if unload_ok:
hass.data[DOMAIN].pop(config_entry.entry_id)
return unload_ok
class HereTravelTimeDataUpdateCoordinator(DataUpdateCoordinator):
"""HERETravelTime DataUpdateCoordinator."""
@ -135,33 +222,40 @@ class HereTravelTimeDataUpdateCoordinator(DataUpdateCoordinator):
) -> tuple[list[str], list[str], str | None, str | None]:
"""Prepare parameters for the HERE api."""
if self.config.origin_entity_id is not None:
origin = find_coordinates(self.hass, self.config.origin_entity_id)
else:
origin = self.config.origin
def _from_entity_id(entity_id: str) -> list[str]:
coordinates = find_coordinates(self.hass, entity_id)
if coordinates is None:
raise InvalidCoordinatesException(
f"No coordinatnes found for {entity_id}"
)
try:
here_formatted_coordinates = coordinates.split(",")
vol.Schema(cv.gps(here_formatted_coordinates))
except (AttributeError, vol.Invalid) as ex:
raise InvalidCoordinatesException(
f"{coordinates} are not valid coordinates"
) from ex
return here_formatted_coordinates
# Destination
if self.config.destination_entity_id is not None:
destination = find_coordinates(self.hass, self.config.destination_entity_id)
destination = _from_entity_id(self.config.destination_entity_id)
else:
destination = self.config.destination
if destination is None:
raise InvalidCoordinatesException("Destination must be configured")
try:
here_formatted_destination = destination.split(",")
vol.Schema(cv.gps(here_formatted_destination))
except (vol.Invalid) as ex:
raise InvalidCoordinatesException(
f"{destination} are not valid coordinates"
) from ex
if origin is None:
raise InvalidCoordinatesException("Origin must be configured")
try:
here_formatted_origin = origin.split(",")
vol.Schema(cv.gps(here_formatted_origin))
except (AttributeError, vol.Invalid) as ex:
raise InvalidCoordinatesException(
f"{origin} are not valid coordinates"
) from ex
destination = [
str(self.config.destination_latitude),
str(self.config.destination_longitude),
]
# Origin
if self.config.origin_entity_id is not None:
origin = _from_entity_id(self.config.origin_entity_id)
else:
origin = [
str(self.config.origin_latitude),
str(self.config.origin_longitude),
]
# Arrival/Departure
arrival: str | None = None
departure: str | None = None
if self.config.arrival is not None:
@ -172,7 +266,7 @@ class HereTravelTimeDataUpdateCoordinator(DataUpdateCoordinator):
if arrival is None and departure is None:
departure = "now"
return (here_formatted_origin, here_formatted_destination, arrival, departure)
return (origin, destination, arrival, departure)
def build_hass_attribution(source_attribution: dict) -> str | None:

369
homeassistant/components/here_travel_time/config_flow.py Normal file
View file

@ -0,0 +1,369 @@
"""Config flow for HERE Travel Time integration."""
from __future__ import annotations
import logging
from typing import Any
from herepy import HEREError, InvalidCredentialsError, RouteMode, RoutingApi
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_API_KEY, CONF_MODE, CONF_NAME, CONF_UNIT_SYSTEM
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.selector import (
EntitySelector,
LocationSelector,
TimeSelector,
selector,
)
from .const import (
CONF_ARRIVAL,
CONF_ARRIVAL_TIME,
CONF_DEPARTURE,
CONF_DEPARTURE_TIME,
CONF_ROUTE_MODE,
CONF_TRAFFIC_MODE,
DEFAULT_NAME,
DOMAIN,
ROUTE_MODE_FASTEST,
ROUTE_MODES,
TRAFFIC_MODE_DISABLED,
TRAFFIC_MODE_ENABLED,
TRAFFIC_MODES,
TRAVEL_MODE_CAR,
TRAVEL_MODE_PUBLIC_TIME_TABLE,
TRAVEL_MODES,
UNITS,
)
from .sensor import (
CONF_DESTINATION_ENTITY_ID,
CONF_DESTINATION_LATITUDE,
CONF_DESTINATION_LONGITUDE,
CONF_ORIGIN_ENTITY_ID,
CONF_ORIGIN_LATITUDE,
CONF_ORIGIN_LONGITUDE,
)
_LOGGER = logging.getLogger(__name__)
def is_dupe_import(
entry: config_entries.ConfigEntry,
user_input: dict[str, Any],
options: dict[str, Any],
) -> bool:
"""Return whether imported config already exists."""
# Check the main data keys
if any(
user_input[key] != entry.data[key]
for key in (CONF_API_KEY, CONF_MODE, CONF_NAME)
):
return False
# Check origin/destination
for key in (
CONF_DESTINATION_LATITUDE,
CONF_DESTINATION_LONGITUDE,
CONF_ORIGIN_LATITUDE,
CONF_ORIGIN_LONGITUDE,
CONF_DESTINATION_ENTITY_ID,
CONF_ORIGIN_ENTITY_ID,
):
if user_input.get(key) != entry.data.get(key):
return False
# We have to check for options that don't have defaults
for key in (
CONF_TRAFFIC_MODE,
CONF_UNIT_SYSTEM,
CONF_ROUTE_MODE,
CONF_ARRIVAL_TIME,
CONF_DEPARTURE_TIME,
):
if options.get(key) != entry.options.get(key):
return False
return True
def validate_api_key(api_key: str) -> None:
"""Validate the user input allows us to connect."""
known_working_origin = [38.9, -77.04833]
known_working_destination = [39.0, -77.1]
RoutingApi(api_key).public_transport_timetable(
known_working_origin,
known_working_destination,
True,
[
RouteMode[ROUTE_MODE_FASTEST],
RouteMode[TRAVEL_MODE_CAR],
RouteMode[TRAFFIC_MODE_ENABLED],
],
arrival=None,
departure="now",
)
def get_user_step_schema(data: dict[str, Any]) -> vol.Schema:
"""Get a populated schema or default."""
return vol.Schema(
{
vol.Optional(
CONF_NAME, default=data.get(CONF_NAME, DEFAULT_NAME)
): cv.string,
vol.Required(CONF_API_KEY, default=data.get(CONF_API_KEY)): cv.string,
vol.Optional(
CONF_MODE, default=data.get(CONF_MODE, TRAVEL_MODE_CAR)
): vol.In(TRAVEL_MODES),
}
)
class HERETravelTimeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for HERE Travel Time."""
VERSION = 1
_config: dict[str, Any] = {}
@staticmethod
@callback
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> HERETravelTimeOptionsFlow:
"""Get the options flow."""
return HERETravelTimeOptionsFlow(config_entry)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
errors = {}
user_input = user_input or {}
if user_input:
try:
await self.hass.async_add_executor_job(
validate_api_key, user_input[CONF_API_KEY]
)
except InvalidCredentialsError:
errors["base"] = "invalid_auth"
except HEREError as error:
_LOGGER.exception("Unexpected exception: %s", error)
errors["base"] = "unknown"
if not errors:
self._config = user_input
return self.async_show_menu(
step_id="origin_menu",
menu_options=["origin_coordinates", "origin_entity"],
)
return self.async_show_form(
step_id="user", data_schema=get_user_step_schema(user_input), errors=errors
)
async def async_step_origin_coordinates(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Configure origin by using gps coordinates."""
if user_input is not None:
self._config[CONF_ORIGIN_LATITUDE] = user_input["origin"]["latitude"]
self._config[CONF_ORIGIN_LONGITUDE] = user_input["origin"]["longitude"]
return self.async_show_menu(
step_id="destination_menu",
menu_options=["destination_coordinates", "destination_entity"],
)
schema = vol.Schema({"origin": selector({LocationSelector.selector_type: {}})})
return self.async_show_form(step_id="origin_coordinates", data_schema=schema)
async def async_step_origin_entity(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Configure origin by using an entity."""
if user_input is not None:
self._config[CONF_ORIGIN_ENTITY_ID] = user_input[CONF_ORIGIN_ENTITY_ID]
return self.async_show_menu(
step_id="destination_menu",
menu_options=["destination_coordinates", "destination_entity"],
)
schema = vol.Schema(
{CONF_ORIGIN_ENTITY_ID: selector({EntitySelector.selector_type: {}})}
)
return self.async_show_form(step_id="origin_entity", data_schema=schema)
async def async_step_destination_coordinates(
self,
user_input: dict[str, Any] | None = None,
) -> FlowResult:
"""Configure destination by using gps coordinates."""
if user_input is not None:
self._config[CONF_DESTINATION_LATITUDE] = user_input["destination"][
"latitude"
]
self._config[CONF_DESTINATION_LONGITUDE] = user_input["destination"][
"longitude"
]
return self.async_create_entry(
title=self._config[CONF_NAME], data=self._config
)
schema = vol.Schema(
{"destination": selector({LocationSelector.selector_type: {}})}
)
return self.async_show_form(
step_id="destination_coordinates", data_schema=schema
)
async def async_step_destination_entity(
self,
user_input: dict[str, Any] | None = None,
) -> FlowResult:
"""Configure destination by using an entity."""
if user_input is not None:
self._config[CONF_DESTINATION_ENTITY_ID] = user_input[
CONF_DESTINATION_ENTITY_ID
]
return self.async_create_entry(
title=self._config[CONF_NAME], data=self._config
)
schema = vol.Schema(
{CONF_DESTINATION_ENTITY_ID: selector({EntitySelector.selector_type: {}})}
)
return self.async_show_form(step_id="destination_entity", data_schema=schema)
async def async_step_import(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Import from configuration.yaml."""
options: dict[str, Any] = {}
user_input, options = self._transform_import_input(user_input)
# We need to prevent duplicate imports
if any(
is_dupe_import(entry, user_input, options)
for entry in self.hass.config_entries.async_entries(DOMAIN)
if entry.source == config_entries.SOURCE_IMPORT
):
return self.async_abort(reason="already_configured")
return self.async_create_entry(
title=user_input[CONF_NAME], data=user_input, options=options
)
def _transform_import_input(
self, user_input
) -> tuple[dict[str, Any], dict[str, Any]]:
"""Transform platform schema input to new model."""
options: dict[str, Any] = {}
if user_input.get(CONF_ORIGIN_LATITUDE) is not None:
user_input[CONF_ORIGIN_LATITUDE] = user_input.pop(CONF_ORIGIN_LATITUDE)
user_input[CONF_ORIGIN_LONGITUDE] = user_input.pop(CONF_ORIGIN_LONGITUDE)
else:
user_input[CONF_ORIGIN_ENTITY_ID] = user_input.pop(CONF_ORIGIN_ENTITY_ID)
if user_input.get(CONF_DESTINATION_LATITUDE) is not None:
user_input[CONF_DESTINATION_LATITUDE] = user_input.pop(
CONF_DESTINATION_LATITUDE
)
user_input[CONF_DESTINATION_LONGITUDE] = user_input.pop(
CONF_DESTINATION_LONGITUDE
)
else:
user_input[CONF_DESTINATION_ENTITY_ID] = user_input.pop(
CONF_DESTINATION_ENTITY_ID
)
options[CONF_TRAFFIC_MODE] = (
TRAFFIC_MODE_ENABLED
if user_input.pop(CONF_TRAFFIC_MODE, False)
else TRAFFIC_MODE_DISABLED
)
options[CONF_ROUTE_MODE] = user_input.pop(CONF_ROUTE_MODE)
options[CONF_UNIT_SYSTEM] = user_input.pop(
CONF_UNIT_SYSTEM, self.hass.config.units.name
)
options[CONF_ARRIVAL_TIME] = user_input.pop(CONF_ARRIVAL, None)
options[CONF_DEPARTURE_TIME] = user_input.pop(CONF_DEPARTURE, None)
return user_input, options
class HERETravelTimeOptionsFlow(config_entries.OptionsFlow):
"""Handle HERE Travel Time options."""
_config: dict[str, Any] = {}
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize HERE Travel Time options flow."""
self.config_entry = config_entry
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage the HERE Travel Time options."""
if user_input is not None:
self._config = user_input
if self.config_entry.data[CONF_MODE] == TRAVEL_MODE_PUBLIC_TIME_TABLE:
return self.async_show_menu(
step_id="time_menu",
menu_options=["departure_time", "arrival_time", "no_time"],
)
return self.async_show_menu(
step_id="time_menu",
menu_options=["departure_time", "no_time"],
)
options = {
vol.Optional(
CONF_TRAFFIC_MODE,
default=self.config_entry.options.get(
CONF_TRAFFIC_MODE, TRAFFIC_MODE_ENABLED
),
): vol.In(TRAFFIC_MODES),
vol.Optional(
CONF_ROUTE_MODE,
default=self.config_entry.options.get(
CONF_ROUTE_MODE, ROUTE_MODE_FASTEST
),
): vol.In(ROUTE_MODES),
vol.Optional(
CONF_UNIT_SYSTEM,
default=self.config_entry.options.get(
CONF_UNIT_SYSTEM, self.hass.config.units.name
),
): vol.In(UNITS),
}
return self.async_show_form(step_id="init", data_schema=vol.Schema(options))
async def async_step_no_time(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Create Options Entry."""
return self.async_create_entry(title="", data=self._config)
async def async_step_arrival_time(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Configure arrival time."""
if user_input is not None:
self._config[CONF_ARRIVAL_TIME] = user_input[CONF_ARRIVAL_TIME]
return self.async_create_entry(title="", data=self._config)
options = {"arrival_time": selector({TimeSelector.selector_type: {}})}
return self.async_show_form(
step_id="arrival_time", data_schema=vol.Schema(options)
)
async def async_step_departure_time(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Configure departure time."""
if user_input is not None:
self._config[CONF_DEPARTURE_TIME] = user_input[CONF_DEPARTURE_TIME]
return self.async_create_entry(title="", data=self._config)
options = {"departure_time": selector({TimeSelector.selector_type: {}})}
return self.async_show_form(
step_id="departure_time", data_schema=vol.Schema(options)
)

15
homeassistant/components/here_travel_time/const.py
View file

@ -8,20 +8,19 @@ from homeassistant.const import (
DOMAIN = "here_travel_time"
DEFAULT_SCAN_INTERVAL = 300
CONF_DESTINATION = "destination"
CONF_ORIGIN = "origin"
CONF_DESTINATION_LATITUDE = "destination_latitude"
CONF_DESTINATION_LONGITUDE = "destination_longitude"
CONF_DESTINATION_ENTITY_ID = "destination_entity_id"
CONF_ORIGIN_LATITUDE = "origin_latitude"
CONF_ORIGIN_LONGITUDE = "origin_longitude"
CONF_ORIGIN_ENTITY_ID = "origin_entity_id"
CONF_TRAFFIC_MODE = "traffic_mode"
CONF_ROUTE_MODE = "route_mode"
CONF_ARRIVAL = "arrival"
CONF_DEPARTURE = "departure"
CONF_ARRIVAL_TIME = "arrival_time"
CONF_DEPARTURE_TIME = "departure_time"
CONF_TIME_TYPE = "time_type"
CONF_TIME = "time"
ARRIVAL_TIME = "Arrival Time"
DEPARTURE_TIME = "Departure Time"
TIME_TYPES = [ARRIVAL_TIME, DEPARTURE_TIME]
DEFAULT_NAME = "HERE Travel Time"

1
homeassistant/components/here_travel_time/manifest.json
View file

@ -1,6 +1,7 @@
{
"domain": "here_travel_time",
"name": "HERE Travel Time",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/here_travel_time",
"requirements": ["herepy==2.0.0"],
"codeowners": ["@eifinger"],

8
homeassistant/components/here_travel_time/model.py
View file

@ -24,10 +24,12 @@ class HERERoutingData(TypedDict):
class HERETravelTimeConfig:
"""Configuration for HereTravelTimeDataUpdateCoordinator."""
origin: str | None
destination: str | None
origin_entity_id: str | None
destination_latitude: float | None
destination_longitude: float | None
destination_entity_id: str | None
origin_latitude: float | None
origin_longitude: float | None
origin_entity_id: str | None
travel_mode: str
route_mode: str
units: str

191
homeassistant/components/here_travel_time/sensor.py
View file

@ -4,11 +4,10 @@ from __future__ import annotations
from datetime import timedelta
import logging
import herepy
from herepy.here_enum import RouteMode
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
ATTR_ATTRIBUTION,
ATTR_MODE,
@ -16,8 +15,6 @@ from homeassistant.const import (
CONF_MODE,
CONF_NAME,
CONF_UNIT_SYSTEM,
CONF_UNIT_SYSTEM_IMPERIAL,
CONF_UNIT_SYSTEM_METRIC,
TIME_MINUTES,
)
from homeassistant.core import HomeAssistant
@ -28,74 +25,47 @@ from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import HereTravelTimeDataUpdateCoordinator
from .model import HERETravelTimeConfig
_LOGGER = logging.getLogger(__name__)
CONF_DESTINATION_LATITUDE = "destination_latitude"
CONF_DESTINATION_LONGITUDE = "destination_longitude"
CONF_DESTINATION_ENTITY_ID = "destination_entity_id"
CONF_ORIGIN_LATITUDE = "origin_latitude"
CONF_ORIGIN_LONGITUDE = "origin_longitude"
CONF_ORIGIN_ENTITY_ID = "origin_entity_id"
CONF_TRAFFIC_MODE = "traffic_mode"
CONF_ROUTE_MODE = "route_mode"
CONF_ARRIVAL = "arrival"
CONF_DEPARTURE = "departure"
DEFAULT_NAME = "HERE Travel Time"
TRAVEL_MODE_BICYCLE = "bicycle"
TRAVEL_MODE_CAR = "car"
TRAVEL_MODE_PEDESTRIAN = "pedestrian"
TRAVEL_MODE_PUBLIC = "publicTransport"
TRAVEL_MODE_PUBLIC_TIME_TABLE = "publicTransportTimeTable"
TRAVEL_MODE_TRUCK = "truck"
TRAVEL_MODE = [
from .const import (
ATTR_DURATION,
ATTR_DURATION_IN_TRAFFIC,
ATTR_TRAFFIC_MODE,
ATTR_UNIT_SYSTEM,
CONF_ARRIVAL,
CONF_DEPARTURE,
CONF_DESTINATION_ENTITY_ID,
CONF_DESTINATION_LATITUDE,
CONF_DESTINATION_LONGITUDE,
CONF_ORIGIN_ENTITY_ID,
CONF_ORIGIN_LATITUDE,
CONF_ORIGIN_LONGITUDE,
CONF_ROUTE_MODE,
CONF_TRAFFIC_MODE,
DEFAULT_NAME,
DOMAIN,
ICON_BICYCLE,
ICON_CAR,
ICON_PEDESTRIAN,
ICON_PUBLIC,
ICON_TRUCK,
ROUTE_MODE_FASTEST,
ROUTE_MODES,
TRAFFIC_MODE_ENABLED,
TRAVEL_MODE_BICYCLE,
TRAVEL_MODE_CAR,
TRAVEL_MODE_PEDESTRIAN,
TRAVEL_MODE_PUBLIC,
TRAVEL_MODE_PUBLIC_TIME_TABLE,
TRAVEL_MODE_TRUCK,
]
TRAVEL_MODES,
TRAVEL_MODES_PUBLIC,
UNITS,
)
TRAVEL_MODES_PUBLIC = [TRAVEL_MODE_PUBLIC, TRAVEL_MODE_PUBLIC_TIME_TABLE]
TRAVEL_MODES_VEHICLE = [TRAVEL_MODE_CAR, TRAVEL_MODE_TRUCK]
TRAVEL_MODES_NON_VEHICLE = [TRAVEL_MODE_BICYCLE, TRAVEL_MODE_PEDESTRIAN]
_LOGGER = logging.getLogger(__name__)
TRAFFIC_MODE_ENABLED = "traffic_enabled"
TRAFFIC_MODE_DISABLED = "traffic_disabled"
ROUTE_MODE_FASTEST = "fastest"
ROUTE_MODE_SHORTEST = "shortest"
ROUTE_MODE = [ROUTE_MODE_FASTEST, ROUTE_MODE_SHORTEST]
ICON_BICYCLE = "mdi:bike"
ICON_CAR = "mdi:car"
ICON_PEDESTRIAN = "mdi:walk"
ICON_PUBLIC = "mdi:bus"
ICON_TRUCK = "mdi:truck"
UNITS = [CONF_UNIT_SYSTEM_METRIC, CONF_UNIT_SYSTEM_IMPERIAL]
ATTR_DURATION = "duration"
ATTR_DISTANCE = "distance"
ATTR_ROUTE = "route"
ATTR_ORIGIN = "origin"
ATTR_DESTINATION = "destination"
ATTR_UNIT_SYSTEM = CONF_UNIT_SYSTEM
ATTR_TRAFFIC_MODE = CONF_TRAFFIC_MODE
ATTR_DURATION_IN_TRAFFIC = "duration_in_traffic"
ATTR_ORIGIN_NAME = "origin_name"
ATTR_DESTINATION_NAME = "destination_name"
SCAN_INTERVAL = timedelta(minutes=5)
NO_ROUTE_ERROR_MESSAGE = "HERE could not find a route based on the input"
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_API_KEY): cv.string,
@ -113,8 +83,8 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
vol.Exclusive(CONF_ORIGIN_ENTITY_ID, "origin"): cv.entity_id,
vol.Optional(CONF_DEPARTURE): cv.time,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_MODE, default=TRAVEL_MODE_CAR): vol.In(TRAVEL_MODE),
vol.Optional(CONF_ROUTE_MODE, default=ROUTE_MODE_FASTEST): vol.In(ROUTE_MODE),
vol.Optional(CONF_MODE, default=TRAVEL_MODE_CAR): vol.In(TRAVEL_MODES),
vol.Optional(CONF_ROUTE_MODE, default=ROUTE_MODE_FASTEST): vol.In(ROUTE_MODES),
vol.Optional(CONF_TRAFFIC_MODE, default=False): cv.boolean,
vol.Optional(CONF_UNIT_SYSTEM): vol.In(UNITS),
}
@ -150,79 +120,36 @@ async def async_setup_platform(
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the HERE travel time platform."""
api_key = config[CONF_API_KEY]
here_client = herepy.RoutingApi(api_key)
if not await hass.async_add_executor_job(
_are_valid_client_credentials, here_client
):
_LOGGER.error(
"Invalid credentials. This error is returned if the specified token was invalid or no contract could be found for this token"
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=config,
)
return
if config.get(CONF_ORIGIN_LATITUDE) is not None:
origin = f"{config[CONF_ORIGIN_LATITUDE]},{config[CONF_ORIGIN_LONGITUDE]}"
origin_entity_id = None
else:
origin = None
origin_entity_id = config[CONF_ORIGIN_ENTITY_ID]
if config.get(CONF_DESTINATION_LATITUDE) is not None:
destination = (
f"{config[CONF_DESTINATION_LATITUDE]},{config[CONF_DESTINATION_LONGITUDE]}"
)
destination_entity_id = None
else:
destination = None
destination_entity_id = config[CONF_DESTINATION_ENTITY_ID]
traffic_mode = config[CONF_TRAFFIC_MODE]
name = config[CONF_NAME]
here_travel_time_config = HERETravelTimeConfig(
origin=origin,
destination=destination,
origin_entity_id=origin_entity_id,
destination_entity_id=destination_entity_id,
travel_mode=config[CONF_MODE],
route_mode=config[CONF_ROUTE_MODE],
units=config.get(CONF_UNIT_SYSTEM, hass.config.units.name),
arrival=config.get(CONF_ARRIVAL),
departure=config.get(CONF_DEPARTURE),
)
coordinator = HereTravelTimeDataUpdateCoordinator(
hass,
here_client,
here_travel_time_config,
_LOGGER.warning(
"Your HERE travel time configuration has been imported into the UI; "
"please remove it from configuration.yaml as support for it will be "
"removed in a future release"
)
sensor = HERETravelTimeSensor(name, traffic_mode, coordinator)
async_add_entities([sensor])
def _are_valid_client_credentials(here_client: herepy.RoutingApi) -> bool:
"""Check if the provided credentials are correct using defaults."""
known_working_origin = [38.9, -77.04833]
known_working_destination = [39.0, -77.1]
try:
here_client.public_transport_timetable(
known_working_origin,
known_working_destination,
True,
[
RouteMode[ROUTE_MODE_FASTEST],
RouteMode[TRAVEL_MODE_CAR],
RouteMode[TRAFFIC_MODE_ENABLED],
],
arrival=None,
departure="now",
)
except herepy.InvalidCredentialsError:
return False
return True
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Add HERE travel time entities from a config_entry."""
async_add_entities(
[
HERETravelTimeSensor(
config_entry.data[CONF_NAME],
config_entry.options[CONF_TRAFFIC_MODE],
hass.data[DOMAIN][config_entry.entry_id],
)
],
)
class HERETravelTimeSensor(SensorEntity, CoordinatorEntity):
@ -231,12 +158,12 @@ class HERETravelTimeSensor(SensorEntity, CoordinatorEntity):
def __init__(
self,
name: str,
traffic_mode: bool,
traffic_mode: str,
coordinator: HereTravelTimeDataUpdateCoordinator,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self._traffic_mode = traffic_mode
self._traffic_mode = traffic_mode == TRAFFIC_MODE_ENABLED
self._attr_native_unit_of_measurement = TIME_MINUTES
self._attr_name = name

82
homeassistant/components/here_travel_time/strings.json Normal file
View file

@ -0,0 +1,82 @@
{
"config": {
"step": {
"user": {
"data": {
"name": "[%key:common::config_flow::data::name%]",
"api_key": "[%key:common::config_flow::data::api_key%]",
"mode": "Travel Mode"
}
},
"origin_coordinates": {
"title": "Choose Origin",
"data": {
"origin": "Origin as GPS coordinates"
}
},
"origin_entity_id": {
"title": "Choose Origin",
"data": {
"origin_entity_id": "Origin using an entity"
}
},
"destination_menu": {
"title": "Choose Destination",
"menu_options": {
"destination_coordinates": "Using a map location",
"destination_entity": "Using an entity"
}
},
"destination_coordinates": {
"title": "Choose Destination",
"data": {
"destination": "Destination as GPS coordinates"
}
},
"destination_entity_id": {
"title": "Choose Destination",
"data": {
"destination_entity_id": "Destination using an entity"
}
}
},
"error": {
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
},
"options": {
"step": {
"init": {
"data": {
"traffic_mode": "Traffic Mode",
"route_mode": "Route Mode",
"unit_system": "Unit system"
}
},
"time_menu": {
"title": "Choose Time Type",
"menu_options": {
"departure_time": "Configure a departure time",
"arrival_time": "Configure an arrival time",
"no_time": "Do not configure a time"
}
},
"departure_time": {
"title": "Choose Departure Time",
"data": {
"departure_time": "Departure Time"
}
},
"arrival_time": {
"title": "Choose Arrival Time",
"data": {
"arrival_time": "Arrival Time"
}
}
}
}
}

1
homeassistant/generated/config_flows.py
View file

@ -143,6 +143,7 @@ FLOWS = {
"hangouts",
"harmony",
"heos",
"here_travel_time",
"hisense_aehw4a1",
"hive",
"hlk_sw16",

15
tests/components/here_travel_time/conftest.py
View file

@ -12,6 +12,11 @@ RESPONSE = RoutingResponse.new_from_jsondict(
)
RESPONSE.route_short = "US-29 - K St NW; US-29 - Whitehurst Fwy; I-495 N - Capital Beltway; MD-187 S - Old Georgetown Rd"
EMPTY_ATTRIBUTION_RESPONSE = RoutingResponse.new_from_jsondict(
json.loads(load_fixture("here_travel_time/empty_attribution_response.json"))
)
EMPTY_ATTRIBUTION_RESPONSE.route_short = "US-29 - K St NW; US-29 - Whitehurst Fwy; I-495 N - Capital Beltway; MD-187 S - Old Georgetown Rd"
@pytest.fixture(name="valid_response")
def valid_response_fixture():
@ -21,3 +26,13 @@ def valid_response_fixture():
return_value=RESPONSE,
) as mock:
yield mock
@pytest.fixture(name="empty_attribution_response")
def empty_attribution_response_fixture():
"""Return valid api response with an empty attribution."""
with patch(
"herepy.RoutingApi.public_transport_timetable",
return_value=EMPTY_ATTRIBUTION_RESPONSE,
) as mock:
yield mock

131
tests/components/here_travel_time/fixtures/empty_attribution_response.json Normal file
View file

@ -0,0 +1,131 @@
{
"response": {
"metaInfo": {
"timestamp": "2019-07-19T07:38:39Z",
"mapVersion": "8.30.98.154",
"moduleVersion": "7.2.201928-4446",
"interfaceVersion": "2.6.64",
"availableMapVersion": ["8.30.98.154"]
},
"route": [
{
"waypoint": [
{
"linkId": "+732182239",
"mappedPosition": {
"latitude": 38.9,
"longitude": -77.0488358
},
"originalPosition": {
"latitude": 38.9,
"longitude": -77.0483301
},
"type": "stopOver",
"spot": 0.4946237,
"sideOfStreet": "right",
"mappedRoadName": "22nd St NW",
"label": "22nd St NW",
"shapeIndex": 0,
"source": "user"
},
{
"linkId": "+942865877",
"mappedPosition": {
"latitude": 38.9999735,
"longitude": -77.100141
},
"originalPosition": {
"latitude": 38.9999999,
"longitude": -77.1000001
},
"type": "stopOver",
"spot": 1,
"sideOfStreet": "left",
"mappedRoadName": "Service Rd S",
"label": "Service Rd S",
"shapeIndex": 279,
"source": "user"
}
],
"mode": {
"type": "fastest",
"transportModes": ["car"],
"trafficMode": "enabled",
"feature": []
},
"leg": [
{
"start": {
"linkId": "+732182239",
"mappedPosition": {
"latitude": 38.9,
"longitude": -77.0488358
},
"originalPosition": {
"latitude": 38.9,
"longitude": -77.0483301
},
"type": "stopOver",
"spot": 0.4946237,
"sideOfStreet": "right",
"mappedRoadName": "22nd St NW",
"label": "22nd St NW",
"shapeIndex": 0,
"source": "user"
},
"end": {
"linkId": "+942865877",
"mappedPosition": {
"latitude": 38.9999735,
"longitude": -77.100141
},
"originalPosition": {
"latitude": 38.9999999,
"longitude": -77.1000001
},
"type": "stopOver",
"spot": 1,
"sideOfStreet": "left",
"mappedRoadName": "Service Rd S",
"label": "Service Rd S",
"shapeIndex": 279,
"source": "user"
},
"length": 23903,
"travelTime": 1884,
"maneuver": [
{
"position": {
"latitude": 38.9999735,
"longitude": -77.100141
},
"instruction": "Arrive at <span class=\"street\">Service Rd S</span>. Your destination is on the left.",
"travelTime": 0,
"length": 0,
"id": "M16",
"_type": "PrivateTransportManeuverType"
}
]
}
],
"summary": {
"distance": 23903,
"trafficTime": 1861,
"baseTime": 1803,
"flags": [
"noThroughRoad",
"motorway",
"builtUpArea",
"park",
"privateRoad"
],
"text": "The trip takes <span class=\"length\">23.9 km</span> and <span class=\"time\">31 mins</span>.",
"travelTime": 1861,
"_type": "RouteSummaryType"
}
}
],
"language": "en-us",
"sourceAttribution": {}
}
}

589
tests/components/here_travel_time/test_config_flow.py Normal file
View file

@ -0,0 +1,589 @@
"""Test the HERE Travel Time config flow."""
from unittest.mock import patch
from herepy import HEREError
from herepy.routing_api import InvalidCredentialsError
import pytest
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.here_travel_time.const import (
CONF_ARRIVAL,
CONF_ARRIVAL_TIME,
CONF_DEPARTURE,
CONF_DEPARTURE_TIME,
CONF_ROUTE_MODE,
CONF_TRAFFIC_MODE,
DOMAIN,
ROUTE_MODE_FASTEST,
TRAFFIC_MODE_ENABLED,
TRAVEL_MODE_CAR,
TRAVEL_MODE_PUBLIC_TIME_TABLE,
)
from homeassistant.components.here_travel_time.sensor import (
CONF_DESTINATION_ENTITY_ID,
CONF_DESTINATION_LATITUDE,
CONF_DESTINATION_LONGITUDE,
CONF_ORIGIN_ENTITY_ID,
CONF_ORIGIN_LATITUDE,
CONF_ORIGIN_LONGITUDE,
)
from homeassistant.const import (
CONF_API_KEY,
CONF_MODE,
CONF_NAME,
CONF_UNIT_SYSTEM,
CONF_UNIT_SYSTEM_IMPERIAL,
CONF_UNIT_SYSTEM_METRIC,
)
from homeassistant.core import HomeAssistant
from .const import (
API_KEY,
CAR_DESTINATION_LATITUDE,
CAR_DESTINATION_LONGITUDE,
CAR_ORIGIN_LATITUDE,
CAR_ORIGIN_LONGITUDE,
)
from tests.common import MockConfigEntry
@pytest.fixture(name="user_step_result")
async def user_step_result_fixture(hass: HomeAssistant) -> data_entry_flow.FlowResult:
"""Provide the result of a completed user step."""
init_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
user_step_result = await hass.config_entries.flow.async_configure(
init_result["flow_id"],
{
CONF_API_KEY: API_KEY,
CONF_MODE: TRAVEL_MODE_CAR,
CONF_NAME: "test",
},
)
await hass.async_block_till_done()
yield user_step_result
@pytest.fixture(name="option_init_result")
async def option_init_result_fixture(hass: HomeAssistant) -> data_entry_flow.FlowResult:
"""Provide the result of a completed options init step."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="0123456789",
data={
CONF_ORIGIN_LATITUDE: float(CAR_ORIGIN_LATITUDE),
CONF_ORIGIN_LONGITUDE: float(CAR_ORIGIN_LONGITUDE),
CONF_DESTINATION_LATITUDE: float(CAR_DESTINATION_LATITUDE),
CONF_DESTINATION_LONGITUDE: float(CAR_DESTINATION_LONGITUDE),
CONF_API_KEY: API_KEY,
CONF_MODE: TRAVEL_MODE_PUBLIC_TIME_TABLE,
CONF_NAME: "test",
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
flow = await hass.config_entries.options.async_init(entry.entry_id)
result = await hass.config_entries.options.async_configure(
flow["flow_id"],
user_input={
CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED,
CONF_ROUTE_MODE: ROUTE_MODE_FASTEST,
CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC,
},
)
yield result
@pytest.fixture(name="origin_step_result")
async def origin_step_result_fixture(
hass: HomeAssistant, user_step_result: data_entry_flow.FlowResult
) -> data_entry_flow.FlowResult:
"""Provide the result of a completed origin by coordinates step."""
origin_menu_result = await hass.config_entries.flow.async_configure(
user_step_result["flow_id"], {"next_step_id": "origin_coordinates"}
)
location_selector_result = await hass.config_entries.flow.async_configure(
origin_menu_result["flow_id"],
{
"origin": {
"latitude": float(CAR_ORIGIN_LATITUDE),
"longitude": float(CAR_ORIGIN_LONGITUDE),
"radius": 3.0,
}
},
)
yield location_selector_result
@pytest.mark.parametrize(
"menu_options",
(["origin_coordinates", "origin_entity"],),
)
@pytest.mark.usefixtures("valid_response")
async def test_step_user(hass: HomeAssistant, menu_options) -> None:
"""Test the user step."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {}
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_API_KEY: API_KEY,
CONF_MODE: TRAVEL_MODE_CAR,
CONF_NAME: "test",
},
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_MENU
assert result2["menu_options"] == menu_options
@pytest.mark.usefixtures("valid_response")
async def test_step_origin_coordinates(
hass: HomeAssistant, user_step_result: data_entry_flow.FlowResult
) -> None:
"""Test the origin coordinates step."""
menu_result = await hass.config_entries.flow.async_configure(
user_step_result["flow_id"], {"next_step_id": "origin_coordinates"}
)
assert menu_result["type"] == data_entry_flow.RESULT_TYPE_FORM
location_selector_result = await hass.config_entries.flow.async_configure(
menu_result["flow_id"],
{
"origin": {
"latitude": float(CAR_ORIGIN_LATITUDE),
"longitude": float(CAR_ORIGIN_LONGITUDE),
"radius": 3.0,
}
},
)
assert location_selector_result["type"] == data_entry_flow.RESULT_TYPE_MENU
@pytest.mark.usefixtures("valid_response")
async def test_step_origin_entity(
hass: HomeAssistant, user_step_result: data_entry_flow.FlowResult
) -> None:
"""Test the origin coordinates step."""
menu_result = await hass.config_entries.flow.async_configure(
user_step_result["flow_id"], {"next_step_id": "origin_entity"}
)
assert menu_result["type"] == data_entry_flow.RESULT_TYPE_FORM
entity_selector_result = await hass.config_entries.flow.async_configure(
menu_result["flow_id"],
{"origin_entity_id": "zone.home"},
)
assert entity_selector_result["type"] == data_entry_flow.RESULT_TYPE_MENU
@pytest.mark.usefixtures("valid_response")
async def test_step_destination_coordinates(
hass: HomeAssistant, origin_step_result: data_entry_flow.FlowResult
) -> None:
"""Test the origin coordinates step."""
menu_result = await hass.config_entries.flow.async_configure(
origin_step_result["flow_id"], {"next_step_id": "destination_coordinates"}
)
assert menu_result["type"] == data_entry_flow.RESULT_TYPE_FORM
location_selector_result = await hass.config_entries.flow.async_configure(
menu_result["flow_id"],
{
"destination": {
"latitude": float(CAR_DESTINATION_LATITUDE),
"longitude": float(CAR_DESTINATION_LONGITUDE),
"radius": 3.0,
}
},
)
assert location_selector_result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
entry = hass.config_entries.async_entries(DOMAIN)[0]
assert entry.data == {
CONF_NAME: "test",
CONF_API_KEY: API_KEY,
CONF_ORIGIN_LATITUDE: float(CAR_ORIGIN_LATITUDE),
CONF_ORIGIN_LONGITUDE: float(CAR_ORIGIN_LONGITUDE),
CONF_DESTINATION_LATITUDE: float(CAR_DESTINATION_LATITUDE),
CONF_DESTINATION_LONGITUDE: float(CAR_DESTINATION_LONGITUDE),
CONF_MODE: TRAVEL_MODE_CAR,
}
@pytest.mark.usefixtures("valid_response")
async def test_step_destination_entity(
hass: HomeAssistant, origin_step_result: data_entry_flow.FlowResult
) -> None:
"""Test the origin coordinates step."""
menu_result = await hass.config_entries.flow.async_configure(
origin_step_result["flow_id"], {"next_step_id": "destination_entity"}
)
assert menu_result["type"] == data_entry_flow.RESULT_TYPE_FORM
entity_selector_result = await hass.config_entries.flow.async_configure(
menu_result["flow_id"],
{"destination_entity_id": "zone.home"},
)
assert entity_selector_result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
entry = hass.config_entries.async_entries(DOMAIN)[0]
assert entry.data == {
CONF_NAME: "test",
CONF_API_KEY: API_KEY,
CONF_ORIGIN_LATITUDE: float(CAR_ORIGIN_LATITUDE),
CONF_ORIGIN_LONGITUDE: float(CAR_ORIGIN_LONGITUDE),
CONF_DESTINATION_ENTITY_ID: "zone.home",
CONF_MODE: TRAVEL_MODE_CAR,
}
async def test_form_invalid_auth(hass: HomeAssistant) -> None:
"""Test we handle invalid auth."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"herepy.RoutingApi.public_transport_timetable",
side_effect=InvalidCredentialsError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_API_KEY: API_KEY,
CONF_MODE: TRAVEL_MODE_CAR,
CONF_NAME: "test",
},
)
assert result2["type"] == "form"
assert result2["errors"] == {"base": "invalid_auth"}
async def test_form_unknown_error(hass: HomeAssistant) -> None:
"""Test we handle invalid auth."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"herepy.RoutingApi.public_transport_timetable",
side_effect=HEREError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_API_KEY: API_KEY,
CONF_MODE: TRAVEL_MODE_CAR,
CONF_NAME: "test",
},
)
assert result2["type"] == "form"
assert result2["errors"] == {"base": "unknown"}
@pytest.mark.usefixtures("valid_response")
async def test_options_flow(hass: HomeAssistant) -> None:
"""Test the options flow."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="0123456789",
data={
CONF_ORIGIN_LATITUDE: float(CAR_ORIGIN_LATITUDE),
CONF_ORIGIN_LONGITUDE: float(CAR_ORIGIN_LONGITUDE),
CONF_DESTINATION_LATITUDE: float(CAR_DESTINATION_LATITUDE),
CONF_DESTINATION_LONGITUDE: float(CAR_DESTINATION_LONGITUDE),
CONF_API_KEY: API_KEY,
CONF_MODE: TRAVEL_MODE_CAR,
CONF_NAME: "test",
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED,
CONF_ROUTE_MODE: ROUTE_MODE_FASTEST,
CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_IMPERIAL,
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_MENU
@pytest.mark.usefixtures("valid_response")
async def test_options_flow_arrival_time_step(
hass: HomeAssistant, option_init_result: data_entry_flow.FlowResult
) -> None:
"""Test the options flow arrival time type."""
menu_result = await hass.config_entries.options.async_configure(
option_init_result["flow_id"], {"next_step_id": "arrival_time"}
)
assert menu_result["type"] == data_entry_flow.RESULT_TYPE_FORM
time_selector_result = await hass.config_entries.options.async_configure(
option_init_result["flow_id"],
user_input={
"arrival_time": "08:00:00",
},
)
assert time_selector_result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
entry = hass.config_entries.async_entries(DOMAIN)[0]
assert entry.options == {
CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC,
CONF_ROUTE_MODE: ROUTE_MODE_FASTEST,
CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED,
CONF_ARRIVAL_TIME: "08:00:00",
}
@pytest.mark.usefixtures("valid_response")
async def test_options_flow_departure_time_step(
hass: HomeAssistant, option_init_result: data_entry_flow.FlowResult
) -> None:
"""Test the options flow departure time type."""
menu_result = await hass.config_entries.options.async_configure(
option_init_result["flow_id"], {"next_step_id": "departure_time"}
)
assert menu_result["type"] == data_entry_flow.RESULT_TYPE_FORM
time_selector_result = await hass.config_entries.options.async_configure(
option_init_result["flow_id"],
user_input={
"departure_time": "08:00:00",
},
)
assert time_selector_result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
entry = hass.config_entries.async_entries(DOMAIN)[0]
assert entry.options == {
CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC,
CONF_ROUTE_MODE: ROUTE_MODE_FASTEST,
CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED,
CONF_DEPARTURE_TIME: "08:00:00",
}
@pytest.mark.usefixtures("valid_response")
async def test_options_flow_no_time_step(
hass: HomeAssistant, option_init_result: data_entry_flow.FlowResult
) -> None:
"""Test the options flow arrival time type."""
menu_result = await hass.config_entries.options.async_configure(
option_init_result["flow_id"], {"next_step_id": "no_time"}
)
assert menu_result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
entry = hass.config_entries.async_entries(DOMAIN)[0]
assert entry.options == {
CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC,
CONF_ROUTE_MODE: ROUTE_MODE_FASTEST,
CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED,
}
@pytest.mark.usefixtures("valid_response")
async def test_import_flow_entity_id(hass: HomeAssistant) -> None:
"""Test import_flow with entity ids."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: CONF_API_KEY,
CONF_ORIGIN_ENTITY_ID: "sensor.origin",
CONF_DESTINATION_ENTITY_ID: "sensor.destination",
CONF_NAME: "test_name",
CONF_MODE: TRAVEL_MODE_CAR,
CONF_DEPARTURE: "08:00:00",
CONF_ROUTE_MODE: ROUTE_MODE_FASTEST,
CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED,
},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "test_name"
entry = hass.config_entries.async_entries(DOMAIN)[0]
assert entry.data == {
CONF_NAME: "test_name",
CONF_API_KEY: CONF_API_KEY,
CONF_ORIGIN_ENTITY_ID: "sensor.origin",
CONF_DESTINATION_ENTITY_ID: "sensor.destination",
CONF_MODE: TRAVEL_MODE_CAR,
}
assert entry.options == {
CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_ROUTE_MODE: ROUTE_MODE_FASTEST,
CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED,
CONF_DEPARTURE_TIME: "08:00:00",
CONF_ARRIVAL_TIME: None,
}
@pytest.mark.usefixtures("valid_response")
async def test_import_flow_coordinates(hass: HomeAssistant) -> None:
"""Test import_flow with coordinates."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: CONF_API_KEY,
CONF_ORIGIN_LATITUDE: CAR_ORIGIN_LATITUDE,
CONF_ORIGIN_LONGITUDE: CAR_ORIGIN_LONGITUDE,
CONF_DESTINATION_LATITUDE: CAR_DESTINATION_LATITUDE,
CONF_DESTINATION_LONGITUDE: CAR_DESTINATION_LONGITUDE,
CONF_NAME: "test_name",
CONF_MODE: TRAVEL_MODE_CAR,
CONF_ARRIVAL: "08:00:00",
CONF_ROUTE_MODE: ROUTE_MODE_FASTEST,
CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC,
CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED,
},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "test_name"
entry = hass.config_entries.async_entries(DOMAIN)[0]
assert entry.data == {
CONF_NAME: "test_name",
CONF_API_KEY: CONF_API_KEY,
CONF_ORIGIN_LATITUDE: CAR_ORIGIN_LATITUDE,
CONF_ORIGIN_LONGITUDE: CAR_ORIGIN_LONGITUDE,
CONF_DESTINATION_LATITUDE: CAR_DESTINATION_LATITUDE,
CONF_DESTINATION_LONGITUDE: CAR_DESTINATION_LONGITUDE,
CONF_MODE: TRAVEL_MODE_CAR,
}
assert entry.options == {
CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC,
CONF_ROUTE_MODE: ROUTE_MODE_FASTEST,
CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED,
CONF_DEPARTURE_TIME: None,
CONF_ARRIVAL_TIME: "08:00:00",
CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC,
}
@pytest.mark.usefixtures("valid_response")
async def test_dupe_import(hass: HomeAssistant) -> None:
"""Test duplicate import."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: CONF_API_KEY,
CONF_ORIGIN_LATITUDE: CAR_ORIGIN_LATITUDE,
CONF_ORIGIN_LONGITUDE: CAR_ORIGIN_LONGITUDE,
CONF_DESTINATION_LATITUDE: CAR_DESTINATION_LATITUDE,
CONF_DESTINATION_LONGITUDE: CAR_DESTINATION_LONGITUDE,
CONF_NAME: "test_name",
CONF_MODE: TRAVEL_MODE_CAR,
CONF_ARRIVAL: "08:00:00",
CONF_ROUTE_MODE: ROUTE_MODE_FASTEST,
CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC,
CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED,
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: CONF_API_KEY,
CONF_ORIGIN_LATITUDE: CAR_ORIGIN_LATITUDE,
CONF_ORIGIN_LONGITUDE: CAR_ORIGIN_LONGITUDE,
CONF_DESTINATION_LATITUDE: CAR_DESTINATION_LATITUDE,
CONF_DESTINATION_LONGITUDE: CAR_DESTINATION_LONGITUDE,
CONF_NAME: "test_name2",
CONF_MODE: TRAVEL_MODE_CAR,
CONF_ARRIVAL: "08:00:00",
CONF_ROUTE_MODE: ROUTE_MODE_FASTEST,
CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC,
CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED,
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: CONF_API_KEY,
CONF_ORIGIN_LATITUDE: CAR_ORIGIN_LATITUDE,
CONF_ORIGIN_LONGITUDE: CAR_ORIGIN_LONGITUDE,
CONF_DESTINATION_LATITUDE: CAR_DESTINATION_LATITUDE,
CONF_DESTINATION_LONGITUDE: CAR_DESTINATION_LONGITUDE,
CONF_NAME: "test_name",
CONF_MODE: TRAVEL_MODE_CAR,
CONF_ARRIVAL: "08:00:01",
CONF_ROUTE_MODE: ROUTE_MODE_FASTEST,
CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC,
CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED,
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: CONF_API_KEY,
CONF_ORIGIN_LATITUDE: CAR_ORIGIN_LATITUDE,
CONF_ORIGIN_LONGITUDE: CAR_ORIGIN_LONGITUDE,
CONF_DESTINATION_LATITUDE: "40.0",
CONF_DESTINATION_LONGITUDE: CAR_DESTINATION_LONGITUDE,
CONF_NAME: "test_name",
CONF_MODE: TRAVEL_MODE_CAR,
CONF_ARRIVAL: "08:00:01",
CONF_ROUTE_MODE: ROUTE_MODE_FASTEST,
CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC,
CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED,
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: CONF_API_KEY,
CONF_ORIGIN_LATITUDE: CAR_ORIGIN_LATITUDE,
CONF_ORIGIN_LONGITUDE: CAR_ORIGIN_LONGITUDE,
CONF_DESTINATION_LATITUDE: CAR_DESTINATION_LATITUDE,
CONF_DESTINATION_LONGITUDE: CAR_DESTINATION_LONGITUDE,
CONF_NAME: "test_name",
CONF_MODE: TRAVEL_MODE_CAR,
CONF_ARRIVAL: "08:00:00",
CONF_ROUTE_MODE: ROUTE_MODE_FASTEST,
CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC,
CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED,
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"

48
tests/components/here_travel_time/test_init.py Normal file
View file

@ -0,0 +1,48 @@
"""The test for the HERE Travel Time integration."""
import pytest
from homeassistant.components.here_travel_time.const import (
CONF_DESTINATION_LATITUDE,
CONF_DESTINATION_LONGITUDE,
CONF_ORIGIN_LATITUDE,
CONF_ORIGIN_LONGITUDE,
DOMAIN,
TRAVEL_MODE_CAR,
)
from homeassistant.const import CONF_API_KEY, CONF_MODE, CONF_NAME
from homeassistant.core import HomeAssistant
from .const import (
API_KEY,
CAR_DESTINATION_LATITUDE,
CAR_DESTINATION_LONGITUDE,
CAR_ORIGIN_LATITUDE,
CAR_ORIGIN_LONGITUDE,
)
from tests.common import MockConfigEntry
@pytest.mark.usefixtures("valid_response")
async def test_unload_entry(hass: HomeAssistant) -> None:
"""Test that unloading an entry works."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="0123456789",
data={
CONF_ORIGIN_LATITUDE: float(CAR_ORIGIN_LATITUDE),
CONF_ORIGIN_LONGITUDE: float(CAR_ORIGIN_LONGITUDE),
CONF_DESTINATION_LATITUDE: float(CAR_DESTINATION_LATITUDE),
CONF_DESTINATION_LONGITUDE: float(CAR_DESTINATION_LONGITUDE),
CONF_API_KEY: API_KEY,
CONF_MODE: TRAVEL_MODE_CAR,
CONF_NAME: "test",
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert await hass.config_entries.async_unload(entry.entry_id)
assert not hass.data[DOMAIN]

484
tests/components/here_travel_time/test_sensor.py
View file

@ -2,15 +2,10 @@
from unittest.mock import MagicMock, patch
from herepy.here_enum import RouteMode
from herepy.routing_api import InvalidCredentialsError, NoRouteFoundError
from herepy.routing_api import NoRouteFoundError
import pytest
from homeassistant.components.here_travel_time.const import (
ROUTE_MODE_FASTEST,
TRAFFIC_MODE_ENABLED,
)
from homeassistant.components.here_travel_time.sensor import (
ATTR_ATTRIBUTION,
ATTR_DESTINATION,
ATTR_DESTINATION_NAME,
ATTR_DISTANCE,
@ -19,16 +14,27 @@ from homeassistant.components.here_travel_time.sensor import (
ATTR_ORIGIN,
ATTR_ORIGIN_NAME,
ATTR_ROUTE,
CONF_MODE,
CONF_ARRIVAL_TIME,
CONF_DEPARTURE_TIME,
CONF_DESTINATION_ENTITY_ID,
CONF_DESTINATION_LATITUDE,
CONF_DESTINATION_LONGITUDE,
CONF_ORIGIN_ENTITY_ID,
CONF_ORIGIN_LATITUDE,
CONF_ORIGIN_LONGITUDE,
CONF_ROUTE_MODE,
CONF_TRAFFIC_MODE,
CONF_UNIT_SYSTEM,
DOMAIN,
ICON_BICYCLE,
ICON_CAR,
ICON_PEDESTRIAN,
ICON_PUBLIC,
ICON_TRUCK,
NO_ROUTE_ERROR_MESSAGE,
TIME_MINUTES,
ROUTE_MODE_FASTEST,
TRAFFIC_MODE_DISABLED,
TRAFFIC_MODE_ENABLED,
TRAVEL_MODE_BICYCLE,
TRAVEL_MODE_CAR,
TRAVEL_MODE_PEDESTRIAN,
@ -36,11 +42,20 @@ from homeassistant.components.here_travel_time.sensor import (
TRAVEL_MODE_TRUCK,
TRAVEL_MODES_VEHICLE,
)
from homeassistant.const import ATTR_ICON, EVENT_HOMEASSISTANT_START
from homeassistant.const import (
ATTR_ATTRIBUTION,
ATTR_ICON,
CONF_API_KEY,
CONF_MODE,
CONF_NAME,
EVENT_HOMEASSISTANT_START,
TIME_MINUTES,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from tests.components.here_travel_time.const import (
from .const import (
API_KEY,
CAR_DESTINATION_LATITUDE,
CAR_DESTINATION_LONGITUDE,
@ -48,21 +63,41 @@ from tests.components.here_travel_time.const import (
CAR_ORIGIN_LONGITUDE,
)
DOMAIN = "sensor"
PLATFORM = "here_travel_time"
from tests.common import MockConfigEntry
@pytest.mark.parametrize(
"mode,icon,traffic_mode,unit_system,expected_state,expected_distance,expected_duration_in_traffic",
"mode,icon,traffic_mode,unit_system,arrival_time,departure_time,expected_state,expected_distance,expected_duration_in_traffic",
[
(TRAVEL_MODE_CAR, ICON_CAR, True, "metric", "31", 23.903, 31.016666666666666),
(TRAVEL_MODE_BICYCLE, ICON_BICYCLE, False, "metric", "30", 23.903, 30.05),
(
TRAVEL_MODE_CAR,
ICON_CAR,
TRAFFIC_MODE_ENABLED,
"metric",
None,
None,
"31",
23.903,
31.016666666666666,
),
(
TRAVEL_MODE_BICYCLE,
ICON_BICYCLE,
TRAFFIC_MODE_DISABLED,
"metric",
None,
None,
"30",
23.903,
30.05,
),
(
TRAVEL_MODE_PEDESTRIAN,
ICON_PEDESTRIAN,
False,
TRAFFIC_MODE_DISABLED,
"imperial",
None,
None,
"30",
14.852635608048994,
30.05,
@ -70,8 +105,10 @@ PLATFORM = "here_travel_time"
(
TRAVEL_MODE_PUBLIC_TIME_TABLE,
ICON_PUBLIC,
False,
TRAFFIC_MODE_DISABLED,
"imperial",
"08:00:00",
None,
"30",
14.852635608048994,
30.05,
@ -79,41 +116,52 @@ PLATFORM = "here_travel_time"
(
TRAVEL_MODE_TRUCK,
ICON_TRUCK,
True,
TRAFFIC_MODE_ENABLED,
"metric",
None,
"08:00:00",
"31",
23.903,
31.016666666666666,
),
],
)
@pytest.mark.usefixtures("valid_response")
async def test_sensor(
hass,
hass: HomeAssistant,
mode,
icon,
traffic_mode,
unit_system,
arrival_time,
departure_time,
expected_state,
expected_distance,
expected_duration_in_traffic,
valid_response,
):
"""Test that sensor works."""
config = {
DOMAIN: {
"platform": PLATFORM,
"name": "test",
"origin_latitude": CAR_ORIGIN_LATITUDE,
"origin_longitude": CAR_ORIGIN_LONGITUDE,
"destination_latitude": CAR_DESTINATION_LATITUDE,
"destination_longitude": CAR_DESTINATION_LONGITUDE,
"api_key": API_KEY,
"traffic_mode": traffic_mode,
"unit_system": unit_system,
"mode": mode,
}
}
assert await async_setup_component(hass, DOMAIN, config)
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="0123456789",
data={
CONF_ORIGIN_LATITUDE: float(CAR_ORIGIN_LATITUDE),
CONF_ORIGIN_LONGITUDE: float(CAR_ORIGIN_LONGITUDE),
CONF_DESTINATION_LATITUDE: float(CAR_DESTINATION_LATITUDE),
CONF_DESTINATION_LONGITUDE: float(CAR_DESTINATION_LONGITUDE),
CONF_API_KEY: API_KEY,
CONF_MODE: mode,
CONF_NAME: "test",
},
options={
CONF_TRAFFIC_MODE: traffic_mode,
CONF_ROUTE_MODE: ROUTE_MODE_FASTEST,
CONF_ARRIVAL_TIME: arrival_time,
CONF_DEPARTURE_TIME: departure_time,
CONF_UNIT_SYSTEM: unit_system,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
await hass.async_block_till_done()
@ -145,7 +193,9 @@ async def test_sensor(
assert sensor.attributes.get(ATTR_ORIGIN_NAME) == "22nd St NW"
assert sensor.attributes.get(ATTR_DESTINATION_NAME) == "Service Rd S"
assert sensor.attributes.get(CONF_MODE) == mode
assert sensor.attributes.get(CONF_TRAFFIC_MODE) is traffic_mode
assert sensor.attributes.get(CONF_TRAFFIC_MODE) is (
traffic_mode == TRAFFIC_MODE_ENABLED
)
assert sensor.attributes.get(ATTR_ICON) == icon
@ -156,7 +206,63 @@ async def test_sensor(
)
async def test_entity_ids(hass, valid_response: MagicMock):
@pytest.mark.usefixtures("valid_response")
async def test_circular_ref(hass: HomeAssistant, caplog):
"""Test that a circular ref is handled."""
hass.states.async_set(
"test.first",
"test.second",
)
hass.states.async_set("test.second", "test.first")
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="0123456789",
data={
CONF_ORIGIN_ENTITY_ID: "test.first",
CONF_DESTINATION_LATITUDE: float(CAR_DESTINATION_LATITUDE),
CONF_DESTINATION_LONGITUDE: float(CAR_DESTINATION_LONGITUDE),
CONF_API_KEY: API_KEY,
CONF_MODE: TRAVEL_MODE_TRUCK,
CONF_NAME: "test",
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
await hass.async_block_till_done()
assert "No coordinatnes found for test.first" in caplog.text
@pytest.mark.usefixtures("empty_attribution_response")
async def test_no_attribution(hass: HomeAssistant):
"""Test that an empty attribution is handled."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="0123456789",
data={
CONF_ORIGIN_LATITUDE: float(CAR_ORIGIN_LATITUDE),
CONF_ORIGIN_LONGITUDE: float(CAR_ORIGIN_LONGITUDE),
CONF_DESTINATION_LATITUDE: float(CAR_DESTINATION_LATITUDE),
CONF_DESTINATION_LONGITUDE: float(CAR_DESTINATION_LONGITUDE),
CONF_API_KEY: API_KEY,
CONF_MODE: TRAVEL_MODE_TRUCK,
CONF_NAME: "test",
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
await hass.async_block_till_done()
assert hass.states.get("sensor.test").attributes.get(ATTR_ATTRIBUTION) is None
async def test_entity_ids(hass: HomeAssistant, valid_response: MagicMock):
"""Test that origin/destination supplied by entities works."""
utcnow = dt_util.utcnow()
# Patching 'utcnow' to gain more control over the timed update.
@ -181,17 +287,19 @@ async def test_entity_ids(hass, valid_response: MagicMock):
"longitude": float(CAR_DESTINATION_LONGITUDE),
},
)
config = {
DOMAIN: {
"platform": PLATFORM,
"name": "test",
"origin_entity_id": "zone.origin",
"destination_entity_id": "device_tracker.test",
"api_key": API_KEY,
"mode": TRAVEL_MODE_TRUCK,
}
}
assert await async_setup_component(hass, DOMAIN, config)
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="0123456789",
data={
CONF_ORIGIN_ENTITY_ID: "zone.origin",
CONF_DESTINATION_ENTITY_ID: "device_tracker.test",
CONF_API_KEY: API_KEY,
CONF_MODE: TRAVEL_MODE_TRUCK,
CONF_NAME: "test",
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
@ -214,20 +322,23 @@ async def test_entity_ids(hass, valid_response: MagicMock):
)
async def test_destination_entity_not_found(hass, caplog, valid_response: MagicMock):
@pytest.mark.usefixtures("valid_response")
async def test_destination_entity_not_found(hass: HomeAssistant, caplog):
"""Test that a not existing destination_entity_id is caught."""
config = {
DOMAIN: {
"platform": PLATFORM,
"name": "test",
"origin_latitude": CAR_ORIGIN_LATITUDE,
"origin_longitude": CAR_ORIGIN_LONGITUDE,
"destination_entity_id": "device_tracker.test",
"api_key": API_KEY,
"mode": TRAVEL_MODE_TRUCK,
}
}
assert await async_setup_component(hass, DOMAIN, config)
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="0123456789",
data={
CONF_ORIGIN_LATITUDE: float(CAR_ORIGIN_LATITUDE),
CONF_ORIGIN_LONGITUDE: float(CAR_ORIGIN_LONGITUDE),
CONF_DESTINATION_ENTITY_ID: "device_tracker.test",
CONF_API_KEY: API_KEY,
CONF_MODE: TRAVEL_MODE_TRUCK,
CONF_NAME: "test",
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
@ -236,20 +347,23 @@ async def test_destination_entity_not_found(hass, caplog, valid_response: MagicM
assert "device_tracker.test are not valid coordinates" in caplog.text
async def test_origin_entity_not_found(hass, caplog, valid_response: MagicMock):
@pytest.mark.usefixtures("valid_response")
async def test_origin_entity_not_found(hass: HomeAssistant, caplog):
"""Test that a not existing origin_entity_id is caught."""
config = {
DOMAIN: {
"platform": PLATFORM,
"name": "test",
"origin_entity_id": "device_tracker.test",
"destination_latitude": CAR_ORIGIN_LATITUDE,
"destination_longitude": CAR_ORIGIN_LONGITUDE,
"api_key": API_KEY,
"mode": TRAVEL_MODE_TRUCK,
}
}
assert await async_setup_component(hass, DOMAIN, config)
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="0123456789",
data={
CONF_ORIGIN_ENTITY_ID: "device_tracker.test",
CONF_DESTINATION_LATITUDE: float(CAR_DESTINATION_LATITUDE),
CONF_DESTINATION_LONGITUDE: float(CAR_DESTINATION_LONGITUDE),
CONF_API_KEY: API_KEY,
CONF_MODE: TRAVEL_MODE_TRUCK,
CONF_NAME: "test",
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
@ -258,26 +372,27 @@ async def test_origin_entity_not_found(hass, caplog, valid_response: MagicMock):
assert "device_tracker.test are not valid coordinates" in caplog.text
async def test_invalid_destination_entity_state(
hass, caplog, valid_response: MagicMock
):
@pytest.mark.usefixtures("valid_response")
async def test_invalid_destination_entity_state(hass: HomeAssistant, caplog):
"""Test that an invalid state of the destination_entity_id is caught."""
hass.states.async_set(
"device_tracker.test",
"test_state",
)
config = {
DOMAIN: {
"platform": PLATFORM,
"name": "test",
"origin_latitude": CAR_ORIGIN_LATITUDE,
"origin_longitude": CAR_ORIGIN_LONGITUDE,
"destination_entity_id": "device_tracker.test",
"api_key": API_KEY,
"mode": TRAVEL_MODE_TRUCK,
}
}
assert await async_setup_component(hass, DOMAIN, config)
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="0123456789",
data={
CONF_ORIGIN_LATITUDE: float(CAR_ORIGIN_LATITUDE),
CONF_ORIGIN_LONGITUDE: float(CAR_ORIGIN_LONGITUDE),
CONF_DESTINATION_ENTITY_ID: "device_tracker.test",
CONF_API_KEY: API_KEY,
CONF_MODE: TRAVEL_MODE_TRUCK,
CONF_NAME: "test",
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
@ -286,24 +401,27 @@ async def test_invalid_destination_entity_state(
assert "test_state are not valid coordinates" in caplog.text
async def test_invalid_origin_entity_state(hass, caplog, valid_response: MagicMock):
@pytest.mark.usefixtures("valid_response")
async def test_invalid_origin_entity_state(hass: HomeAssistant, caplog):
"""Test that an invalid state of the origin_entity_id is caught."""
hass.states.async_set(
"device_tracker.test",
"test_state",
)
config = {
DOMAIN: {
"platform": PLATFORM,
"name": "test",
"origin_entity_id": "device_tracker.test",
"destination_latitude": CAR_ORIGIN_LATITUDE,
"destination_longitude": CAR_ORIGIN_LONGITUDE,
"api_key": API_KEY,
"mode": TRAVEL_MODE_TRUCK,
}
}
assert await async_setup_component(hass, DOMAIN, config)
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="0123456789",
data={
CONF_ORIGIN_ENTITY_ID: "device_tracker.test",
CONF_DESTINATION_LATITUDE: float(CAR_DESTINATION_LATITUDE),
CONF_DESTINATION_LONGITUDE: float(CAR_DESTINATION_LONGITUDE),
CONF_API_KEY: API_KEY,
CONF_MODE: TRAVEL_MODE_TRUCK,
CONF_NAME: "test",
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
@ -312,27 +430,30 @@ async def test_invalid_origin_entity_state(hass, caplog, valid_response: MagicMo
assert "test_state are not valid coordinates" in caplog.text
async def test_route_not_found(hass, caplog):
async def test_route_not_found(hass: HomeAssistant, caplog):
"""Test that route not found error is correctly handled."""
config = {
DOMAIN: {
"platform": PLATFORM,
"name": "test",
"origin_latitude": CAR_ORIGIN_LATITUDE,
"origin_longitude": CAR_ORIGIN_LONGITUDE,
"destination_latitude": CAR_DESTINATION_LATITUDE,
"destination_longitude": CAR_DESTINATION_LONGITUDE,
"api_key": API_KEY,
}
}
with patch(
"homeassistant.components.here_travel_time.sensor._are_valid_client_credentials",
return_value=True,
"homeassistant.components.here_travel_time.config_flow.validate_api_key",
return_value=None,
), patch(
"herepy.RoutingApi.public_transport_timetable",
side_effect=NoRouteFoundError,
):
assert await async_setup_component(hass, DOMAIN, config)
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="0123456789",
data={
CONF_ORIGIN_LATITUDE: float(CAR_ORIGIN_LATITUDE),
CONF_ORIGIN_LONGITUDE: float(CAR_ORIGIN_LONGITUDE),
CONF_DESTINATION_LATITUDE: float(CAR_DESTINATION_LATITUDE),
CONF_DESTINATION_LONGITUDE: float(CAR_DESTINATION_LONGITUDE),
CONF_API_KEY: API_KEY,
CONF_MODE: TRAVEL_MODE_TRUCK,
CONF_NAME: "test",
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
await hass.async_block_till_done()
@ -340,113 +461,26 @@ async def test_route_not_found(hass, caplog):
assert NO_ROUTE_ERROR_MESSAGE in caplog.text
async def test_invalid_credentials(hass, caplog):
"""Test that invalid credentials error is correctly handled."""
@pytest.mark.usefixtures("valid_response")
async def test_setup_platform(hass: HomeAssistant, caplog):
"""Test that setup platform migration works."""
config = {
"sensor": {
"platform": DOMAIN,
"name": "test",
"origin_latitude": CAR_ORIGIN_LATITUDE,
"origin_longitude": CAR_ORIGIN_LONGITUDE,
"destination_latitude": CAR_DESTINATION_LATITUDE,
"destination_longitude": CAR_DESTINATION_LONGITUDE,
"api_key": API_KEY,
}
}
with patch(
"herepy.RoutingApi.public_transport_timetable",
side_effect=InvalidCredentialsError,
"homeassistant.components.here_travel_time.async_setup_entry", return_value=True
):
config = {
DOMAIN: {
"platform": PLATFORM,
"name": "test",
"origin_latitude": CAR_ORIGIN_LATITUDE,
"origin_longitude": CAR_ORIGIN_LONGITUDE,
"destination_latitude": CAR_DESTINATION_LATITUDE,
"destination_longitude": CAR_DESTINATION_LONGITUDE,
"api_key": API_KEY,
}
}
assert await async_setup_component(hass, DOMAIN, config)
await async_setup_component(hass, "sensor", config)
await hass.async_block_till_done()
assert "Invalid credentials" in caplog.text
async def test_arrival(hass, valid_response):
"""Test that arrival works."""
config = {
DOMAIN: {
"platform": PLATFORM,
"name": "test",
"origin_latitude": CAR_ORIGIN_LATITUDE,
"origin_longitude": CAR_ORIGIN_LONGITUDE,
"destination_latitude": CAR_DESTINATION_LATITUDE,
"destination_longitude": CAR_DESTINATION_LONGITUDE,
"api_key": API_KEY,
"mode": TRAVEL_MODE_PUBLIC_TIME_TABLE,
"arrival": "01:00:00",
}
}
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
await hass.async_block_till_done()
sensor = hass.states.get("sensor.test")
assert sensor.state == "30"
async def test_departure(hass, valid_response):
"""Test that departure works."""
config = {
DOMAIN: {
"platform": PLATFORM,
"name": "test",
"origin_latitude": CAR_ORIGIN_LATITUDE,
"origin_longitude": CAR_ORIGIN_LONGITUDE,
"destination_latitude": CAR_DESTINATION_LATITUDE,
"destination_longitude": CAR_DESTINATION_LONGITUDE,
"api_key": API_KEY,
"mode": TRAVEL_MODE_PUBLIC_TIME_TABLE,
"departure": "23:00:00",
}
}
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
await hass.async_block_till_done()
sensor = hass.states.get("sensor.test")
assert sensor.state == "30"
async def test_arrival_only_allowed_for_timetable(hass, caplog):
"""Test that arrival is only allowed when mode is publicTransportTimeTable."""
config = {
DOMAIN: {
"platform": PLATFORM,
"name": "test",
"origin_latitude": CAR_ORIGIN_LATITUDE,
"origin_longitude": CAR_ORIGIN_LONGITUDE,
"destination_latitude": CAR_DESTINATION_LATITUDE,
"destination_longitude": CAR_DESTINATION_LONGITUDE,
"api_key": API_KEY,
"arrival": "01:00:00",
}
}
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
assert "[arrival] is an invalid option" in caplog.text
async def test_exclusive_arrival_and_departure(hass, caplog):
"""Test that arrival and departure are exclusive."""
config = {
DOMAIN: {
"platform": PLATFORM,
"name": "test",
"origin_latitude": CAR_ORIGIN_LATITUDE,
"origin_longitude": CAR_ORIGIN_LONGITUDE,
"destination_latitude": CAR_DESTINATION_LATITUDE,
"destination_longitude": CAR_DESTINATION_LONGITUDE,
"api_key": API_KEY,
"arrival": "01:00:00",
"mode": TRAVEL_MODE_PUBLIC_TIME_TABLE,
"departure": "01:00:00",
}
}
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
assert "two or more values in the same group of exclusion" in caplog.text
assert (
"Your HERE travel time configuration has been imported into the UI"
in caplog.text
)