diff --git a/.strict-typing b/.strict-typing index 059ef850ab723f..180c77c946740e 100644 --- a/.strict-typing +++ b/.strict-typing @@ -55,6 +55,7 @@ homeassistant.components.goalzero.* homeassistant.components.greeneye_monitor.* homeassistant.components.group.* homeassistant.components.guardian.* +homeassistant.components.here_travel_time.* homeassistant.components.history.* homeassistant.components.homeassistant.triggers.event homeassistant.components.http.* diff --git a/homeassistant/components/here_travel_time/__init__.py b/homeassistant/components/here_travel_time/__init__.py index 9a5c8ec32aca43..359659044088b2 100644 --- a/homeassistant/components/here_travel_time/__init__.py +++ b/homeassistant/components/here_travel_time/__init__.py @@ -1 +1,226 @@ -"""The here_travel_time component.""" +"""The HERE Travel Time integration.""" +from __future__ import annotations + +from datetime import datetime, timedelta +import logging + +import async_timeout +from herepy import NoRouteFoundError, RouteMode, RoutingApi, RoutingResponse + +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import ( + CONF_API_KEY, + CONF_MODE, + CONF_UNIT_SYSTEM, + CONF_UNIT_SYSTEM_IMPERIAL, +) +from homeassistant.core import HomeAssistant +from homeassistant.helpers.location import find_coordinates +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed +from homeassistant.util import dt + +from .const import ( + ARRIVAL_TIME, + CONF_DESTINATION, + CONF_ORIGIN, + CONF_ROUTE_MODE, + CONF_TIME, + CONF_TIME_TYPE, + CONF_TRAFFIC_MODE, + DEFAULT_SCAN_INTERVAL, + DEPARTURE_TIME, + DOMAIN, + NO_ROUTE_ERROR_MESSAGE, + ROUTE_MODE_FASTEST, + TRACKABLE_DOMAINS, + TRAFFIC_MODE_ENABLED, + HERERoutingData, +) + +PLATFORMS = ["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.""" + here_data = HERETravelTimeData(hass, config_entry) + await here_data.async_setup() + hass.data.setdefault(DOMAIN, {})[config_entry.entry_id] = here_data + hass.config_entries.async_setup_platforms(config_entry, PLATFORMS) + + return True + + +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 HERETravelTimeData: + """HERETravelTime data object.""" + + def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None: + """Initialize.""" + self._hass = hass + self._config_entry = config_entry + self._api = RoutingApi(config_entry.data[CONF_API_KEY]) + self.coordinator: DataUpdateCoordinator[HERERoutingData | None] | None = None + + async def async_update(self) -> HERERoutingData | None: + """Get the latest data from the HERE Routing API.""" + try: + async with async_timeout.timeout(10): + return await self._hass.async_add_executor_job(self._update) + except NoRouteFoundError as error: + raise UpdateFailed(NO_ROUTE_ERROR_MESSAGE) from error + + async def async_setup(self) -> None: + """Set up the HERETravelTime integration.""" + if not self._config_entry.options: + options = { + CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED, + CONF_ROUTE_MODE: ROUTE_MODE_FASTEST, + CONF_TIME_TYPE: DEPARTURE_TIME, + CONF_UNIT_SYSTEM: self._hass.config.units.name, + CONF_TIME: "now", + } + self._hass.config_entries.async_update_entry( + self._config_entry, options=options + ) + self.coordinator = DataUpdateCoordinator( + self._hass, + _LOGGER, + name=DOMAIN, + update_method=self.async_update, + update_interval=timedelta(seconds=DEFAULT_SCAN_INTERVAL), + ) + await self.coordinator.async_config_entry_first_refresh() + + def _update(self) -> HERERoutingData | None: + """Get the latest data from the HERE Routing API.""" + if (destination := self._config_entry.data[CONF_DESTINATION]).split(".", 1)[ + 0 + ] in TRACKABLE_DOMAINS: + destination = find_coordinates(self._hass, destination) + + if (origin := self._config_entry.data[CONF_ORIGIN]).split(".", 1)[ + 0 + ] in TRACKABLE_DOMAINS: + origin = find_coordinates(self._hass, origin) + + if destination is not None and origin is not None: + # Convert location to HERE friendly location + destination = destination.split(",") + origin = origin.split(",") + arrival: str | None = None + departure: str | None = "now" + if self._config_entry.options[CONF_TIME_TYPE] == ARRIVAL_TIME: + if (conf_arrival := self._config_entry.options[CONF_TIME]) != "": + arrival = convert_time_to_isodate(conf_arrival) + if arrival is None: + arrival = "" + _LOGGER.warning( + "Supplied arrival time could not be parsed. It was ignored" + ) + if self._config_entry.options[CONF_TIME_TYPE] == DEPARTURE_TIME: + if (conf_departure := self._config_entry.options[CONF_TIME]) not in [ + "", + "now", + ]: + departure = convert_time_to_isodate(conf_departure) + if departure is None: + departure = "now" + _LOGGER.warning( + "Supplied departure time could not be parsed. It was ignored" + ) + + _LOGGER.debug( + "Requesting route for origin: %s, destination: %s, route_mode: %s, mode: %s, traffic_mode: %s, arrival: %s, departure: %s", + origin, + destination, + RouteMode[self._config_entry.options[CONF_ROUTE_MODE]], + RouteMode[self._config_entry.data[CONF_MODE]], + RouteMode[TRAFFIC_MODE_ENABLED], + arrival, + departure, + ) + + response: RoutingResponse = self._api.public_transport_timetable( + origin, + destination, + True, + [ + RouteMode[self._config_entry.options[CONF_ROUTE_MODE]], + RouteMode[self._config_entry.data[CONF_MODE]], + RouteMode[TRAFFIC_MODE_ENABLED], + ], + arrival=arrival, + departure=departure, + ) + + _LOGGER.debug( + "Raw response is: %s", response.response # pylint: disable=no-member + ) + + source_attribution = response.response.get( # pylint: disable=no-member + "sourceAttribution" + ) + attribution: str | None = None + if source_attribution is not None: + attribution = build_hass_attribution(source_attribution) + route: list = response.response["route"] # pylint: disable=no-member + summary: dict = route[0]["summary"] + waypoint: list = route[0]["waypoint"] + distance: float = summary["distance"] + if ( + self._config_entry.options[CONF_UNIT_SYSTEM] + == CONF_UNIT_SYSTEM_IMPERIAL + ): + # Convert to miles. + distance = distance / 1609.344 + else: + # Convert to kilometers + distance = distance / 1000 + return { + "attribution": attribution, + "base_time": summary["baseTime"], + "traffic_time": summary["trafficTime"], + "distance": distance, + "route": response.route_short, + "origin": ",".join(origin), + "destination": ",".join(destination), + "origin_name": waypoint[0]["mappedRoadName"], + "destination_name": waypoint[1]["mappedRoadName"], + } + return None + + +def build_hass_attribution(source_attribution: dict) -> str | None: + """Build a hass frontend ready string out of the sourceAttribution.""" + if (suppliers := source_attribution.get("supplier")) is not None: + supplier_titles = [] + for supplier in suppliers: + if (title := supplier.get("title")) is not None: + supplier_titles.append(title) + joined_supplier_titles = ",".join(supplier_titles) + attribution = f"With the support of {joined_supplier_titles}. All information is provided without warranty of any kind." + return attribution + return None + + +def convert_time_to_isodate(timestr: str) -> str | None: + """Take a string like 08:00:00 and combine it with the current date.""" + if (parsed_time := dt.parse_time(timestr)) is None: + return None + if ( + combined := datetime.combine(dt.start_of_local_day(), parsed_time) + ) < datetime.now(): + combined = combined + timedelta(days=1) + return combined.isoformat() diff --git a/homeassistant/components/here_travel_time/config_flow.py b/homeassistant/components/here_travel_time/config_flow.py new file mode 100644 index 00000000000000..af36a2a45fcae3 --- /dev/null +++ b/homeassistant/components/here_travel_time/config_flow.py @@ -0,0 +1,238 @@ +"""Config flow for HERE Travel Time integration.""" +from __future__ import annotations + +import logging +from typing import Any + +from herepy import InvalidCredentialsError, RouteMode, RoutingApi +import voluptuous as vol + +from homeassistant import config_entries +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 +from homeassistant.core import HomeAssistant, callback +from homeassistant.data_entry_flow import FlowResult +from homeassistant.exceptions import HomeAssistantError +import homeassistant.helpers.config_validation as cv + +from .const import ( + ARRIVAL_TIME, + CONF_ARRIVAL, + CONF_DEPARTURE, + CONF_DESTINATION, + CONF_ORIGIN, + CONF_ROUTE_MODE, + CONF_TIME, + CONF_TIME_TYPE, + CONF_TRAFFIC_MODE, + DEFAULT_NAME, + DEPARTURE_TIME, + DOMAIN, + ROUTE_MODE_FASTEST, + ROUTE_MODES, + TIME_TYPES, + TRAFFIC_MODE_DISABLED, + TRAFFIC_MODE_ENABLED, + TRAFFIC_MODES, + TRAVEL_MODE_CAR, + TRAVEL_MODES, + UNITS, +) + +_LOGGER = logging.getLogger(__name__) + + +def is_dupe_import( + hass: HomeAssistant, + 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( + entry.data[key] != user_input[key] + for key in (CONF_API_KEY, CONF_DESTINATION, CONF_ORIGIN, CONF_MODE, CONF_NAME) + ): + 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_TIME_TYPE, + CONF_TIME, + ): + if options.get(key) != entry.options.get(key): + return False + + return True + + +def validate_input(data: dict[str, Any]) -> None: + """Validate the user input allows us to connect.""" + known_working_origin = [38.9, -77.04833] + known_working_destination = [39.0, -77.1] + RoutingApi(data[CONF_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.""" + name = DEFAULT_NAME if data.get(CONF_NAME) is None else data.get(CONF_NAME) + mode = TRAVEL_MODE_CAR if data.get(CONF_MODE) is None else data.get(CONF_MODE) + + return vol.Schema( + { + vol.Optional(CONF_NAME, default=name): cv.string, + vol.Required(CONF_API_KEY, default=data.get(CONF_API_KEY)): cv.string, + vol.Required( + CONF_DESTINATION, default=data.get(CONF_DESTINATION) + ): cv.string, + vol.Required(CONF_ORIGIN, default=data.get(CONF_ORIGIN)): cv.string, + vol.Optional(CONF_MODE, default=mode): vol.In(TRAVEL_MODES), + } + ) + + +class HERETravelTimeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): + """Handle a config flow for HERE Travel Time.""" + + VERSION = 1 + + @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 = {} + options = {} + user_input = user_input or {} + if user_input: + if self.source == config_entries.SOURCE_IMPORT: + if user_input.get(CONF_ORIGIN_LATITUDE) is not None: + user_input[ + CONF_ORIGIN + ] = f"{user_input.pop(CONF_ORIGIN_LATITUDE)},{user_input.pop(CONF_ORIGIN_LONGITUDE)}" + else: + user_input[CONF_ORIGIN] = user_input.pop(CONF_ORIGIN_ENTITY_ID) + + if user_input.get(CONF_DESTINATION_LATITUDE) is not None: + user_input[ + CONF_DESTINATION + ] = f"{user_input.pop(CONF_DESTINATION_LATITUDE)},{user_input.pop(CONF_DESTINATION_LONGITUDE)}" + else: + user_input[CONF_DESTINATION] = 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_TIME_TYPE] = ( + ARRIVAL_TIME if CONF_ARRIVAL in user_input else DEPARTURE_TIME + ) + if (arrival_time := user_input.pop(CONF_ARRIVAL, None)) is not None: + options[CONF_TIME] = arrival_time + if (departure_time := user_input.pop(CONF_DEPARTURE, None)) is not None: + options[CONF_TIME] = departure_time + + # We need to prevent duplicate imports + if any( + is_dupe_import(self.hass, 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") + try: + await self.hass.async_add_executor_job(validate_input, user_input) + except InvalidCredentialsError: + errors["base"] = "invalid_auth" + except Exception: # pylint: disable=broad-except + _LOGGER.exception("Unexpected exception") + errors["base"] = "unknown" + else: + return self.async_create_entry( + title=user_input["name"], data=user_input, options=options + ) + + return self.async_show_form( + step_id="user", data_schema=get_user_step_schema(user_input), errors=errors + ) + + async_step_import = async_step_user + + +class HERETravelTimeOptionsFlow(config_entries.OptionsFlow): + """Handle HERE Travel Time options.""" + + 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: + return self.async_create_entry(title="", data=user_input) + + 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_TIME_TYPE, default=DEPARTURE_TIME): vol.In(TIME_TYPES), + vol.Optional(CONF_TIME, default=""): cv.string, + 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)) + + +class InvalidAuth(HomeAssistantError): + """Error to indicate there is invalid auth.""" diff --git a/homeassistant/components/here_travel_time/const.py b/homeassistant/components/here_travel_time/const.py new file mode 100644 index 00000000000000..d11997b96e151c --- /dev/null +++ b/homeassistant/components/here_travel_time/const.py @@ -0,0 +1,95 @@ +"""Constants for the HERE Travel Time integration.""" +from __future__ import annotations + +from typing import TypedDict + +from homeassistant.const import ( + CONF_UNIT_SYSTEM, + CONF_UNIT_SYSTEM_IMPERIAL, + CONF_UNIT_SYSTEM_METRIC, +) + + +class HERERoutingData(TypedDict): + """Routing information calculated from a herepy.RoutingResponse.""" + + attribution: str | None + base_time: float + traffic_time: float + distance: float + route: str + origin: str + destination: str + origin_name: str + destination_name: str + + +DOMAIN = "here_travel_time" +DEFAULT_SCAN_INTERVAL = 300 + +CONF_DESTINATION = "destination" +CONF_ORIGIN = "origin" +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" + +TRACKABLE_DOMAINS = ["device_tracker", "sensor", "zone", "person"] + +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_MODES = [ + TRAVEL_MODE_BICYCLE, + TRAVEL_MODE_CAR, + TRAVEL_MODE_PEDESTRIAN, + TRAVEL_MODE_PUBLIC, + TRAVEL_MODE_PUBLIC_TIME_TABLE, + TRAVEL_MODE_TRUCK, +] + +TRAVEL_MODES_PUBLIC = [TRAVEL_MODE_PUBLIC, TRAVEL_MODE_PUBLIC_TIME_TABLE] + +TRAFFIC_MODE_ENABLED = "traffic_enabled" +TRAFFIC_MODE_DISABLED = "traffic_disabled" +TRAFFIC_MODES = [TRAFFIC_MODE_ENABLED, TRAFFIC_MODE_DISABLED] + +ROUTE_MODE_FASTEST = "fastest" +ROUTE_MODE_SHORTEST = "shortest" +ROUTE_MODES = [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" + +NO_ROUTE_ERROR_MESSAGE = "HERE could not find a route based on the input" diff --git a/homeassistant/components/here_travel_time/manifest.json b/homeassistant/components/here_travel_time/manifest.json index 9a3e8bd482717b..fd554f7452bace 100644 --- a/homeassistant/components/here_travel_time/manifest.json +++ b/homeassistant/components/here_travel_time/manifest.json @@ -1,8 +1,13 @@ { "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"], + "requirements": [ + "herepy==3.5.2" + ], + "codeowners": [ + "@eifinger" + ], "iot_class": "cloud_polling" -} +} \ No newline at end of file diff --git a/homeassistant/components/here_travel_time/sensor.py b/homeassistant/components/here_travel_time/sensor.py index 1d47bbaf89f63c..b6cf6b13f12fea 100644 --- a/homeassistant/components/here_travel_time/sensor.py +++ b/homeassistant/components/here_travel_time/sensor.py @@ -1,33 +1,66 @@ """Support for HERE travel time sensors.""" from __future__ import annotations -from datetime import datetime, timedelta import logging -import herepy import voluptuous as vol +from homeassistant.components.here_travel_time import HERETravelTimeData +from homeassistant.components.here_travel_time.const import ( + ATTR_DESTINATION, + ATTR_DESTINATION_NAME, + ATTR_DISTANCE, + ATTR_DURATION, + ATTR_DURATION_IN_TRAFFIC, + ATTR_ORIGIN, + ATTR_ORIGIN_NAME, + ATTR_ROUTE, + ATTR_TRAFFIC_MODE, + ATTR_UNIT_SYSTEM, + CONF_ARRIVAL, + CONF_DEPARTURE, + CONF_DESTINATION, + CONF_ORIGIN, + 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, +) from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity +from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry from homeassistant.const import ( ATTR_ATTRIBUTION, - ATTR_LATITUDE, - ATTR_LONGITUDE, ATTR_MODE, CONF_API_KEY, CONF_MODE, CONF_NAME, + CONF_SCAN_INTERVAL, CONF_UNIT_SYSTEM, - CONF_UNIT_SYSTEM_IMPERIAL, - CONF_UNIT_SYSTEM_METRIC, - EVENT_HOMEASSISTANT_START, TIME_MINUTES, ) -from homeassistant.core import HomeAssistant, State, callback -from homeassistant.helpers import location +from homeassistant.core import HomeAssistant import homeassistant.helpers.config_validation as cv +from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.typing import DiscoveryInfoType -from homeassistant.util import dt +from homeassistant.helpers.update_coordinator import CoordinatorEntity _LOGGER = logging.getLogger(__name__) @@ -37,63 +70,6 @@ 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 = [ - TRAVEL_MODE_BICYCLE, - TRAVEL_MODE_CAR, - TRAVEL_MODE_PEDESTRIAN, - TRAVEL_MODE_PUBLIC, - TRAVEL_MODE_PUBLIC_TIME_TABLE, - TRAVEL_MODE_TRUCK, -] - -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] - -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( { @@ -112,11 +88,13 @@ 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), - } + vol.Remove(CONF_SCAN_INTERVAL): cv.time_period, + }, + extra=vol.REMOVE_EXTRA, ) PLATFORM_SCHEMA = vol.All( @@ -133,8 +111,8 @@ TRAVEL_MODE_TRUCK: PLATFORM_SCHEMA, TRAVEL_MODE_PUBLIC_TIME_TABLE: PLATFORM_SCHEMA.extend( { - vol.Exclusive(CONF_ARRIVAL, "arrival_departure"): cv.time, - vol.Exclusive(CONF_DEPARTURE, "arrival_departure"): cv.time, + vol.Exclusive(CONF_ARRIVAL, "arrival_departure"): cv.string, + vol.Exclusive(CONF_DEPARTURE, "arrival_departure"): cv.string, } ), }, @@ -144,354 +122,125 @@ async def async_setup_platform( hass: HomeAssistant, - config: dict[str, str | bool], - async_add_entities: AddEntitiesCallback, + config: ConfigEntry, + add_entities_callback: AddEntitiesCallback, 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] - - travel_mode = config[CONF_MODE] - traffic_mode = config[CONF_TRAFFIC_MODE] - route_mode = config[CONF_ROUTE_MODE] - name = config[CONF_NAME] - units = config.get(CONF_UNIT_SYSTEM, hass.config.units.name) - arrival = config.get(CONF_ARRIVAL) - departure = config.get(CONF_DEPARTURE) - - here_data = HERETravelTimeData( - here_client, travel_mode, traffic_mode, route_mode, units, arrival, departure ) - sensor = HERETravelTimeSensor( - name, origin, destination, origin_entity_id, destination_entity_id, here_data + _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" ) - async_add_entities([sensor]) +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Add HERE travel time entities from a config_entry.""" -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.car_route( - known_working_origin, - known_working_destination, - [ - herepy.RouteMode[ROUTE_MODE_FASTEST], - herepy.RouteMode[TRAVEL_MODE_CAR], - herepy.RouteMode[TRAFFIC_MODE_DISABLED], - ], - ) - except herepy.InvalidCredentialsError: - return False - return True + async_add_entities( + [HERETravelTimeSensor(hass.data[DOMAIN][config_entry.entry_id], config_entry)], + True, + ) -class HERETravelTimeSensor(SensorEntity): +class HERETravelTimeSensor(SensorEntity, CoordinatorEntity): """Representation of a HERE travel time sensor.""" def __init__( self, - name: str, - origin: str, - destination: str, - origin_entity_id: str, - destination_entity_id: str, here_data: HERETravelTimeData, + config_entry: ConfigEntry, ) -> None: """Initialize the sensor.""" - self._name = name - self._origin_entity_id = origin_entity_id - self._destination_entity_id = destination_entity_id - self._here_data = here_data - self._unit_of_measurement = TIME_MINUTES - self._attrs = { - ATTR_UNIT_SYSTEM: self._here_data.units, - ATTR_MODE: self._here_data.travel_mode, - ATTR_TRAFFIC_MODE: self._here_data.traffic_mode, - } - if self._origin_entity_id is None: - self._here_data.origin = origin - - if self._destination_entity_id is None: - self._here_data.destination = destination - - async def async_added_to_hass(self) -> None: - """Delay the sensor update to avoid entity not found warnings.""" - - @callback - def delayed_sensor_update(event): - """Update sensor after Home Assistant started.""" - self.async_schedule_update_ha_state(True) - - self.hass.bus.async_listen_once( - EVENT_HOMEASSISTANT_START, delayed_sensor_update - ) + self._config_entry = config_entry + assert here_data.coordinator is not None + super().__init__(here_data.coordinator) @property def native_value(self) -> str | None: """Return the state of the sensor.""" - if self._here_data.traffic_mode and self._here_data.traffic_time is not None: - return str(round(self._here_data.traffic_time / 60)) - if self._here_data.base_time is not None: - return str(round(self._here_data.base_time / 60)) - + if self.coordinator.data is not None: + if (time := self.coordinator.data.get("traffic_time")) is not None: + return str(round(time / 60)) return None - @property - def name(self) -> str: - """Get the name of the sensor.""" - return self._name - @property def extra_state_attributes( self, ) -> dict[str, None | float | str | bool] | None: """Return the state attributes.""" - if self._here_data.base_time is None: - return None + if self.coordinator.data is not None: + res = { + ATTR_UNIT_SYSTEM: self.hass.config.units.name, + ATTR_MODE: self._config_entry.data[CONF_MODE], + ATTR_TRAFFIC_MODE: self._config_entry.options[CONF_TRAFFIC_MODE] + == TRAFFIC_MODE_ENABLED, + ATTR_DURATION: self.coordinator.data["base_time"] / 60, + ATTR_DISTANCE: self.coordinator.data["distance"], + ATTR_ROUTE: self.coordinator.data["route"], + ATTR_DURATION_IN_TRAFFIC: self.coordinator.data["traffic_time"] / 60, + ATTR_ORIGIN: self.coordinator.data["origin"], + ATTR_DESTINATION: self.coordinator.data["destination"], + ATTR_ORIGIN_NAME: self.coordinator.data["origin_name"], + ATTR_DESTINATION_NAME: self.coordinator.data["destination_name"], + } + if (attribution := self.coordinator.data.get("attribution")) is not None: + res[ATTR_ATTRIBUTION] = attribution + return res + return None - res = self._attrs - if self._here_data.attribution is not None: - res[ATTR_ATTRIBUTION] = self._here_data.attribution - res[ATTR_DURATION] = self._here_data.base_time / 60 - res[ATTR_DISTANCE] = self._here_data.distance - res[ATTR_ROUTE] = self._here_data.route - res[ATTR_DURATION_IN_TRAFFIC] = self._here_data.traffic_time / 60 - res[ATTR_ORIGIN] = self._here_data.origin - res[ATTR_DESTINATION] = self._here_data.destination - res[ATTR_ORIGIN_NAME] = self._here_data.origin_name - res[ATTR_DESTINATION_NAME] = self._here_data.destination_name - return res + @property + def name(self) -> str: + """Get the name of the sensor.""" + return str(self._config_entry.data[CONF_NAME]) + + @property + def unique_id(self) -> str: + """Return unique ID of entity.""" + return self._config_entry.entry_id @property def native_unit_of_measurement(self) -> str: """Return the unit this state is expressed in.""" - return self._unit_of_measurement + return TIME_MINUTES @property def icon(self) -> str: """Icon to use in the frontend depending on travel_mode.""" - if self._here_data.travel_mode == TRAVEL_MODE_BICYCLE: + travel_mode = self._config_entry.data[CONF_MODE] + if travel_mode == TRAVEL_MODE_BICYCLE: return ICON_BICYCLE - if self._here_data.travel_mode == TRAVEL_MODE_PEDESTRIAN: + if travel_mode == TRAVEL_MODE_PEDESTRIAN: return ICON_PEDESTRIAN - if self._here_data.travel_mode in TRAVEL_MODES_PUBLIC: + if travel_mode in TRAVEL_MODES_PUBLIC: return ICON_PUBLIC - if self._here_data.travel_mode == TRAVEL_MODE_TRUCK: + if travel_mode == TRAVEL_MODE_TRUCK: return ICON_TRUCK return ICON_CAR - async def async_update(self) -> None: - """Update Sensor Information.""" - # Convert device_trackers to HERE friendly location - if self._origin_entity_id is not None: - self._here_data.origin = await self._get_location_from_entity( - self._origin_entity_id - ) - - if self._destination_entity_id is not None: - self._here_data.destination = await self._get_location_from_entity( - self._destination_entity_id - ) - - await self.hass.async_add_executor_job(self._here_data.update) - - async def _get_location_from_entity(self, entity_id: str) -> str | None: - """Get the location from the entity state or attributes.""" - if (entity := self.hass.states.get(entity_id)) is None: - _LOGGER.error("Unable to find entity %s", entity_id) - return None - - # Check if the entity has location attributes - if location.has_location(entity): - return self._get_location_from_attributes(entity) - - # Check if device is in a zone - zone_entity = self.hass.states.get(f"zone.{entity.state}") - if location.has_location(zone_entity): - _LOGGER.debug( - "%s is in %s, getting zone location", entity_id, zone_entity.entity_id - ) - return self._get_location_from_attributes(zone_entity) - - # Check if state is valid coordinate set - if self._entity_state_is_valid_coordinate_set(entity.state): - return entity.state - - _LOGGER.error( - "The state of %s is not a valid set of coordinates: %s", - entity_id, - entity.state, - ) - return None - - @staticmethod - def _entity_state_is_valid_coordinate_set(state: str) -> bool: - """Check that the given string is a valid set of coordinates.""" - schema = vol.Schema(cv.gps) - try: - coordinates = state.split(",") - schema(coordinates) - return True - except (vol.MultipleInvalid): - return False - - @staticmethod - def _get_location_from_attributes(entity: State) -> str: - """Get the lat/long string from an entities attributes.""" - attr = entity.attributes - return f"{attr.get(ATTR_LATITUDE)},{attr.get(ATTR_LONGITUDE)}" - - -class HERETravelTimeData: - """HERETravelTime data object.""" - - def __init__( - self, - here_client: herepy.RoutingApi, - travel_mode: str, - traffic_mode: bool, - route_mode: str, - units: str, - arrival: datetime, - departure: datetime, - ) -> None: - """Initialize herepy.""" - self.origin = None - self.destination = None - self.travel_mode = travel_mode - self.traffic_mode = traffic_mode - self.route_mode = route_mode - self.arrival = arrival - self.departure = departure - self.attribution = None - self.traffic_time = None - self.distance = None - self.route = None - self.base_time = None - self.origin_name = None - self.destination_name = None - self.units = units - self._client = here_client - self.combine_change = True - - def update(self) -> None: - """Get the latest data from HERE.""" - if self.traffic_mode: - traffic_mode = TRAFFIC_MODE_ENABLED - else: - traffic_mode = TRAFFIC_MODE_DISABLED - - if self.destination is not None and self.origin is not None: - # Convert location to HERE friendly location - destination = self.destination.split(",") - origin = self.origin.split(",") - if (arrival := self.arrival) is not None: - arrival = convert_time_to_isodate(arrival) - if (departure := self.departure) is not None: - departure = convert_time_to_isodate(departure) - - if departure is None and arrival is None: - departure = "now" - - _LOGGER.debug( - "Requesting route for origin: %s, destination: %s, route_mode: %s, mode: %s, traffic_mode: %s, arrival: %s, departure: %s", - origin, - destination, - herepy.RouteMode[self.route_mode], - herepy.RouteMode[self.travel_mode], - herepy.RouteMode[traffic_mode], - arrival, - departure, - ) - - try: - response = self._client.public_transport_timetable( - origin, - destination, - self.combine_change, - [ - herepy.RouteMode[self.route_mode], - herepy.RouteMode[self.travel_mode], - herepy.RouteMode[traffic_mode], - ], - arrival=arrival, - departure=departure, + @property + def device_info(self) -> DeviceInfo: + """Return device specific attributes.""" + return { + "name": self.name, + "identifiers": { + ( + DOMAIN, + f"{self._config_entry.data[CONF_ORIGIN]}_{self._config_entry.data[CONF_DESTINATION]}", ) - except herepy.NoRouteFoundError: - # Better error message for cryptic no route error codes - _LOGGER.error(NO_ROUTE_ERROR_MESSAGE) - return - - _LOGGER.debug("Raw response is: %s", response.response) - - source_attribution = response.response.get("sourceAttribution") - if source_attribution is not None: - self.attribution = self._build_hass_attribution(source_attribution) - route = response.response["route"] - summary = route[0]["summary"] - waypoint = route[0]["waypoint"] - self.base_time = summary["baseTime"] - if self.travel_mode in TRAVEL_MODES_VEHICLE: - self.traffic_time = summary["trafficTime"] - else: - self.traffic_time = self.base_time - distance = summary["distance"] - if self.units == CONF_UNIT_SYSTEM_IMPERIAL: - # Convert to miles. - self.distance = distance / 1609.344 - else: - # Convert to kilometers - self.distance = distance / 1000 - self.route = response.route_short - self.origin_name = waypoint[0]["mappedRoadName"] - self.destination_name = waypoint[1]["mappedRoadName"] - - @staticmethod - def _build_hass_attribution(source_attribution: dict) -> str | None: - """Build a hass frontend ready string out of the sourceAttribution.""" - suppliers = source_attribution.get("supplier") - if suppliers is not None: - supplier_titles = [] - for supplier in suppliers: - if (title := supplier.get("title")) is not None: - supplier_titles.append(title) - joined_supplier_titles = ",".join(supplier_titles) - attribution = f"With the support of {joined_supplier_titles}. All information is provided without warranty of any kind." - return attribution - - -def convert_time_to_isodate(timestr: str) -> str: - """Take a string like 08:00:00 and combine it with the current date.""" - combined = datetime.combine(dt.start_of_local_day(), dt.parse_time(timestr)) - if combined < datetime.now(): - combined = combined + timedelta(days=1) - return combined.isoformat() + }, + "manufacturer": "HERE", + "entry_type": "service", + } diff --git a/homeassistant/components/here_travel_time/strings.json b/homeassistant/components/here_travel_time/strings.json new file mode 100644 index 00000000000000..90c9c28ef90aa4 --- /dev/null +++ b/homeassistant/components/here_travel_time/strings.json @@ -0,0 +1,37 @@ +{ + "config": { + "step": { + "user": { + "description": "When specifying the origin and destination, you can supply the location in the form of an address, latitude/longitude coordinates, or an entity id.", + "data": { + "destination": "Destination as GPS coordinates or an entity id", + "origin": "Origin as GPS coordinates or an entity id", + "name": "[%key:common::config_flow::data::name%]", + "api_key": "[%key:common::config_flow::data::api_key%]", + "mode": "Travel Mode" + } + } + }, + "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", + "time_type": "Time Type", + "time": "Time", + "unit_system": "Unit system" + }, + "description": "You can optionally specify either a Departure Time or Arrival Time. If specifying a departure time, you can enter `now`, a Unix timestamp, or a 24 hour time string like `08:00:00`. If specifying an arrival time, you can use a Unix timestamp or a 24 hour time string like `08:00:00`" + } + } + } +} \ No newline at end of file diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index bff1305504f880..c36dc3c3cdad87 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -118,6 +118,7 @@ "hangouts", "harmony", "heos", + "here_travel_time", "hisense_aehw4a1", "hive", "hlk_sw16", diff --git a/mypy.ini b/mypy.ini index f52ee13b689f2d..4e23c806cedbca 100644 --- a/mypy.ini +++ b/mypy.ini @@ -616,6 +616,17 @@ no_implicit_optional = true warn_return_any = true warn_unreachable = true +[mypy-homeassistant.components.here_travel_time.*] +check_untyped_defs = true +disallow_incomplete_defs = true +disallow_subclassing_any = true +disallow_untyped_calls = true +disallow_untyped_decorators = true +disallow_untyped_defs = true +no_implicit_optional = true +warn_return_any = true +warn_unreachable = true + [mypy-homeassistant.components.history.*] check_untyped_defs = true disallow_incomplete_defs = true @@ -1665,7 +1676,7 @@ ignore_errors = true [mypy-homeassistant.components.hassio.*] ignore_errors = true -[mypy-homeassistant.components.here_travel_time.*] +[mypy-homeassistant.components.hdmi_cec.*] ignore_errors = true [mypy-homeassistant.components.hisense_aehw4a1.*] diff --git a/requirements_all.txt b/requirements_all.txt index c84aac7e3ec9e3..a0181473a583f6 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -801,7 +801,7 @@ hdate==0.10.4 heatmiserV3==1.1.18 # homeassistant.components.here_travel_time -herepy==2.0.0 +herepy==3.5.2 # homeassistant.components.hikvisioncam hikvision==0.4 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index c5b5ad5f9cdba0..bd0b9c47e2442c 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -494,7 +494,7 @@ hatasmota==0.3.0 hdate==0.10.4 # homeassistant.components.here_travel_time -herepy==2.0.0 +herepy==3.5.2 # homeassistant.components.hlk_sw16 hlk-sw16==0.0.9 diff --git a/script/hassfest/mypy_config.py b/script/hassfest/mypy_config.py index 7d2899843355e5..788cd1589d05ba 100644 --- a/script/hassfest/mypy_config.py +++ b/script/hassfest/mypy_config.py @@ -41,7 +41,7 @@ "homeassistant.components.habitica.*", "homeassistant.components.harmony.*", "homeassistant.components.hassio.*", - "homeassistant.components.here_travel_time.*", + "homeassistant.components.hdmi_cec.*", "homeassistant.components.hisense_aehw4a1.*", "homeassistant.components.home_connect.*", "homeassistant.components.home_plus_control.*", diff --git a/tests/components/here_travel_time/conftest.py b/tests/components/here_travel_time/conftest.py new file mode 100644 index 00000000000000..a98fda57316f23 --- /dev/null +++ b/tests/components/here_travel_time/conftest.py @@ -0,0 +1,42 @@ +"""Fixtures for HERE Travel Time tests.""" +import json +from unittest.mock import patch + +from herepy.models import RoutingResponse +import pytest + +from tests.common import load_fixture + +RESPONSE = RoutingResponse.new_from_jsondict( + json.loads(load_fixture("here_travel_time/car_response.json")) +) +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(): + """Return valid api response.""" + with patch( + "herepy.RoutingApi.public_transport_timetable", + return_value=RESPONSE, + ): + yield + + +@pytest.fixture(name="skip_notifications", autouse=True) +def skip_notifications_fixture(): + """Skip notification calls.""" + with patch("homeassistant.components.persistent_notification.async_create"), patch( + "homeassistant.components.persistent_notification.async_dismiss" + ): + yield + + +@pytest.fixture(name="validate_config_entry") +def validate_config_entry_fixture(): + """Return valid config entry.""" + with patch( + "herepy.RoutingApi.public_transport_timetable", + return_value=None, + ): + yield diff --git a/tests/components/here_travel_time/const.py b/tests/components/here_travel_time/const.py new file mode 100644 index 00000000000000..e589c09ef54657 --- /dev/null +++ b/tests/components/here_travel_time/const.py @@ -0,0 +1,8 @@ +"""Constants for here_travel_time tests.""" + +API_KEY = "test" + +CAR_ORIGIN_LATITUDE = "38.9" +CAR_ORIGIN_LONGITUDE = "-77.04833" +CAR_DESTINATION_LATITUDE = "39.0" +CAR_DESTINATION_LONGITUDE = "-77.1" diff --git a/tests/components/here_travel_time/fixtures/car_response.json b/tests/components/here_travel_time/fixtures/car_response.json index bda8454f3f3780..ef050b78362326 100644 --- a/tests/components/here_travel_time/fixtures/car_response.json +++ b/tests/components/here_travel_time/fixtures/car_response.json @@ -294,6 +294,15 @@ } } ], - "language": "en-us" + "language": "en-us", + "sourceAttribution": { + "attribution": "With the support of HERE Technologies. All information is provided without warranty of any kind.", + "supplier": [ + { + "title": "HERE Technologies", + "href": "https://transit.api.here.com/r?appId=Mt1bOYh3m9uxE7r3wuUx&u=https://wego.here.com" + } + ] + } } } \ No newline at end of file diff --git a/tests/components/here_travel_time/test_config_flow.py b/tests/components/here_travel_time/test_config_flow.py new file mode 100644 index 00000000000000..40b2e4c4dac894 --- /dev/null +++ b/tests/components/here_travel_time/test_config_flow.py @@ -0,0 +1,350 @@ +"""Test the HERE Travel Time config flow.""" +from unittest.mock import patch + +from herepy.routing_api import InvalidCredentialsError + +from homeassistant import config_entries, data_entry_flow, setup +from homeassistant.components.here_travel_time.const import ( + ARRIVAL_TIME, + CONF_ARRIVAL, + CONF_DEPARTURE, + CONF_DESTINATION, + CONF_ORIGIN, + CONF_ROUTE_MODE, + CONF_TIME, + CONF_TIME_TYPE, + CONF_TRAFFIC_MODE, + DEPARTURE_TIME, + DOMAIN, + ROUTE_MODE_FASTEST, + TRAFFIC_MODE_ENABLED, + TRAVEL_MODE_CAR, +) +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 tests.common import MockConfigEntry +from tests.components.here_travel_time.const import ( + API_KEY, + CAR_DESTINATION_LATITUDE, + CAR_DESTINATION_LONGITUDE, + CAR_ORIGIN_LATITUDE, + CAR_ORIGIN_LONGITUDE, +) + + +async def test_form(hass: HomeAssistant) -> 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["type"] == "form" + assert result["errors"] == {} + + with patch( + "herepy.RoutingApi.public_transport_timetable", + return_value=None, + ), patch( + "homeassistant.components.here_travel_time.async_setup_entry", + return_value=True, + ) as mock_setup_entry: + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_ORIGIN: f"{CAR_ORIGIN_LATITUDE},{CAR_ORIGIN_LONGITUDE}", + CONF_DESTINATION: f"{CAR_DESTINATION_LATITUDE},{CAR_DESTINATION_LONGITUDE}", + CONF_API_KEY: API_KEY, + CONF_MODE: TRAVEL_MODE_CAR, + CONF_NAME: "test", + }, + ) + await hass.async_block_till_done() + + assert result2["type"] == "create_entry" + assert result2["title"] == "test" + assert result2["data"] == { + CONF_ORIGIN: f"{CAR_ORIGIN_LATITUDE},{CAR_ORIGIN_LONGITUDE}", + CONF_DESTINATION: f"{CAR_DESTINATION_LATITUDE},{CAR_DESTINATION_LONGITUDE}", + CONF_API_KEY: API_KEY, + CONF_MODE: TRAVEL_MODE_CAR, + CONF_NAME: "test", + } + assert len(mock_setup_entry.mock_calls) == 1 + + +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_ORIGIN: f"{CAR_ORIGIN_LATITUDE},{CAR_ORIGIN_LONGITUDE}", + CONF_DESTINATION: f"{CAR_DESTINATION_LATITUDE},{CAR_DESTINATION_LONGITUDE}", + 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=Exception, + ): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_ORIGIN: f"{CAR_ORIGIN_LATITUDE},{CAR_ORIGIN_LONGITUDE}", + CONF_DESTINATION: f"{CAR_DESTINATION_LATITUDE},{CAR_DESTINATION_LONGITUDE}", + CONF_API_KEY: API_KEY, + CONF_MODE: TRAVEL_MODE_CAR, + CONF_NAME: "test", + }, + ) + + assert result2["type"] == "form" + assert result2["errors"] == {"base": "unknown"} + + +async def test_options_flow(hass: HomeAssistant, valid_response) -> None: + """Test the options flow.""" + entry = MockConfigEntry( + domain=DOMAIN, + unique_id="0123456789", + data={ + CONF_ORIGIN: f"{CAR_ORIGIN_LATITUDE},{CAR_ORIGIN_LONGITUDE}", + CONF_DESTINATION: f"{CAR_DESTINATION_LATITUDE},{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_TIME_TYPE: DEPARTURE_TIME, + CONF_TIME: "", + }, + ) + + assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY + assert entry.options == { + CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED, + CONF_ROUTE_MODE: ROUTE_MODE_FASTEST, + CONF_TIME_TYPE: DEPARTURE_TIME, + CONF_TIME: "", + CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC, + } + + +async def test_import_flow_entity_id( + hass: HomeAssistant, validate_config_entry +) -> 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: "sensor.origin", + CONF_DESTINATION: "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_TIME_TYPE: DEPARTURE_TIME, + CONF_TIME: "08:00:00", + } + + +async def test_import_flow_coordinates( + hass: HomeAssistant, validate_config_entry +) -> 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: f"{CAR_ORIGIN_LATITUDE},{CAR_ORIGIN_LONGITUDE}", + CONF_DESTINATION: f"{CAR_DESTINATION_LATITUDE},{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_TIME_TYPE: ARRIVAL_TIME, + CONF_TIME: "08:00:00", + CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC, + } + + +async def test_dupe_import(hass: HomeAssistant, validate_config_entry) -> 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: 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" diff --git a/tests/components/here_travel_time/test_init.py b/tests/components/here_travel_time/test_init.py new file mode 100644 index 00000000000000..340ca75d5ba513 --- /dev/null +++ b/tests/components/here_travel_time/test_init.py @@ -0,0 +1,40 @@ +"""Test init of here travel time.""" + +from homeassistant.components.here_travel_time.const import ( + CONF_DESTINATION, + CONF_ORIGIN, + DOMAIN, + TRAVEL_MODE_CAR, +) +from homeassistant.const import CONF_API_KEY, CONF_MODE, CONF_NAME +from homeassistant.core import HomeAssistant + +from tests.common import MockConfigEntry +from tests.components.here_travel_time.const import ( + API_KEY, + CAR_DESTINATION_LATITUDE, + CAR_DESTINATION_LONGITUDE, + CAR_ORIGIN_LATITUDE, + CAR_ORIGIN_LONGITUDE, +) + + +async def test_unload_entry(hass: HomeAssistant, valid_response): + """Test that unloading an entry works.""" + entry = MockConfigEntry( + domain=DOMAIN, + unique_id="0123456789", + data={ + CONF_ORIGIN: f"{CAR_ORIGIN_LATITUDE},{CAR_ORIGIN_LONGITUDE}", + CONF_DESTINATION: f"{CAR_DESTINATION_LATITUDE},{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] diff --git a/tests/components/here_travel_time/test_sensor.py b/tests/components/here_travel_time/test_sensor.py index ce0c2d9ca6d7bb..bf5220c0102357 100644 --- a/tests/components/here_travel_time/test_sensor.py +++ b/tests/components/here_travel_time/test_sensor.py @@ -1,13 +1,11 @@ """The test for the here_travel_time sensor platform.""" -import logging from unittest.mock import patch -import urllib -import herepy +from herepy.routing_api import NoRouteFoundError import pytest -from homeassistant.components.here_travel_time.sensor import ( - ATTR_ATTRIBUTION, +from homeassistant.components.here_travel_time.const import ( + ARRIVAL_TIME, ATTR_DESTINATION, ATTR_DESTINATION_NAME, ATTR_DISTANCE, @@ -16,9 +14,12 @@ ATTR_ORIGIN, ATTR_ORIGIN_NAME, ATTR_ROUTE, - CONF_MODE, + CONF_ROUTE_MODE, + CONF_TIME, + CONF_TIME_TYPE, CONF_TRAFFIC_MODE, - CONF_UNIT_SYSTEM, + DEPARTURE_TIME, + DOMAIN, ICON_BICYCLE, ICON_CAR, ICON_PEDESTRIAN, @@ -26,163 +27,79 @@ ICON_TRUCK, NO_ROUTE_ERROR_MESSAGE, ROUTE_MODE_FASTEST, - ROUTE_MODE_SHORTEST, - SCAN_INTERVAL, - TIME_MINUTES, - TRAFFIC_MODE_DISABLED, TRAFFIC_MODE_ENABLED, TRAVEL_MODE_BICYCLE, TRAVEL_MODE_CAR, TRAVEL_MODE_PEDESTRIAN, - TRAVEL_MODE_PUBLIC, TRAVEL_MODE_PUBLIC_TIME_TABLE, TRAVEL_MODE_TRUCK, - convert_time_to_isodate, ) -from homeassistant.const import ATTR_ICON, EVENT_HOMEASSISTANT_START +from homeassistant.const import ( + ATTR_ATTRIBUTION, + ATTR_ICON, + CONF_MODE, + CONF_UNIT_SYSTEM, + CONF_UNIT_SYSTEM_IMPERIAL, + TIME_MINUTES, +) from homeassistant.setup import async_setup_component -import homeassistant.util.dt as dt_util - -from tests.common import async_fire_time_changed, load_fixture - -DOMAIN = "sensor" - -PLATFORM = "here_travel_time" - -API_KEY = "test" - -TRUCK_ORIGIN_LATITUDE = "41.9798" -TRUCK_ORIGIN_LONGITUDE = "-87.8801" -TRUCK_DESTINATION_LATITUDE = "41.9043" -TRUCK_DESTINATION_LONGITUDE = "-87.9216" - -BIKE_ORIGIN_LATITUDE = "41.9798" -BIKE_ORIGIN_LONGITUDE = "-87.8801" -BIKE_DESTINATION_LATITUDE = "41.9043" -BIKE_DESTINATION_LONGITUDE = "-87.9216" - -CAR_ORIGIN_LATITUDE = "38.9" -CAR_ORIGIN_LONGITUDE = "-77.04833" -CAR_DESTINATION_LATITUDE = "39.0" -CAR_DESTINATION_LONGITUDE = "-77.1" - - -def _build_mock_url(origin, destination, modes, api_key, departure=None, arrival=None): - """Construct a url for HERE.""" - base_url = "https://route.ls.hereapi.com/routing/7.2/calculateroute.json?" - parameters = { - "waypoint0": f"geo!{origin}", - "waypoint1": f"geo!{destination}", - "mode": ";".join(str(herepy.RouteMode[mode]) for mode in modes), - "apikey": api_key, - } - if arrival is not None: - parameters["arrival"] = arrival - if departure is not None: - parameters["departure"] = departure - if departure is None and arrival is None: - parameters["departure"] = "now" - url = base_url + urllib.parse.urlencode(parameters) - return url - - -def _assert_truck_sensor(sensor): - """Assert that states and attributes are correct for truck_response.""" - assert sensor.state == "14" - assert sensor.attributes.get("unit_of_measurement") == TIME_MINUTES - - assert sensor.attributes.get(ATTR_ATTRIBUTION) is None - assert sensor.attributes.get(ATTR_DURATION) == 13.533333333333333 - assert sensor.attributes.get(ATTR_DISTANCE) == 13.049 - assert sensor.attributes.get(ATTR_ROUTE) == ( - "I-190; I-294 S - Tri-State Tollway; I-290 W - Eisenhower Expy W; " - "IL-64 W - E North Ave; I-290 E - Eisenhower Expy E; I-290" - ) - assert sensor.attributes.get(CONF_UNIT_SYSTEM) == "metric" - assert sensor.attributes.get(ATTR_DURATION_IN_TRAFFIC) == 13.533333333333333 - assert sensor.attributes.get(ATTR_ORIGIN) == ",".join( - [TRUCK_ORIGIN_LATITUDE, TRUCK_ORIGIN_LONGITUDE] - ) - assert sensor.attributes.get(ATTR_DESTINATION) == ",".join( - [TRUCK_DESTINATION_LATITUDE, TRUCK_DESTINATION_LONGITUDE] - ) - assert sensor.attributes.get(ATTR_ORIGIN_NAME) == "" - assert sensor.attributes.get(ATTR_DESTINATION_NAME) == "Eisenhower Expy E" - assert sensor.attributes.get(CONF_MODE) == TRAVEL_MODE_TRUCK - assert sensor.attributes.get(CONF_TRAFFIC_MODE) is False - - assert sensor.attributes.get(ATTR_ICON) == ICON_TRUCK - - -@pytest.fixture -def requests_mock_credentials_check(requests_mock): - """Add the url used in the api validation to all requests mock.""" - modes = [ROUTE_MODE_FASTEST, TRAVEL_MODE_CAR, TRAFFIC_MODE_DISABLED] - response_url = _build_mock_url( - ",".join([CAR_ORIGIN_LATITUDE, CAR_ORIGIN_LONGITUDE]), - ",".join([CAR_DESTINATION_LATITUDE, CAR_DESTINATION_LONGITUDE]), - modes, - API_KEY, - ) - requests_mock.get( - response_url, text=load_fixture("here_travel_time/car_response.json") - ) - return requests_mock - -@pytest.fixture -def requests_mock_truck_response(requests_mock_credentials_check): - """Return a requests_mock for truck response.""" - modes = [ROUTE_MODE_FASTEST, TRAVEL_MODE_TRUCK, TRAFFIC_MODE_DISABLED] - response_url = _build_mock_url( - ",".join([TRUCK_ORIGIN_LATITUDE, TRUCK_ORIGIN_LONGITUDE]), - ",".join([TRUCK_DESTINATION_LATITUDE, TRUCK_DESTINATION_LONGITUDE]), - modes, - API_KEY, - ) - requests_mock_credentials_check.get( - response_url, text=load_fixture("here_travel_time/truck_response.json") - ) +from tests.common import MockConfigEntry +from tests.components.here_travel_time.const import ( + API_KEY, + CAR_DESTINATION_LATITUDE, + CAR_DESTINATION_LONGITUDE, + CAR_ORIGIN_LATITUDE, + CAR_ORIGIN_LONGITUDE, +) -@pytest.fixture -def requests_mock_car_disabled_response(requests_mock_credentials_check): - """Return a requests_mock for truck response.""" - modes = [ROUTE_MODE_FASTEST, TRAVEL_MODE_CAR, TRAFFIC_MODE_DISABLED] - response_url = _build_mock_url( - ",".join([CAR_ORIGIN_LATITUDE, CAR_ORIGIN_LONGITUDE]), - ",".join([CAR_DESTINATION_LATITUDE, CAR_DESTINATION_LONGITUDE]), - modes, - API_KEY, - ) - requests_mock_credentials_check.get( - response_url, text=load_fixture("here_travel_time/car_response.json") +@pytest.fixture(name="failed_response") +def failed_response_fixture(): + """Return failed api response.""" + with patch( + "herepy.RoutingApi.public_transport_timetable", + side_effect=NoRouteFoundError, + ): + yield + + +@pytest.mark.parametrize( + "mode,icon", + [ + (TRAVEL_MODE_CAR, ICON_CAR), + (TRAVEL_MODE_BICYCLE, ICON_BICYCLE), + (TRAVEL_MODE_PEDESTRIAN, ICON_PEDESTRIAN), + (TRAVEL_MODE_PUBLIC_TIME_TABLE, ICON_PUBLIC), + (TRAVEL_MODE_TRUCK, ICON_TRUCK), + ], +) +async def test_sensor(hass, mode, icon, valid_response): + """Test that sensor works.""" + entry = MockConfigEntry( + domain=DOMAIN, + unique_id="0123456789", + data={ + "origin": f"{CAR_ORIGIN_LATITUDE},{CAR_ORIGIN_LONGITUDE}", + "destination": f"{CAR_DESTINATION_LATITUDE},{CAR_DESTINATION_LONGITUDE}", + "api_key": API_KEY, + "mode": mode, + "name": "test", + }, ) + entry.add_to_hass(hass) + await hass.config_entries.async_setup(entry.entry_id) -async def test_car(hass, requests_mock_car_disabled_response): - """Test that car 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, - } - } - 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" + assert sensor.state == "31" assert sensor.attributes.get("unit_of_measurement") == TIME_MINUTES - assert sensor.attributes.get(ATTR_ATTRIBUTION) is None + assert ( + sensor.attributes.get(ATTR_ATTRIBUTION) + == "With the support of HERE Technologies. All information is provided without warranty of any kind." + ) assert sensor.attributes.get(ATTR_DURATION) == 30.05 assert sensor.attributes.get(ATTR_DISTANCE) == 23.903 assert sensor.attributes.get(ATTR_ROUTE) == ( @@ -199,10 +116,10 @@ async def test_car(hass, requests_mock_car_disabled_response): ) 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) == TRAVEL_MODE_CAR - assert sensor.attributes.get(CONF_TRAFFIC_MODE) is False + assert sensor.attributes.get(CONF_MODE) == mode + assert sensor.attributes.get(CONF_TRAFFIC_MODE) is True - assert sensor.attributes.get(ATTR_ICON) == ICON_CAR + assert sensor.attributes.get(ATTR_ICON) == icon # Test traffic mode disabled assert sensor.attributes.get(ATTR_DURATION) != sensor.attributes.get( @@ -210,743 +127,122 @@ async def test_car(hass, requests_mock_car_disabled_response): ) -async def test_traffic_mode_enabled(hass, requests_mock_credentials_check): - """Test that traffic mode enabled works.""" - modes = [ROUTE_MODE_FASTEST, TRAVEL_MODE_CAR, TRAFFIC_MODE_ENABLED] - response_url = _build_mock_url( - ",".join([CAR_ORIGIN_LATITUDE, CAR_ORIGIN_LONGITUDE]), - ",".join([CAR_DESTINATION_LATITUDE, CAR_DESTINATION_LONGITUDE]), - modes, - API_KEY, - ) - requests_mock_credentials_check.get( - response_url, text=load_fixture("here_travel_time/car_enabled_response.json") - ) - - 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": True, - } - } - 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") - # Test traffic mode enabled - assert sensor.attributes.get(ATTR_DURATION) != sensor.attributes.get( - ATTR_DURATION_IN_TRAFFIC - ) - - -async def test_imperial(hass, requests_mock_car_disabled_response): - """Test that imperial units work.""" - 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, - "unit_system": "imperial", - } - } - 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.attributes.get(ATTR_DISTANCE) == 14.852635608048994 - - -async def test_route_mode_shortest(hass, requests_mock_credentials_check): - """Test that route mode shortest works.""" - origin = "38.902981,-77.048338" - destination = "39.042158,-77.119116" - modes = [ROUTE_MODE_SHORTEST, TRAVEL_MODE_CAR, TRAFFIC_MODE_DISABLED] - response_url = _build_mock_url(origin, destination, modes, API_KEY) - requests_mock_credentials_check.get( - response_url, text=load_fixture("here_travel_time/car_shortest_response.json") - ) - - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_latitude": origin.split(",")[0], - "origin_longitude": origin.split(",")[1], - "destination_latitude": destination.split(",")[0], - "destination_longitude": destination.split(",")[1], - "api_key": API_KEY, - "route_mode": ROUTE_MODE_SHORTEST, - } - } - 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.attributes.get(ATTR_DISTANCE) == 18.388 - - -async def test_route_mode_fastest(hass, requests_mock_credentials_check): - """Test that route mode fastest works.""" - origin = "38.902981,-77.048338" - destination = "39.042158,-77.119116" - modes = [ROUTE_MODE_FASTEST, TRAVEL_MODE_CAR, TRAFFIC_MODE_ENABLED] - response_url = _build_mock_url(origin, destination, modes, API_KEY) - requests_mock_credentials_check.get( - response_url, text=load_fixture("here_travel_time/car_enabled_response.json") - ) - - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_latitude": origin.split(",")[0], - "origin_longitude": origin.split(",")[1], - "destination_latitude": destination.split(",")[0], - "destination_longitude": destination.split(",")[1], - "api_key": API_KEY, - "traffic_mode": True, - } - } - 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.attributes.get(ATTR_DISTANCE) == 23.381 - - -async def test_truck(hass, requests_mock_truck_response): - """Test that truck works.""" - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_latitude": TRUCK_ORIGIN_LATITUDE, - "origin_longitude": TRUCK_ORIGIN_LONGITUDE, - "destination_latitude": TRUCK_DESTINATION_LATITUDE, - "destination_longitude": TRUCK_DESTINATION_LONGITUDE, - "api_key": API_KEY, - "mode": TRAVEL_MODE_TRUCK, - } - } - 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_truck_sensor(sensor) - - -async def test_public_transport(hass, requests_mock_credentials_check): - """Test that publicTransport works.""" - origin = "41.9798,-87.8801" - destination = "41.9043,-87.9216" - modes = [ROUTE_MODE_FASTEST, TRAVEL_MODE_PUBLIC, TRAFFIC_MODE_DISABLED] - response_url = _build_mock_url(origin, destination, modes, API_KEY) - requests_mock_credentials_check.get( - response_url, text=load_fixture("here_travel_time/public_response.json") - ) - - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_latitude": origin.split(",")[0], - "origin_longitude": origin.split(",")[1], - "destination_latitude": destination.split(",")[0], - "destination_longitude": destination.split(",")[1], - "api_key": API_KEY, - "mode": TRAVEL_MODE_PUBLIC, - } - } - 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 == "89" - assert sensor.attributes.get("unit_of_measurement") == TIME_MINUTES - - assert sensor.attributes.get(ATTR_ATTRIBUTION) is None - assert sensor.attributes.get(ATTR_DURATION) == 89.16666666666667 - assert sensor.attributes.get(ATTR_DISTANCE) == 22.325 - assert sensor.attributes.get(ATTR_ROUTE) == ( - "332 - Palmer/Schiller; 332 - Cargo Rd./Delta Cargo; 332 - Palmer/Schiller" - ) - assert sensor.attributes.get(CONF_UNIT_SYSTEM) == "metric" - assert sensor.attributes.get(ATTR_DURATION_IN_TRAFFIC) == 89.16666666666667 - assert sensor.attributes.get(ATTR_ORIGIN) == origin - assert sensor.attributes.get(ATTR_DESTINATION) == destination - assert sensor.attributes.get(ATTR_ORIGIN_NAME) == "Mannheim Rd" - assert sensor.attributes.get(ATTR_DESTINATION_NAME) == "" - assert sensor.attributes.get(CONF_MODE) == TRAVEL_MODE_PUBLIC - assert sensor.attributes.get(CONF_TRAFFIC_MODE) is False - - assert sensor.attributes.get(ATTR_ICON) == ICON_PUBLIC - - -async def test_public_transport_time_table(hass, requests_mock_credentials_check): - """Test that publicTransportTimeTable works.""" - origin = "41.9798,-87.8801" - destination = "41.9043,-87.9216" - modes = [ROUTE_MODE_FASTEST, TRAVEL_MODE_PUBLIC_TIME_TABLE, TRAFFIC_MODE_DISABLED] - response_url = _build_mock_url(origin, destination, modes, API_KEY) - requests_mock_credentials_check.get( - response_url, - text=load_fixture("here_travel_time/public_time_table_response.json"), - ) - - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_latitude": origin.split(",")[0], - "origin_longitude": origin.split(",")[1], - "destination_latitude": destination.split(",")[0], - "destination_longitude": destination.split(",")[1], +@pytest.mark.parametrize( + "time_type,time", + [ + (DEPARTURE_TIME, "08:00:00"), + (DEPARTURE_TIME, "now"), + (DEPARTURE_TIME, ""), + (ARRIVAL_TIME, "08:00:00"), + ], +) +async def test_options(hass, time_type, time, valid_response): + """Test that different options work.""" + entry = MockConfigEntry( + domain=DOMAIN, + unique_id="0123456789", + data={ + "origin": f"{CAR_ORIGIN_LATITUDE},{CAR_ORIGIN_LONGITUDE}", + "destination": f"{CAR_DESTINATION_LATITUDE},{CAR_DESTINATION_LONGITUDE}", "api_key": API_KEY, - "mode": TRAVEL_MODE_PUBLIC_TIME_TABLE, - } - } - 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 == "80" - assert sensor.attributes.get("unit_of_measurement") == TIME_MINUTES - - assert sensor.attributes.get(ATTR_ATTRIBUTION) is None - assert sensor.attributes.get(ATTR_DURATION) == 79.73333333333333 - assert sensor.attributes.get(ATTR_DISTANCE) == 14.775 - assert sensor.attributes.get(ATTR_ROUTE) == ( - "330 - Archer/Harlem (Terminal); 309 - Elmhurst Metra Station" - ) - assert sensor.attributes.get(CONF_UNIT_SYSTEM) == "metric" - assert sensor.attributes.get(ATTR_DURATION_IN_TRAFFIC) == 79.73333333333333 - assert sensor.attributes.get(ATTR_ORIGIN) == origin - assert sensor.attributes.get(ATTR_DESTINATION) == destination - assert sensor.attributes.get(ATTR_ORIGIN_NAME) == "Mannheim Rd" - assert sensor.attributes.get(ATTR_DESTINATION_NAME) == "" - assert sensor.attributes.get(CONF_MODE) == TRAVEL_MODE_PUBLIC_TIME_TABLE - assert sensor.attributes.get(CONF_TRAFFIC_MODE) is False - - assert sensor.attributes.get(ATTR_ICON) == ICON_PUBLIC - - -async def test_pedestrian(hass, requests_mock_credentials_check): - """Test that pedestrian works.""" - origin = "41.9798,-87.8801" - destination = "41.9043,-87.9216" - modes = [ROUTE_MODE_FASTEST, TRAVEL_MODE_PEDESTRIAN, TRAFFIC_MODE_DISABLED] - response_url = _build_mock_url(origin, destination, modes, API_KEY) - requests_mock_credentials_check.get( - response_url, text=load_fixture("here_travel_time/pedestrian_response.json") - ) - - config = { - DOMAIN: { - "platform": PLATFORM, + "mode": TRAVEL_MODE_CAR, "name": "test", - "origin_latitude": origin.split(",")[0], - "origin_longitude": origin.split(",")[1], - "destination_latitude": destination.split(",")[0], - "destination_longitude": destination.split(",")[1], - "api_key": API_KEY, - "mode": TRAVEL_MODE_PEDESTRIAN, - } - } - - 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 == "211" - assert sensor.attributes.get("unit_of_measurement") == TIME_MINUTES - - assert sensor.attributes.get(ATTR_ATTRIBUTION) is None - assert sensor.attributes.get(ATTR_DURATION) == 210.51666666666668 - assert sensor.attributes.get(ATTR_DISTANCE) == 12.533 - assert sensor.attributes.get(ATTR_ROUTE) == ( - "Mannheim Rd; W Belmont Ave; Cullerton St; E Fullerton Ave; " - "La Porte Ave; E Palmer Ave; N Railroad Ave; W North Ave; " - "E North Ave; E Third St" + }, + options={ + CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED, + CONF_ROUTE_MODE: ROUTE_MODE_FASTEST, + CONF_TIME_TYPE: time_type, + CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_IMPERIAL, + CONF_TIME: time, + }, ) - assert sensor.attributes.get(CONF_UNIT_SYSTEM) == "metric" - assert sensor.attributes.get(ATTR_DURATION_IN_TRAFFIC) == 210.51666666666668 - assert sensor.attributes.get(ATTR_ORIGIN) == origin - assert sensor.attributes.get(ATTR_DESTINATION) == destination - assert sensor.attributes.get(ATTR_ORIGIN_NAME) == "Mannheim Rd" - assert sensor.attributes.get(ATTR_DESTINATION_NAME) == "" - assert sensor.attributes.get(CONF_MODE) == TRAVEL_MODE_PEDESTRIAN - assert sensor.attributes.get(CONF_TRAFFIC_MODE) is False - - assert sensor.attributes.get(ATTR_ICON) == ICON_PEDESTRIAN + entry.add_to_hass(hass) + await hass.config_entries.async_setup(entry.entry_id) -async def test_bicycle(hass, requests_mock_credentials_check): - """Test that bicycle works.""" - origin = "41.9798,-87.8801" - destination = "41.9043,-87.9216" - modes = [ROUTE_MODE_FASTEST, TRAVEL_MODE_BICYCLE, TRAFFIC_MODE_DISABLED] - response_url = _build_mock_url(origin, destination, modes, API_KEY) - requests_mock_credentials_check.get( - response_url, text=load_fixture("here_travel_time/bike_response.json") - ) - - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_latitude": origin.split(",")[0], - "origin_longitude": origin.split(",")[1], - "destination_latitude": destination.split(",")[0], - "destination_longitude": destination.split(",")[1], - "api_key": API_KEY, - "mode": TRAVEL_MODE_BICYCLE, - } - } - 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 == "55" - assert sensor.attributes.get("unit_of_measurement") == TIME_MINUTES - - assert sensor.attributes.get(ATTR_ATTRIBUTION) is None - assert sensor.attributes.get(ATTR_DURATION) == 54.86666666666667 - assert sensor.attributes.get(ATTR_DISTANCE) == 12.613 - assert sensor.attributes.get(ATTR_ROUTE) == ( - "Mannheim Rd; W Belmont Ave; Cullerton St; N Landen Dr; " - "E Fullerton Ave; N Wolf Rd; W North Ave; N Clinton Ave; " - "E Third St; N Caroline Ave" - ) - assert sensor.attributes.get(CONF_UNIT_SYSTEM) == "metric" - assert sensor.attributes.get(ATTR_DURATION_IN_TRAFFIC) == 54.86666666666667 - assert sensor.attributes.get(ATTR_ORIGIN) == origin - assert sensor.attributes.get(ATTR_DESTINATION) == destination - assert sensor.attributes.get(ATTR_ORIGIN_NAME) == "Mannheim Rd" - assert sensor.attributes.get(ATTR_DESTINATION_NAME) == "" - assert sensor.attributes.get(CONF_MODE) == TRAVEL_MODE_BICYCLE - assert sensor.attributes.get(CONF_TRAFFIC_MODE) is False - - assert sensor.attributes.get(ATTR_ICON) == ICON_BICYCLE - - -async def test_location_zone(hass, requests_mock_truck_response, legacy_patchable_time): - """Test that origin/destination supplied by a zone works.""" - utcnow = dt_util.utcnow() - # Patching 'utcnow' to gain more control over the timed update. - with patch("homeassistant.util.dt.utcnow", return_value=utcnow): - zone_config = { - "zone": [ - { - "name": "Destination", - "latitude": TRUCK_DESTINATION_LATITUDE, - "longitude": TRUCK_DESTINATION_LONGITUDE, - "radius": 250, - "passive": False, - }, - { - "name": "Origin", - "latitude": TRUCK_ORIGIN_LATITUDE, - "longitude": TRUCK_ORIGIN_LONGITUDE, - "radius": 250, - "passive": False, - }, - ] - } - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_entity_id": "zone.origin", - "destination_entity_id": "zone.destination", - "api_key": API_KEY, - "mode": TRAVEL_MODE_TRUCK, - } - } - assert await async_setup_component(hass, "zone", zone_config) - 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_truck_sensor(sensor) - - # Test that update works more than once - async_fire_time_changed(hass, utcnow + SCAN_INTERVAL) - await hass.async_block_till_done() - - sensor = hass.states.get("sensor.test") - _assert_truck_sensor(sensor) - - -async def test_location_sensor( - hass, requests_mock_truck_response, legacy_patchable_time -): - """Test that origin/destination supplied by a sensor works.""" - utcnow = dt_util.utcnow() - # Patching 'utcnow' to gain more control over the timed update. - with patch("homeassistant.util.dt.utcnow", return_value=utcnow): - hass.states.async_set( - "sensor.origin", ",".join([TRUCK_ORIGIN_LATITUDE, TRUCK_ORIGIN_LONGITUDE]) - ) - hass.states.async_set( - "sensor.destination", - ",".join([TRUCK_DESTINATION_LATITUDE, TRUCK_DESTINATION_LONGITUDE]), - ) - - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_entity_id": "sensor.origin", - "destination_entity_id": "sensor.destination", - "api_key": API_KEY, - "mode": TRAVEL_MODE_TRUCK, - } - } - 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_truck_sensor(sensor) - - # Test that update works more than once - async_fire_time_changed(hass, utcnow + SCAN_INTERVAL) - await hass.async_block_till_done() - - sensor = hass.states.get("sensor.test") - _assert_truck_sensor(sensor) - - -async def test_location_person( - hass, requests_mock_truck_response, legacy_patchable_time -): - """Test that origin/destination supplied by a person works.""" - utcnow = dt_util.utcnow() - # Patching 'utcnow' to gain more control over the timed update. - with patch("homeassistant.util.dt.utcnow", return_value=utcnow): - hass.states.async_set( - "person.origin", - "unknown", - { - "latitude": float(TRUCK_ORIGIN_LATITUDE), - "longitude": float(TRUCK_ORIGIN_LONGITUDE), - }, - ) - hass.states.async_set( - "person.destination", - "unknown", - { - "latitude": float(TRUCK_DESTINATION_LATITUDE), - "longitude": float(TRUCK_DESTINATION_LONGITUDE), - }, - ) - - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_entity_id": "person.origin", - "destination_entity_id": "person.destination", - "api_key": API_KEY, - "mode": TRAVEL_MODE_TRUCK, - } - } - 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_truck_sensor(sensor) - - # Test that update works more than once - async_fire_time_changed(hass, utcnow + SCAN_INTERVAL) - await hass.async_block_till_done() - - sensor = hass.states.get("sensor.test") - _assert_truck_sensor(sensor) - - -async def test_location_device_tracker( - hass, requests_mock_truck_response, legacy_patchable_time -): - """Test that origin/destination supplied by a device_tracker works.""" - utcnow = dt_util.utcnow() - # Patching 'utcnow' to gain more control over the timed update. - with patch("homeassistant.util.dt.utcnow", return_value=utcnow): - hass.states.async_set( - "device_tracker.origin", - "unknown", - { - "latitude": float(TRUCK_ORIGIN_LATITUDE), - "longitude": float(TRUCK_ORIGIN_LONGITUDE), - }, - ) - hass.states.async_set( - "device_tracker.destination", - "unknown", - { - "latitude": float(TRUCK_DESTINATION_LATITUDE), - "longitude": float(TRUCK_DESTINATION_LONGITUDE), - }, - ) - - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_entity_id": "device_tracker.origin", - "destination_entity_id": "device_tracker.destination", - "api_key": API_KEY, - "mode": TRAVEL_MODE_TRUCK, - } - } - 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_truck_sensor(sensor) - - # Test that update works more than once - async_fire_time_changed(hass, utcnow + SCAN_INTERVAL) - await hass.async_block_till_done() - - sensor = hass.states.get("sensor.test") - _assert_truck_sensor(sensor) - - -async def test_location_device_tracker_added_after_update( - hass, requests_mock_truck_response, legacy_patchable_time, caplog -): - """Test that device_tracker added after first update works.""" - caplog.set_level(logging.ERROR) - utcnow = dt_util.utcnow() - # Patching 'utcnow' to gain more control over the timed update. - with patch("homeassistant.util.dt.utcnow", return_value=utcnow): - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_entity_id": "device_tracker.origin", - "destination_entity_id": "device_tracker.destination", - "api_key": API_KEY, - "mode": TRAVEL_MODE_TRUCK, - } - } - 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 "Unable to find entity" in caplog.text - caplog.clear() - - # Device tracker appear after first update - hass.states.async_set( - "device_tracker.origin", - "unknown", - { - "latitude": float(TRUCK_ORIGIN_LATITUDE), - "longitude": float(TRUCK_ORIGIN_LONGITUDE), - }, - ) - hass.states.async_set( - "device_tracker.destination", - "unknown", - { - "latitude": float(TRUCK_DESTINATION_LATITUDE), - "longitude": float(TRUCK_DESTINATION_LONGITUDE), - }, - ) - - # Test that update works more than once - async_fire_time_changed(hass, utcnow + SCAN_INTERVAL) - await hass.async_block_till_done() - - sensor = hass.states.get("sensor.test") - _assert_truck_sensor(sensor) - assert len(caplog.records) == 0 + assert sensor.attributes.get(ATTR_DISTANCE) == 14.852635608048994 -async def test_location_device_tracker_in_zone( - hass, requests_mock_truck_response, caplog -): - """Test that device_tracker in zone uses device_tracker state works.""" - caplog.set_level(logging.DEBUG) +async def test_entity_ids(hass, valid_response): + """Test that resolving an entity works.""" zone_config = { "zone": [ + { + "name": "Destination", + "latitude": CAR_ORIGIN_LATITUDE, + "longitude": CAR_ORIGIN_LONGITUDE, + "radius": 250, + "passive": False, + }, { "name": "Origin", - "latitude": TRUCK_ORIGIN_LATITUDE, - "longitude": TRUCK_ORIGIN_LONGITUDE, + "latitude": CAR_DESTINATION_LATITUDE, + "longitude": CAR_DESTINATION_LONGITUDE, "radius": 250, "passive": False, - } + }, ] } assert await async_setup_component(hass, "zone", zone_config) - hass.states.async_set( - "device_tracker.origin", "origin", {"latitude": None, "longitude": None} - ) - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_entity_id": "device_tracker.origin", - "destination_latitude": TRUCK_DESTINATION_LATITUDE, - "destination_longitude": TRUCK_DESTINATION_LONGITUDE, - "api_key": API_KEY, - "mode": TRAVEL_MODE_TRUCK, - } - } - 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_truck_sensor(sensor) - assert ", getting zone location" in caplog.text - -async def test_route_not_found(hass, requests_mock_credentials_check, caplog): - """Test that route not found error is correctly handled.""" - caplog.set_level(logging.ERROR) - origin = "52.516,13.3779" - destination = "47.013399,-10.171986" - modes = [ROUTE_MODE_FASTEST, TRAVEL_MODE_CAR, TRAFFIC_MODE_DISABLED] - response_url = _build_mock_url(origin, destination, modes, API_KEY) - requests_mock_credentials_check.get( - response_url, - text=load_fixture("here_travel_time/routing_error_no_route_found.json"), + entry = MockConfigEntry( + domain=DOMAIN, + unique_id="0123456789", + data={ + "origin": "zone.origin", + "destination": "zone.destination", + "api_key": API_KEY, + "mode": TRAVEL_MODE_CAR, + "name": "test", + }, + options={ + CONF_TRAFFIC_MODE: TRAFFIC_MODE_ENABLED, + CONF_ROUTE_MODE: ROUTE_MODE_FASTEST, + CONF_TIME_TYPE: DEPARTURE_TIME, + CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_IMPERIAL, + CONF_TIME: "now", + }, ) + entry.add_to_hass(hass) - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_latitude": origin.split(",")[0], - "origin_longitude": origin.split(",")[1], - "destination_latitude": destination.split(",")[0], - "destination_longitude": destination.split(",")[1], - "api_key": API_KEY, - } - } - assert await async_setup_component(hass, DOMAIN, config) - await hass.async_block_till_done() + await hass.config_entries.async_setup(entry.entry_id) - hass.bus.async_fire(EVENT_HOMEASSISTANT_START) await hass.async_block_till_done() - assert len(caplog.records) == 1 - assert NO_ROUTE_ERROR_MESSAGE in caplog.text + sensor = hass.states.get("sensor.test") + assert sensor.attributes.get(ATTR_DISTANCE) == 14.852635608048994 -async def test_pattern_origin(hass, caplog): - """Test that pattern matching the origin works.""" - caplog.set_level(logging.ERROR) - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_latitude": "138.90", - "origin_longitude": "-77.04833", - "destination_latitude": CAR_DESTINATION_LATITUDE, - "destination_longitude": CAR_DESTINATION_LONGITUDE, +async def test_route_not_found(hass, caplog, failed_response): + """Test that no route error is logged.""" + entry = MockConfigEntry( + domain=DOMAIN, + unique_id="0123456789", + data={ + "origin": f"{CAR_ORIGIN_LATITUDE},{CAR_ORIGIN_LONGITUDE}", + "destination": f"{CAR_DESTINATION_LATITUDE},{CAR_DESTINATION_LONGITUDE}", "api_key": API_KEY, - } - } - assert await async_setup_component(hass, DOMAIN, config) - await hass.async_block_till_done() - assert "invalid latitude" in caplog.text + "mode": TRAVEL_MODE_CAR, + "name": "test", + }, + ) + entry.add_to_hass(hass) + await hass.config_entries.async_setup(entry.entry_id) -async def test_pattern_destination(hass, caplog): - """Test that pattern matching the destination works.""" - caplog.set_level(logging.ERROR) - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_latitude": CAR_ORIGIN_LATITUDE, - "origin_longitude": CAR_ORIGIN_LONGITUDE, - "destination_latitude": "139.0", - "destination_longitude": "-77.1", - "api_key": API_KEY, - } - } - assert await async_setup_component(hass, DOMAIN, config) await hass.async_block_till_done() - assert "invalid latitude" in caplog.text + assert NO_ROUTE_ERROR_MESSAGE in caplog.text -async def test_invalid_credentials(hass, requests_mock, caplog): - """Test that invalid credentials error is correctly handled.""" - caplog.set_level(logging.ERROR) - modes = [ROUTE_MODE_FASTEST, TRAVEL_MODE_CAR, TRAFFIC_MODE_DISABLED] - response_url = _build_mock_url( - ",".join([CAR_ORIGIN_LATITUDE, CAR_ORIGIN_LONGITUDE]), - ",".join([CAR_DESTINATION_LATITUDE, CAR_DESTINATION_LONGITUDE]), - modes, - API_KEY, - ) - requests_mock.get( - response_url, - text=load_fixture("here_travel_time/routing_error_invalid_credentials.json"), - ) +async def test_setup_platform(hass, caplog, valid_response): + """Test that setup platform migration works.""" config = { - DOMAIN: { - "platform": PLATFORM, + "sensor": { + "platform": DOMAIN, "name": "test", "origin_latitude": CAR_ORIGIN_LATITUDE, "origin_longitude": CAR_ORIGIN_LONGITUDE, @@ -955,249 +251,13 @@ async def test_invalid_credentials(hass, requests_mock, caplog): "api_key": API_KEY, } } - assert await async_setup_component(hass, DOMAIN, config) - await hass.async_block_till_done() - assert len(caplog.records) == 1 - assert "Invalid credentials" in caplog.text - - -async def test_attribution(hass, requests_mock_credentials_check): - """Test that attributions are correctly displayed.""" - origin = "50.037751372637686,14.39233448220898" - destination = "50.07993838201255,14.42582157361062" - modes = [ROUTE_MODE_SHORTEST, TRAVEL_MODE_PUBLIC_TIME_TABLE, TRAFFIC_MODE_ENABLED] - response_url = _build_mock_url(origin, destination, modes, API_KEY) - requests_mock_credentials_check.get( - response_url, text=load_fixture("here_travel_time/attribution_response.json") - ) - - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_latitude": origin.split(",")[0], - "origin_longitude": origin.split(",")[1], - "destination_latitude": destination.split(",")[0], - "destination_longitude": destination.split(",")[1], - "api_key": API_KEY, - "traffic_mode": True, - "route_mode": ROUTE_MODE_SHORTEST, - "mode": TRAVEL_MODE_PUBLIC_TIME_TABLE, - } - } - 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() + with patch( + "homeassistant.components.here_travel_time.async_setup_entry", return_value=True + ): + await async_setup_component(hass, "sensor", config) + await hass.async_block_till_done() - sensor = hass.states.get("sensor.test") assert ( - sensor.attributes.get(ATTR_ATTRIBUTION) - == "With the support of HERE Technologies. All information is provided without warranty of any kind." - ) - - -async def test_pattern_entity_state(hass, requests_mock_truck_response, caplog): - """Test that pattern matching the state of an entity works.""" - caplog.set_level(logging.ERROR) - hass.states.async_set("sensor.origin", "invalid") - - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_entity_id": "sensor.origin", - "destination_latitude": TRUCK_DESTINATION_LATITUDE, - "destination_longitude": TRUCK_DESTINATION_LONGITUDE, - "api_key": API_KEY, - "mode": TRAVEL_MODE_TRUCK, - } - } - 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() - - assert len(caplog.records) == 1 - assert "is not a valid set of coordinates" in caplog.text - - -async def test_pattern_entity_state_with_space(hass, requests_mock_truck_response): - """Test that pattern matching the state including a space of an entity works.""" - hass.states.async_set( - "sensor.origin", ", ".join([TRUCK_ORIGIN_LATITUDE, TRUCK_ORIGIN_LONGITUDE]) - ) - - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_entity_id": "sensor.origin", - "destination_latitude": TRUCK_DESTINATION_LATITUDE, - "destination_longitude": TRUCK_DESTINATION_LONGITUDE, - "api_key": API_KEY, - "mode": TRAVEL_MODE_TRUCK, - } - } - assert await async_setup_component(hass, DOMAIN, config) - await hass.async_block_till_done() - - -async def test_delayed_update(hass, requests_mock_truck_response, caplog): - """Test that delayed update does not complain about missing entities.""" - caplog.set_level(logging.WARNING) - - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_entity_id": "sensor.origin", - "destination_latitude": TRUCK_DESTINATION_LATITUDE, - "destination_longitude": TRUCK_DESTINATION_LONGITUDE, - "api_key": API_KEY, - "mode": TRAVEL_MODE_TRUCK, - } - } - sensor_config = { - "sensor": { - "platform": "template", - "sensors": [ - {"template_sensor": {"value_template": "{{states('sensor.origin')}}"}} - ], - } - } - assert await async_setup_component(hass, DOMAIN, config) - await hass.async_block_till_done() - assert await async_setup_component(hass, "sensor", sensor_config) - hass.states.async_set( - "sensor.origin", ",".join([TRUCK_ORIGIN_LATITUDE, TRUCK_ORIGIN_LONGITUDE]) - ) - - hass.bus.async_fire(EVENT_HOMEASSISTANT_START) - await hass.async_block_till_done() - - assert "Unable to find entity" not in caplog.text - - -async def test_arrival(hass, requests_mock_credentials_check): - """Test that arrival works.""" - origin = "41.9798,-87.8801" - destination = "41.9043,-87.9216" - arrival = "01:00:00" - arrival_isodate = convert_time_to_isodate(arrival) - modes = [ROUTE_MODE_FASTEST, TRAVEL_MODE_PUBLIC_TIME_TABLE, TRAFFIC_MODE_DISABLED] - response_url = _build_mock_url( - origin, destination, modes, API_KEY, arrival=arrival_isodate - ) - requests_mock_credentials_check.get( - response_url, - text=load_fixture("here_travel_time/public_time_table_response.json"), + "Your HERE travel time configuration has been imported into the UI" + in caplog.text ) - - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_latitude": origin.split(",")[0], - "origin_longitude": origin.split(",")[1], - "destination_latitude": destination.split(",")[0], - "destination_longitude": destination.split(",")[1], - "api_key": API_KEY, - "mode": TRAVEL_MODE_PUBLIC_TIME_TABLE, - "arrival": arrival, - } - } - 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 == "80" - - -async def test_departure(hass, requests_mock_credentials_check): - """Test that arrival works.""" - origin = "41.9798,-87.8801" - destination = "41.9043,-87.9216" - departure = "23:00:00" - departure_isodate = convert_time_to_isodate(departure) - modes = [ROUTE_MODE_FASTEST, TRAVEL_MODE_PUBLIC_TIME_TABLE, TRAFFIC_MODE_DISABLED] - response_url = _build_mock_url( - origin, destination, modes, API_KEY, departure=departure_isodate - ) - requests_mock_credentials_check.get( - response_url, - text=load_fixture("here_travel_time/public_time_table_response.json"), - ) - - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_latitude": origin.split(",")[0], - "origin_longitude": origin.split(",")[1], - "destination_latitude": destination.split(",")[0], - "destination_longitude": destination.split(",")[1], - "api_key": API_KEY, - "mode": TRAVEL_MODE_PUBLIC_TIME_TABLE, - "departure": departure, - } - } - 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 == "80" - - -async def test_arrival_only_allowed_for_timetable(hass, caplog): - """Test that arrival is only allowed when mode is publicTransportTimeTable.""" - caplog.set_level(logging.ERROR) - origin = "41.9798,-87.8801" - destination = "41.9043,-87.9216" - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_latitude": origin.split(",")[0], - "origin_longitude": origin.split(",")[1], - "destination_latitude": destination.split(",")[0], - "destination_longitude": destination.split(",")[1], - "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.""" - caplog.set_level(logging.ERROR) - origin = "41.9798,-87.8801" - destination = "41.9043,-87.9216" - config = { - DOMAIN: { - "platform": PLATFORM, - "name": "test", - "origin_latitude": origin.split(",")[0], - "origin_longitude": origin.split(",")[1], - "destination_latitude": destination.split(",")[0], - "destination_longitude": destination.split(",")[1], - "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