jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 11:17:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Don't enforce uniqueness requirements for Waze and Google Travel Time (#50254)

Browse Source 
Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
This commit is contained in:
Raman Gupta 2021-05-26 11:32:26 -04:00 committed by GitHub
parent a36935dcee
commit 8edf7f0407
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 651 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,14 +1,29 @@
"""The google_travel_time component."""
import logging
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_registry import (
async_entries_for_config_entry,
async_get,
)
PLATFORMS = ["sensor"]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry):
"""Set up Google Maps Travel Time from a config entry."""
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
if config_entry.unique_id is not None:
hass.config_entries.async_update_entry(config_entry, unique_id=None)
ent_reg = async_get(hass)
for entity in async_entries_for_config_entry(ent_reg, config_entry.entry_id):
ent_reg.async_update_entity(
entity.entity_id, new_unique_id=config_entry.entry_id
)
hass.config_entries.async_setup_platforms(config_entry, PLATFORMS)
return True

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

@ -1,13 +1,15 @@
"""Config flow for Google Maps Travel Time integration."""
from __future__ import annotations
import logging
from typing import Any
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_API_KEY, CONF_MODE, CONF_NAME
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
import homeassistant.helpers.config_validation as cv
from homeassistant.util import slugify
from .const import (
ALL_LANGUAGES,
@ -18,12 +20,14 @@ from .const import (
CONF_DEPARTURE_TIME,
CONF_DESTINATION,
CONF_LANGUAGE,
CONF_OPTIONS,
CONF_ORIGIN,
CONF_TIME,
CONF_TIME_TYPE,
CONF_TRAFFIC_MODEL,
CONF_TRANSIT_MODE,
CONF_TRANSIT_ROUTING_PREFERENCE,
CONF_TRAVEL_MODE,
CONF_UNITS,
DEFAULT_NAME,
DEPARTURE_TIME,
@ -40,6 +44,47 @@ from .helpers import is_valid_config_entry
_LOGGER = logging.getLogger(__name__)
def is_dupe_import(
hass: HomeAssistant, entry: config_entries.ConfigEntry, user_input: 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)
):
return False
options = user_input.get(CONF_OPTIONS, {})
# We have to check for units differently because there is a default
units = options.get(CONF_UNITS) or hass.config.units.name
if entry.options[CONF_UNITS] != units:
return False
# We have to check for travel mode differently because of the default and because
# it can be provided in two different ways. We have to give mode preference over
# travel mode because that's the way that entry setup works.
mode = options.get(CONF_MODE) or user_input.get(CONF_TRAVEL_MODE) or "driving"
if entry.options[CONF_MODE] != mode:
return False
# We have to check for options that don't have defaults
for key in (
CONF_LANGUAGE,
CONF_AVOID,
CONF_ARRIVAL_TIME,
CONF_DEPARTURE_TIME,
CONF_TRAFFIC_MODEL,
CONF_TRANSIT_MODE,
CONF_TRANSIT_ROUTING_PREFERENCE,
):
if options.get(key) != entry.options.get(key):
return False
return True
class GoogleOptionsFlow(config_entries.OptionsFlow):
"""Handle an options flow for Google Travel Time."""
@ -126,12 +171,14 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors = {}
user_input = user_input or {}
if user_input:
await self.async_set_unique_id(
slugify(
f"{DOMAIN}_{user_input[CONF_ORIGIN]}_{user_input[CONF_DESTINATION]}"
)
)
self._abort_if_unique_id_configured()
# We need to prevent duplicate imports
if self.source == config_entries.SOURCE_IMPORT and any(
is_dupe_import(self.hass, entry, user_input)
for entry in self.hass.config_entries.async_entries(DOMAIN)
if entry.source == config_entries.SOURCE_IMPORT
):
return self.async_abort(reason="already_configured")
if (
self.source == config_entries.SOURCE_IMPORT
or await self.hass.async_add_executor_job(

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

@ -172,7 +172,7 @@ class GoogleTravelTimeSensor(SensorEntity):
self._unit_of_measurement = TIME_MINUTES
self._matrix = None
self._api_key = api_key
self._unique_id = config_entry.unique_id
self._unique_id = config_entry.entry_id
self._client = client
# Check if location is a trackable entity

15
homeassistant/components/waze_travel_time/__init__.py
View File

@ -1,13 +1,28 @@
"""The waze_travel_time component."""
import logging
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_registry import (
async_entries_for_config_entry,
async_get,
)
PLATFORMS = ["sensor"]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Load the saved entities."""
if config_entry.unique_id is not None:
hass.config_entries.async_update_entry(config_entry, unique_id=None)
ent_reg = async_get(hass)
for entity in async_entries_for_config_entry(ent_reg, config_entry.entry_id):
ent_reg.async_update_entity(
entity.entity_id, new_unique_id=config_entry.entry_id
)
hass.config_entries.async_setup_platforms(config_entry, PLATFORMS)
return True

69
homeassistant/components/waze_travel_time/config_flow.py
View File

@ -1,13 +1,15 @@
"""Config flow for Waze Travel Time integration."""
from __future__ import annotations
import logging
from typing import Any
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_NAME, CONF_REGION
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
import homeassistant.helpers.config_validation as cv
from homeassistant.util import slugify
from .const import (
CONF_AVOID_FERRIES,
@ -20,7 +22,12 @@ from .const import (
CONF_REALTIME,
CONF_UNITS,
CONF_VEHICLE_TYPE,
DEFAULT_AVOID_FERRIES,
DEFAULT_AVOID_SUBSCRIPTION_ROADS,
DEFAULT_AVOID_TOLL_ROADS,
DEFAULT_NAME,
DEFAULT_REALTIME,
DEFAULT_VEHICLE_TYPE,
DOMAIN,
REGIONS,
UNITS,
@ -31,6 +38,50 @@ from .helpers import is_valid_config_entry
_LOGGER = logging.getLogger(__name__)
def is_dupe_import(
hass: HomeAssistant, entry: config_entries.ConfigEntry, user_input: dict[str, Any]
) -> bool:
"""Return whether imported config already exists."""
entry_data = {**entry.data, **entry.options}
defaults = {
CONF_REALTIME: DEFAULT_REALTIME,
CONF_VEHICLE_TYPE: DEFAULT_VEHICLE_TYPE,
CONF_UNITS: hass.config.units.name,
CONF_AVOID_FERRIES: DEFAULT_AVOID_FERRIES,
CONF_AVOID_SUBSCRIPTION_ROADS: DEFAULT_AVOID_SUBSCRIPTION_ROADS,
CONF_AVOID_TOLL_ROADS: DEFAULT_AVOID_TOLL_ROADS,
}
for key in (
CONF_ORIGIN,
CONF_DESTINATION,
CONF_REGION,
CONF_INCL_FILTER,
CONF_EXCL_FILTER,
CONF_REALTIME,
CONF_VEHICLE_TYPE,
CONF_UNITS,
CONF_AVOID_FERRIES,
CONF_AVOID_SUBSCRIPTION_ROADS,
CONF_AVOID_TOLL_ROADS,
):
# If the key is present the check is simple
if key in user_input and user_input[key] != entry_data[key]:
return False
# If the key is not present, then we have to check if the key has a default and
# if the default is in the options. If it doesn't have a default, we have to check
# if the key is in the options
if key not in user_input:
if key in defaults and defaults[key] != entry_data[key]:
return False
if key not in defaults and key in entry_data:
return False
return True
class WazeOptionsFlow(config_entries.OptionsFlow):
"""Handle an options flow for Waze Travel Time."""
@ -108,12 +159,14 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
user_input = user_input or {}
if user_input:
await self.async_set_unique_id(
slugify(
f"{DOMAIN}_{user_input[CONF_ORIGIN]}_{user_input[CONF_DESTINATION]}"
)
)
self._abort_if_unique_id_configured()
# We need to prevent duplicate imports
if self.source == config_entries.SOURCE_IMPORT and any(
is_dupe_import(self.hass, entry, user_input)
for entry in self.hass.config_entries.async_entries(DOMAIN)
if entry.source == config_entries.SOURCE_IMPORT
):
return self.async_abort(reason="already_configured")
if (
self.source == config_entries.SOURCE_IMPORT
or await self.hass.async_add_executor_job(

2
homeassistant/components/waze_travel_time/sensor.py
View File

@ -158,7 +158,7 @@ async def async_setup_entry(
config_entry,
)
sensor = WazeTravelTime(config_entry.unique_id, name, origin, destination, data)
sensor = WazeTravelTime(config_entry.entry_id, name, origin, destination, data)
async_add_entities([sensor], False)

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

@ -37,6 +37,16 @@ def bypass_setup_fixture():
yield
@pytest.fixture(name="bypass_platform_setup")
def bypass_platform_setup_fixture():
"""Bypass platform setup."""
with patch(
"homeassistant.components.google_travel_time.sensor.async_setup_entry",
return_value=True,
):
yield
@pytest.fixture(name="bypass_update")
def bypass_update_fixture():
"""Bypass sensor update."""

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

@ -14,6 +14,7 @@ from homeassistant.components.google_travel_time.const import (
CONF_TRAFFIC_MODEL,
CONF_TRANSIT_MODE,
CONF_TRANSIT_ROUTING_PREFERENCE,
CONF_TRAVEL_MODE,
CONF_UNITS,
DEFAULT_NAME,
DEPARTURE_TIME,
@ -24,6 +25,7 @@ from homeassistant.const import (
CONF_MODE,
CONF_NAME,
CONF_UNIT_SYSTEM_IMPERIAL,
CONF_UNIT_SYSTEM_METRIC,
)
from tests.common import MockConfigEntry
@ -197,8 +199,8 @@ async def test_options_flow_departure_time(hass, validate_config_entry, bypass_u
}
async def test_dupe_id(hass, validate_config_entry, bypass_setup):
"""Test setting up the same entry twice fails."""
async def test_dupe(hass, validate_config_entry, bypass_setup):
"""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}
)
@ -233,8 +235,7 @@ async def test_dupe_id(hass, validate_config_entry, bypass_setup):
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result2["reason"] == "already_configured"
assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
async def test_import_flow(hass, validate_config_entry, bypass_update):
@ -297,3 +298,317 @@ async def test_import_flow(hass, validate_config_entry, bypass_update):
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
}
async def test_dupe_import_no_options(hass, bypass_update):
"""Test duplicate import with no options."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_NAME: "test_name",
},
)
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: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_NAME: "test_name",
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_dupe_import_default_options(hass, bypass_update):
"""Test duplicate import with default options."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_NAME: "test_name",
CONF_OPTIONS: {
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_ARRIVAL_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
},
},
)
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: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_NAME: "test_name",
CONF_OPTIONS: {
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_ARRIVAL_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
},
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def _setup_dupe_import(hass, bypass_update):
"""Set up dupe import."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_NAME: "test_name",
CONF_OPTIONS: {
CONF_MODE: "walking",
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_ARRIVAL_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
},
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
await hass.async_block_till_done()
async def test_dupe_import(hass, bypass_update):
"""Test duplicate import."""
await _setup_dupe_import(hass, bypass_update)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_NAME: "test_name",
CONF_OPTIONS: {
CONF_MODE: "walking",
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_ARRIVAL_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
},
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_dupe_import_false_check_data_keys(hass, bypass_update):
"""Test false duplicate import check when data keys differ."""
await _setup_dupe_import(hass, bypass_update)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: "api_key2",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_NAME: "test_name",
CONF_OPTIONS: {
CONF_MODE: "walking",
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_ARRIVAL_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
},
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
async def test_dupe_import_false_check_no_units(hass, bypass_update):
"""Test false duplicate import check when units aren't provided."""
await _setup_dupe_import(hass, bypass_update)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_NAME: "test_name",
CONF_OPTIONS: {
CONF_MODE: "walking",
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_ARRIVAL_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
},
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
async def test_dupe_import_false_check_units(hass, bypass_update):
"""Test false duplicate import check when units are provided but different."""
await _setup_dupe_import(hass, bypass_update)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_NAME: "test_name",
CONF_OPTIONS: {
CONF_MODE: "walking",
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: CONF_UNIT_SYSTEM_METRIC,
CONF_ARRIVAL_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
},
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
async def test_dupe_import_false_check_travel_mode(hass, bypass_update):
"""Test false duplicate import check when travel mode differs."""
await _setup_dupe_import(hass, bypass_update)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_NAME: "test_name",
CONF_TRAVEL_MODE: "driving",
CONF_OPTIONS: {
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_ARRIVAL_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
},
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
async def test_dupe_import_false_check_mode(hass, bypass_update):
"""Test false duplicate import check when mode diiffers."""
await _setup_dupe_import(hass, bypass_update)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_NAME: "test_name",
CONF_OPTIONS: {
CONF_MODE: "driving",
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_ARRIVAL_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
},
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
async def test_dupe_import_false_check_no_mode(hass, bypass_update):
"""Test false duplicate import check when no mode is provided."""
await _setup_dupe_import(hass, bypass_update)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_NAME: "test_name",
CONF_OPTIONS: {
CONF_LANGUAGE: "en",
CONF_AVOID: "tolls",
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_ARRIVAL_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
},
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
async def test_dupe_import_false_check_options(hass, bypass_update):
"""Test false duplicate import check when options differ."""
await _setup_dupe_import(hass, bypass_update)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_API_KEY: "api_key",
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_NAME: "test_name",
CONF_OPTIONS: {
CONF_MODE: "walking",
CONF_AVOID: "tolls",
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_ARRIVAL_TIME: "test",
CONF_TRAFFIC_MODEL: "best_guess",
CONF_TRANSIT_MODE: "train",
CONF_TRANSIT_ROUTING_PREFERENCE: "less_walking",
},
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY

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

@ -0,0 +1,21 @@
"""Test Google Maps Travel Time initialization."""
from homeassistant.components.google_travel_time.const import DOMAIN
from homeassistant.helpers.entity_registry import async_get
from tests.common import MockConfigEntry
async def test_migration(hass, bypass_platform_setup):
"""Test migration logic for unique id."""
config_entry = MockConfigEntry(
domain=DOMAIN, version=1, entry_id="test", unique_id="test"
)
ent_reg = async_get(hass)
ent_entry = ent_reg.async_get_or_create(
"sensor", DOMAIN, unique_id="replaceable_unique_id", config_entry=config_entry
)
entity_id = ent_entry.entity_id
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
assert config_entry.unique_id is None
assert ent_reg.async_get(entity_id).unique_id == config_entry.entry_id

10
tests/components/waze_travel_time/conftest.py
View File

@ -35,6 +35,16 @@ def bypass_setup_fixture():
yield
@pytest.fixture(name="bypass_platform_setup")
def bypass_platform_setup_fixture():
"""Bypass platform setup."""
with patch(
"homeassistant.components.waze_travel_time.sensor.async_setup_entry",
return_value=True,
):
yield
@pytest.fixture(name="mock_update")
def mock_update_fixture():
"""Mock an update to the sensor."""

124
tests/components/waze_travel_time/test_config_flow.py
View File

@ -145,8 +145,125 @@ async def test_import(hass, validate_config_entry, mock_update):
}
async def test_dupe_id(hass, validate_config_entry, bypass_setup):
"""Test setting up the same entry twice fails."""
async def _setup_dupe_import(hass, mock_update):
"""Set up dupe import."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_REGION: "US",
CONF_AVOID_FERRIES: True,
CONF_AVOID_SUBSCRIPTION_ROADS: True,
CONF_AVOID_TOLL_ROADS: True,
CONF_EXCL_FILTER: "exclude",
CONF_INCL_FILTER: "include",
CONF_REALTIME: False,
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_VEHICLE_TYPE: "taxi",
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
await hass.async_block_till_done()
async def test_dupe_import(hass, mock_update):
"""Test duplicate import."""
await _setup_dupe_import(hass, mock_update)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_REGION: "US",
CONF_AVOID_FERRIES: True,
CONF_AVOID_SUBSCRIPTION_ROADS: True,
CONF_AVOID_TOLL_ROADS: True,
CONF_EXCL_FILTER: "exclude",
CONF_INCL_FILTER: "include",
CONF_REALTIME: False,
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_VEHICLE_TYPE: "taxi",
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_dupe_import_false_check_different_options_value(hass, mock_update):
"""Test false duplicate import check when options value differs."""
await _setup_dupe_import(hass, mock_update)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_REGION: "US",
CONF_AVOID_FERRIES: True,
CONF_AVOID_SUBSCRIPTION_ROADS: True,
CONF_AVOID_TOLL_ROADS: True,
CONF_EXCL_FILTER: "exclude",
CONF_INCL_FILTER: "include",
CONF_REALTIME: False,
CONF_UNITS: CONF_UNIT_SYSTEM_IMPERIAL,
CONF_VEHICLE_TYPE: "car",
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
async def test_dupe_import_false_check_default_option(hass, mock_update):
"""Test false duplicate import check when option with a default is missing."""
await _setup_dupe_import(hass, mock_update)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_REGION: "US",
CONF_AVOID_FERRIES: True,
CONF_AVOID_SUBSCRIPTION_ROADS: True,
CONF_AVOID_TOLL_ROADS: True,
CONF_EXCL_FILTER: "exclude",
CONF_INCL_FILTER: "include",
CONF_REALTIME: False,
CONF_VEHICLE_TYPE: "taxi",
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
async def test_dupe_import_false_check_no_default_option(hass, mock_update):
"""Test false duplicate import check option when option with no default is miissing."""
await _setup_dupe_import(hass, mock_update)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_ORIGIN: "location1",
CONF_DESTINATION: "location2",
CONF_REGION: "US",
CONF_AVOID_FERRIES: True,
CONF_AVOID_SUBSCRIPTION_ROADS: True,
CONF_AVOID_TOLL_ROADS: True,
CONF_EXCL_FILTER: "exclude",
CONF_REALTIME: False,
CONF_VEHICLE_TYPE: "taxi",
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
async def test_dupe(hass, validate_config_entry, bypass_setup):
"""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}
)
@ -182,8 +299,7 @@ async def test_dupe_id(hass, validate_config_entry, bypass_setup):
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result2["reason"] == "already_configured"
assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
async def test_invalid_config_entry(hass, invalidate_config_entry):

21
tests/components/waze_travel_time/test_init.py Normal file
View File

@ -0,0 +1,21 @@
"""Test Waze Travel Time initialization."""
from homeassistant.components.waze_travel_time.const import DOMAIN
from homeassistant.helpers.entity_registry import async_get
from tests.common import MockConfigEntry
async def test_migration(hass, bypass_platform_setup):
"""Test migration logic for unique id."""
config_entry = MockConfigEntry(
domain=DOMAIN, version=1, entry_id="test", unique_id="test"
)
ent_reg = async_get(hass)
ent_entry = ent_reg.async_get_or_create(
"sensor", DOMAIN, unique_id="replaceable_unique_id", config_entry=config_entry
)
entity_id = ent_entry.entity_id
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
assert config_entry.unique_id is None
assert ent_reg.async_get(entity_id).unique_id == config_entry.entry_id