home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-06-25 01:21:51 +02:00
Code Issues Packages Projects Releases Wiki Activity

Use google-maps-routing in google_travel_time (#140691)

Browse Source 
Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
Kevin Stillhammer
2025-04-30 18:22:15 +02:00
committed by GitHub
parent 2c118d4850
commit 02bd8d67c8
14 changed files with 609 additions and 600 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
homeassistant/components/google_travel_time/__init__.py
View File

@ -1,11 +1,18 @@
"""The google_travel_time component."""
import logging
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
from .const import CONF_TIME
PLATFORMS = [Platform.SENSOR]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Google Maps Travel Time from a config entry."""
@ -16,3 +23,37 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Migrate an old config entry."""
if config_entry.version == 1:
_LOGGER.debug(
"Migrating from version %s.%s",
config_entry.version,
config_entry.minor_version,
)
options = dict(config_entry.options)
if options.get(CONF_TIME) == "now":
options[CONF_TIME] = None
elif options.get(CONF_TIME) is not None:
if dt_util.parse_time(options[CONF_TIME]) is None:
try:
from_timestamp = dt_util.utc_from_timestamp(int(options[CONF_TIME]))
options[CONF_TIME] = (
f"{from_timestamp.time().hour:02}:{from_timestamp.time().minute:02}"
)
except ValueError:
_LOGGER.error(
"Invalid time format found while migrating: %s. The old config never worked. Reset to default (empty)",
options[CONF_TIME],
)
options[CONF_TIME] = None
hass.config_entries.async_update_entry(config_entry, options=options, version=2)
_LOGGER.debug(
"Migration to version %s.%s successful",
config_entry.version,
config_entry.minor_version,
)
return True

8
homeassistant/components/google_travel_time/config_flow.py
View File

@ -19,6 +19,7 @@ from homeassistant.helpers.selector import (
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
TimeSelector,
)
from homeassistant.util.unit_system import US_CUSTOMARY_SYSTEM
@ -106,7 +107,7 @@ OPTIONS_SCHEMA = vol.Schema(
translation_key=CONF_TIME_TYPE,
)
),
vol.Optional(CONF_TIME, default=""): cv.string,
vol.Optional(CONF_TIME): TimeSelector(),
vol.Optional(CONF_TRAFFIC_MODEL): SelectSelector(
SelectSelectorConfig(
options=TRAFFIC_MODELS,
@ -181,8 +182,7 @@ async def validate_input(
) -> dict[str, str] | None:
"""Validate the user input allows us to connect."""
try:
await hass.async_add_executor_job(
validate_config_entry,
await validate_config_entry(
hass,
user_input[CONF_API_KEY],
user_input[CONF_ORIGIN],
@ -201,7 +201,7 @@ async def validate_input(
class GoogleTravelTimeConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Google Maps Travel Time."""
VERSION = 1
VERSION = 2
@staticmethod
@callback

34
homeassistant/components/google_travel_time/const.py
View File

@ -1,5 +1,12 @@
"""Constants for Google Travel Time."""
from google.maps.routing_v2 import (
RouteTravelMode,
TrafficModel,
TransitPreferences,
Units,
)
DOMAIN = "google_travel_time"
ATTRIBUTION = "Powered by Google"
@ -7,7 +14,6 @@ ATTRIBUTION = "Powered by Google"
CONF_DESTINATION = "destination"
CONF_OPTIONS = "options"
CONF_ORIGIN = "origin"
CONF_TRAVEL_MODE = "travel_mode"
CONF_AVOID = "avoid"
CONF_UNITS = "units"
CONF_ARRIVAL_TIME = "arrival_time"
@ -79,11 +85,37 @@ ALL_LANGUAGES = [
AVOID_OPTIONS = ["tolls", "highways", "ferries", "indoor"]
TRANSIT_PREFS = ["less_walking", "fewer_transfers"]
TRANSIT_PREFS_TO_GOOGLE_SDK_ENUM = {
"less_walking": TransitPreferences.TransitRoutingPreference.LESS_WALKING,
"fewer_transfers": TransitPreferences.TransitRoutingPreference.FEWER_TRANSFERS,
}
TRANSPORT_TYPES = ["bus", "subway", "train", "tram", "rail"]
TRANSPORT_TYPES_TO_GOOGLE_SDK_ENUM = {
"bus": TransitPreferences.TransitTravelMode.BUS,
"subway": TransitPreferences.TransitTravelMode.SUBWAY,
"train": TransitPreferences.TransitTravelMode.TRAIN,
"tram": TransitPreferences.TransitTravelMode.LIGHT_RAIL,
"rail": TransitPreferences.TransitTravelMode.RAIL,
}
TRAVEL_MODES = ["driving", "walking", "bicycling", "transit"]
TRAVEL_MODES_TO_GOOGLE_SDK_ENUM = {
"driving": RouteTravelMode.DRIVE,
"walking": RouteTravelMode.WALK,
"bicycling": RouteTravelMode.BICYCLE,
"transit": RouteTravelMode.TRANSIT,
}
TRAFFIC_MODELS = ["best_guess", "pessimistic", "optimistic"]
TRAFFIC_MODELS_TO_GOOGLE_SDK_ENUM = {
"best_guess": TrafficModel.BEST_GUESS,
"pessimistic": TrafficModel.PESSIMISTIC,
"optimistic": TrafficModel.OPTIMISTIC,
}
# googlemaps library uses "metric" or "imperial" terminology in distance_matrix
UNITS_METRIC = "metric"
UNITS_IMPERIAL = "imperial"
UNITS = [UNITS_METRIC, UNITS_IMPERIAL]
UNITS_TO_GOOGLE_SDK_ENUM = {
UNITS_METRIC: Units.METRIC,
UNITS_IMPERIAL: Units.IMPERIAL,
}

83
homeassistant/components/google_travel_time/helpers.py
View File

@ -2,41 +2,80 @@
import logging
from googlemaps import Client
from googlemaps.distance_matrix import distance_matrix
from googlemaps.exceptions import ApiError, Timeout, TransportError
from google.api_core.client_options import ClientOptions
from google.api_core.exceptions import (
Forbidden,
GatewayTimeout,
GoogleAPIError,
Unauthorized,
)
from google.maps.routing_v2 import (
ComputeRoutesRequest,
Location,
RoutesAsyncClient,
RouteTravelMode,
Waypoint,
)
from google.type import latlng_pb2
import voluptuous as vol
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.location import find_coordinates
_LOGGER = logging.getLogger(__name__)
def validate_config_entry(
def convert_to_waypoint(hass: HomeAssistant, location: str) -> Waypoint | None:
"""Convert a location to a Waypoint.
Will either use coordinates or if none are found, use the location as an address.
"""
coordinates = find_coordinates(hass, location)
if coordinates is None:
return None
try:
formatted_coordinates = coordinates.split(",")
vol.Schema(cv.gps(formatted_coordinates))
except (AttributeError, vol.ExactSequenceInvalid):
return Waypoint(address=location)
return Waypoint(
location=Location(
lat_lng=latlng_pb2.LatLng(
latitude=float(formatted_coordinates[0]),
longitude=float(formatted_coordinates[1]),
)
)
)
async def validate_config_entry(
hass: HomeAssistant, api_key: str, origin: str, destination: str
) -> None:
"""Return whether the config entry data is valid."""
resolved_origin = find_coordinates(hass, origin)
resolved_destination = find_coordinates(hass, destination)
resolved_origin = convert_to_waypoint(hass, origin)
resolved_destination = convert_to_waypoint(hass, destination)
client_options = ClientOptions(api_key=api_key)
client = RoutesAsyncClient(client_options=client_options)
field_mask = "routes.duration"
request = ComputeRoutesRequest(
origin=resolved_origin,
destination=resolved_destination,
travel_mode=RouteTravelMode.DRIVE,
)
try:
client = Client(api_key, timeout=10)
except ValueError as value_error:
_LOGGER.error("Malformed API key")
raise InvalidApiKeyException from value_error
try:
distance_matrix(client, resolved_origin, resolved_destination, mode="driving")
except ApiError as api_error:
if api_error.status == "REQUEST_DENIED":
_LOGGER.error("Request denied: %s", api_error.message)
raise InvalidApiKeyException from api_error
_LOGGER.error("Unknown error: %s", api_error.message)
raise UnknownException from api_error
except TransportError as transport_error:
_LOGGER.error("Unknown error: %s", transport_error)
raise UnknownException from transport_error
except Timeout as timeout_error:
await client.compute_routes(
request, metadata=[("x-goog-fieldmask", field_mask)]
)
except (Unauthorized, Forbidden) as unauthorized_error:
_LOGGER.error("Request denied: %s", unauthorized_error.message)
raise InvalidApiKeyException from unauthorized_error
except GatewayTimeout as timeout_error:
_LOGGER.error("Timeout error")
raise TimeoutError from timeout_error
except GoogleAPIError as unknown_error:
_LOGGER.error("Unknown error: %s", unknown_error)
raise UnknownException from unknown_error
class InvalidApiKeyException(Exception):

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

@ -5,6 +5,6 @@
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/google_travel_time",
"iot_class": "cloud_polling",
"loggers": ["googlemaps", "homeassistant.helpers.location"],
"requirements": ["googlemaps==2.5.1"]
"loggers": ["google", "homeassistant.helpers.location"],
"requirements": ["google-maps-routing==0.6.14"]
}

225
homeassistant/components/google_travel_time/sensor.py
View File

@ -2,12 +2,22 @@
from __future__ import annotations
from datetime import datetime, timedelta
import datetime
import logging
from typing import TYPE_CHECKING, Any
from googlemaps import Client
from googlemaps.distance_matrix import distance_matrix
from googlemaps.exceptions import ApiError, Timeout, TransportError
from google.api_core.client_options import ClientOptions
from google.api_core.exceptions import GoogleAPIError
from google.maps.routing_v2 import (
ComputeRoutesRequest,
Route,
RouteModifiers,
RoutesAsyncClient,
RouteTravelMode,
RoutingPreference,
TransitPreferences,
)
from google.protobuf import timestamp_pb2
from homeassistant.components.sensor import (
SensorDeviceClass,
@ -17,6 +27,8 @@ from homeassistant.components.sensor import (
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_API_KEY,
CONF_LANGUAGE,
CONF_MODE,
CONF_NAME,
EVENT_HOMEASSISTANT_STARTED,
UnitOfTime,
@ -30,26 +42,49 @@ from homeassistant.util import dt as dt_util
from .const import (
ATTRIBUTION,
CONF_ARRIVAL_TIME,
CONF_AVOID,
CONF_DEPARTURE_TIME,
CONF_DESTINATION,
CONF_ORIGIN,
CONF_TRAFFIC_MODEL,
CONF_TRANSIT_MODE,
CONF_TRANSIT_ROUTING_PREFERENCE,
CONF_UNITS,
DEFAULT_NAME,
DOMAIN,
TRAFFIC_MODELS_TO_GOOGLE_SDK_ENUM,
TRANSIT_PREFS_TO_GOOGLE_SDK_ENUM,
TRANSPORT_TYPES_TO_GOOGLE_SDK_ENUM,
TRAVEL_MODES_TO_GOOGLE_SDK_ENUM,
UNITS_TO_GOOGLE_SDK_ENUM,
)
from .helpers import convert_to_waypoint
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=5)
SCAN_INTERVAL = datetime.timedelta(minutes=10)
FIELD_MASK = "routes.duration,routes.localized_values"
def convert_time_to_utc(timestr):
"""Take a string like 08:00:00 and convert it to a unix timestamp."""
combined = datetime.combine(
dt_util.start_of_local_day(), dt_util.parse_time(timestr)
def convert_time(time_str: str) -> timestamp_pb2.Timestamp | None:
"""Convert a string like '08:00' to a google pb2 Timestamp.
If the time is in the past, it will be shifted to the next day.
"""
parsed_time = dt_util.parse_time(time_str)
if TYPE_CHECKING:
assert parsed_time is not None
start_of_day = dt_util.start_of_local_day()
combined = datetime.datetime.combine(
start_of_day,
parsed_time,
start_of_day.tzinfo,
)
if combined < datetime.now():
combined = combined + timedelta(days=1)
return dt_util.as_timestamp(combined)
if combined < dt_util.now():
combined = combined + datetime.timedelta(days=1)
timestamp = timestamp_pb2.Timestamp()
timestamp.FromDatetime(dt=combined)
return timestamp
async def async_setup_entry(
@ -63,7 +98,8 @@ async def async_setup_entry(
destination = config_entry.data[CONF_DESTINATION]
name = config_entry.data.get(CONF_NAME, DEFAULT_NAME)
client = Client(api_key, timeout=10)
client_options = ClientOptions(api_key=api_key)
client = RoutesAsyncClient(client_options=client_options)
sensor = GoogleTravelTimeSensor(
config_entry, name, api_key, origin, destination, client
@ -80,7 +116,15 @@ class GoogleTravelTimeSensor(SensorEntity):
_attr_device_class = SensorDeviceClass.DURATION
_attr_state_class = SensorStateClass.MEASUREMENT
def __init__(self, config_entry, name, api_key, origin, destination, client):
def __init__(
self,
config_entry: ConfigEntry,
name: str,
api_key: str,
origin: str,
destination: str,
client: RoutesAsyncClient,
) -> None:
"""Initialize the sensor."""
self._attr_name = name
self._attr_unique_id = config_entry.entry_id
@ -91,13 +135,12 @@ class GoogleTravelTimeSensor(SensorEntity):
)
self._config_entry = config_entry
self._matrix = None
self._api_key = api_key
self._route: Route | None = None
self._client = client
self._origin = origin
self._destination = destination
self._resolved_origin = None
self._resolved_destination = None
self._resolved_origin: str | None = None
self._resolved_destination: str | None = None
async def async_added_to_hass(self) -> None:
"""Handle when entity is added."""
@ -109,77 +152,127 @@ class GoogleTravelTimeSensor(SensorEntity):
await self.first_update()
@property
def native_value(self):
def native_value(self) -> float | None:
"""Return the state of the sensor."""
if self._matrix is None:
if self._route is None:
return None
_data = self._matrix["rows"][0]["elements"][0]
if "duration_in_traffic" in _data:
return round(_data["duration_in_traffic"]["value"] / 60)
if "duration" in _data:
return round(_data["duration"]["value"] / 60)
return None
return round(self._route.duration.seconds / 60)
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return the state attributes."""
if self._matrix is None:
if self._route is None:
return None
res = self._matrix.copy()
options = self._config_entry.options.copy()
res.update(options)
del res["rows"]
_data = self._matrix["rows"][0]["elements"][0]
if "duration_in_traffic" in _data:
res["duration_in_traffic"] = _data["duration_in_traffic"]["text"]
if "duration" in _data:
res["duration"] = _data["duration"]["text"]
if "distance" in _data:
res["distance"] = _data["distance"]["text"]
res["origin"] = self._resolved_origin
res["destination"] = self._resolved_destination
return res
result = self._config_entry.options.copy()
result["duration_in_traffic"] = self._route.localized_values.duration.text
result["duration"] = self._route.localized_values.static_duration.text
result["distance"] = self._route.localized_values.distance.text
async def first_update(self, _=None):
result["origin"] = self._resolved_origin
result["destination"] = self._resolved_destination
return result
async def first_update(self, _=None) -> None:
"""Run the first update and write the state."""
await self.hass.async_add_executor_job(self.update)
await self.async_update()
self.async_write_ha_state()
def update(self) -> None:
async def async_update(self) -> None:
"""Get the latest data from Google."""
options_copy = self._config_entry.options.copy()
dtime = options_copy.get(CONF_DEPARTURE_TIME)
atime = options_copy.get(CONF_ARRIVAL_TIME)
if dtime is not None and ":" in dtime:
options_copy[CONF_DEPARTURE_TIME] = convert_time_to_utc(dtime)
elif dtime is not None:
options_copy[CONF_DEPARTURE_TIME] = dtime
elif atime is None:
options_copy[CONF_DEPARTURE_TIME] = "now"
travel_mode = TRAVEL_MODES_TO_GOOGLE_SDK_ENUM[
self._config_entry.options[CONF_MODE]
]
if atime is not None and ":" in atime:
options_copy[CONF_ARRIVAL_TIME] = convert_time_to_utc(atime)
elif atime is not None:
options_copy[CONF_ARRIVAL_TIME] = atime
if (
departure_time := self._config_entry.options.get(CONF_DEPARTURE_TIME)
) is not None:
departure_time = convert_time(departure_time)
if (
arrival_time := self._config_entry.options.get(CONF_ARRIVAL_TIME)
) is not None:
arrival_time = convert_time(arrival_time)
if travel_mode != RouteTravelMode.TRANSIT:
arrival_time = None
traffic_model = None
routing_preference = None
route_modifiers = None
if travel_mode == RouteTravelMode.DRIVE:
if (
options_traffic_model := self._config_entry.options.get(
CONF_TRAFFIC_MODEL
)
) is not None:
traffic_model = TRAFFIC_MODELS_TO_GOOGLE_SDK_ENUM[options_traffic_model]
routing_preference = RoutingPreference.TRAFFIC_AWARE_OPTIMAL
route_modifiers = RouteModifiers(
avoid_tolls=self._config_entry.options.get(CONF_AVOID) == "tolls",
avoid_ferries=self._config_entry.options.get(CONF_AVOID) == "ferries",
avoid_highways=self._config_entry.options.get(CONF_AVOID) == "highways",
avoid_indoor=self._config_entry.options.get(CONF_AVOID) == "indoor",
)
transit_preferences = None
if travel_mode == RouteTravelMode.TRANSIT:
transit_routing_preference = None
transit_travel_mode = (
TransitPreferences.TransitTravelMode.TRANSIT_TRAVEL_MODE_UNSPECIFIED
)
if (
option_transit_preferences := self._config_entry.options.get(
CONF_TRANSIT_ROUTING_PREFERENCE
)
) is not None:
transit_routing_preference = TRANSIT_PREFS_TO_GOOGLE_SDK_ENUM[
option_transit_preferences
]
if (
option_transit_mode := self._config_entry.options.get(CONF_TRANSIT_MODE)
) is not None:
transit_travel_mode = TRANSPORT_TYPES_TO_GOOGLE_SDK_ENUM[
option_transit_mode
]
transit_preferences = TransitPreferences(
routing_preference=transit_routing_preference,
allowed_travel_modes=[transit_travel_mode],
)
language = None
if (
options_language := self._config_entry.options.get(CONF_LANGUAGE)
) is not None:
language = options_language
self._resolved_origin = find_coordinates(self.hass, self._origin)
self._resolved_destination = find_coordinates(self.hass, self._destination)
_LOGGER.debug(
"Getting update for origin: %s destination: %s",
self._resolved_origin,
self._resolved_destination,
)
if self._resolved_destination is not None and self._resolved_origin is not None:
request = ComputeRoutesRequest(
origin=convert_to_waypoint(self.hass, self._resolved_origin),
destination=convert_to_waypoint(self.hass, self._resolved_destination),
travel_mode=travel_mode,
routing_preference=routing_preference,
departure_time=departure_time,
arrival_time=arrival_time,
route_modifiers=route_modifiers,
language_code=language,
units=UNITS_TO_GOOGLE_SDK_ENUM[self._config_entry.options[CONF_UNITS]],
traffic_model=traffic_model,
transit_preferences=transit_preferences,
)
try:
self._matrix = distance_matrix(
self._client,
self._resolved_origin,
self._resolved_destination,
**options_copy,
response = await self._client.compute_routes(
request, metadata=[("x-goog-fieldmask", FIELD_MASK)]
)
except (ApiError, TransportError, Timeout) as ex:
if response is not None and len(response.routes) > 0:
self._route = response.routes[0]
except GoogleAPIError as ex:
_LOGGER.error("Error getting travel time: %s", ex)
self._matrix = None
self._route = None

28
homeassistant/components/google_travel_time/strings.json
View File

@ -3,7 +3,7 @@
"config": {
"step": {
"user": {
"description": "You can specify the origin and destination in the form of an address, latitude/longitude coordinates, or a Google place ID. When specifying the location using a Google place ID, the ID must be prefixed with `place_id:`.",
"description": "You can specify the origin and destination in the form of an address, latitude/longitude coordinates or an entity ID that provides this information in its state, an entity ID with latitude and longitude attributes, or zone friendly name (case sensitive)",
"data": {
"name": "[%key:common::config_flow::data::name%]",
"api_key": "[%key:common::config_flow::data::api_key%]",
@ -33,16 +33,16 @@
"options": {
"step": {
"init": {
"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`",
"description": "You can optionally specify either a departure time or arrival time in the form of a 24 hour time string like `08:00:00`",
"data": {
"mode": "Travel Mode",
"mode": "Travel mode",
"language": "[%key:common::config_flow::data::language%]",
"time_type": "Time Type",
"time_type": "Time type",
"time": "Time",
"avoid": "Avoid",
"traffic_model": "Traffic Model",
"transit_mode": "Transit Mode",
"transit_routing_preference": "Transit Routing Preference",
"traffic_model": "Traffic model",
"transit_mode": "Transit mode",
"transit_routing_preference": "Transit routing preference",
"units": "Units"
}
}
@ -68,19 +68,19 @@
},
"units": {
"options": {
"metric": "Metric System",
"imperial": "Imperial System"
"metric": "Metric system",
"imperial": "Imperial system"
}
},
"time_type": {
"options": {
"arrival_time": "Arrival Time",
"departure_time": "Departure Time"
"arrival_time": "Arrival time",
"departure_time": "Departure time"
}
},
"traffic_model": {
"options": {
"best_guess": "Best Guess",
"best_guess": "Best guess",
"pessimistic": "Pessimistic",
"optimistic": "Optimistic"
}
@ -96,8 +96,8 @@
},
"transit_routing_preference": {
"options": {
"less_walking": "Less Walking",
"fewer_transfers": "Fewer Transfers"
"less_walking": "Less walking",
"fewer_transfers": "Fewer transfers"
}
}
}

6
requirements_all.txt generated
View File

@ -1047,15 +1047,15 @@ google-cloud-texttospeech==2.25.1
# homeassistant.components.google_generative_ai_conversation
google-genai==1.7.0
# homeassistant.components.google_travel_time
google-maps-routing==0.6.14
# homeassistant.components.nest
google-nest-sdm==7.1.4
# homeassistant.components.google_photos
google-photos-library-api==0.12.1
# homeassistant.components.google_travel_time
googlemaps==2.5.1
# homeassistant.components.slide
# homeassistant.components.slide_local
goslide-api==0.7.0

6
requirements_test_all.txt generated
View File

@ -898,15 +898,15 @@ google-cloud-texttospeech==2.25.1
# homeassistant.components.google_generative_ai_conversation
google-genai==1.7.0
# homeassistant.components.google_travel_time
google-maps-routing==0.6.14
# homeassistant.components.nest
google-nest-sdm==7.1.4
# homeassistant.components.google_photos
google-photos-library-api==0.12.1
# homeassistant.components.google_travel_time
googlemaps==2.5.1
# homeassistant.components.slide
# homeassistant.components.slide_local
goslide-api==0.7.0

88
tests/components/google_travel_time/conftest.py
View File

@ -2,9 +2,11 @@
from collections.abc import Generator
from typing import Any
from unittest.mock import MagicMock, patch
from unittest.mock import AsyncMock, patch
from googlemaps.exceptions import ApiError, Timeout, TransportError
from google.maps.routing_v2 import ComputeRoutesResponse, Route
from google.protobuf import duration_pb2
from google.type import localized_text_pb2
import pytest
from homeassistant.components.google_travel_time.const import DOMAIN
@ -30,8 +32,8 @@ async def mock_config_fixture(
return config_entry
@pytest.fixture(name="bypass_setup")
def bypass_setup_fixture() -> Generator[None]:
@pytest.fixture
def mock_setup_entry() -> Generator[None]:
"""Bypass entry setup."""
with patch(
"homeassistant.components.google_travel_time.async_setup_entry",
@ -40,48 +42,42 @@ def bypass_setup_fixture() -> Generator[None]:
yield
@pytest.fixture(name="bypass_platform_setup")
def bypass_platform_setup_fixture() -> Generator[None]:
"""Bypass platform setup."""
with patch(
"homeassistant.components.google_travel_time.sensor.async_setup_entry",
return_value=True,
):
yield
@pytest.fixture(name="validate_config_entry")
def validate_config_entry_fixture() -> Generator[MagicMock]:
"""Return valid config entry."""
@pytest.fixture
def routes_mock() -> Generator[AsyncMock]:
"""Return valid API result."""
with (
patch("homeassistant.components.google_travel_time.helpers.Client"),
patch(
"homeassistant.components.google_travel_time.helpers.distance_matrix"
) as distance_matrix_mock,
"homeassistant.components.google_travel_time.helpers.RoutesAsyncClient",
autospec=True,
) as mock_client,
patch(
"homeassistant.components.google_travel_time.sensor.RoutesAsyncClient",
new=mock_client,
),
):
distance_matrix_mock.return_value = None
yield distance_matrix_mock
@pytest.fixture(name="invalidate_config_entry")
def invalidate_config_entry_fixture(validate_config_entry: MagicMock) -> None:
"""Return invalid config entry."""
validate_config_entry.side_effect = ApiError("test")
@pytest.fixture(name="invalid_api_key")
def invalid_api_key_fixture(validate_config_entry: MagicMock) -> None:
"""Throw a REQUEST_DENIED ApiError."""
validate_config_entry.side_effect = ApiError("REQUEST_DENIED", "Invalid API key.")
@pytest.fixture(name="timeout")
def timeout_fixture(validate_config_entry: MagicMock) -> None:
"""Throw a Timeout exception."""
validate_config_entry.side_effect = Timeout()
@pytest.fixture(name="transport_error")
def transport_error_fixture(validate_config_entry: MagicMock) -> None:
"""Throw a TransportError exception."""
validate_config_entry.side_effect = TransportError("Unknown.")
client_mock = mock_client.return_value
client_mock.compute_routes.return_value = ComputeRoutesResponse(
mapping={
"routes": [
Route(
mapping={
"localized_values": Route.RouteLocalizedValues(
mapping={
"distance": localized_text_pb2.LocalizedText(
text="21.3 km"
),
"duration": localized_text_pb2.LocalizedText(
text="27 mins"
),
"static_duration": localized_text_pb2.LocalizedText(
text="26 mins"
),
}
),
"duration": duration_pb2.Duration(seconds=1620),
}
)
]
}
)
yield client_mock

8
tests/components/google_travel_time/const.py
View File

@ -3,13 +3,15 @@
from homeassistant.components.google_travel_time.const import (
CONF_DESTINATION,
CONF_ORIGIN,
CONF_UNITS,
UNITS_METRIC,
)
from homeassistant.const import CONF_API_KEY
from homeassistant.const import CONF_API_KEY, CONF_MODE
MOCK_CONFIG = {
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_DESTINATION: "49.983862755708444,8.223882827079068",
}
RECONFIGURE_CONFIG = {
@ -17,3 +19,5 @@ RECONFIGURE_CONFIG = {
CONF_ORIGIN: "location3",
CONF_DESTINATION: "location4",
}
DEFAULT_OPTIONS = {CONF_MODE: "driving", CONF_UNITS: UNITS_METRIC}

417
tests/components/google_travel_time/test_config_flow.py
View File

@ -1,10 +1,10 @@
"""Test the Google Maps Travel Time config flow."""
from unittest.mock import patch
from unittest.mock import AsyncMock, patch
from google.api_core.exceptions import GatewayTimeout, GoogleAPIError, Unauthorized
import pytest
from homeassistant import config_entries
from homeassistant.components.google_travel_time.const import (
ARRIVAL_TIME,
CONF_ARRIVAL_TIME,
@ -23,26 +23,32 @@ from homeassistant.components.google_travel_time.const import (
DOMAIN,
UNITS_IMPERIAL,
)
from homeassistant.config_entries import SOURCE_USER, ConfigFlowResult
from homeassistant.const import CONF_API_KEY, CONF_LANGUAGE, CONF_MODE, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .const import MOCK_CONFIG, RECONFIGURE_CONFIG
from .const import DEFAULT_OPTIONS, MOCK_CONFIG, RECONFIGURE_CONFIG
from tests.common import MockConfigEntry
async def assert_common_reconfigure_steps(
hass: HomeAssistant, reconfigure_result: config_entries.ConfigFlowResult
hass: HomeAssistant, reconfigure_result: ConfigFlowResult
) -> None:
"""Step through and assert the happy case reconfigure flow."""
client_mock = AsyncMock()
with (
patch("homeassistant.components.google_travel_time.helpers.Client"),
patch(
"homeassistant.components.google_travel_time.helpers.distance_matrix",
return_value=None,
"homeassistant.components.google_travel_time.helpers.RoutesAsyncClient",
return_value=client_mock,
),
patch(
"homeassistant.components.google_travel_time.sensor.RoutesAsyncClient",
return_value=client_mock,
),
):
client_mock.compute_routes.return_value = None
reconfigure_successful_result = await hass.config_entries.flow.async_configure(
reconfigure_result["flow_id"],
RECONFIGURE_CONFIG,
@ -56,38 +62,28 @@ async def assert_common_reconfigure_steps(
async def assert_common_create_steps(
hass: HomeAssistant, user_step_result: config_entries.ConfigFlowResult
hass: HomeAssistant, result: ConfigFlowResult
) -> None:
"""Step through and assert the happy case create flow."""
with (
patch("homeassistant.components.google_travel_time.helpers.Client"),
patch(
"homeassistant.components.google_travel_time.helpers.distance_matrix",
return_value=None,
),
):
create_result = await hass.config_entries.flow.async_configure(
user_step_result["flow_id"],
MOCK_CONFIG,
)
assert create_result["type"] is FlowResultType.CREATE_ENTRY
await hass.async_block_till_done()
entry = hass.config_entries.async_entries(DOMAIN)[0]
assert entry.title == DEFAULT_NAME
assert entry.data == {
CONF_NAME: DEFAULT_NAME,
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_CONFIG,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == DEFAULT_NAME
assert result["data"] == {
CONF_NAME: DEFAULT_NAME,
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "49.983862755708444,8.223882827079068",
}
@pytest.mark.usefixtures("validate_config_entry", "bypass_setup")
@pytest.mark.usefixtures("routes_mock", "mock_setup_entry")
async def test_minimum_fields(hass: HomeAssistant) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] is None
@ -95,255 +91,101 @@ async def test_minimum_fields(hass: HomeAssistant) -> None:
await assert_common_create_steps(hass, result)
@pytest.mark.usefixtures("invalidate_config_entry")
async def test_invalid_config_entry(hass: HomeAssistant) -> None:
"""Test we get the form."""
@pytest.mark.usefixtures("mock_setup_entry")
@pytest.mark.parametrize(
("exception", "error"),
[
(GoogleAPIError("test"), "cannot_connect"),
(GatewayTimeout("Timeout error."), "timeout_connect"),
(Unauthorized("Invalid API key."), "invalid_auth"),
],
)
async def test_errors(
hass: HomeAssistant, routes_mock: AsyncMock, exception: Exception, error: str
) -> None:
"""Test errors in the flow."""
routes_mock.compute_routes.side_effect = exception
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] is None
result2 = await hass.config_entries.flow.async_configure(
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_CONFIG,
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "cannot_connect"}
await assert_common_create_steps(hass, result2)
@pytest.mark.usefixtures("invalid_api_key")
async def test_invalid_api_key(hass: HomeAssistant) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] is None
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_CONFIG,
)
assert result["errors"] == {"base": error}
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "invalid_auth"}
await assert_common_create_steps(hass, result2)
@pytest.mark.usefixtures("transport_error")
async def test_transport_error(hass: HomeAssistant) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] is None
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_CONFIG,
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "cannot_connect"}
await assert_common_create_steps(hass, result2)
@pytest.mark.usefixtures("timeout")
async def test_timeout(hass: HomeAssistant) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] is None
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_CONFIG,
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "timeout_connect"}
await assert_common_create_steps(hass, result2)
async def test_malformed_api_key(hass: HomeAssistant) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] is None
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_CONFIG,
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "invalid_auth"}
routes_mock.compute_routes.side_effect = None
await assert_common_create_steps(hass, result)
@pytest.mark.parametrize(
("data", "options"),
[
(
MOCK_CONFIG,
{
CONF_MODE: "driving",
CONF_UNITS: UNITS_IMPERIAL,
},
)
],
[(MOCK_CONFIG, DEFAULT_OPTIONS)],
)
@pytest.mark.usefixtures("validate_config_entry", "bypass_setup")
@pytest.mark.usefixtures("routes_mock", "mock_setup_entry")
async def test_reconfigure(hass: HomeAssistant, mock_config: MockConfigEntry) -> None:
"""Test reconfigure flow."""
reconfigure_result = await mock_config.start_reconfigure_flow(hass)
assert reconfigure_result["type"] is FlowResultType.FORM
assert reconfigure_result["step_id"] == "reconfigure"
result = await mock_config.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
await assert_common_reconfigure_steps(hass, reconfigure_result)
await assert_common_reconfigure_steps(hass, result)
@pytest.mark.usefixtures("mock_setup_entry")
@pytest.mark.parametrize(
("data", "options"),
[(MOCK_CONFIG, DEFAULT_OPTIONS)],
)
@pytest.mark.parametrize(
("exception", "error"),
[
(
MOCK_CONFIG,
{
CONF_MODE: "driving",
CONF_UNITS: UNITS_IMPERIAL,
},
)
(GoogleAPIError("test"), "cannot_connect"),
(GatewayTimeout("Timeout error."), "timeout_connect"),
(Unauthorized("Invalid API key."), "invalid_auth"),
],
)
@pytest.mark.usefixtures("invalidate_config_entry")
async def test_reconfigure_invalid_config_entry(
hass: HomeAssistant, mock_config: MockConfigEntry
hass: HomeAssistant,
mock_config: MockConfigEntry,
routes_mock: AsyncMock,
exception: Exception,
error: str,
) -> None:
"""Test we get the form."""
result = await mock_config.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["errors"] is None
result2 = await hass.config_entries.flow.async_configure(
routes_mock.compute_routes.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
RECONFIGURE_CONFIG,
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "cannot_connect"}
await assert_common_reconfigure_steps(hass, result2)
@pytest.mark.parametrize(
("data", "options"),
[
(
MOCK_CONFIG,
{
CONF_MODE: "driving",
CONF_UNITS: UNITS_IMPERIAL,
},
)
],
)
@pytest.mark.usefixtures("invalid_api_key")
async def test_reconfigure_invalid_api_key(
hass: HomeAssistant, mock_config: MockConfigEntry
) -> None:
"""Test we get the form."""
result = await mock_config.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["errors"] is None
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
RECONFIGURE_CONFIG,
)
assert result["errors"] == {"base": error}
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "invalid_auth"}
await assert_common_reconfigure_steps(hass, result2)
routes_mock.compute_routes.side_effect = None
await assert_common_reconfigure_steps(hass, result)
@pytest.mark.parametrize(
("data", "options"),
[
(
MOCK_CONFIG,
{
CONF_MODE: "driving",
CONF_UNITS: UNITS_IMPERIAL,
},
)
],
[(MOCK_CONFIG, DEFAULT_OPTIONS)],
)
@pytest.mark.usefixtures("transport_error")
async def test_reconfigure_transport_error(
hass: HomeAssistant, mock_config: MockConfigEntry
) -> None:
"""Test we get the form."""
result = await mock_config.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["errors"] is None
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
RECONFIGURE_CONFIG,
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "cannot_connect"}
await assert_common_reconfigure_steps(hass, result2)
@pytest.mark.parametrize(
("data", "options"),
[
(
MOCK_CONFIG,
{
CONF_MODE: "driving",
CONF_UNITS: UNITS_IMPERIAL,
},
)
],
)
@pytest.mark.usefixtures("timeout")
async def test_reconfigure_timeout(
hass: HomeAssistant, mock_config: MockConfigEntry
) -> None:
"""Test we get the form."""
result = await mock_config.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["errors"] is None
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
RECONFIGURE_CONFIG,
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "timeout_connect"}
await assert_common_reconfigure_steps(hass, result2)
@pytest.mark.parametrize(
("data", "options"),
[
(
MOCK_CONFIG,
{
CONF_MODE: "driving",
CONF_UNITS: UNITS_IMPERIAL,
},
)
],
)
@pytest.mark.usefixtures("validate_config_entry")
@pytest.mark.usefixtures("routes_mock")
async def test_options_flow(hass: HomeAssistant, mock_config: MockConfigEntry) -> None:
"""Test options flow."""
result = await hass.config_entries.options.async_init(
mock_config.entry_id, data=None
)
result = await hass.config_entries.options.async_init(mock_config.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
@ -356,7 +198,7 @@ async def test_options_flow(hass: HomeAssistant, mock_config: MockConfigEntry) -
CONF_AVOID: "tolls",
CONF_UNITS: UNITS_IMPERIAL,
CONF_TIME_TYPE: ARRIVAL_TIME,
CONF_TIME: "test",
CONF_TIME: "08:00",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
@ -369,7 +211,7 @@ async def test_options_flow(hass: HomeAssistant, mock_config: MockConfigEntry) -
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: UNITS_IMPERIAL,
CONF_ARRIVAL_TIME: "test",
CONF_ARRIVAL_TIME: "08:00",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
@ -380,7 +222,7 @@ async def test_options_flow(hass: HomeAssistant, mock_config: MockConfigEntry) -
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: UNITS_IMPERIAL,
CONF_ARRIVAL_TIME: "test",
CONF_ARRIVAL_TIME: "08:00",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
@ -389,24 +231,14 @@ async def test_options_flow(hass: HomeAssistant, mock_config: MockConfigEntry) -
@pytest.mark.parametrize(
("data", "options"),
[
(
MOCK_CONFIG,
{
CONF_MODE: "driving",
CONF_UNITS: UNITS_IMPERIAL,
},
)
],
[(MOCK_CONFIG, DEFAULT_OPTIONS)],
)
@pytest.mark.usefixtures("validate_config_entry")
@pytest.mark.usefixtures("routes_mock")
async def test_options_flow_departure_time(
hass: HomeAssistant, mock_config: MockConfigEntry
) -> None:
"""Test options flow with departure time."""
result = await hass.config_entries.options.async_init(
mock_config.entry_id, data=None
)
result = await hass.config_entries.options.async_init(mock_config.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
@ -419,7 +251,7 @@ async def test_options_flow_departure_time(
CONF_AVOID: "tolls",
CONF_UNITS: UNITS_IMPERIAL,
CONF_TIME_TYPE: DEPARTURE_TIME,
CONF_TIME: "test",
CONF_TIME: "08:00",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
@ -432,7 +264,7 @@ async def test_options_flow_departure_time(
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: UNITS_IMPERIAL,
CONF_DEPARTURE_TIME: "test",
CONF_DEPARTURE_TIME: "08:00",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
@ -443,7 +275,7 @@ async def test_options_flow_departure_time(
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: UNITS_IMPERIAL,
CONF_DEPARTURE_TIME: "test",
CONF_DEPARTURE_TIME: "08:00",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
@ -458,7 +290,7 @@ async def test_options_flow_departure_time(
{
CONF_MODE: "driving",
CONF_UNITS: UNITS_IMPERIAL,
CONF_DEPARTURE_TIME: "test",
CONF_DEPARTURE_TIME: "08:00",
},
),
(
@ -466,19 +298,17 @@ async def test_options_flow_departure_time(
{
CONF_MODE: "driving",
CONF_UNITS: UNITS_IMPERIAL,
CONF_ARRIVAL_TIME: "test",
CONF_ARRIVAL_TIME: "08:00",
},
),
],
)
@pytest.mark.usefixtures("validate_config_entry")
@pytest.mark.usefixtures("routes_mock")
async def test_reset_departure_time(
hass: HomeAssistant, mock_config: MockConfigEntry
) -> None:
"""Test resetting departure time."""
result = await hass.config_entries.options.async_init(
mock_config.entry_id, data=None
)
result = await hass.config_entries.options.async_init(mock_config.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
@ -492,6 +322,8 @@ async def test_reset_departure_time(
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert mock_config.options == {
CONF_MODE: "driving",
CONF_UNITS: UNITS_IMPERIAL,
@ -506,7 +338,7 @@ async def test_reset_departure_time(
{
CONF_MODE: "driving",
CONF_UNITS: UNITS_IMPERIAL,
CONF_ARRIVAL_TIME: "test",
CONF_ARRIVAL_TIME: "08:00",
},
),
(
@ -514,19 +346,17 @@ async def test_reset_departure_time(
{
CONF_MODE: "driving",
CONF_UNITS: UNITS_IMPERIAL,
CONF_DEPARTURE_TIME: "test",
CONF_DEPARTURE_TIME: "08:00",
},
),
],
)
@pytest.mark.usefixtures("validate_config_entry")
@pytest.mark.usefixtures("routes_mock")
async def test_reset_arrival_time(
hass: HomeAssistant, mock_config: MockConfigEntry
) -> None:
"""Test resetting arrival time."""
result = await hass.config_entries.options.async_init(
mock_config.entry_id, data=None
)
result = await hass.config_entries.options.async_init(mock_config.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
@ -540,6 +370,8 @@ async def test_reset_arrival_time(
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert mock_config.options == {
CONF_MODE: "driving",
CONF_UNITS: UNITS_IMPERIAL,
@ -557,7 +389,7 @@ async def test_reset_arrival_time(
CONF_AVOID: "tolls",
CONF_UNITS: UNITS_IMPERIAL,
CONF_TIME_TYPE: ARRIVAL_TIME,
CONF_TIME: "test",
CONF_TIME: "08:00",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
@ -565,14 +397,12 @@ async def test_reset_arrival_time(
)
],
)
@pytest.mark.usefixtures("validate_config_entry")
@pytest.mark.usefixtures("routes_mock")
async def test_reset_options_flow_fields(
hass: HomeAssistant, mock_config: MockConfigEntry
) -> None:
"""Test resetting options flow fields that are not time related to None."""
result = await hass.config_entries.options.async_init(
mock_config.entry_id, data=None
)
result = await hass.config_entries.options.async_init(mock_config.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
@ -583,52 +413,39 @@ async def test_reset_options_flow_fields(
CONF_MODE: "driving",
CONF_UNITS: UNITS_IMPERIAL,
CONF_TIME_TYPE: ARRIVAL_TIME,
CONF_TIME: "test",
CONF_TIME: "08:00",
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert mock_config.options == {
CONF_MODE: "driving",
CONF_UNITS: UNITS_IMPERIAL,
CONF_ARRIVAL_TIME: "test",
CONF_ARRIVAL_TIME: "08:00",
}
@pytest.mark.usefixtures("validate_config_entry", "bypass_setup")
async def test_dupe(hass: HomeAssistant) -> None:
@pytest.mark.parametrize(
("data", "options"),
[(MOCK_CONFIG, DEFAULT_OPTIONS)],
)
@pytest.mark.usefixtures("routes_mock", "mock_setup_entry")
async def test_dupe(hass: HomeAssistant, mock_config: MockConfigEntry) -> None:
"""Test setting up the same entry data twice is OK."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] is None
result2 = await hass.config_entries.flow.async_configure(
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_API_KEY: "test",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_DESTINATION: "49.983862755708444,8.223882827079068",
},
)
assert result2["type"] is FlowResultType.CREATE_ENTRY
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] is None
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_API_KEY: "test",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
},
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert result["type"] is FlowResultType.CREATE_ENTRY

82
tests/components/google_travel_time/test_init.py Normal file
View File

@ -0,0 +1,82 @@
"""Tests for Google Maps Travel Time init."""
import pytest
from homeassistant.components.google_travel_time.const import (
ARRIVAL_TIME,
CONF_TIME,
CONF_TIME_TYPE,
DOMAIN,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from .const import DEFAULT_OPTIONS, MOCK_CONFIG
from tests.common import MockConfigEntry
@pytest.mark.parametrize(
("v1", "v2"),
[
("08:00", "08:00"),
("08:00:00", "08:00:00"),
("1742144400", "17:00"),
("now", None),
(None, None),
],
)
@pytest.mark.usefixtures("routes_mock", "mock_setup_entry")
async def test_migrate_entry_v1_v2(
hass: HomeAssistant,
v1: str,
v2: str | None,
) -> None:
"""Test successful migration of entry data."""
mock_entry = MockConfigEntry(
domain=DOMAIN,
version=1,
data=MOCK_CONFIG,
options={
**DEFAULT_OPTIONS,
CONF_TIME_TYPE: ARRIVAL_TIME,
CONF_TIME: v1,
},
)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
updated_entry = hass.config_entries.async_get_entry(mock_entry.entry_id)
assert updated_entry.state is ConfigEntryState.LOADED
assert updated_entry.version == 2
assert updated_entry.options[CONF_TIME] == v2
@pytest.mark.usefixtures("routes_mock", "mock_setup_entry")
async def test_migrate_entry_v1_v2_invalid_time(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test successful migration of entry data."""
mock_entry = MockConfigEntry(
domain=DOMAIN,
version=1,
data=MOCK_CONFIG,
options={
**DEFAULT_OPTIONS,
CONF_TIME_TYPE: ARRIVAL_TIME,
CONF_TIME: "invalid",
},
)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
updated_entry = hass.config_entries.async_get_entry(mock_entry.entry_id)
assert updated_entry.state is ConfigEntryState.LOADED
assert updated_entry.version == 2
assert updated_entry.options[CONF_TIME] is None
assert "Invalid time format found while migrating" in caplog.text

179
tests/components/google_travel_time/test_sensor.py
View File

@ -1,97 +1,48 @@
"""Test the Google Maps Travel Time sensors."""
from collections.abc import Generator
from unittest.mock import MagicMock, patch
from unittest.mock import AsyncMock
from googlemaps.exceptions import ApiError, Timeout, TransportError
from freezegun.api import FrozenDateTimeFactory
from google.api_core.exceptions import GoogleAPIError
from google.maps.routing_v2 import Units
import pytest
from homeassistant.components.google_travel_time.config_flow import default_options
from homeassistant.components.google_travel_time.const import (
CONF_ARRIVAL_TIME,
CONF_DEPARTURE_TIME,
CONF_TRANSIT_MODE,
CONF_TRANSIT_ROUTING_PREFERENCE,
CONF_UNITS,
DOMAIN,
UNITS_IMPERIAL,
UNITS_METRIC,
)
from homeassistant.components.google_travel_time.sensor import SCAN_INTERVAL
from homeassistant.const import CONF_MODE, STATE_UNKNOWN
from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
from homeassistant.util.unit_system import (
METRIC_SYSTEM,
US_CUSTOMARY_SYSTEM,
UnitSystem,
)
from .const import MOCK_CONFIG
from .const import DEFAULT_OPTIONS, MOCK_CONFIG
from tests.common import MockConfigEntry, async_fire_time_changed
@pytest.fixture(name="mock_update")
def mock_update_fixture() -> Generator[MagicMock]:
"""Mock an update to the sensor."""
with (
patch("homeassistant.components.google_travel_time.sensor.Client"),
patch(
"homeassistant.components.google_travel_time.sensor.distance_matrix"
) as distance_matrix_mock,
):
distance_matrix_mock.return_value = {
"rows": [
{
"elements": [
{
"duration_in_traffic": {
"value": 1620,
"text": "27 mins",
},
"duration": {
"value": 1560,
"text": "26 mins",
},
"distance": {"text": "21.3 km"},
}
]
}
]
}
yield distance_matrix_mock
@pytest.fixture(name="mock_update_duration")
def mock_update_duration_fixture(mock_update: MagicMock) -> MagicMock:
"""Mock an update to the sensor returning no duration_in_traffic."""
mock_update.return_value = {
"rows": [
{
"elements": [
{
"duration": {
"value": 1560,
"text": "26 mins",
},
"distance": {"text": "21.3 km"},
}
]
}
]
}
return mock_update
@pytest.fixture(name="mock_update_empty")
def mock_update_empty_fixture(mock_update: MagicMock) -> MagicMock:
def mock_update_empty_fixture(routes_mock: AsyncMock) -> AsyncMock:
"""Mock an update to the sensor with an empty response."""
mock_update.return_value = None
return mock_update
routes_mock.compute_routes.return_value = None
return routes_mock
@pytest.mark.parametrize(
("data", "options"),
[(MOCK_CONFIG, {})],
[(MOCK_CONFIG, DEFAULT_OPTIONS)],
)
@pytest.mark.usefixtures("mock_update", "mock_config")
@pytest.mark.usefixtures("routes_mock", "mock_config")
async def test_sensor(hass: HomeAssistant) -> None:
"""Test that sensor works."""
assert hass.states.get("sensor.google_travel_time").state == "27"
@ -114,7 +65,7 @@ async def test_sensor(hass: HomeAssistant) -> None:
)
assert (
hass.states.get("sensor.google_travel_time").attributes["destination"]
== "location2"
== "49.983862755708444,8.223882827079068"
)
assert (
hass.states.get("sensor.google_travel_time").attributes["unit_of_measurement"]
@ -122,24 +73,14 @@ async def test_sensor(hass: HomeAssistant) -> None:
)
@pytest.mark.parametrize(
("data", "options"),
[(MOCK_CONFIG, {})],
)
@pytest.mark.usefixtures("mock_update_duration", "mock_config")
async def test_sensor_duration(hass: HomeAssistant) -> None:
"""Test that sensor works with no duration_in_traffic in response."""
assert hass.states.get("sensor.google_travel_time").state == "26"
@pytest.mark.parametrize(
("data", "options"),
[(MOCK_CONFIG, {})],
)
@pytest.mark.usefixtures("mock_update_empty", "mock_config")
@pytest.mark.parametrize(
("data", "options"),
[(MOCK_CONFIG, DEFAULT_OPTIONS)],
)
async def test_sensor_empty_response(hass: HomeAssistant) -> None:
"""Test that sensor works for an empty response."""
assert hass.states.get("sensor.google_travel_time").state == "unknown"
assert hass.states.get("sensor.google_travel_time").state == STATE_UNKNOWN
@pytest.mark.parametrize(
@ -148,12 +89,13 @@ async def test_sensor_empty_response(hass: HomeAssistant) -> None:
(
MOCK_CONFIG,
{
**DEFAULT_OPTIONS,
CONF_DEPARTURE_TIME: "10:00",
},
),
],
)
@pytest.mark.usefixtures("mock_update", "mock_config")
@pytest.mark.usefixtures("routes_mock", "mock_config")
async def test_sensor_departure_time(hass: HomeAssistant) -> None:
"""Test that sensor works for departure time."""
assert hass.states.get("sensor.google_travel_time").state == "27"
@ -165,60 +107,31 @@ async def test_sensor_departure_time(hass: HomeAssistant) -> None:
(
MOCK_CONFIG,
{
CONF_DEPARTURE_TIME: "custom_timestamp",
},
),
],
)
@pytest.mark.usefixtures("mock_update", "mock_config")
async def test_sensor_departure_time_custom_timestamp(hass: HomeAssistant) -> None:
"""Test that sensor works for departure time with a custom timestamp."""
assert hass.states.get("sensor.google_travel_time").state == "27"
@pytest.mark.parametrize(
("data", "options"),
[
(
MOCK_CONFIG,
{
CONF_MODE: "transit",
CONF_UNITS: UNITS_METRIC,
CONF_TRANSIT_ROUTING_PREFERENCE: "fewer_transfers",
CONF_TRANSIT_MODE: "bus",
CONF_ARRIVAL_TIME: "10:00",
},
),
],
)
@pytest.mark.usefixtures("mock_update", "mock_config")
@pytest.mark.usefixtures("routes_mock", "mock_config")
async def test_sensor_arrival_time(hass: HomeAssistant) -> None:
"""Test that sensor works for arrival time."""
assert hass.states.get("sensor.google_travel_time").state == "27"
@pytest.mark.parametrize(
("data", "options"),
[
(
MOCK_CONFIG,
{
CONF_ARRIVAL_TIME: "custom_timestamp",
},
),
],
)
@pytest.mark.usefixtures("mock_update", "mock_config")
async def test_sensor_arrival_time_custom_timestamp(hass: HomeAssistant) -> None:
"""Test that sensor works for arrival time with a custom timestamp."""
assert hass.states.get("sensor.google_travel_time").state == "27"
@pytest.mark.parametrize(
("unit_system", "expected_unit_option"),
[
(METRIC_SYSTEM, UNITS_METRIC),
(US_CUSTOMARY_SYSTEM, UNITS_IMPERIAL),
(METRIC_SYSTEM, Units.METRIC),
(US_CUSTOMARY_SYSTEM, Units.IMPERIAL),
],
)
async def test_sensor_unit_system(
hass: HomeAssistant,
routes_mock: AsyncMock,
unit_system: UnitSystem,
expected_unit_option: str,
) -> None:
@ -232,36 +145,28 @@ async def test_sensor_unit_system(
entry_id="test",
)
config_entry.add_to_hass(hass)
with (
patch("homeassistant.components.google_travel_time.sensor.Client"),
patch(
"homeassistant.components.google_travel_time.sensor.distance_matrix"
) as distance_matrix_mock,
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
distance_matrix_mock.assert_called_once()
assert distance_matrix_mock.call_args.kwargs["units"] == expected_unit_option
routes_mock.compute_routes.assert_called_once()
assert routes_mock.compute_routes.call_args.args[0].units == expected_unit_option
@pytest.mark.parametrize(
("exception"),
[(ApiError), (TransportError), (Timeout)],
)
@pytest.mark.parametrize(
("data", "options"),
[(MOCK_CONFIG, {})],
[(MOCK_CONFIG, DEFAULT_OPTIONS)],
)
async def test_sensor_exception(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
mock_update: MagicMock,
mock_config: MagicMock,
exception: Exception,
routes_mock: AsyncMock,
mock_config: MockConfigEntry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test that exception gets caught."""
mock_update.side_effect = exception("Errormessage")
async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL)
routes_mock.compute_routes.side_effect = GoogleAPIError("Errormessage")
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("sensor.google_travel_time").state == STATE_UNKNOWN
assert "Error getting travel time" in caplog.text