Add get_url helper, deprecate base_url (#35224)

This commit is contained in:
Franck Nijhof 2020-05-08 02:29:47 +02:00 committed by GitHub
parent 7e4aa2409f
commit 2223592486
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
51 changed files with 1416 additions and 198 deletions

View File

@ -8,6 +8,8 @@ import sys
import threading
from typing import List
import yarl
from homeassistant.const import REQUIRED_PYTHON_VER, RESTART_EXIT_CODE, __version__
@ -256,10 +258,17 @@ async def setup_and_run_hass(config_dir: str, args: argparse.Namespace) -> int:
if hass is None:
return 1
if args.open_ui and hass.config.api is not None:
if args.open_ui:
import webbrowser # pylint: disable=import-outside-toplevel
hass.add_job(webbrowser.open, hass.config.api.base_url)
if hass.config.api is not None:
scheme = "https" if hass.config.api.use_ssl else "http"
url = str(
yarl.URL.build(
scheme=scheme, host="127.0.0.1", port=hass.config.api.port
)
)
hass.add_job(webbrowser.open, url)
return await hass.async_run()

View File

@ -1,7 +1,6 @@
"""Alexa entity adapters."""
import logging
from typing import List
from urllib.parse import urlparse
from homeassistant.components import (
alarm_control_panel,
@ -799,8 +798,15 @@ class CameraCapabilities(AlexaEntity):
)
return False
url = urlparse(network.async_get_external_url(self.hass))
if url.scheme != "https":
try:
network.async_get_url(
self.hass,
allow_internal=False,
allow_ip=False,
require_ssl=True,
require_standard_port=True,
)
except network.NoURLAvailableError:
_LOGGER.debug(
"%s requires HTTPS for AlexaCameraStreamController", self.entity_id
)

View File

@ -1533,7 +1533,18 @@ async def async_api_initialize_camera_stream(hass, config, directive, context):
entity = directive.entity
stream_source = await camera.async_request_stream(hass, entity.entity_id, fmt="hls")
camera_image = hass.states.get(entity.entity_id).attributes["entity_picture"]
external_url = network.async_get_external_url(hass)
try:
external_url = network.async_get_url(
hass,
allow_internal=False,
allow_ip=False,
require_ssl=True,
require_standard_port=True,
)
except network.NoURLAvailableError:
raise AlexaInvalidValueError("Failed to find suitable URL to serve to Alexa")
payload = {
"cameraStreams": [
{

View File

@ -147,16 +147,17 @@ async def _configure_almond_for_ha(
hass: HomeAssistant, entry: config_entries.ConfigEntry, api: WebAlmondAPI
):
"""Configure Almond to connect to HA."""
if entry.data["type"] == TYPE_OAUTH2:
# If we're connecting over OAuth2, we will only set up connection
# with Home Assistant if we're remotely accessible.
hass_url = network.async_get_external_url(hass)
else:
hass_url = hass.config.api.base_url
# If hass_url is None, we're not going to configure Almond to connect to HA.
if hass_url is None:
try:
if entry.data["type"] == TYPE_OAUTH2:
# If we're connecting over OAuth2, we will only set up connection
# with Home Assistant if we're remotely accessible.
hass_url = network.async_get_url(
hass, allow_internal=False, prefer_cloud=True
)
else:
hass_url = network.async_get_url(hass)
except network.NoURLAvailableError:
# If no URL is available, we're not going to configure Almond to connect to HA.
return
_LOGGER.debug("Configuring Almond to connect to Home Assistant at %s", hass_url)

View File

@ -7,6 +7,7 @@ from homeassistant import config_entries
from homeassistant.components.http import HomeAssistantView
from homeassistant.core import callback
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.network import async_get_url
from .const import (
AUTH_CALLBACK_NAME,
@ -122,16 +123,15 @@ class AmbiclimateFlowHandler(config_entries.ConfigFlow):
clientsession = async_get_clientsession(self.hass)
callback_url = self._cb_url()
oauth = ambiclimate.AmbiclimateOAuth(
return ambiclimate.AmbiclimateOAuth(
config.get(CONF_CLIENT_ID),
config.get(CONF_CLIENT_SECRET),
callback_url,
clientsession,
)
return oauth
def _cb_url(self):
return f"{self.hass.config.api.base_url}{AUTH_CALLBACK_PATH}"
return f"{async_get_url(self.hass)}{AUTH_CALLBACK_PATH}"
async def _get_authorize_url(self):
oauth = self._generate_oauth()

View File

@ -46,6 +46,7 @@ from homeassistant.helpers.config_validation import ( # noqa: F401
)
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.network import async_get_url
from homeassistant.loader import bind_hass
from homeassistant.setup import async_when_setup
@ -684,7 +685,7 @@ async def async_handle_play_stream_service(camera, service_call):
)
data = {
ATTR_ENTITY_ID: entity_ids,
ATTR_MEDIA_CONTENT_ID: f"{hass.config.api.base_url}{url}",
ATTR_MEDIA_CONTENT_ID: f"{async_get_url(hass)}{url}",
ATTR_MEDIA_CONTENT_TYPE: FORMAT_CONTENT_TYPE[fmt],
}

View File

@ -7,6 +7,7 @@ import voluptuous as vol
from homeassistant import auth, config_entries, core
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.helpers import config_validation as cv, dispatcher
from homeassistant.helpers.network import async_get_url
from .const import DOMAIN, SIGNAL_HASS_CAST_SHOW_VIEW
@ -40,15 +41,7 @@ async def async_setup_ha_cast(
async def handle_show_view(call: core.ServiceCall):
"""Handle a Show View service call."""
hass_url = hass.config.api.base_url
# Home Assistant Cast only works with https urls. If user has no configured
# base url, use their remote url.
if not hass_url.lower().startswith("https://"):
try:
hass_url = hass.components.cloud.async_remote_ui_url()
except hass.components.cloud.CloudNotAvailable:
pass
hass_url = async_get_url(hass, require_ssl=True)
controller = HomeAssistantController(
# If you are developing Home Assistant Cast, uncomment and set to your dev app id.

View File

@ -44,6 +44,8 @@ class CheckConfigView(HomeAssistantView):
vol.Optional("unit_system"): cv.unit_system,
vol.Optional("location_name"): str,
vol.Optional("time_zone"): cv.time_zone,
vol.Optional("external_url"): vol.Any(cv.url, None),
vol.Optional("internal_url"): vol.Any(cv.url, None),
}
)
async def websocket_update_config(hass, connection, msg):

View File

@ -23,6 +23,7 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.network import async_get_url
from homeassistant.util import dt as dt_util, slugify
from .const import CONF_EVENTS, DOMAIN, DOOR_STATION, DOOR_STATION_INFO, PLATFORMS
@ -252,7 +253,7 @@ class ConfiguredDoorBird:
def register_events(self, hass):
"""Register events on device."""
# Get the URL of this server
hass_url = hass.config.api.base_url
hass_url = async_get_url(hass)
# Override url if another is specified in the configuration
if self.custom_url is not None:

View File

@ -24,6 +24,7 @@ from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.icon import icon_for_battery_level
from homeassistant.helpers.network import async_get_url
from homeassistant.util.json import load_json, save_json
_CONFIGURING = {}
@ -180,7 +181,7 @@ def request_app_setup(hass, config, add_entities, config_path, discovery_info=No
else:
setup_platform(hass, config, add_entities, discovery_info)
start_url = f"{hass.config.api.base_url}{FITBIT_AUTH_CALLBACK_PATH}"
start_url = f"{async_get_url(hass)}{FITBIT_AUTH_CALLBACK_PATH}"
description = f"""Please create a Fitbit developer app at
https://dev.fitbit.com/apps/new.
@ -215,7 +216,7 @@ def request_oauth_completion(hass):
def fitbit_configuration_callback(callback_data):
"""Handle configuration updates."""
start_url = f"{hass.config.api.base_url}{FITBIT_AUTH_START}"
start_url = f"{async_get_url(hass)}{FITBIT_AUTH_START}"
description = f"Please authorize Fitbit by visiting {start_url}"
@ -307,7 +308,7 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
config_file.get(ATTR_CLIENT_ID), config_file.get(ATTR_CLIENT_SECRET)
)
redirect_uri = f"{hass.config.api.base_url}{FITBIT_AUTH_CALLBACK_PATH}"
redirect_uri = f"{async_get_url(hass)}{FITBIT_AUTH_CALLBACK_PATH}"
fitbit_auth_start_url, _ = oauth.authorize_token_url(
redirect_uri=redirect_uri,
@ -352,7 +353,7 @@ class FitbitAuthCallbackView(HomeAssistantView):
result = None
if data.get("code") is not None:
redirect_uri = f"{hass.config.api.base_url}{FITBIT_AUTH_CALLBACK_PATH}"
redirect_uri = f"{async_get_url(hass)}{FITBIT_AUTH_CALLBACK_PATH}"
try:
result = self.oauth.fetch_access_token(data.get("code"), redirect_uri)

View File

@ -18,6 +18,7 @@ from homeassistant.const import (
)
from homeassistant.core import Context, HomeAssistant, State, callback
from homeassistant.helpers.event import async_call_later
from homeassistant.helpers.network import async_get_url
from homeassistant.helpers.storage import Store
from . import trait
@ -425,7 +426,7 @@ class GoogleEntity:
"webhookId": self.config.local_sdk_webhook_id,
"httpPort": self.hass.http.server_port,
"httpSSL": self.hass.config.api.use_ssl,
"baseUrl": self.hass.config.api.base_url,
"baseUrl": async_get_url(self.hass, prefer_external=True),
"proxyDeviceId": agent_user_id,
}

View File

@ -50,6 +50,7 @@ from homeassistant.const import (
TEMP_FAHRENHEIT,
)
from homeassistant.core import DOMAIN as HA_DOMAIN
from homeassistant.helpers.network import async_get_url
from homeassistant.util import color as color_util, temperature as temp_util
from .const import (
@ -247,9 +248,7 @@ class CameraStreamTrait(_Trait):
url = await self.hass.components.camera.async_request_stream(
self.state.entity_id, "hls"
)
self.stream_info = {
"cameraStreamAccessUrl": self.hass.config.api.base_url + url
}
self.stream_info = {"cameraStreamAccessUrl": f"{async_get_url(self.hass)}{url}"}
@register_trait

View File

@ -63,29 +63,32 @@ STORAGE_KEY = DOMAIN
STORAGE_VERSION = 1
HTTP_SCHEMA = vol.Schema(
{
vol.Optional(CONF_SERVER_HOST, default=DEFAULT_SERVER_HOST): cv.string,
vol.Optional(CONF_SERVER_PORT, default=SERVER_PORT): cv.port,
vol.Optional(CONF_BASE_URL): cv.string,
vol.Optional(CONF_SSL_CERTIFICATE): cv.isfile,
vol.Optional(CONF_SSL_PEER_CERTIFICATE): cv.isfile,
vol.Optional(CONF_SSL_KEY): cv.isfile,
vol.Optional(CONF_CORS_ORIGINS, default=[DEFAULT_CORS]): vol.All(
cv.ensure_list, [cv.string]
),
vol.Inclusive(CONF_USE_X_FORWARDED_FOR, "proxy"): cv.boolean,
vol.Inclusive(CONF_TRUSTED_PROXIES, "proxy"): vol.All(
cv.ensure_list, [ip_network]
),
vol.Optional(
CONF_LOGIN_ATTEMPTS_THRESHOLD, default=NO_LOGIN_ATTEMPT_THRESHOLD
): vol.Any(cv.positive_int, NO_LOGIN_ATTEMPT_THRESHOLD),
vol.Optional(CONF_IP_BAN_ENABLED, default=True): cv.boolean,
vol.Optional(CONF_SSL_PROFILE, default=SSL_MODERN): vol.In(
[SSL_INTERMEDIATE, SSL_MODERN]
),
}
HTTP_SCHEMA = vol.All(
cv.deprecated(CONF_BASE_URL),
vol.Schema(
{
vol.Optional(CONF_SERVER_HOST, default=DEFAULT_SERVER_HOST): cv.string,
vol.Optional(CONF_SERVER_PORT, default=SERVER_PORT): cv.port,
vol.Optional(CONF_BASE_URL): cv.string,
vol.Optional(CONF_SSL_CERTIFICATE): cv.isfile,
vol.Optional(CONF_SSL_PEER_CERTIFICATE): cv.isfile,
vol.Optional(CONF_SSL_KEY): cv.isfile,
vol.Optional(CONF_CORS_ORIGINS, default=[DEFAULT_CORS]): vol.All(
cv.ensure_list, [cv.string]
),
vol.Inclusive(CONF_USE_X_FORWARDED_FOR, "proxy"): cv.boolean,
vol.Inclusive(CONF_TRUSTED_PROXIES, "proxy"): vol.All(
cv.ensure_list, [ip_network]
),
vol.Optional(
CONF_LOGIN_ATTEMPTS_THRESHOLD, default=NO_LOGIN_ATTEMPT_THRESHOLD
): vol.Any(cv.positive_int, NO_LOGIN_ATTEMPT_THRESHOLD),
vol.Optional(CONF_IP_BAN_ENABLED, default=True): cv.boolean,
vol.Optional(CONF_SSL_PROFILE, default=SSL_MODERN): vol.In(
[SSL_INTERMEDIATE, SSL_MODERN]
),
}
),
)
CONFIG_SCHEMA = vol.Schema({DOMAIN: HTTP_SCHEMA}, extra=vol.ALLOW_EXTRA)
@ -102,9 +105,14 @@ class ApiConfig:
"""Configuration settings for API server."""
def __init__(
self, host: str, port: Optional[int] = SERVER_PORT, use_ssl: bool = False
self,
local_ip: str,
host: str,
port: Optional[int] = SERVER_PORT,
use_ssl: bool = False,
) -> None:
"""Initialize a new API config object."""
self.local_ip = local_ip
self.host = host
self.port = port
self.use_ssl = use_ssl
@ -182,6 +190,7 @@ async def async_setup(hass, config):
hass.http = server
host = conf.get(CONF_BASE_URL)
local_ip = await hass.async_add_executor_job(hass_util.get_local_ip)
if host:
port = None
@ -189,10 +198,10 @@ async def async_setup(hass, config):
host = server_host
port = server_port
else:
host = hass_util.get_local_ip()
host = local_ip
port = server_port
hass.config.api = ApiConfig(host, port, ssl_certificate is not None)
hass.config.api = ApiConfig(local_ip, host, port, ssl_certificate is not None)
return True

View File

@ -23,6 +23,7 @@ from homeassistant.const import (
from homeassistant.core import callback
from homeassistant.helpers import aiohttp_client, device_registry as dr
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.network import async_get_url
from .const import (
CONF_ACTIVATION,
@ -297,7 +298,7 @@ class AlarmPanel:
# keeping self.hass.data check for backwards compatibility
# newly configured integrations store this in the config entry
desired_api_host = self.options.get(CONF_API_HOST) or (
self.hass.data[DOMAIN].get(CONF_API_HOST) or self.hass.config.api.base_url
self.hass.data[DOMAIN].get(CONF_API_HOST) or async_get_url(self.hass)
)
desired_api_endpoint = desired_api_host + ENDPOINT_ROOT

View File

@ -48,6 +48,7 @@ from homeassistant.helpers.config_validation import ( # noqa: F401
)
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.network import async_get_url
from homeassistant.loader import bind_hass
from .const import (
@ -820,7 +821,7 @@ async def _async_fetch_image(hass, url):
cache_maxsize = ENTITY_IMAGE_CACHE[CACHE_MAXSIZE]
if urlparse(url).hostname is None:
url = hass.config.api.base_url + url
url = f"{async_get_url(hass)}{url}"
if url not in cache_images:
cache_images[url] = {CACHE_LOCK: asyncio.Lock()}

View File

@ -22,6 +22,7 @@ from homeassistant.const import (
from homeassistant.core import callback
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.network import async_get_url
from .const import ( # pylint: disable=unused-import
AUTH_CALLBACK_NAME,
@ -278,7 +279,9 @@ class PlexFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
session = async_get_clientsession(self.hass)
self.plexauth = PlexAuth(payload, session)
await self.plexauth.initiate_auth()
forward_url = f"{self.hass.config.api.base_url}{AUTH_CALLBACK_PATH}?flow_id={self.flow_id}"
forward_url = (
f"{async_get_url(self.hass)}{AUTH_CALLBACK_PATH}?flow_id={self.flow_id}"
)
auth_url = self.plexauth.auth_url(forward_url)
return self.async_external_step(step_id="obtain_token", url=auth_url)

View File

@ -30,6 +30,7 @@ from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
async_dispatcher_send,
)
from homeassistant.helpers.network import NoURLAvailableError, async_get_url
from homeassistant.helpers.typing import HomeAssistantType
from .const import (
@ -111,7 +112,11 @@ def get_webhook_url(hass: HomeAssistantType) -> str:
def _get_app_template(hass: HomeAssistantType):
endpoint = f"at {hass.config.api.base_url}"
try:
endpoint = f"at {async_get_url(hass, allow_cloud=False, prefer_external=True)}"
except NoURLAvailableError:
endpoint = ""
cloudhook_url = hass.data[DOMAIN][CONF_CLOUDHOOK_URL]
if cloudhook_url is not None:
endpoint = "via Nabu Casa"

View File

@ -11,6 +11,7 @@ from homeassistant.const import (
HTTP_BAD_REQUEST,
HTTP_UNAUTHORIZED,
)
from homeassistant.helpers.network import async_get_url
from . import (
CONF_ALLOWED_CHAT_IDS,
@ -32,7 +33,9 @@ async def async_setup_platform(hass, config):
bot = initialize_bot(config)
current_status = await hass.async_add_job(bot.getWebhookInfo)
base_url = config.get(CONF_URL, hass.config.api.base_url)
base_url = config.get(
CONF_URL, async_get_url(hass, require_ssl=True, allow_internal=False)
)
# Some logging of Bot current status:
last_error_date = getattr(current_status, "last_error_date", None)

View File

@ -33,6 +33,7 @@ from homeassistant.core import callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_per_platform, discovery
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.network import async_get_url
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.setup import async_prepare_setup_platform
@ -114,7 +115,7 @@ async def async_setup(hass, config):
use_cache = conf.get(CONF_CACHE, DEFAULT_CACHE)
cache_dir = conf.get(CONF_CACHE_DIR, DEFAULT_CACHE_DIR)
time_memory = conf.get(CONF_TIME_MEMORY, DEFAULT_TIME_MEMORY)
base_url = conf.get(CONF_BASE_URL) or hass.config.api.base_url
base_url = conf.get(CONF_BASE_URL) or async_get_url(hass)
await tts.async_init_cache(use_cache, cache_dir, time_memory, base_url)
except (HomeAssistantError, KeyError) as err:

View File

@ -10,6 +10,7 @@ from homeassistant.components.http.const import KEY_REAL_IP
from homeassistant.components.http.view import HomeAssistantView
from homeassistant.const import HTTP_OK
from homeassistant.core import callback
from homeassistant.helpers.network import async_get_url
from homeassistant.loader import bind_hass
_LOGGER = logging.getLogger(__name__)
@ -55,7 +56,10 @@ def async_generate_id():
@bind_hass
def async_generate_url(hass, webhook_id):
"""Generate the full URL for a webhook_id."""
return "{}{}".format(hass.config.api.base_url, async_generate_path(webhook_id))
return "{}{}".format(
async_get_url(hass, prefer_external=True, allow_cloud=False),
async_generate_path(webhook_id),
)
@callback

View File

@ -29,6 +29,7 @@ from homeassistant.helpers.config_validation import make_entity_service_schema
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.event import track_time_interval
from homeassistant.helpers.network import async_get_url
from homeassistant.util.json import load_json, save_json
_LOGGER = logging.getLogger(__name__)
@ -231,7 +232,7 @@ def _request_app_setup(hass, config):
_configurator = hass.data[DOMAIN]["configuring"][DOMAIN]
configurator.notify_errors(_configurator, error_msg)
start_url = f"{hass.config.api.base_url}{WINK_AUTH_CALLBACK_PATH}"
start_url = f"{async_get_url(hass)}{WINK_AUTH_CALLBACK_PATH}"
description = f"""Please create a Wink developer app at
https://developer.wink.com.
@ -269,7 +270,7 @@ def _request_oauth_completion(hass, config):
"""Call setup again."""
setup(hass, config)
start_url = f"{hass.config.api.base_url}{WINK_AUTH_START}"
start_url = f"{async_get_url(hass)}{WINK_AUTH_START}"
description = f"Please authorize Wink by visiting {start_url}"
@ -349,7 +350,7 @@ def setup(hass, config):
# Home .
else:
redirect_uri = f"{hass.config.api.base_url}{WINK_AUTH_CALLBACK_PATH}"
redirect_uri = f"{async_get_url(hass)}{WINK_AUTH_CALLBACK_PATH}"
wink_auth_start_url = pywink.get_authorization_url(
config_file.get(ATTR_CLIENT_ID), redirect_uri

View File

@ -22,6 +22,7 @@ from homeassistant.const import (
)
from homeassistant.generated.zeroconf import HOMEKIT, ZEROCONF
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.network import NoURLAvailableError, async_get_url
_LOGGER = logging.getLogger(__name__)
@ -63,11 +64,27 @@ def setup(hass, config):
params = {
"version": __version__,
"base_url": hass.config.api.base_url,
"external_url": None,
"internal_url": None,
# Old base URL, for backward compatibility
"base_url": None,
# Always needs authentication
"requires_api_password": True,
}
try:
params["external_url"] = async_get_url(hass, allow_internal=False)
except NoURLAvailableError:
pass
try:
params["internal_url"] = async_get_url(hass, allow_external=False)
except NoURLAvailableError:
pass
# Set old base URL based on external or internal
params["base_url"] = params["external_url"] or params["internal_url"]
host_ip = util.get_local_ip()
try:

View File

@ -27,7 +27,9 @@ from homeassistant.const import (
CONF_CUSTOMIZE_DOMAIN,
CONF_CUSTOMIZE_GLOB,
CONF_ELEVATION,
CONF_EXTERNAL_URL,
CONF_ID,
CONF_INTERNAL_URL,
CONF_LATITUDE,
CONF_LONGITUDE,
CONF_NAME,
@ -74,10 +76,6 @@ DEFAULT_CONFIG = f"""
# Configure a default setup of Home Assistant (frontend, api, etc)
default_config:
# Uncomment this if you are using SSL/TLS, running in Docker container, etc.
# http:
# base_url: example.duckdns.org:8123
# Text to speech
tts:
- platform: google_translate
@ -183,6 +181,8 @@ CORE_CONFIG_SCHEMA = CUSTOMIZE_CONFIG_SCHEMA.extend(
vol.Optional(CONF_TEMPERATURE_UNIT): cv.temperature_unit,
CONF_UNIT_SYSTEM: cv.unit_system,
CONF_TIME_ZONE: cv.time_zone,
vol.Optional(CONF_INTERNAL_URL): cv.url,
vol.Optional(CONF_EXTERNAL_URL): cv.url,
vol.Optional(CONF_WHITELIST_EXTERNAL_DIRS):
# pylint: disable=no-value-for-parameter
vol.All(cv.ensure_list, [vol.IsDir()]),
@ -478,6 +478,8 @@ async def async_process_ha_core_config(hass: HomeAssistant, config: Dict) -> Non
CONF_ELEVATION,
CONF_TIME_ZONE,
CONF_UNIT_SYSTEM,
CONF_EXTERNAL_URL,
CONF_INTERNAL_URL,
]
):
hac.config_source = SOURCE_YAML
@ -487,6 +489,8 @@ async def async_process_ha_core_config(hass: HomeAssistant, config: Dict) -> Non
(CONF_LONGITUDE, "longitude"),
(CONF_NAME, "location_name"),
(CONF_ELEVATION, "elevation"),
(CONF_INTERNAL_URL, "internal_url"),
(CONF_EXTERNAL_URL, "external_url"),
):
if key in config:
setattr(hac, attr, config[key])
@ -529,10 +533,7 @@ async def async_process_ha_core_config(hass: HomeAssistant, config: Dict) -> Non
hac.units = METRIC_SYSTEM
elif CONF_TEMPERATURE_UNIT in config:
unit = config[CONF_TEMPERATURE_UNIT]
if unit == TEMP_CELSIUS:
hac.units = METRIC_SYSTEM
else:
hac.units = IMPERIAL_SYSTEM
hac.units = METRIC_SYSTEM if unit == TEMP_CELSIUS else IMPERIAL_SYSTEM
_LOGGER.warning(
"Found deprecated temperature unit in core "
"configuration expected unit system. Replace '%s: %s' "

View File

@ -88,6 +88,7 @@ CONF_EVENT = "event"
CONF_EVENT_DATA = "event_data"
CONF_EVENT_DATA_TEMPLATE = "event_data_template"
CONF_EXCLUDE = "exclude"
CONF_EXTERNAL_URL = "external_url"
CONF_FILE_PATH = "file_path"
CONF_FILENAME = "filename"
CONF_FOR = "for"
@ -102,6 +103,7 @@ CONF_ICON = "icon"
CONF_ICON_TEMPLATE = "icon_template"
CONF_ID = "id"
CONF_INCLUDE = "include"
CONF_INTERNAL_URL = "internal_url"
CONF_IP_ADDRESS = "ip_address"
CONF_LATITUDE = "latitude"
CONF_LIGHTS = "lights"

View File

@ -9,6 +9,7 @@ from concurrent.futures import ThreadPoolExecutor
import datetime
import enum
import functools
from ipaddress import ip_address
import logging
import os
import pathlib
@ -29,12 +30,14 @@ from typing import (
Set,
TypeVar,
Union,
cast,
)
import uuid
from async_timeout import timeout
import attr
import voluptuous as vol
import yarl
from homeassistant import block_async_io, loader, util
from homeassistant.const import (
@ -68,7 +71,7 @@ from homeassistant.exceptions import (
ServiceNotFound,
Unauthorized,
)
from homeassistant.util import location
from homeassistant.util import location, network
from homeassistant.util.async_ import fire_coroutine_threadsafe, run_callback_threadsafe
import homeassistant.util.dt as dt_util
from homeassistant.util.thread import fix_threading_exception_logging
@ -84,6 +87,7 @@ block_async_io.enable()
fix_threading_exception_logging()
T = TypeVar("T")
_UNDEF: dict = {}
# pylint: disable=invalid-name
CALLABLE_T = TypeVar("CALLABLE_T", bound=Callable)
CALLBACK_TYPE = Callable[[], None]
@ -426,7 +430,7 @@ class HomeAssistant:
# regardless of the state of the loop.
if self.state == CoreState.not_running: # just ignore
return
if self.state == CoreState.stopping or self.state == CoreState.final_write:
if self.state in [CoreState.stopping, CoreState.final_write]:
_LOGGER.info("async_stop called twice: ignored")
return
if self.state == CoreState.starting:
@ -1301,6 +1305,8 @@ class Config:
self.location_name: str = "Home"
self.time_zone: datetime.tzinfo = dt_util.UTC
self.units: UnitSystem = METRIC_SYSTEM
self.internal_url: Optional[str] = None
self.external_url: Optional[str] = None
self.config_source: str = "default"
@ -1385,6 +1391,8 @@ class Config:
"version": __version__,
"config_source": self.config_source,
"safe_mode": self.safe_mode,
"external_url": self.external_url,
"internal_url": self.internal_url,
}
def set_time_zone(self, time_zone_str: str) -> None:
@ -1408,6 +1416,8 @@ class Config:
unit_system: Optional[str] = None,
location_name: Optional[str] = None,
time_zone: Optional[str] = None,
external_url: Optional[Union[str, dict]] = _UNDEF,
internal_url: Optional[Union[str, dict]] = _UNDEF,
) -> None:
"""Update the configuration from a dictionary."""
self.config_source = source
@ -1426,6 +1436,10 @@ class Config:
self.location_name = location_name
if time_zone is not None:
self.set_time_zone(time_zone)
if external_url is not _UNDEF:
self.external_url = cast(Optional[str], external_url)
if internal_url is not _UNDEF:
self.internal_url = cast(Optional[str], internal_url)
async def async_update(self, **kwargs: Any) -> None:
"""Update the configuration from a dictionary."""
@ -1439,10 +1453,42 @@ class Config:
CORE_STORAGE_VERSION, CORE_STORAGE_KEY, private=True
)
data = await store.async_load()
if not data:
if data and "external_url" in data:
self._update(source=SOURCE_STORAGE, **data)
return
self._update(source=SOURCE_STORAGE, **data)
async def migrate_base_url(_: Event) -> None:
"""Migrate base_url to internal_url/external_url."""
if self.hass.config.api is None:
return
base_url = yarl.URL(self.hass.config.api.base_url)
# Check if this is an internal URL
if str(base_url.host).endswith(".local") or (
network.is_ip_address(str(base_url.host))
and network.is_private(ip_address(base_url.host))
):
await self.async_update(
internal_url=network.normalize_url(str(base_url))
)
return
# External, ensure this is not a loopback address
if not (
network.is_ip_address(str(base_url.host))
and network.is_loopback(ip_address(base_url.host))
):
await self.async_update(
external_url=network.normalize_url(str(base_url))
)
# Try to migrate base_url to internal_url/external_url
self.hass.bus.async_listen_once(EVENT_HOMEASSISTANT_START, migrate_base_url)
if data:
self._update(source=SOURCE_STORAGE, **data)
async def async_store(self) -> None:
"""Store [homeassistant] core config."""
@ -1457,6 +1503,8 @@ class Config:
"unit_system": self.units.name,
"location_name": self.location_name,
"time_zone": time_zone,
"external_url": self.external_url,
"internal_url": self.internal_url,
}
store = self.hass.helpers.storage.Store(

View File

@ -21,6 +21,7 @@ from yarl import URL
from homeassistant import config_entries
from homeassistant.components.http import HomeAssistantView
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.network import async_get_url
from .aiohttp_client import async_get_clientsession
@ -117,7 +118,7 @@ class LocalOAuth2Implementation(AbstractOAuth2Implementation):
@property
def redirect_uri(self) -> str:
"""Return the redirect uri."""
return f"{self.hass.config.api.base_url}{AUTH_CALLBACK_PATH}" # type: ignore
return f"{async_get_url(self.hass)}{AUTH_CALLBACK_PATH}"
async def async_generate_authorize_url(self, flow_id: str) -> str:
"""Generate a url for the user to authorize."""

View File

@ -1,38 +1,230 @@
"""Network helpers."""
from ipaddress import ip_address
from typing import Optional, cast
from typing import cast
import yarl
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.loader import bind_hass
from homeassistant.util.network import is_local
from homeassistant.util.network import (
is_ip_address,
is_local,
is_loopback,
is_private,
normalize_url,
)
TYPE_URL_INTERNAL = "internal_url"
TYPE_URL_EXTERNAL = "external_url"
class NoURLAvailableError(HomeAssistantError):
"""An URL to the Home Assistant instance is not available."""
@bind_hass
@callback
def async_get_external_url(hass: HomeAssistant) -> Optional[str]:
"""Get external url of this instance.
def async_get_url(
hass: HomeAssistant,
*,
require_ssl: bool = False,
require_standard_port: bool = False,
allow_internal: bool = True,
allow_external: bool = True,
allow_cloud: bool = True,
allow_ip: bool = True,
prefer_external: bool = False,
prefer_cloud: bool = False,
) -> str:
"""Get a URL to this instance."""
order = [TYPE_URL_INTERNAL, TYPE_URL_EXTERNAL]
if prefer_external:
order.reverse()
Note: currently it takes 30 seconds after Home Assistant starts for
cloud.async_remote_ui_url to work.
"""
# Try finding an URL in the order specified
for url_type in order:
if allow_internal and url_type == TYPE_URL_INTERNAL:
try:
return _async_get_internal_url(
hass,
allow_ip=allow_ip,
require_ssl=require_ssl,
require_standard_port=require_standard_port,
)
except NoURLAvailableError:
pass
if allow_external and url_type == TYPE_URL_EXTERNAL:
try:
return _async_get_external_url(
hass,
allow_cloud=allow_cloud,
allow_ip=allow_ip,
prefer_cloud=prefer_cloud,
require_ssl=require_ssl,
require_standard_port=require_standard_port,
)
except NoURLAvailableError:
pass
# We have to be honest now, we have no viable option available
raise NoURLAvailableError
@bind_hass
@callback
def _async_get_internal_url(
hass: HomeAssistant,
*,
allow_ip: bool = True,
require_ssl: bool = False,
require_standard_port: bool = False,
) -> str:
"""Get internal URL of this instance."""
if hass.config.internal_url:
internal_url = yarl.URL(hass.config.internal_url)
if (
(not require_ssl or internal_url.scheme == "https")
and (not require_standard_port or internal_url.is_default_port())
and (allow_ip or not is_ip_address(str(internal_url.host)))
):
return normalize_url(str(internal_url))
# Fallback to old base_url
try:
return _async_get_deprecated_base_url(
hass,
internal=True,
allow_ip=allow_ip,
require_ssl=require_ssl,
require_standard_port=require_standard_port,
)
except NoURLAvailableError:
pass
# Fallback to detected local IP
if allow_ip and not (
require_ssl or hass.config.api is None or hass.config.api.use_ssl
):
ip_url = yarl.URL.build(
scheme="http", host=hass.config.api.local_ip, port=hass.config.api.port
)
if not is_loopback(ip_address(ip_url.host)) and (
not require_standard_port or ip_url.is_default_port()
):
return normalize_url(str(ip_url))
raise NoURLAvailableError
@bind_hass
@callback
def _async_get_external_url(
hass: HomeAssistant,
*,
allow_cloud: bool = True,
allow_ip: bool = True,
prefer_cloud: bool = False,
require_ssl: bool = False,
require_standard_port: bool = False,
) -> str:
"""Get external URL of this instance."""
if prefer_cloud and allow_cloud:
try:
return _async_get_cloud_url(hass)
except NoURLAvailableError:
pass
if hass.config.external_url:
external_url = yarl.URL(hass.config.external_url)
if (
(allow_ip or not is_ip_address(str(external_url.host)))
and (not require_standard_port or external_url.is_default_port())
and (
not require_ssl
or (
external_url.scheme == "https"
and not is_ip_address(str(external_url.host))
)
)
):
return normalize_url(str(external_url))
try:
return _async_get_deprecated_base_url(
hass,
allow_ip=allow_ip,
require_ssl=require_ssl,
require_standard_port=require_standard_port,
)
except NoURLAvailableError:
pass
if allow_cloud:
try:
return _async_get_cloud_url(hass)
except NoURLAvailableError:
pass
raise NoURLAvailableError
@bind_hass
@callback
def _async_get_cloud_url(hass: HomeAssistant) -> str:
"""Get external Home Assistant Cloud URL of this instance."""
if "cloud" in hass.config.components:
try:
return cast(str, hass.components.cloud.async_remote_ui_url())
except hass.components.cloud.CloudNotAvailable:
pass
if hass.config.api is None:
return None
raise NoURLAvailableError
@bind_hass
@callback
def _async_get_deprecated_base_url(
hass: HomeAssistant,
*,
internal: bool = False,
allow_ip: bool = True,
require_ssl: bool = False,
require_standard_port: bool = False,
) -> str:
"""Work with the deprecated `base_url`, used as fallback."""
if hass.config.api is None or not hass.config.api.base_url:
raise NoURLAvailableError
base_url = yarl.URL(hass.config.api.base_url)
# Rules that apply to both internal and external
if (
(allow_ip or not is_ip_address(str(base_url.host)))
and (not require_ssl or base_url.scheme == "https")
and (not require_standard_port or base_url.is_default_port())
):
# Check to ensure an internal URL
if internal and (
str(base_url.host).endswith(".local")
or (
is_ip_address(str(base_url.host))
and not is_loopback(ip_address(base_url.host))
and is_private(ip_address(base_url.host))
)
):
return normalize_url(str(base_url))
try:
if is_local(ip_address(base_url.host)):
return None
except ValueError:
# ip_address raises ValueError if host is not an IP address
pass
# Check to ensure an external URL (a little)
if (
not internal
and not str(base_url.host).endswith(".local")
and not (
is_ip_address(str(base_url.host))
and is_local(ip_address(str(base_url.host)))
)
):
return normalize_url(str(base_url))
return str(base_url)
raise NoURLAvailableError

View File

@ -1,7 +1,9 @@
"""Network utilities."""
from ipaddress import IPv4Address, IPv6Address, ip_network
from ipaddress import IPv4Address, IPv6Address, ip_address, ip_network
from typing import Union
import yarl
# RFC6890 - IP addresses of loopback interfaces
LOOPBACK_NETWORKS = (
ip_network("127.0.0.0/8"),
@ -39,3 +41,21 @@ def is_link_local(address: Union[IPv4Address, IPv6Address]) -> bool:
def is_local(address: Union[IPv4Address, IPv6Address]) -> bool:
"""Check if an address is loopback or private."""
return is_loopback(address) or is_private(address)
def is_ip_address(address: str) -> bool:
"""Check if a given string is an IP address."""
try:
ip_address(address)
except ValueError:
return False
return True
def normalize_url(address: str) -> str:
"""Normalize a given URL."""
url = yarl.URL(address.rstrip("/"))
if url.is_default_port():
return str(url.with_port(None))
return str(url)

View File

@ -22,6 +22,7 @@ from homeassistant.components.media_player.const import (
SUPPORT_VOLUME_STEP,
)
import homeassistant.components.vacuum as vacuum
from homeassistant.config import async_process_ha_core_config
from homeassistant.const import TEMP_CELSIUS, TEMP_FAHRENHEIT
from homeassistant.core import Context, callback
from homeassistant.helpers import entityfilter
@ -3784,8 +3785,11 @@ async def test_camera_discovery(hass, mock_stream):
"idle",
{"friendly_name": "Test camera", "supported_features": 3},
)
with patch(
"homeassistant.helpers.network.async_get_external_url",
hass.config.components.add("cloud")
with patch.object(
hass.components.cloud,
"async_remote_ui_url",
return_value="https://example.nabu.casa",
):
appliance = await discovery_test(device, hass)
@ -3812,8 +3816,11 @@ async def test_camera_discovery_without_stream(hass):
"idle",
{"friendly_name": "Test camera", "supported_features": 3},
)
with patch(
"homeassistant.helpers.network.async_get_external_url",
hass.config.components.add("cloud")
with patch.object(
hass.components.cloud,
"async_remote_ui_url",
return_value="https://example.nabu.casa",
):
appliance = await discovery_test(device, hass)
@ -3826,8 +3833,7 @@ async def test_camera_discovery_without_stream(hass):
[
("http://nohttpswrongport.org:8123", 2),
("http://nohttpsport443.org:443", 2),
("tls://nohttpsport443.org:443", 2),
("https://httpsnnonstandport.org:8123", 3),
("https://httpsnnonstandport.org:8123", 2),
("https://correctschemaandport.org:443", 3),
("https://correctschemaandport.org", 3),
],
@ -3839,11 +3845,12 @@ async def test_camera_hass_urls(hass, mock_stream, url, result):
"idle",
{"friendly_name": "Test camera", "supported_features": 3},
)
with patch(
"homeassistant.helpers.network.async_get_external_url", return_value=url
):
appliance = await discovery_test(device, hass)
assert len(appliance["capabilities"]) == result
await async_process_ha_core_config(
hass, {"external_url": url},
)
appliance = await discovery_test(device, hass)
assert len(appliance["capabilities"]) == result
async def test_initialize_camera_stream(hass, mock_camera, mock_stream):
@ -3852,12 +3859,13 @@ async def test_initialize_camera_stream(hass, mock_camera, mock_stream):
"Alexa.CameraStreamController", "InitializeCameraStreams", "camera#demo_camera"
)
await async_process_ha_core_config(
hass, {"external_url": "https://mycamerastream.test"},
)
with patch(
"homeassistant.components.demo.camera.DemoCamera.stream_source",
return_value="rtsp://example.local",
), patch(
"homeassistant.helpers.network.async_get_external_url",
return_value="https://mycamerastream.test",
):
msg = await smart_home.async_handle_message(hass, DEFAULT_CONFIG, request)
await hass.async_block_till_done()

View File

@ -5,6 +5,7 @@ import pytest
from homeassistant import config_entries, core
from homeassistant.components.almond import const
from homeassistant.config import async_process_ha_core_config
from homeassistant.const import EVENT_HOMEASSISTANT_START
from homeassistant.setup import async_setup_component
from homeassistant.util.dt import utcnow
@ -39,8 +40,9 @@ async def test_set_up_oauth_remote_url(hass, aioclient_mock):
assert entry.state == config_entries.ENTRY_STATE_LOADED
hass.config.components.add("cloud")
with patch("homeassistant.components.almond.ALMOND_SETUP_DELAY", 0), patch(
"homeassistant.helpers.network.async_get_external_url",
"homeassistant.helpers.network.async_get_url",
return_value="https://example.nabu.casa",
), patch("pyalmond.WebAlmondAPI.async_create_device") as mock_create_device:
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
@ -93,7 +95,13 @@ async def test_set_up_hassio(hass, aioclient_mock):
async def test_set_up_local(hass, aioclient_mock):
"""Test we do not set up Almond to connect to HA if we use Hass.io."""
"""Test we do not set up Almond to connect to HA if we use local."""
# Set up an internal URL, as Almond won't be set up if there is no URL available
await async_process_ha_core_config(
hass, {"internal_url": "https://192.168.0.1"},
)
entry = MockConfigEntry(
domain="almond",
data={"type": const.TYPE_LOCAL, "host": "http://localhost:9999"},

View File

@ -9,6 +9,7 @@ from homeassistant.components import camera
from homeassistant.components.camera.const import DOMAIN, PREF_PRELOAD_STREAM
from homeassistant.components.camera.prefs import CameraEntityPreferences
from homeassistant.components.websocket_api.const import TYPE_RESULT
from homeassistant.config import async_process_ha_core_config
from homeassistant.const import ATTR_ENTITY_ID, EVENT_HOMEASSISTANT_START
from homeassistant.exceptions import HomeAssistantError
from homeassistant.setup import async_setup_component
@ -241,6 +242,9 @@ async def test_play_stream_service_no_source(hass, mock_camera, mock_stream):
async def test_handle_play_stream_service(hass, mock_camera, mock_stream):
"""Test camera play_stream service."""
await async_process_ha_core_config(
hass, {"external_url": "https://example.com"},
)
await async_setup_component(hass, "media_player", {})
with patch(
"homeassistant.components.camera.request_stream"

View File

@ -7,7 +7,7 @@ from tests.common import MockConfigEntry, async_mock_signal
async def test_service_show_view(hass):
"""Test we don't set app id in prod."""
hass.config.api = Mock(base_url="http://example.com")
hass.config.api = Mock(base_url="https://example.com")
await home_assistant_cast.async_setup_ha_cast(hass, MockConfigEntry())
calls = async_mock_signal(hass, home_assistant_cast.SIGNAL_HASS_CAST_SHOW_VIEW)
@ -20,7 +20,7 @@ async def test_service_show_view(hass):
assert len(calls) == 1
controller, entity_id, view_path, url_path = calls[0]
assert controller.hass_url == "http://example.com"
assert controller.hass_url == "https://example.com"
assert controller.client_id is None
# Verify user did not accidentally submit their dev app id
assert controller.supporting_app_id == "B12CE3CA"
@ -31,7 +31,7 @@ async def test_service_show_view(hass):
async def test_service_show_view_dashboard(hass):
"""Test casting a specific dashboard."""
hass.config.api = Mock(base_url="http://example.com")
hass.config.api = Mock(base_url="https://example.com")
await home_assistant_cast.async_setup_ha_cast(hass, MockConfigEntry())
calls = async_mock_signal(hass, home_assistant_cast.SIGNAL_HASS_CAST_SHOW_VIEW)
@ -56,12 +56,14 @@ async def test_service_show_view_dashboard(hass):
async def test_use_cloud_url(hass):
"""Test that we fall back to cloud url."""
hass.config.api = Mock(base_url="http://example.com")
hass.config.components.add("cloud")
await home_assistant_cast.async_setup_ha_cast(hass, MockConfigEntry())
calls = async_mock_signal(hass, home_assistant_cast.SIGNAL_HASS_CAST_SHOW_VIEW)
with patch(
"homeassistant.components.cloud.async_remote_ui_url",
return_value="https://something.nabu.acas",
return_value="https://something.nabu.casa",
):
await hass.services.async_call(
"cast",
@ -72,4 +74,4 @@ async def test_use_cloud_url(hass):
assert len(calls) == 1
controller = calls[0][0]
assert controller.hass_url == "https://something.nabu.acas"
assert controller.hass_url == "https://something.nabu.casa"

View File

@ -58,6 +58,8 @@ async def test_websocket_core_update(hass, client):
assert hass.config.location_name != "Huis"
assert hass.config.units.name != CONF_UNIT_SYSTEM_IMPERIAL
assert hass.config.time_zone.zone != "America/New_York"
assert hass.config.external_url != "https://www.example.com"
assert hass.config.internal_url != "http://example.com"
await client.send_json(
{
@ -69,6 +71,8 @@ async def test_websocket_core_update(hass, client):
"location_name": "Huis",
CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_IMPERIAL,
"time_zone": "America/New_York",
"external_url": "https://www.example.com",
"internal_url": "http://example.local",
}
)
@ -83,6 +87,8 @@ async def test_websocket_core_update(hass, client):
assert hass.config.location_name == "Huis"
assert hass.config.units.name == CONF_UNIT_SYSTEM_IMPERIAL
assert hass.config.time_zone.zone == "America/New_York"
assert hass.config.external_url == "https://www.example.com"
assert hass.config.internal_url == "http://example.local"
dt_util.set_default_time_zone(ORIG_TIME_ZONE)

View File

@ -20,6 +20,7 @@ from homeassistant.components.google_assistant import (
smart_home as sh,
trait,
)
from homeassistant.config import async_process_ha_core_config
from homeassistant.const import ATTR_UNIT_OF_MEASUREMENT, TEMP_CELSIUS, __version__
from homeassistant.core import EVENT_CALL_SERVICE, State
from homeassistant.helpers import device_registry
@ -27,7 +28,7 @@ from homeassistant.setup import async_setup_component
from . import BASIC_CONFIG, MockConfig
from tests.async_mock import Mock, patch
from tests.async_mock import patch
from tests.common import mock_area_registry, mock_device_registry, mock_registry
REQ_ID = "ff36a3cc-ec34-11e6-b1a0-64510650abcf"
@ -798,7 +799,9 @@ async def test_query_disconnect(hass):
async def test_trait_execute_adding_query_data(hass):
"""Test a trait execute influencing query data."""
hass.config.api = Mock(base_url="http://1.1.1.1:8123")
await async_process_ha_core_config(
hass, {"external_url": "https://example.com"},
)
hass.states.async_set(
"camera.office", "idle", {"supported_features": camera.SUPPORT_STREAM}
)
@ -852,7 +855,7 @@ async def test_trait_execute_adding_query_data(hass):
"status": "SUCCESS",
"states": {
"online": True,
"cameraStreamAccessUrl": "http://1.1.1.1:8123/api/streams/bla",
"cameraStreamAccessUrl": "https://example.com/api/streams/bla",
},
}
]

View File

@ -22,6 +22,7 @@ from homeassistant.components import (
)
from homeassistant.components.climate import const as climate
from homeassistant.components.google_assistant import const, error, helpers, trait
from homeassistant.config import async_process_ha_core_config
from homeassistant.const import (
ATTR_ASSUMED_STATE,
ATTR_DEVICE_CLASS,
@ -44,7 +45,7 @@ from homeassistant.util import color
from . import BASIC_CONFIG, MockConfig
from tests.async_mock import Mock, patch
from tests.async_mock import patch
from tests.common import async_mock_service
_LOGGER = logging.getLogger(__name__)
@ -99,7 +100,9 @@ async def test_brightness_light(hass):
async def test_camera_stream(hass):
"""Test camera stream trait support for camera domain."""
hass.config.api = Mock(base_url="http://1.1.1.1:8123")
await async_process_ha_core_config(
hass, {"external_url": "https://example.com"},
)
assert helpers.get_google_type(camera.DOMAIN, None) is not None
assert trait.CameraStreamTrait.supported(camera.DOMAIN, camera.SUPPORT_STREAM, None)
@ -122,7 +125,7 @@ async def test_camera_stream(hass):
await trt.execute(trait.COMMAND_GET_CAMERA_STREAM, BASIC_DATA, {}, {})
assert trt.query_attributes() == {
"cameraStreamAccessUrl": "http://1.1.1.1:8123/api/streams/bla"
"cameraStreamAccessUrl": "https://example.com/api/streams/bla"
}

View File

@ -9,6 +9,7 @@ from homeassistant.components.media_player.const import (
SERVICE_PLAY_MEDIA,
)
import homeassistant.components.tts as tts
from homeassistant.config import async_process_ha_core_config
from homeassistant.setup import setup_component
from tests.async_mock import patch
@ -23,6 +24,13 @@ class TestTTSGooglePlatform:
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
asyncio.run_coroutine_threadsafe(
async_process_ha_core_config(
self.hass, {"internal_url": "http://example.local:8123"}
),
self.hass.loop,
)
self.url = "https://translate.google.com/translate_tts"
self.url_param = {
"tl": "en",

View File

@ -39,49 +39,54 @@ class TestApiConfig(unittest.TestCase):
def test_api_base_url_with_domain(hass):
"""Test setting API URL with domain."""
api_config = http.ApiConfig("example.com")
api_config = http.ApiConfig("127.0.0.1", "example.com")
assert api_config.base_url == "http://example.com:8123"
def test_api_base_url_with_ip(hass):
"""Test setting API URL with IP."""
api_config = http.ApiConfig("1.1.1.1")
api_config = http.ApiConfig("127.0.0.1", "1.1.1.1")
assert api_config.base_url == "http://1.1.1.1:8123"
def test_api_base_url_with_ip_and_port(hass):
"""Test setting API URL with IP and port."""
api_config = http.ApiConfig("1.1.1.1", 8124)
api_config = http.ApiConfig("127.0.0.1", "1.1.1.1", 8124)
assert api_config.base_url == "http://1.1.1.1:8124"
def test_api_base_url_with_protocol(hass):
"""Test setting API URL with protocol."""
api_config = http.ApiConfig("https://example.com")
api_config = http.ApiConfig("127.0.0.1", "https://example.com")
assert api_config.base_url == "https://example.com:8123"
def test_api_base_url_with_protocol_and_port(hass):
"""Test setting API URL with protocol and port."""
api_config = http.ApiConfig("https://example.com", 433)
api_config = http.ApiConfig("127.0.0.1", "https://example.com", 433)
assert api_config.base_url == "https://example.com:433"
def test_api_base_url_with_ssl_enable(hass):
"""Test setting API URL with use_ssl enabled."""
api_config = http.ApiConfig("example.com", use_ssl=True)
api_config = http.ApiConfig("127.0.0.1", "example.com", use_ssl=True)
assert api_config.base_url == "https://example.com:8123"
def test_api_base_url_with_ssl_enable_and_port(hass):
"""Test setting API URL with use_ssl enabled and port."""
api_config = http.ApiConfig("1.1.1.1", use_ssl=True, port=8888)
api_config = http.ApiConfig("127.0.0.1", "1.1.1.1", use_ssl=True, port=8888)
assert api_config.base_url == "https://1.1.1.1:8888"
def test_api_base_url_with_protocol_and_ssl_enable(hass):
"""Test setting API URL with specific protocol and use_ssl enabled."""
api_config = http.ApiConfig("http://example.com", use_ssl=True)
api_config = http.ApiConfig("127.0.0.1", "http://example.com", use_ssl=True)
assert api_config.base_url == "http://example.com:8123"
def test_api_base_url_removes_trailing_slash(hass):
"""Test a trialing slash is removed when setting the API URL."""
api_config = http.ApiConfig("http://example.com/")
api_config = http.ApiConfig("127.0.0.1", "http://example.com/")
assert api_config.base_url == "http://example.com:8123"
def test_api_local_ip(hass):
"""Test a trialing slash is removed when setting the API URL."""
api_config = http.ApiConfig("127.0.0.1", "http://example.com/")
assert api_config.local_ip == "127.0.0.1"
async def test_api_base_url_with_domain(hass):
"""Test setting API URL."""
@ -117,6 +122,13 @@ async def test_api_no_base_url(hass):
assert hass.config.api.base_url == "http://127.0.0.1:8123"
async def test_api_local_ip(hass):
"""Test setting api url."""
result = await async_setup_component(hass, "http", {"http": {}})
assert result
assert hass.config.api.local_ip == "127.0.0.1"
async def test_api_base_url_removes_trailing_slash(hass):
"""Test setting api url."""
result = await async_setup_component(

View File

@ -3,6 +3,7 @@ import pytest
from homeassistant.components import konnected
from homeassistant.components.konnected import config_flow
from homeassistant.config import async_process_ha_core_config
from homeassistant.const import HTTP_NOT_FOUND
from homeassistant.setup import async_setup_component
@ -385,6 +386,9 @@ async def test_config_passed_to_config_entry(hass):
async def test_unload_entry(hass, mock_panel):
"""Test being able to unload an entry."""
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local:8123"},
)
entry = MockConfigEntry(
domain=konnected.DOMAIN, data={konnected.CONF_ID: "aabbccddeeff"}
)
@ -563,7 +567,9 @@ async def test_api(hass, aiohttp_client, mock_panel):
async def test_state_updates_zone(hass, aiohttp_client, mock_panel):
"""Test callback view."""
await async_setup_component(hass, "http", {"http": {}})
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local:8123"},
)
device_config = config_flow.CONFIG_ENTRY_SCHEMA(
{
@ -711,7 +717,9 @@ async def test_state_updates_zone(hass, aiohttp_client, mock_panel):
async def test_state_updates_pin(hass, aiohttp_client, mock_panel):
"""Test callback view."""
await async_setup_component(hass, "http", {"http": {}})
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local:8123"},
)
device_config = config_flow.CONFIG_ENTRY_SCHEMA(
{

View File

@ -1,4 +1,5 @@
"""The tests for the MaryTTS speech platform."""
import asyncio
import os
import shutil
from urllib.parse import urlencode
@ -11,6 +12,7 @@ from homeassistant.components.media_player.const import (
SERVICE_PLAY_MEDIA,
)
import homeassistant.components.tts as tts
from homeassistant.config import async_process_ha_core_config
from homeassistant.const import HTTP_INTERNAL_SERVER_ERROR
from homeassistant.setup import setup_component
@ -24,6 +26,13 @@ class TestTTSMaryTTSPlatform:
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
asyncio.run_coroutine_threadsafe(
async_process_ha_core_config(
self.hass, {"internal_url": "http://example.local:8123"}
),
self.hass.loop,
)
self.host = "localhost"
self.port = 59125
self.params = {

View File

@ -5,6 +5,7 @@ from homeassistant import data_entry_flow
from homeassistant.components.owntracks import config_flow
from homeassistant.components.owntracks.config_flow import CONF_CLOUDHOOK, CONF_SECRET
from homeassistant.components.owntracks.const import DOMAIN
from homeassistant.config import async_process_ha_core_config
from homeassistant.const import CONF_WEBHOOK_ID
from homeassistant.setup import async_setup_component
@ -86,6 +87,10 @@ async def test_import(hass, webhook_id, secret):
async def test_import_setup(hass):
"""Test that we automatically create a config flow."""
await async_process_ha_core_config(
hass, {"external_url": "http://example.com"},
)
assert not hass.config_entries.async_entries(DOMAIN)
assert await async_setup_component(hass, DOMAIN, {"owntracks": {}})
await hass.async_block_till_done()
@ -124,6 +129,10 @@ async def test_user_not_supports_encryption(hass, not_supports_encryption):
async def test_unload(hass):
"""Test unloading a config flow."""
await async_process_ha_core_config(
hass, {"external_url": "http://example.com"},
)
with patch(
"homeassistant.config_entries.ConfigEntries.async_forward_entry_setup"
) as mock_forward:

View File

@ -20,6 +20,7 @@ from homeassistant.components.plex.const import (
PLEX_UPDATE_PLATFORMS_SIGNAL,
SERVERS,
)
from homeassistant.config import async_process_ha_core_config
from homeassistant.config_entries import ENTRY_STATE_LOADED
from homeassistant.const import (
CONF_HOST,
@ -30,7 +31,6 @@ from homeassistant.const import (
CONF_VERIFY_SSL,
)
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.setup import async_setup_component
from .const import DEFAULT_DATA, DEFAULT_OPTIONS, MOCK_SERVERS, MOCK_TOKEN
from .mock_classes import MockPlexAccount, MockPlexServer
@ -41,6 +41,9 @@ from tests.common import MockConfigEntry
async def test_bad_credentials(hass):
"""Test when provided credentials are rejected."""
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local:8123"},
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}
@ -111,6 +114,9 @@ async def test_import_bad_hostname(hass):
async def test_unknown_exception(hass):
"""Test when an unknown exception is encountered."""
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local:8123"},
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}
@ -137,7 +143,9 @@ async def test_unknown_exception(hass):
async def test_no_servers_found(hass):
"""Test when no servers are on an account."""
await async_setup_component(hass, "http", {"http": {}})
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local:8123"},
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}
@ -169,7 +177,9 @@ async def test_single_available_server(hass):
mock_plex_server = MockPlexServer()
await async_setup_component(hass, "http", {"http": {}})
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local:8123"},
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}
@ -206,7 +216,9 @@ async def test_multiple_servers_with_selection(hass):
mock_plex_server = MockPlexServer()
await async_setup_component(hass, "http", {"http": {}})
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local:8123"},
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}
@ -251,7 +263,9 @@ async def test_adding_last_unconfigured_server(hass):
mock_plex_server = MockPlexServer()
await async_setup_component(hass, "http", {"http": {}})
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local:8123"},
)
MockConfigEntry(
domain=DOMAIN,
@ -325,7 +339,9 @@ async def test_already_configured(hass):
async def test_all_available_servers_configured(hass):
"""Test when all available servers are already configured."""
await async_setup_component(hass, "http", {"http": {}})
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local:8123"},
)
MockConfigEntry(
domain=DOMAIN,
@ -467,7 +483,9 @@ async def test_option_flow_new_users_available(hass, caplog):
async def test_external_timed_out(hass):
"""Test when external flow times out."""
await async_setup_component(hass, "http", {"http": {}})
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local:8123"},
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}
@ -494,7 +512,9 @@ async def test_external_timed_out(hass):
async def test_callback_view(hass, aiohttp_client):
"""Test callback view."""
await async_setup_component(hass, "http", {"http": {}})
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local:8123"},
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}
@ -534,6 +554,9 @@ async def test_multiple_servers_with_import(hass):
async def test_manual_config(hass):
"""Test creating via manual configuration."""
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local:8123"},
)
class WrongCertValidaitionException(requests.exceptions.SSLError):
"""Mock the exception showing an unmatched error."""

View File

@ -3,12 +3,17 @@ from datetime import timedelta
import io
from homeassistant import core as ha
from homeassistant.config import async_process_ha_core_config
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
async def test_bad_posting(hass, aiohttp_client):
"""Test that posting to wrong api endpoint fails."""
await async_process_ha_core_config(
hass, {"external_url": "http://example.com"},
)
await async_setup_component(
hass,
"camera",
@ -35,6 +40,10 @@ async def test_bad_posting(hass, aiohttp_client):
async def test_posting_url(hass, aiohttp_client):
"""Test that posting to api endpoint works."""
await async_process_ha_core_config(
hass, {"external_url": "http://example.com"},
)
await async_setup_component(
hass,
"camera",

View File

@ -14,6 +14,7 @@ from homeassistant.components.media_player.const import (
)
import homeassistant.components.tts as tts
from homeassistant.components.tts import _get_cache_files
from homeassistant.config import async_process_ha_core_config
from homeassistant.const import HTTP_NOT_FOUND
from homeassistant.setup import async_setup_component
@ -84,6 +85,14 @@ def mutagen_mock():
yield
@pytest.fixture(autouse=True)
async def internal_url_mock(hass):
"""Mock internal URL of the instance."""
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local:8123"},
)
async def test_setup_component_demo(hass):
"""Set up the demo platform with defaults."""
config = {tts.DOMAIN: {"platform": "demo"}}
@ -127,10 +136,9 @@ async def test_setup_component_and_test_service(hass, empty_cache_dir):
assert len(calls) == 1
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC
assert calls[0].data[
ATTR_MEDIA_CONTENT_ID
] == "{}/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3".format(
hass.config.api.base_url
assert (
calls[0].data[ATTR_MEDIA_CONTENT_ID]
== "http://example.local:8123/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
)
await hass.async_block_till_done()
assert (
@ -160,10 +168,9 @@ async def test_setup_component_and_test_service_with_config_language(
)
assert len(calls) == 1
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC
assert calls[0].data[
ATTR_MEDIA_CONTENT_ID
] == "{}/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_-_demo.mp3".format(
hass.config.api.base_url
assert (
calls[0].data[ATTR_MEDIA_CONTENT_ID]
== "http://example.local:8123/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_-_demo.mp3"
)
await hass.async_block_till_done()
assert (
@ -202,10 +209,9 @@ async def test_setup_component_and_test_service_with_service_language(
)
assert len(calls) == 1
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC
assert calls[0].data[
ATTR_MEDIA_CONTENT_ID
] == "{}/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_-_demo.mp3".format(
hass.config.api.base_url
assert (
calls[0].data[ATTR_MEDIA_CONTENT_ID]
== "http://example.local:8123/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_-_demo.mp3"
)
await hass.async_block_till_done()
assert (
@ -267,10 +273,9 @@ async def test_setup_component_and_test_service_with_service_options(
assert len(calls) == 1
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC
assert calls[0].data[
ATTR_MEDIA_CONTENT_ID
] == "{}/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_{}_demo.mp3".format(
hass.config.api.base_url, opt_hash
assert (
calls[0].data[ATTR_MEDIA_CONTENT_ID]
== f"http://example.local:8123/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_{opt_hash}_demo.mp3"
)
await hass.async_block_till_done()
assert (
@ -305,10 +310,9 @@ async def test_setup_component_and_test_with_service_options_def(hass, empty_cac
assert len(calls) == 1
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC
assert calls[0].data[
ATTR_MEDIA_CONTENT_ID
] == "{}/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_{}_demo.mp3".format(
hass.config.api.base_url, opt_hash
assert (
calls[0].data[ATTR_MEDIA_CONTENT_ID]
== f"http://example.local:8123/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_{opt_hash}_demo.mp3"
)
await hass.async_block_till_done()
assert (
@ -603,10 +607,9 @@ async def test_setup_component_test_with_cache_dir(
blocking=True,
)
assert len(calls) == 1
assert calls[0].data[
ATTR_MEDIA_CONTENT_ID
] == "{}/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3".format(
hass.config.api.base_url
assert (
calls[0].data[ATTR_MEDIA_CONTENT_ID]
== "http://example.local:8123/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
)
@ -662,9 +665,7 @@ async def test_setup_component_and_web_get_url(hass, hass_client):
assert req.status == 200
response = await req.json()
assert response.get("url") == (
"{}/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3".format(
hass.config.api.base_url
)
"http://example.local:8123/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
)

View File

@ -9,6 +9,7 @@ from homeassistant.components.media_player.const import (
SERVICE_PLAY_MEDIA,
)
import homeassistant.components.tts as tts
from homeassistant.config import async_process_ha_core_config
from homeassistant.setup import setup_component
from tests.common import assert_setup_component, get_test_home_assistant, mock_service
@ -22,6 +23,13 @@ class TestTTSVoiceRSSPlatform:
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
asyncio.run_coroutine_threadsafe(
async_process_ha_core_config(
self.hass, {"internal_url": "http://example.local:8123"}
),
self.hass.loop,
)
self.url = "https://api.voicerss.org/"
self.form_data = {
"key": "1234567xx",

View File

@ -124,7 +124,7 @@ async def configure_integration(
assert result["url"] == (
"https://account.withings.com/oauth2_user/authorize2?"
"response_type=code&client_id=my_client_id&"
"redirect_uri=http://127.0.0.1:8080/auth/external/callback&"
"redirect_uri=http://example.local/auth/external/callback&"
f"state={state}"
"&scope=user.info,user.metrics,user.activity"
)

View File

@ -14,6 +14,7 @@ from homeassistant.components.withings import (
async_setup_entry,
const,
)
from homeassistant.config import async_process_ha_core_config
from homeassistant.const import STATE_UNKNOWN
from homeassistant.core import HomeAssistant
@ -164,6 +165,10 @@ async def test_upgrade_token(
config = await setup_hass(hass)
profiles = config[const.DOMAIN][const.PROFILES]
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local"},
)
await configure_integration(
hass=hass,
aiohttp_client=aiohttp_client,
@ -234,6 +239,10 @@ async def test_auth_failure(
config = await setup_hass(hass)
profiles = config[const.DOMAIN][const.PROFILES]
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local"},
)
await configure_integration(
hass=hass,
aiohttp_client=aiohttp_client,
@ -269,6 +278,10 @@ async def test_full_setup(hass: HomeAssistant, aiohttp_client, aioclient_mock) -
config = await setup_hass(hass)
profiles = config[const.DOMAIN][const.PROFILES]
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local"},
)
await configure_integration(
hass=hass,
aiohttp_client=aiohttp_client,

View File

@ -8,6 +8,7 @@ from homeassistant.components.media_player.const import (
SERVICE_PLAY_MEDIA,
)
import homeassistant.components.tts as tts
from homeassistant.config import async_process_ha_core_config
from homeassistant.const import HTTP_FORBIDDEN
from homeassistant.setup import setup_component
@ -25,6 +26,13 @@ class TestTTSYandexPlatform:
self.hass = get_test_home_assistant()
self._base_url = "https://tts.voicetech.yandex.net/generate?"
asyncio.run_coroutine_threadsafe(
async_process_ha_core_config(
self.hass, {"internal_url": "http://example.local:8123"}
),
self.hass.loop,
)
def teardown_method(self):
"""Stop everything that was started."""
default_tts = self.hass.config.path(tts.DEFAULT_CACHE_DIR)

View File

@ -1,34 +1,687 @@
"""Test network helper."""
import pytest
from homeassistant.components import cloud
from homeassistant.helpers import network
from homeassistant.config import async_process_ha_core_config
from homeassistant.core import HomeAssistant
from homeassistant.helpers.network import (
NoURLAvailableError,
_async_get_cloud_url,
_async_get_deprecated_base_url,
_async_get_external_url,
_async_get_internal_url,
async_get_url,
)
from tests.async_mock import Mock, patch
async def test_get_external_url(hass):
"""Test get_external_url."""
hass.config.api = Mock(base_url="http://192.168.1.100:8123")
async def test_get_url_internal(hass: HomeAssistant):
"""Test getting an instance URL when the user has set an internal URL."""
assert hass.config.internal_url is None
assert network.async_get_external_url(hass) is None
# Test with internal URL: http://example.local:8123
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local:8123"},
)
hass.config.api = Mock(base_url="http://example.duckdns.org:8123")
assert hass.config.internal_url == "http://example.local:8123"
assert _async_get_internal_url(hass) == "http://example.local:8123"
assert _async_get_internal_url(hass, allow_ip=False) == "http://example.local:8123"
assert network.async_get_external_url(hass) == "http://example.duckdns.org:8123"
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, require_standard_port=True)
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, require_ssl=True)
# Test with internal URL: https://example.local:8123
await async_process_ha_core_config(
hass, {"internal_url": "https://example.local:8123"},
)
assert hass.config.internal_url == "https://example.local:8123"
assert _async_get_internal_url(hass) == "https://example.local:8123"
assert _async_get_internal_url(hass, allow_ip=False) == "https://example.local:8123"
assert (
_async_get_internal_url(hass, require_ssl=True) == "https://example.local:8123"
)
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, require_standard_port=True)
# Test with internal URL: http://example.local:80/
await async_process_ha_core_config(
hass, {"internal_url": "http://example.local:80/"},
)
assert hass.config.internal_url == "http://example.local:80/"
assert _async_get_internal_url(hass) == "http://example.local"
assert _async_get_internal_url(hass, allow_ip=False) == "http://example.local"
assert (
_async_get_internal_url(hass, require_standard_port=True)
== "http://example.local"
)
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, require_ssl=True)
# Test with internal URL: https://example.local:443
await async_process_ha_core_config(
hass, {"internal_url": "https://example.local:443"},
)
assert hass.config.internal_url == "https://example.local:443"
assert _async_get_internal_url(hass) == "https://example.local"
assert _async_get_internal_url(hass, allow_ip=False) == "https://example.local"
assert (
_async_get_internal_url(hass, require_standard_port=True)
== "https://example.local"
)
assert _async_get_internal_url(hass, require_ssl=True) == "https://example.local"
# Test with internal URL: https://192.168.0.1
await async_process_ha_core_config(
hass, {"internal_url": "https://192.168.0.1"},
)
assert hass.config.internal_url == "https://192.168.0.1"
assert _async_get_internal_url(hass) == "https://192.168.0.1"
assert (
_async_get_internal_url(hass, require_standard_port=True)
== "https://192.168.0.1"
)
assert _async_get_internal_url(hass, require_ssl=True) == "https://192.168.0.1"
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, allow_ip=False)
# Test with internal URL: http://192.168.0.1:8123
await async_process_ha_core_config(
hass, {"internal_url": "http://192.168.0.1:8123"},
)
assert hass.config.internal_url == "http://192.168.0.1:8123"
assert _async_get_internal_url(hass) == "http://192.168.0.1:8123"
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, require_standard_port=True)
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, require_ssl=True)
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, allow_ip=False)
async def test_get_url_internal_fallback(hass: HomeAssistant):
"""Test getting an instance URL when the user has not set an internal URL."""
assert hass.config.internal_url is None
hass.config.api = Mock(
use_ssl=False, port=8123, base_url=None, local_ip="192.168.123.123"
)
assert _async_get_internal_url(hass) == "http://192.168.123.123:8123"
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, allow_ip=False)
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, require_standard_port=True)
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, require_ssl=True)
hass.config.api = Mock(
use_ssl=False, port=80, base_url=None, local_ip="192.168.123.123"
)
assert _async_get_internal_url(hass) == "http://192.168.123.123"
assert (
_async_get_internal_url(hass, require_standard_port=True)
== "http://192.168.123.123"
)
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, allow_ip=False)
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, require_ssl=True)
hass.config.api = Mock(use_ssl=True, port=443, base_url=None)
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass)
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, require_standard_port=True)
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, allow_ip=False)
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, require_ssl=True)
# Do no accept any local loopback address as fallback
hass.config.api = Mock(use_ssl=False, port=80, base_url=None, local_ip="127.0.0.1")
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass)
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, require_standard_port=True)
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, allow_ip=False)
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, require_ssl=True)
async def test_get_url_external(hass: HomeAssistant):
"""Test getting an instance URL when the user has set an external URL."""
assert hass.config.external_url is None
# Test with external URL: http://example.com:8123
await async_process_ha_core_config(
hass, {"external_url": "http://example.com:8123"},
)
assert hass.config.external_url == "http://example.com:8123"
assert _async_get_external_url(hass) == "http://example.com:8123"
assert _async_get_external_url(hass, allow_cloud=False) == "http://example.com:8123"
assert _async_get_external_url(hass, allow_ip=False) == "http://example.com:8123"
assert _async_get_external_url(hass, prefer_cloud=True) == "http://example.com:8123"
with pytest.raises(NoURLAvailableError):
_async_get_external_url(hass, require_standard_port=True)
with pytest.raises(NoURLAvailableError):
_async_get_external_url(hass, require_ssl=True)
# Test with external URL: http://example.com:80/
await async_process_ha_core_config(
hass, {"external_url": "http://example.com:80/"},
)
assert hass.config.external_url == "http://example.com:80/"
assert _async_get_external_url(hass) == "http://example.com"
assert _async_get_external_url(hass, allow_cloud=False) == "http://example.com"
assert _async_get_external_url(hass, allow_ip=False) == "http://example.com"
assert _async_get_external_url(hass, prefer_cloud=True) == "http://example.com"
assert (
_async_get_external_url(hass, require_standard_port=True)
== "http://example.com"
)
with pytest.raises(NoURLAvailableError):
_async_get_external_url(hass, require_ssl=True)
# Test with external url: https://example.com:443/
await async_process_ha_core_config(
hass, {"external_url": "https://example.com:443/"},
)
assert hass.config.external_url == "https://example.com:443/"
assert _async_get_external_url(hass) == "https://example.com"
assert _async_get_external_url(hass, allow_cloud=False) == "https://example.com"
assert _async_get_external_url(hass, allow_ip=False) == "https://example.com"
assert _async_get_external_url(hass, prefer_cloud=True) == "https://example.com"
assert _async_get_external_url(hass, require_ssl=False) == "https://example.com"
assert (
_async_get_external_url(hass, require_standard_port=True)
== "https://example.com"
)
# Test with external URL: https://example.com:80
await async_process_ha_core_config(
hass, {"external_url": "https://example.com:80"},
)
assert hass.config.external_url == "https://example.com:80"
assert _async_get_external_url(hass) == "https://example.com:80"
assert _async_get_external_url(hass, allow_cloud=False) == "https://example.com:80"
assert _async_get_external_url(hass, allow_ip=False) == "https://example.com:80"
assert _async_get_external_url(hass, prefer_cloud=True) == "https://example.com:80"
assert _async_get_external_url(hass, require_ssl=True) == "https://example.com:80"
with pytest.raises(NoURLAvailableError):
_async_get_external_url(hass, require_standard_port=True)
# Test with external URL: https://192.168.0.1
await async_process_ha_core_config(
hass, {"external_url": "https://192.168.0.1"},
)
assert hass.config.external_url == "https://192.168.0.1"
assert _async_get_external_url(hass) == "https://192.168.0.1"
assert _async_get_external_url(hass, allow_cloud=False) == "https://192.168.0.1"
assert _async_get_external_url(hass, prefer_cloud=True) == "https://192.168.0.1"
assert (
_async_get_external_url(hass, require_standard_port=True)
== "https://192.168.0.1"
)
with pytest.raises(NoURLAvailableError):
_async_get_external_url(hass, allow_ip=False)
with pytest.raises(NoURLAvailableError):
_async_get_external_url(hass, require_ssl=True)
async def test_get_cloud_url(hass: HomeAssistant):
"""Test getting an instance URL when the user has set an external URL."""
assert hass.config.external_url is None
hass.config.components.add("cloud")
assert network.async_get_external_url(hass) == "http://example.duckdns.org:8123"
with patch.object(
hass.components.cloud,
"async_remote_ui_url",
side_effect=cloud.CloudNotAvailable,
):
assert network.async_get_external_url(hass) == "http://example.duckdns.org:8123"
with patch.object(
hass.components.cloud,
"async_remote_ui_url",
return_value="https://example.nabu.casa",
):
assert network.async_get_external_url(hass) == "https://example.nabu.casa"
assert _async_get_cloud_url(hass) == "https://example.nabu.casa"
with patch.object(
hass.components.cloud,
"async_remote_ui_url",
side_effect=cloud.CloudNotAvailable,
):
with pytest.raises(NoURLAvailableError):
_async_get_cloud_url(hass)
async def test_get_external_url_cloud_fallback(hass: HomeAssistant):
"""Test getting an external instance URL with cloud fallback."""
assert hass.config.external_url is None
# Test with external URL: http://1.1.1.1:8123
await async_process_ha_core_config(
hass, {"external_url": "http://1.1.1.1:8123"},
)
assert hass.config.external_url == "http://1.1.1.1:8123"
assert _async_get_external_url(hass, prefer_cloud=True) == "http://1.1.1.1:8123"
# Add Cloud to the previous test
hass.config.components.add("cloud")
with patch.object(
hass.components.cloud,
"async_remote_ui_url",
return_value="https://example.nabu.casa",
):
assert _async_get_external_url(hass, allow_cloud=False) == "http://1.1.1.1:8123"
assert (
_async_get_external_url(hass, allow_ip=False) == "https://example.nabu.casa"
)
assert (
_async_get_external_url(hass, prefer_cloud=False) == "http://1.1.1.1:8123"
)
assert (
_async_get_external_url(hass, prefer_cloud=True)
== "https://example.nabu.casa"
)
assert (
_async_get_external_url(hass, require_ssl=True)
== "https://example.nabu.casa"
)
assert (
_async_get_external_url(hass, require_standard_port=True)
== "https://example.nabu.casa"
)
# Test with external URL: https://example.com
await async_process_ha_core_config(
hass, {"external_url": "https://example.com"},
)
assert hass.config.external_url == "https://example.com"
assert _async_get_external_url(hass, prefer_cloud=True) == "https://example.com"
# Add Cloud to the previous test
hass.config.components.add("cloud")
with patch.object(
hass.components.cloud,
"async_remote_ui_url",
return_value="https://example.nabu.casa",
):
assert _async_get_external_url(hass, allow_cloud=False) == "https://example.com"
assert _async_get_external_url(hass, allow_ip=False) == "https://example.com"
assert (
_async_get_external_url(hass, prefer_cloud=False) == "https://example.com"
)
assert (
_async_get_external_url(hass, prefer_cloud=True)
== "https://example.nabu.casa"
)
assert _async_get_external_url(hass, require_ssl=True) == "https://example.com"
assert (
_async_get_external_url(hass, require_standard_port=True)
== "https://example.com"
)
assert (
_async_get_external_url(hass, prefer_cloud=True, allow_cloud=False)
== "https://example.com"
)
async def test_get_url(hass: HomeAssistant):
"""Test getting an instance URL."""
assert hass.config.external_url is None
assert hass.config.internal_url is None
with pytest.raises(NoURLAvailableError):
async_get_url(hass)
hass.config.api = Mock(
use_ssl=False, port=8123, base_url=None, local_ip="192.168.123.123"
)
assert async_get_url(hass) == "http://192.168.123.123:8123"
assert async_get_url(hass, prefer_external=True) == "http://192.168.123.123:8123"
with pytest.raises(NoURLAvailableError):
async_get_url(hass, allow_internal=False)
# Test only external
hass.config.api = None
await async_process_ha_core_config(
hass, {"external_url": "https://example.com"},
)
assert hass.config.external_url == "https://example.com"
assert hass.config.internal_url is None
assert async_get_url(hass) == "https://example.com"
# Test preference or allowance
await async_process_ha_core_config(
hass,
{"internal_url": "http://example.local", "external_url": "https://example.com"},
)
assert hass.config.external_url == "https://example.com"
assert hass.config.internal_url == "http://example.local"
assert async_get_url(hass) == "http://example.local"
assert async_get_url(hass, prefer_external=True) == "https://example.com"
assert async_get_url(hass, allow_internal=False) == "https://example.com"
assert (
async_get_url(hass, prefer_external=True, allow_external=False)
== "http://example.local"
)
with pytest.raises(NoURLAvailableError):
async_get_url(hass, allow_external=False, require_ssl=True)
with pytest.raises(NoURLAvailableError):
async_get_url(hass, allow_external=False, allow_internal=False)
async def test_get_deprecated_base_url_internal(hass: HomeAssistant):
"""Test getting an internal instance URL from the deprecated base_url."""
# Test with SSL local URL
hass.config.api = Mock(base_url="https://example.local")
assert (
_async_get_deprecated_base_url(hass, internal=True) == "https://example.local"
)
assert (
_async_get_deprecated_base_url(hass, internal=True, allow_ip=False)
== "https://example.local"
)
assert (
_async_get_deprecated_base_url(hass, internal=True, require_ssl=True)
== "https://example.local"
)
assert (
_async_get_deprecated_base_url(hass, internal=True, require_standard_port=True)
== "https://example.local"
)
# Test with no SSL, local IP URL
hass.config.api = Mock(base_url="http://10.10.10.10:8123")
assert (
_async_get_deprecated_base_url(hass, internal=True) == "http://10.10.10.10:8123"
)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, internal=True, allow_ip=False)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, internal=True, require_ssl=True)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, internal=True, require_standard_port=True)
# Test with SSL, local IP URL
hass.config.api = Mock(base_url="https://10.10.10.10")
assert _async_get_deprecated_base_url(hass, internal=True) == "https://10.10.10.10"
assert (
_async_get_deprecated_base_url(hass, internal=True, require_ssl=True)
== "https://10.10.10.10"
)
assert (
_async_get_deprecated_base_url(hass, internal=True, require_standard_port=True)
== "https://10.10.10.10"
)
# Test external URL
hass.config.api = Mock(base_url="https://example.com")
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, internal=True)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, internal=True, require_ssl=True)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, internal=True, require_standard_port=True)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, internal=True, allow_ip=False)
# Test with loopback
hass.config.api = Mock(base_url="https://127.0.0.42")
with pytest.raises(NoURLAvailableError):
assert _async_get_deprecated_base_url(hass, internal=True)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, internal=True, allow_ip=False)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, internal=True, require_ssl=True)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, internal=True, require_standard_port=True)
async def test_get_deprecated_base_url_external(hass: HomeAssistant):
"""Test getting an external instance URL from the deprecated base_url."""
# Test with SSL and external domain on standard port
hass.config.api = Mock(base_url="https://example.com:443/")
assert _async_get_deprecated_base_url(hass) == "https://example.com"
assert (
_async_get_deprecated_base_url(hass, require_ssl=True) == "https://example.com"
)
assert (
_async_get_deprecated_base_url(hass, require_standard_port=True)
== "https://example.com"
)
# Test without SSL and external domain on non-standard port
hass.config.api = Mock(base_url="http://example.com:8123/")
assert _async_get_deprecated_base_url(hass) == "http://example.com:8123"
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, require_ssl=True)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, require_standard_port=True)
# Test SSL on external IP
hass.config.api = Mock(base_url="https://1.1.1.1")
assert _async_get_deprecated_base_url(hass) == "https://1.1.1.1"
assert _async_get_deprecated_base_url(hass, require_ssl=True) == "https://1.1.1.1"
assert (
_async_get_deprecated_base_url(hass, require_standard_port=True)
== "https://1.1.1.1"
)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, allow_ip=False)
# Test with private IP
hass.config.api = Mock(base_url="https://10.10.10.10")
with pytest.raises(NoURLAvailableError):
assert _async_get_deprecated_base_url(hass)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, allow_ip=False)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, require_ssl=True)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, require_standard_port=True)
# Test with local domain
hass.config.api = Mock(base_url="https://example.local")
with pytest.raises(NoURLAvailableError):
assert _async_get_deprecated_base_url(hass)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, allow_ip=False)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, require_ssl=True)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, require_standard_port=True)
# Test with loopback
hass.config.api = Mock(base_url="https://127.0.0.42")
with pytest.raises(NoURLAvailableError):
assert _async_get_deprecated_base_url(hass)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, allow_ip=False)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, require_ssl=True)
with pytest.raises(NoURLAvailableError):
_async_get_deprecated_base_url(hass, require_standard_port=True)
async def test_get_internal_url_with_base_url_fallback(hass: HomeAssistant):
"""Test getting an internal instance URL with the deprecated base_url fallback."""
hass.config.api = Mock(
use_ssl=False, port=8123, base_url=None, local_ip="192.168.123.123"
)
assert hass.config.internal_url is None
assert _async_get_internal_url(hass) == "http://192.168.123.123:8123"
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, allow_ip=False)
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, require_standard_port=True)
with pytest.raises(NoURLAvailableError):
_async_get_internal_url(hass, require_ssl=True)
# Add base_url
hass.config.api = Mock(use_ssl=False, port=8123, base_url="https://example.local")
assert _async_get_internal_url(hass) == "https://example.local"
assert _async_get_internal_url(hass, allow_ip=False) == "https://example.local"
assert (
_async_get_internal_url(hass, require_standard_port=True)
== "https://example.local"
)
assert _async_get_internal_url(hass, require_ssl=True) == "https://example.local"
# Add internal URL
await async_process_ha_core_config(
hass, {"internal_url": "https://internal.local"},
)
assert _async_get_internal_url(hass) == "https://internal.local"
assert _async_get_internal_url(hass, allow_ip=False) == "https://internal.local"
assert (
_async_get_internal_url(hass, require_standard_port=True)
== "https://internal.local"
)
assert _async_get_internal_url(hass, require_ssl=True) == "https://internal.local"
# Add internal URL, mixed results
await async_process_ha_core_config(
hass, {"internal_url": "http://internal.local:8123"},
)
assert _async_get_internal_url(hass) == "http://internal.local:8123"
assert _async_get_internal_url(hass, allow_ip=False) == "http://internal.local:8123"
assert (
_async_get_internal_url(hass, require_standard_port=True)
== "https://example.local"
)
assert _async_get_internal_url(hass, require_ssl=True) == "https://example.local"
# Add internal URL set to an IP
await async_process_ha_core_config(
hass, {"internal_url": "http://10.10.10.10:8123"},
)
assert _async_get_internal_url(hass) == "http://10.10.10.10:8123"
assert _async_get_internal_url(hass, allow_ip=False) == "https://example.local"
assert (
_async_get_internal_url(hass, require_standard_port=True)
== "https://example.local"
)
assert _async_get_internal_url(hass, require_ssl=True) == "https://example.local"
async def test_get_external_url_with_base_url_fallback(hass: HomeAssistant):
"""Test getting an external instance URL with the deprecated base_url fallback."""
hass.config.api = Mock(use_ssl=False, port=8123, base_url=None)
assert hass.config.internal_url is None
with pytest.raises(NoURLAvailableError):
_async_get_external_url(hass)
# Test with SSL and external domain on standard port
hass.config.api = Mock(base_url="https://example.com:443/")
assert _async_get_external_url(hass) == "https://example.com"
assert _async_get_external_url(hass, allow_ip=False) == "https://example.com"
assert _async_get_external_url(hass, require_ssl=True) == "https://example.com"
assert (
_async_get_external_url(hass, require_standard_port=True)
== "https://example.com"
)
# Add external URL
await async_process_ha_core_config(
hass, {"external_url": "https://external.example.com"},
)
assert _async_get_external_url(hass) == "https://external.example.com"
assert (
_async_get_external_url(hass, allow_ip=False) == "https://external.example.com"
)
assert (
_async_get_external_url(hass, require_standard_port=True)
== "https://external.example.com"
)
assert (
_async_get_external_url(hass, require_ssl=True)
== "https://external.example.com"
)
# Add external URL, mixed results
await async_process_ha_core_config(
hass, {"external_url": "http://external.example.com:8123"},
)
assert _async_get_external_url(hass) == "http://external.example.com:8123"
assert (
_async_get_external_url(hass, allow_ip=False)
== "http://external.example.com:8123"
)
assert (
_async_get_external_url(hass, require_standard_port=True)
== "https://example.com"
)
assert _async_get_external_url(hass, require_ssl=True) == "https://example.com"
# Add external URL set to an IP
await async_process_ha_core_config(
hass, {"external_url": "http://1.1.1.1:8123"},
)
assert _async_get_external_url(hass) == "http://1.1.1.1:8123"
assert _async_get_external_url(hass, allow_ip=False) == "https://example.com"
assert (
_async_get_external_url(hass, require_standard_port=True)
== "https://example.com"
)
assert _async_get_external_url(hass, require_ssl=True) == "https://example.com"

View File

@ -178,6 +178,8 @@ def test_core_config_schema():
{"time_zone": "non-exist"},
{"latitude": "91"},
{"longitude": -181},
{"external_url": "not an url"},
{"internal_url": "not an url"},
{"customize": "bla"},
{"customize": {"light.sensor": 100}},
{"customize": {"entity_id": []}},
@ -190,6 +192,8 @@ def test_core_config_schema():
"name": "Test name",
"latitude": "-23.45",
"longitude": "123.45",
"external_url": "https://www.example.com",
"internal_url": "http://example.local",
CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC,
"customize": {"sensor.temperature": {"hidden": True}},
}
@ -342,6 +346,8 @@ async def test_loading_configuration_from_storage(hass, hass_storage):
"longitude": 13,
"time_zone": "Europe/Copenhagen",
"unit_system": "metric",
"external_url": "https://www.example.com",
"internal_url": "http://example.local",
},
"key": "core.config",
"version": 1,
@ -356,6 +362,8 @@ async def test_loading_configuration_from_storage(hass, hass_storage):
assert hass.config.location_name == "Home"
assert hass.config.units.name == CONF_UNIT_SYSTEM_METRIC
assert hass.config.time_zone.zone == "Europe/Copenhagen"
assert hass.config.external_url == "https://www.example.com"
assert hass.config.internal_url == "http://example.local"
assert len(hass.config.whitelist_external_dirs) == 2
assert "/etc" in hass.config.whitelist_external_dirs
assert hass.config.config_source == SOURCE_STORAGE
@ -371,6 +379,8 @@ async def test_updating_configuration(hass, hass_storage):
"longitude": 13,
"time_zone": "Europe/Copenhagen",
"unit_system": "metric",
"external_url": "https://www.example.com",
"internal_url": "http://example.local",
},
"key": "core.config",
"version": 1,
@ -428,6 +438,8 @@ async def test_loading_configuration(hass):
CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_IMPERIAL,
"time_zone": "America/New_York",
"whitelist_external_dirs": "/etc",
"external_url": "https://www.example.com",
"internal_url": "http://example.local",
},
)
@ -437,6 +449,8 @@ async def test_loading_configuration(hass):
assert hass.config.location_name == "Huis"
assert hass.config.units.name == CONF_UNIT_SYSTEM_IMPERIAL
assert hass.config.time_zone.zone == "America/New_York"
assert hass.config.external_url == "https://www.example.com"
assert hass.config.internal_url == "http://example.local"
assert len(hass.config.whitelist_external_dirs) == 2
assert "/etc" in hass.config.whitelist_external_dirs
assert hass.config.config_source == config_util.SOURCE_YAML
@ -453,6 +467,8 @@ async def test_loading_configuration_temperature_unit(hass):
"name": "Huis",
CONF_TEMPERATURE_UNIT: "C",
"time_zone": "America/New_York",
"external_url": "https://www.example.com",
"internal_url": "http://example.local",
},
)
@ -462,6 +478,8 @@ async def test_loading_configuration_temperature_unit(hass):
assert hass.config.location_name == "Huis"
assert hass.config.units.name == CONF_UNIT_SYSTEM_METRIC
assert hass.config.time_zone.zone == "America/New_York"
assert hass.config.external_url == "https://www.example.com"
assert hass.config.internal_url == "http://example.local"
assert hass.config.config_source == config_util.SOURCE_YAML
@ -476,6 +494,8 @@ async def test_loading_configuration_from_packages(hass):
"name": "Huis",
CONF_TEMPERATURE_UNIT: "C",
"time_zone": "Europe/Madrid",
"external_url": "https://www.example.com",
"internal_url": "http://example.local",
"packages": {
"package_1": {"wake_on_lan": None},
"package_2": {

View File

@ -21,6 +21,7 @@ from homeassistant.const import (
EVENT_CORE_CONFIG_UPDATE,
EVENT_HOMEASSISTANT_CLOSE,
EVENT_HOMEASSISTANT_FINAL_WRITE,
EVENT_HOMEASSISTANT_START,
EVENT_HOMEASSISTANT_STOP,
EVENT_SERVICE_REGISTERED,
EVENT_SERVICE_REMOVED,
@ -34,7 +35,7 @@ from homeassistant.exceptions import InvalidEntityFormatError, InvalidStateError
import homeassistant.util.dt as dt_util
from homeassistant.util.unit_system import METRIC_SYSTEM
from tests.async_mock import MagicMock, patch
from tests.async_mock import MagicMock, Mock, patch
from tests.common import async_mock_service, get_test_home_assistant
PST = pytz.timezone("America/Los_Angeles")
@ -913,6 +914,8 @@ class TestConfig(unittest.TestCase):
"version": __version__,
"config_source": "default",
"safe_mode": False,
"external_url": None,
"internal_url": None,
}
assert expected == self.config.as_dict()
@ -948,7 +951,7 @@ class TestConfig(unittest.TestCase):
self.config.is_allowed_path(None)
async def test_event_on_update(hass, hass_storage):
async def test_event_on_update(hass):
"""Test that event is fired on update."""
events = []
@ -1281,3 +1284,38 @@ def test_valid_entity_id():
"light.something_yoo",
]:
assert ha.valid_entity_id(valid), valid
async def test_migration_base_url(hass, hass_storage):
"""Test that we migrate base url to internal/external url."""
config = ha.Config(hass)
stored = {"version": 1, "data": {}}
hass_storage[ha.CORE_STORAGE_KEY] = stored
with patch.object(hass.bus, "async_listen_once") as mock_listen:
# Empty config
await config.async_load()
assert len(mock_listen.mock_calls) == 1
# With just a name
stored["data"] = {"location_name": "Test Name"}
await config.async_load()
assert len(mock_listen.mock_calls) == 2
# With external url
stored["data"]["external_url"] = "https://example.com"
await config.async_load()
assert len(mock_listen.mock_calls) == 2
# Test that the event listener works
assert mock_listen.mock_calls[0][1][0] == EVENT_HOMEASSISTANT_START
# External
hass.config.api = Mock(base_url="https://loaded-example.com")
await mock_listen.mock_calls[0][1][1](None)
assert config.external_url == "https://loaded-example.com"
# Internal
for internal in ("http://hass.local", "http://192.168.1.100:8123"):
hass.config.api = Mock(base_url=internal)
await mock_listen.mock_calls[0][1][1](None)
assert config.internal_url == internal

View File

@ -38,3 +38,34 @@ def test_is_local():
assert network_util.is_local(ip_address("192.168.0.1"))
assert network_util.is_local(ip_address("127.0.0.1"))
assert not network_util.is_local(ip_address("208.5.4.2"))
def test_is_ip_address():
"""Test if strings are IP addresses."""
assert network_util.is_ip_address("192.168.0.1")
assert network_util.is_ip_address("8.8.8.8")
assert network_util.is_ip_address("::ffff:127.0.0.0")
assert not network_util.is_ip_address("192.168.0.999")
assert not network_util.is_ip_address("192.168.0.0/24")
assert not network_util.is_ip_address("example.com")
def test_normalize_url():
"""Test the normalizing of URLs."""
assert network_util.normalize_url("http://example.com") == "http://example.com"
assert network_util.normalize_url("https://example.com") == "https://example.com"
assert network_util.normalize_url("https://example.com/") == "https://example.com"
assert (
network_util.normalize_url("https://example.com:443") == "https://example.com"
)
assert network_util.normalize_url("http://example.com:80") == "http://example.com"
assert (
network_util.normalize_url("https://example.com:80") == "https://example.com:80"
)
assert (
network_util.normalize_url("http://example.com:443") == "http://example.com:443"
)
assert (
network_util.normalize_url("https://example.com:443/test/")
== "https://example.com/test"
)