Remove base_url fallback (#46316)

This commit is contained in:
Franck Nijhof 2021-02-10 14:31:11 +01:00 committed by GitHub
parent a6358430b4
commit 22389043eb
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 38 additions and 706 deletions

View file

@ -4,7 +4,6 @@ from ipaddress import ip_network
import logging import logging
import os import os
import ssl import ssl
from traceback import extract_stack
from typing import Dict, Optional, cast from typing import Dict, Optional, cast
from aiohttp import web from aiohttp import web
@ -125,69 +124,6 @@ class ApiConfig:
self.port = port self.port = port
self.use_ssl = use_ssl self.use_ssl = use_ssl
host = host.rstrip("/")
if host.startswith(("http://", "https://")):
self.deprecated_base_url = host
elif use_ssl:
self.deprecated_base_url = f"https://{host}"
else:
self.deprecated_base_url = f"http://{host}"
if port is not None:
self.deprecated_base_url += f":{port}"
@property
def base_url(self) -> str:
"""Proxy property to find caller of this deprecated property."""
found_frame = None
for frame in reversed(extract_stack()[:-1]):
for path in ("custom_components/", "homeassistant/components/"):
try:
index = frame.filename.index(path)
# Skip webhook from the stack
if frame.filename[index:].startswith(
"homeassistant/components/webhook/"
):
continue
found_frame = frame
break
except ValueError:
continue
if found_frame is not None:
break
# Did not source from an integration? Hard error.
if found_frame is None:
raise RuntimeError(
"Detected use of deprecated `base_url` property in the Home Assistant core. Please report this issue."
)
# If a frame was found, it originated from an integration
if found_frame:
start = index + len(path)
end = found_frame.filename.index("/", start)
integration = found_frame.filename[start:end]
if path == "custom_components/":
extra = " to the custom component author"
else:
extra = ""
_LOGGER.warning(
"Detected use of deprecated `base_url` property, use `homeassistant.helpers.network.get_url` method instead. Please report issue%s for %s using this method at %s, line %s: %s",
extra,
integration,
found_frame.filename[index:],
found_frame.lineno,
found_frame.line.strip(),
)
return self.deprecated_base_url
async def async_setup(hass, config): async def async_setup(hass, config):
"""Set up the HTTP API and debug interface.""" """Set up the HTTP API and debug interface."""
@ -256,20 +192,16 @@ async def async_setup(hass, config):
hass.http = server hass.http = server
host = conf.get(CONF_BASE_URL)
local_ip = await hass.async_add_executor_job(hass_util.get_local_ip) local_ip = await hass.async_add_executor_job(hass_util.get_local_ip)
if host: host = local_ip
port = None if server_host is not None:
elif server_host is not None:
# Assume the first server host name provided as API host # Assume the first server host name provided as API host
host = server_host[0] host = server_host[0]
port = server_port
else:
host = local_ip
port = server_port
hass.config.api = ApiConfig(local_ip, host, port, ssl_certificate is not None) hass.config.api = ApiConfig(
local_ip, host, server_port, ssl_certificate is not None
)
return True return True

View file

@ -8,7 +8,6 @@ import asyncio
import datetime import datetime
import enum import enum
import functools import functools
from ipaddress import ip_address
import logging import logging
import os import os
import pathlib import pathlib
@ -70,7 +69,7 @@ from homeassistant.exceptions import (
ServiceNotFound, ServiceNotFound,
Unauthorized, Unauthorized,
) )
from homeassistant.util import location, network from homeassistant.util import location
from homeassistant.util.async_ import ( from homeassistant.util.async_ import (
fire_coroutine_threadsafe, fire_coroutine_threadsafe,
run_callback_threadsafe, run_callback_threadsafe,
@ -1687,39 +1686,7 @@ class Config:
) )
data = await store.async_load() data = await store.async_load()
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.deprecated_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))
)
if data: if data:
# Try to migrate base_url to internal_url/external_url
if "external_url" not in data:
self.hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_START, migrate_base_url
)
self._update( self._update(
source=SOURCE_STORAGE, source=SOURCE_STORAGE,
latitude=data.get("latitude"), latitude=data.get("latitude"),

View file

@ -8,13 +8,7 @@ from homeassistant.components import http
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError
from homeassistant.loader import bind_hass from homeassistant.loader import bind_hass
from homeassistant.util.network import ( from homeassistant.util.network import is_ip_address, is_loopback, normalize_url
is_ip_address,
is_local,
is_loopback,
is_private,
normalize_url,
)
TYPE_URL_INTERNAL = "internal_url" TYPE_URL_INTERNAL = "internal_url"
TYPE_URL_EXTERNAL = "external_url" TYPE_URL_EXTERNAL = "external_url"
@ -151,19 +145,6 @@ def _get_internal_url(
): ):
return normalize_url(str(internal_url)) return normalize_url(str(internal_url))
# Fallback to old base_url
try:
return _get_deprecated_base_url(
hass,
internal=True,
allow_ip=allow_ip,
require_current_request=require_current_request,
require_ssl=require_ssl,
require_standard_port=require_standard_port,
)
except NoURLAvailableError:
pass
# Fallback to detected local IP # Fallback to detected local IP
if allow_ip and not ( if allow_ip and not (
require_ssl or hass.config.api is None or hass.config.api.use_ssl require_ssl or hass.config.api is None or hass.config.api.use_ssl
@ -217,17 +198,6 @@ def _get_external_url(
): ):
return normalize_url(str(external_url)) return normalize_url(str(external_url))
try:
return _get_deprecated_base_url(
hass,
allow_ip=allow_ip,
require_current_request=require_current_request,
require_ssl=require_ssl,
require_standard_port=require_standard_port,
)
except NoURLAvailableError:
pass
if allow_cloud: if allow_cloud:
try: try:
return _get_cloud_url(hass, require_current_request=require_current_request) return _get_cloud_url(hass, require_current_request=require_current_request)
@ -250,50 +220,3 @@ def _get_cloud_url(hass: HomeAssistant, require_current_request: bool = False) -
return normalize_url(str(cloud_url)) return normalize_url(str(cloud_url))
raise NoURLAvailableError raise NoURLAvailableError
@bind_hass
def _get_deprecated_base_url(
hass: HomeAssistant,
*,
internal: bool = False,
allow_ip: bool = True,
require_current_request: bool = False,
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.deprecated_base_url:
raise NoURLAvailableError
base_url = yarl.URL(hass.config.api.deprecated_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_current_request or base_url.host == _get_request_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))
# 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))
raise NoURLAvailableError

View file

@ -5,6 +5,7 @@ import ambiclimate
from homeassistant import data_entry_flow from homeassistant import data_entry_flow
from homeassistant.components.ambiclimate import config_flow from homeassistant.components.ambiclimate import config_flow
from homeassistant.config import async_process_ha_core_config
from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from homeassistant.util import aiohttp from homeassistant.util import aiohttp
@ -12,9 +13,11 @@ from homeassistant.util import aiohttp
async def init_config_flow(hass): async def init_config_flow(hass):
"""Init a configuration flow.""" """Init a configuration flow."""
await async_setup_component( await async_process_ha_core_config(
hass, "http", {"http": {"base_url": "https://hass.com"}} hass,
{"external_url": "https://example.com"},
) )
await async_setup_component(hass, "http", {})
config_flow.register_flow_implementation(hass, "id", "secret") config_flow.register_flow_implementation(hass, "id", "secret")
flow = config_flow.AmbiclimateFlowHandler() flow = config_flow.AmbiclimateFlowHandler()
@ -58,20 +61,20 @@ async def test_full_flow_implementation(hass):
assert result["step_id"] == "auth" assert result["step_id"] == "auth"
assert ( assert (
result["description_placeholders"]["cb_url"] result["description_placeholders"]["cb_url"]
== "https://hass.com/api/ambiclimate" == "https://example.com/api/ambiclimate"
) )
url = result["description_placeholders"]["authorization_url"] url = result["description_placeholders"]["authorization_url"]
assert "https://api.ambiclimate.com/oauth2/authorize" in url assert "https://api.ambiclimate.com/oauth2/authorize" in url
assert "client_id=id" in url assert "client_id=id" in url
assert "response_type=code" in url assert "response_type=code" in url
assert "redirect_uri=https%3A%2F%2Fhass.com%2Fapi%2Fambiclimate" in url assert "redirect_uri=https%3A%2F%2Fexample.com%2Fapi%2Fambiclimate" in url
with patch("ambiclimate.AmbiclimateOAuth.get_access_token", return_value="test"): with patch("ambiclimate.AmbiclimateOAuth.get_access_token", return_value="test"):
result = await flow.async_step_code("123ABC") result = await flow.async_step_code("123ABC")
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "Ambiclimate" assert result["title"] == "Ambiclimate"
assert result["data"]["callback_url"] == "https://hass.com/api/ambiclimate" assert result["data"]["callback_url"] == "https://example.com/api/ambiclimate"
assert result["data"][CONF_CLIENT_SECRET] == "secret" assert result["data"][CONF_CLIENT_SECRET] == "secret"
assert result["data"][CONF_CLIENT_ID] == "id" assert result["data"][CONF_CLIENT_ID] == "id"

View file

@ -60,73 +60,6 @@ async def test_registering_view_while_running(
hass.http.register_view(TestView) hass.http.register_view(TestView)
def test_api_base_url_with_domain(mock_stack):
"""Test setting API URL with domain."""
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(mock_stack):
"""Test setting API URL with IP."""
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(mock_stack):
"""Test setting API URL with IP and port."""
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(mock_stack):
"""Test setting API URL with protocol."""
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(mock_stack):
"""Test setting API URL with protocol and port."""
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(mock_stack):
"""Test setting API URL with use_ssl enabled."""
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(mock_stack):
"""Test setting API URL with use_ssl enabled and port."""
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(mock_stack):
"""Test setting API URL with specific protocol and use_ssl enabled."""
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(mock_stack):
"""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.base_url == "http://example.com:8123"
def test_api_local_ip(mock_stack):
"""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_no_base_url(hass, mock_stack):
"""Test setting api url."""
result = await async_setup_component(hass, "http", {"http": {}})
assert result
assert hass.config.api.base_url == "http://127.0.0.1:8123"
async def test_not_log_password(hass, aiohttp_client, caplog, legacy_auth): async def test_not_log_password(hass, aiohttp_client, caplog, legacy_auth):
"""Test access with password doesn't get logged.""" """Test access with password doesn't get logged."""
assert await async_setup_component(hass, "api", {"http": {}}) assert await async_setup_component(hass, "api", {"http": {}})
@ -263,127 +196,3 @@ async def test_storing_config(hass, aiohttp_client, aiohttp_unused_port):
restored["trusted_proxies"][0] = ip_network(restored["trusted_proxies"][0]) restored["trusted_proxies"][0] = ip_network(restored["trusted_proxies"][0])
assert restored == http.HTTP_SCHEMA(config) assert restored == http.HTTP_SCHEMA(config)
async def test_use_of_base_url(hass):
"""Test detection base_url usage when called without integration context."""
await async_setup_component(hass, "http", {"http": {}})
with patch(
"homeassistant.components.http.extract_stack",
return_value=[
Mock(
filename="/home/frenck/homeassistant/core.py",
lineno="21",
line="do_something()",
),
Mock(
filename="/home/frenck/homeassistant/core.py",
lineno="42",
line="url = hass.config.api.base_url",
),
Mock(
filename="/home/frenck/example/client.py",
lineno="21",
line="something()",
),
],
), pytest.raises(RuntimeError):
hass.config.api.base_url
async def test_use_of_base_url_integration(hass, caplog):
"""Test detection base_url usage when called with integration context."""
await async_setup_component(hass, "http", {"http": {}})
with patch(
"homeassistant.components.http.extract_stack",
return_value=[
Mock(
filename="/home/frenck/homeassistant/core.py",
lineno="21",
line="do_something()",
),
Mock(
filename="/home/frenck/homeassistant/components/example/__init__.py",
lineno="42",
line="url = hass.config.api.base_url",
),
Mock(
filename="/home/frenck/example/client.py",
lineno="21",
line="something()",
),
],
):
assert hass.config.api.base_url == "http://127.0.0.1:8123"
assert (
"Detected use of deprecated `base_url` property, use `homeassistant.helpers.network.get_url` method instead. Please report issue for example using this method at homeassistant/components/example/__init__.py, line 42: url = hass.config.api.base_url"
in caplog.text
)
async def test_use_of_base_url_integration_webhook(hass, caplog):
"""Test detection base_url usage when called with integration context."""
await async_setup_component(hass, "http", {"http": {}})
with patch(
"homeassistant.components.http.extract_stack",
return_value=[
Mock(
filename="/home/frenck/homeassistant/core.py",
lineno="21",
line="do_something()",
),
Mock(
filename="/home/frenck/homeassistant/components/example/__init__.py",
lineno="42",
line="url = hass.config.api.base_url",
),
Mock(
filename="/home/frenck/homeassistant/components/webhook/__init__.py",
lineno="42",
line="return get_url(hass)",
),
Mock(
filename="/home/frenck/example/client.py",
lineno="21",
line="something()",
),
],
):
assert hass.config.api.base_url == "http://127.0.0.1:8123"
assert (
"Detected use of deprecated `base_url` property, use `homeassistant.helpers.network.get_url` method instead. Please report issue for example using this method at homeassistant/components/example/__init__.py, line 42: url = hass.config.api.base_url"
in caplog.text
)
async def test_use_of_base_url_custom_component(hass, caplog):
"""Test detection base_url usage when called with custom component context."""
await async_setup_component(hass, "http", {"http": {}})
with patch(
"homeassistant.components.http.extract_stack",
return_value=[
Mock(
filename="/home/frenck/homeassistant/core.py",
lineno="21",
line="do_something()",
),
Mock(
filename="/home/frenck/.homeassistant/custom_components/example/__init__.py",
lineno="42",
line="url = hass.config.api.base_url",
),
Mock(
filename="/home/frenck/example/client.py",
lineno="21",
line="something()",
),
],
):
assert hass.config.api.base_url == "http://127.0.0.1:8123"
assert (
"Detected use of deprecated `base_url` property, use `homeassistant.helpers.network.get_url` method instead. Please report issue to the custom component author for example using this method at custom_components/example/__init__.py, line 42: url = hass.config.api.base_url"
in caplog.text
)

View file

@ -1,17 +1,20 @@
"""Test the init file of IFTTT.""" """Test the init file of IFTTT."""
from unittest.mock import patch
from homeassistant import data_entry_flow from homeassistant import data_entry_flow
from homeassistant.components import ifttt from homeassistant.components import ifttt
from homeassistant.config import async_process_ha_core_config
from homeassistant.core import callback from homeassistant.core import callback
async def test_config_flow_registers_webhook(hass, aiohttp_client): async def test_config_flow_registers_webhook(hass, aiohttp_client):
"""Test setting up IFTTT and sending webhook.""" """Test setting up IFTTT and sending webhook."""
with patch("homeassistant.util.get_local_ip", return_value="example.com"): await async_process_ha_core_config(
result = await hass.config_entries.flow.async_init( hass,
"ifttt", context={"source": "user"} {"internal_url": "http://example.local:8123"},
) )
result = await hass.config_entries.flow.async_init(
"ifttt", context={"source": "user"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM, result assert result["type"] == data_entry_flow.RESULT_TYPE_FORM, result
result = await hass.config_entries.flow.async_configure(result["flow_id"], {}) result = await hass.config_entries.flow.async_configure(result["flow_id"], {})

View file

@ -1,17 +1,19 @@
"""Test the init file of Twilio.""" """Test the init file of Twilio."""
from unittest.mock import patch
from homeassistant import data_entry_flow from homeassistant import data_entry_flow
from homeassistant.components import twilio from homeassistant.components import twilio
from homeassistant.config import async_process_ha_core_config
from homeassistant.core import callback from homeassistant.core import callback
async def test_config_flow_registers_webhook(hass, aiohttp_client): async def test_config_flow_registers_webhook(hass, aiohttp_client):
"""Test setting up Twilio and sending webhook.""" """Test setting up Twilio and sending webhook."""
with patch("homeassistant.util.get_local_ip", return_value="example.com"): await async_process_ha_core_config(
result = await hass.config_entries.flow.async_init( hass,
"twilio", context={"source": "user"} {"internal_url": "http://example.local:8123"},
) )
result = await hass.config_entries.flow.async_init(
"twilio", context={"source": "user"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM, result assert result["type"] == data_entry_flow.RESULT_TYPE_FORM, result
result = await hass.config_entries.flow.async_configure(result["flow_id"], {}) result = await hass.config_entries.flow.async_configure(result["flow_id"], {})

View file

@ -9,7 +9,6 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.network import ( from homeassistant.helpers.network import (
NoURLAvailableError, NoURLAvailableError,
_get_cloud_url, _get_cloud_url,
_get_deprecated_base_url,
_get_external_url, _get_external_url,
_get_internal_url, _get_internal_url,
_get_request_host, _get_request_host,
@ -166,9 +165,7 @@ async def test_get_url_internal_fallback(hass: HomeAssistant):
"""Test getting an instance URL when the user has not set an internal URL.""" """Test getting an instance URL when the user has not set an internal URL."""
assert hass.config.internal_url is None assert hass.config.internal_url is None
hass.config.api = Mock( hass.config.api = Mock(use_ssl=False, port=8123, local_ip="192.168.123.123")
use_ssl=False, port=8123, deprecated_base_url=None, local_ip="192.168.123.123"
)
assert _get_internal_url(hass) == "http://192.168.123.123:8123" assert _get_internal_url(hass) == "http://192.168.123.123:8123"
with pytest.raises(NoURLAvailableError): with pytest.raises(NoURLAvailableError):
@ -180,9 +177,7 @@ async def test_get_url_internal_fallback(hass: HomeAssistant):
with pytest.raises(NoURLAvailableError): with pytest.raises(NoURLAvailableError):
_get_internal_url(hass, require_ssl=True) _get_internal_url(hass, require_ssl=True)
hass.config.api = Mock( hass.config.api = Mock(use_ssl=False, port=80, local_ip="192.168.123.123")
use_ssl=False, port=80, deprecated_base_url=None, local_ip="192.168.123.123"
)
assert _get_internal_url(hass) == "http://192.168.123.123" assert _get_internal_url(hass) == "http://192.168.123.123"
assert ( assert (
_get_internal_url(hass, require_standard_port=True) == "http://192.168.123.123" _get_internal_url(hass, require_standard_port=True) == "http://192.168.123.123"
@ -194,7 +189,7 @@ async def test_get_url_internal_fallback(hass: HomeAssistant):
with pytest.raises(NoURLAvailableError): with pytest.raises(NoURLAvailableError):
_get_internal_url(hass, require_ssl=True) _get_internal_url(hass, require_ssl=True)
hass.config.api = Mock(use_ssl=True, port=443, deprecated_base_url=None) hass.config.api = Mock(use_ssl=True, port=443)
with pytest.raises(NoURLAvailableError): with pytest.raises(NoURLAvailableError):
_get_internal_url(hass) _get_internal_url(hass)
@ -208,9 +203,7 @@ async def test_get_url_internal_fallback(hass: HomeAssistant):
_get_internal_url(hass, require_ssl=True) _get_internal_url(hass, require_ssl=True)
# Do no accept any local loopback address as fallback # Do no accept any local loopback address as fallback
hass.config.api = Mock( hass.config.api = Mock(use_ssl=False, port=80, local_ip="127.0.0.1")
use_ssl=False, port=80, deprecated_base_url=None, local_ip="127.0.0.1"
)
with pytest.raises(NoURLAvailableError): with pytest.raises(NoURLAvailableError):
_get_internal_url(hass) _get_internal_url(hass)
@ -457,9 +450,7 @@ async def test_get_url(hass: HomeAssistant):
with pytest.raises(NoURLAvailableError): with pytest.raises(NoURLAvailableError):
get_url(hass) get_url(hass)
hass.config.api = Mock( hass.config.api = Mock(use_ssl=False, port=8123, local_ip="192.168.123.123")
use_ssl=False, port=8123, deprecated_base_url=None, local_ip="192.168.123.123"
)
assert get_url(hass) == "http://192.168.123.123:8123" assert get_url(hass) == "http://192.168.123.123:8123"
assert get_url(hass, prefer_external=True) == "http://192.168.123.123:8123" assert get_url(hass, prefer_external=True) == "http://192.168.123.123:8123"
@ -543,274 +534,11 @@ async def test_get_request_host(hass: HomeAssistant):
assert _get_request_host() == "example.com" assert _get_request_host() == "example.com"
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(deprecated_base_url="https://example.local")
assert _get_deprecated_base_url(hass, internal=True) == "https://example.local"
assert (
_get_deprecated_base_url(hass, internal=True, allow_ip=False)
== "https://example.local"
)
assert (
_get_deprecated_base_url(hass, internal=True, require_ssl=True)
== "https://example.local"
)
assert (
_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(deprecated_base_url="http://10.10.10.10:8123")
assert _get_deprecated_base_url(hass, internal=True) == "http://10.10.10.10:8123"
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, internal=True, allow_ip=False)
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, internal=True, require_ssl=True)
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, internal=True, require_standard_port=True)
# Test with SSL, local IP URL
hass.config.api = Mock(deprecated_base_url="https://10.10.10.10")
assert _get_deprecated_base_url(hass, internal=True) == "https://10.10.10.10"
assert (
_get_deprecated_base_url(hass, internal=True, require_ssl=True)
== "https://10.10.10.10"
)
assert (
_get_deprecated_base_url(hass, internal=True, require_standard_port=True)
== "https://10.10.10.10"
)
# Test external URL
hass.config.api = Mock(deprecated_base_url="https://example.com")
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, internal=True)
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, internal=True, require_ssl=True)
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, internal=True, require_standard_port=True)
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, internal=True, allow_ip=False)
# Test with loopback
hass.config.api = Mock(deprecated_base_url="https://127.0.0.42")
with pytest.raises(NoURLAvailableError):
assert _get_deprecated_base_url(hass, internal=True)
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, internal=True, allow_ip=False)
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, internal=True, require_ssl=True)
with pytest.raises(NoURLAvailableError):
_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(deprecated_base_url="https://example.com:443/")
assert _get_deprecated_base_url(hass) == "https://example.com"
assert _get_deprecated_base_url(hass, require_ssl=True) == "https://example.com"
assert (
_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(deprecated_base_url="http://example.com:8123/")
assert _get_deprecated_base_url(hass) == "http://example.com:8123"
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, require_ssl=True)
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, require_standard_port=True)
# Test SSL on external IP
hass.config.api = Mock(deprecated_base_url="https://1.1.1.1")
assert _get_deprecated_base_url(hass) == "https://1.1.1.1"
assert _get_deprecated_base_url(hass, require_ssl=True) == "https://1.1.1.1"
assert (
_get_deprecated_base_url(hass, require_standard_port=True) == "https://1.1.1.1"
)
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, allow_ip=False)
# Test with private IP
hass.config.api = Mock(deprecated_base_url="https://10.10.10.10")
with pytest.raises(NoURLAvailableError):
assert _get_deprecated_base_url(hass)
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, allow_ip=False)
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, require_ssl=True)
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, require_standard_port=True)
# Test with local domain
hass.config.api = Mock(deprecated_base_url="https://example.local")
with pytest.raises(NoURLAvailableError):
assert _get_deprecated_base_url(hass)
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, allow_ip=False)
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, require_ssl=True)
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, require_standard_port=True)
# Test with loopback
hass.config.api = Mock(deprecated_base_url="https://127.0.0.42")
with pytest.raises(NoURLAvailableError):
assert _get_deprecated_base_url(hass)
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, allow_ip=False)
with pytest.raises(NoURLAvailableError):
_get_deprecated_base_url(hass, require_ssl=True)
with pytest.raises(NoURLAvailableError):
_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, deprecated_base_url=None, local_ip="192.168.123.123"
)
assert hass.config.internal_url is None
assert _get_internal_url(hass) == "http://192.168.123.123:8123"
with pytest.raises(NoURLAvailableError):
_get_internal_url(hass, allow_ip=False)
with pytest.raises(NoURLAvailableError):
_get_internal_url(hass, require_standard_port=True)
with pytest.raises(NoURLAvailableError):
_get_internal_url(hass, require_ssl=True)
# Add base_url
hass.config.api = Mock(
use_ssl=False, port=8123, deprecated_base_url="https://example.local"
)
assert _get_internal_url(hass) == "https://example.local"
assert _get_internal_url(hass, allow_ip=False) == "https://example.local"
assert (
_get_internal_url(hass, require_standard_port=True) == "https://example.local"
)
assert _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 _get_internal_url(hass) == "https://internal.local"
assert _get_internal_url(hass, allow_ip=False) == "https://internal.local"
assert (
_get_internal_url(hass, require_standard_port=True) == "https://internal.local"
)
assert _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 _get_internal_url(hass) == "http://internal.local:8123"
assert _get_internal_url(hass, allow_ip=False) == "http://internal.local:8123"
assert (
_get_internal_url(hass, require_standard_port=True) == "https://example.local"
)
assert _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 _get_internal_url(hass) == "http://10.10.10.10:8123"
assert _get_internal_url(hass, allow_ip=False) == "https://example.local"
assert (
_get_internal_url(hass, require_standard_port=True) == "https://example.local"
)
assert _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, deprecated_base_url=None)
assert hass.config.internal_url is None
with pytest.raises(NoURLAvailableError):
_get_external_url(hass)
# Test with SSL and external domain on standard port
hass.config.api = Mock(deprecated_base_url="https://example.com:443/")
assert _get_external_url(hass) == "https://example.com"
assert _get_external_url(hass, allow_ip=False) == "https://example.com"
assert _get_external_url(hass, require_ssl=True) == "https://example.com"
assert _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 _get_external_url(hass) == "https://external.example.com"
assert _get_external_url(hass, allow_ip=False) == "https://external.example.com"
assert (
_get_external_url(hass, require_standard_port=True)
== "https://external.example.com"
)
assert _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 _get_external_url(hass) == "http://external.example.com:8123"
assert _get_external_url(hass, allow_ip=False) == "http://external.example.com:8123"
assert _get_external_url(hass, require_standard_port=True) == "https://example.com"
assert _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 _get_external_url(hass) == "http://1.1.1.1:8123"
assert _get_external_url(hass, allow_ip=False) == "https://example.com"
assert _get_external_url(hass, require_standard_port=True) == "https://example.com"
assert _get_external_url(hass, require_ssl=True) == "https://example.com"
async def test_get_current_request_url_with_known_host( async def test_get_current_request_url_with_known_host(
hass: HomeAssistant, current_request hass: HomeAssistant, current_request
): ):
"""Test getting current request URL with known hosts addresses.""" """Test getting current request URL with known hosts addresses."""
hass.config.api = Mock( hass.config.api = Mock(use_ssl=False, port=8123, local_ip="127.0.0.1")
use_ssl=False, port=8123, local_ip="127.0.0.1", deprecated_base_url=None
)
assert hass.config.internal_url is None assert hass.config.internal_url is None
with pytest.raises(NoURLAvailableError): with pytest.raises(NoURLAvailableError):

View file

@ -1294,41 +1294,6 @@ def test_valid_entity_id():
assert ha.valid_entity_id(valid), valid 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) == 0
# With just a name
stored["data"] = {"location_name": "Test Name"}
await config.async_load()
assert len(mock_listen.mock_calls) == 1
# With external url
stored["data"]["external_url"] = "https://example.com"
await config.async_load()
assert len(mock_listen.mock_calls) == 1
# Test that the event listener works
assert mock_listen.mock_calls[0][1][0] == EVENT_HOMEASSISTANT_START
# External
hass.config.api = Mock(deprecated_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(deprecated_base_url=internal)
await mock_listen.mock_calls[0][1][1](None)
assert config.internal_url == internal
async def test_additional_data_in_core_config(hass, hass_storage): async def test_additional_data_in_core_config(hass, hass_storage):
"""Test that we can handle additional data in core configuration.""" """Test that we can handle additional data in core configuration."""
config = ha.Config(hass) config = ha.Config(hass)