Add network configuration integration (#50874)

Co-authored-by: Ruslan Sayfutdinov <ruslan@sayfutdinov.com>
Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
J. Nick Koston 2021-05-26 11:06:30 -05:00 committed by GitHub
parent 16e90f12ca
commit 64661ee2b7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
19 changed files with 955 additions and 106 deletions

View file

@ -45,6 +45,7 @@ homeassistant.components.lock.*
homeassistant.components.mailbox.* homeassistant.components.mailbox.*
homeassistant.components.media_player.* homeassistant.components.media_player.*
homeassistant.components.nam.* homeassistant.components.nam.*
homeassistant.components.network.*
homeassistant.components.notify.* homeassistant.components.notify.*
homeassistant.components.number.* homeassistant.components.number.*
homeassistant.components.onewire.* homeassistant.components.onewire.*

View file

@ -19,6 +19,7 @@
"media_source", "media_source",
"mobile_app", "mobile_app",
"my", "my",
"network",
"person", "person",
"scene", "scene",
"script", "script",

View file

@ -0,0 +1,91 @@
"""The Network Configuration integration."""
from __future__ import annotations
import logging
import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.components.websocket_api.connection import ActiveConnection
from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType
from homeassistant.loader import bind_hass
from .const import (
ATTR_ADAPTERS,
ATTR_CONFIGURED_ADAPTERS,
DOMAIN,
NETWORK_CONFIG_SCHEMA,
)
from .models import Adapter
from .network import Network
ZEROCONF_DOMAIN = "zeroconf" # cannot import from zeroconf due to circular dep
_LOGGER = logging.getLogger(__name__)
@bind_hass
async def async_get_adapters(hass: HomeAssistant) -> list[Adapter]:
"""Get the network adapter configuration."""
network: Network = hass.data[DOMAIN]
return network.adapters
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up network for Home Assistant."""
hass.data[DOMAIN] = network = Network(hass)
await network.async_setup()
if ZEROCONF_DOMAIN in config:
await network.async_migrate_from_zeroconf(config[ZEROCONF_DOMAIN])
network.async_configure()
_LOGGER.debug("Adapters: %s", network.adapters)
websocket_api.async_register_command(hass, websocket_network_adapters)
websocket_api.async_register_command(hass, websocket_network_adapters_configure)
return True
@websocket_api.require_admin
@websocket_api.websocket_command({vol.Required("type"): "network"})
@websocket_api.async_response
async def websocket_network_adapters(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
) -> None:
"""Return network preferences."""
network: Network = hass.data[DOMAIN]
connection.send_result(
msg["id"],
{
ATTR_ADAPTERS: network.adapters,
ATTR_CONFIGURED_ADAPTERS: network.configured_adapters,
},
)
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required("type"): "network/configure",
vol.Required("config", default={}): NETWORK_CONFIG_SCHEMA,
}
)
@websocket_api.async_response
async def websocket_network_adapters_configure(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
) -> None:
"""Update network config."""
network: Network = hass.data[DOMAIN]
await network.async_reconfig(msg["config"])
connection.send_result(
msg["id"],
{ATTR_CONFIGURED_ADAPTERS: network.configured_adapters},
)

View file

@ -0,0 +1,27 @@
"""Constants for the network integration."""
from __future__ import annotations
from typing import Final
import voluptuous as vol
import homeassistant.helpers.config_validation as cv
DOMAIN: Final = "network"
STORAGE_KEY: Final = "core.network"
STORAGE_VERSION: Final = 1
ATTR_ADAPTERS: Final = "adapters"
ATTR_CONFIGURED_ADAPTERS: Final = "configured_adapters"
DEFAULT_CONFIGURED_ADAPTERS: list[str] = []
MDNS_TARGET_IP: Final = "224.0.0.251"
NETWORK_CONFIG_SCHEMA = vol.Schema(
{
vol.Optional(
ATTR_CONFIGURED_ADAPTERS, default=DEFAULT_CONFIGURED_ADAPTERS
): vol.Schema(vol.All(cv.ensure_list, [cv.string])),
}
)

View file

@ -0,0 +1,10 @@
{
"domain": "network",
"name": "Network Configuration",
"documentation": "https://www.home-assistant.io/integrations/network",
"requirements": ["ifaddr==0.1.7"],
"codeowners": [],
"dependencies": ["websocket_api"],
"quality_scale": "internal",
"iot_class": "local_push"
}

View file

@ -0,0 +1,31 @@
"""Models helper class for the network integration."""
from __future__ import annotations
from typing import TypedDict
class IPv6ConfiguredAddress(TypedDict):
"""Represent an IPv6 address."""
address: str
flowinfo: int
scope_id: int
network_prefix: int
class IPv4ConfiguredAddress(TypedDict):
"""Represent an IPv4 address."""
address: str
network_prefix: int
class Adapter(TypedDict):
"""Configured network adapters."""
name: str
enabled: bool
auto: bool
default: bool
ipv6: list[IPv6ConfiguredAddress]
ipv4: list[IPv4ConfiguredAddress]

View file

@ -0,0 +1,78 @@
"""Network helper class for the network integration."""
from __future__ import annotations
from typing import Any, cast
from homeassistant.core import HomeAssistant, callback
from .const import (
ATTR_CONFIGURED_ADAPTERS,
DEFAULT_CONFIGURED_ADAPTERS,
NETWORK_CONFIG_SCHEMA,
STORAGE_KEY,
STORAGE_VERSION,
)
from .models import Adapter
from .util import (
adapters_with_exernal_addresses,
async_load_adapters,
enable_adapters,
enable_auto_detected_adapters,
)
class Network:
"""Network helper class for the network integration."""
def __init__(self, hass: HomeAssistant) -> None:
"""Initialize the Network class."""
self._store = hass.helpers.storage.Store(STORAGE_VERSION, STORAGE_KEY)
self._data: dict[str, Any] = {}
self.adapters: list[Adapter] = []
@property
def configured_adapters(self) -> list[str]:
"""Return the configured adapters."""
return self._data.get(ATTR_CONFIGURED_ADAPTERS, DEFAULT_CONFIGURED_ADAPTERS)
async def async_setup(self) -> None:
"""Set up the network config."""
await self.async_load()
self.adapters = await async_load_adapters()
async def async_migrate_from_zeroconf(self, zc_config: dict[str, Any]) -> None:
"""Migrate configuration from zeroconf."""
if self._data or not zc_config:
return
from homeassistant.components.zeroconf import ( # pylint: disable=import-outside-toplevel
CONF_DEFAULT_INTERFACE,
)
if zc_config.get(CONF_DEFAULT_INTERFACE) is False:
self._data[ATTR_CONFIGURED_ADAPTERS] = adapters_with_exernal_addresses(
self.adapters
)
await self._async_save()
@callback
def async_configure(self) -> None:
"""Configure from storage."""
if not enable_adapters(self.adapters, self.configured_adapters):
enable_auto_detected_adapters(self.adapters)
async def async_reconfig(self, config: dict[str, Any]) -> None:
"""Reconfigure network."""
config = NETWORK_CONFIG_SCHEMA(config)
self._data[ATTR_CONFIGURED_ADAPTERS] = config[ATTR_CONFIGURED_ADAPTERS]
self.async_configure()
await self._async_save()
async def async_load(self) -> None:
"""Load config."""
if stored := await self._store.async_load():
self._data = cast(dict, stored)
async def _async_save(self) -> None:
"""Save preferences."""
await self._store.async_save(self._data)

View file

@ -0,0 +1,158 @@
"""Network helper class for the network integration."""
from __future__ import annotations
from ipaddress import IPv4Address, IPv6Address, ip_address
import logging
import socket
from typing import cast
import ifaddr
from homeassistant.core import callback
from .const import MDNS_TARGET_IP
from .models import Adapter, IPv4ConfiguredAddress, IPv6ConfiguredAddress
_LOGGER = logging.getLogger(__name__)
async def async_load_adapters() -> list[Adapter]:
"""Load adapters."""
source_ip = async_get_source_ip(MDNS_TARGET_IP)
source_ip_address = ip_address(source_ip) if source_ip else None
ha_adapters: list[Adapter] = [
_ifaddr_adapter_to_ha(adapter, source_ip_address)
for adapter in ifaddr.get_adapters()
]
if not any(adapter["default"] and adapter["auto"] for adapter in ha_adapters):
for adapter in ha_adapters:
if _adapter_has_external_address(adapter):
adapter["auto"] = True
return ha_adapters
def enable_adapters(adapters: list[Adapter], enabled_interfaces: list[str]) -> bool:
"""Enable configured adapters."""
_reset_enabled_adapters(adapters)
if not enabled_interfaces:
return False
found_adapter = False
for adapter in adapters:
if adapter["name"] in enabled_interfaces:
adapter["enabled"] = True
found_adapter = True
return found_adapter
def enable_auto_detected_adapters(adapters: list[Adapter]) -> None:
"""Enable auto detected adapters."""
enable_adapters(
adapters, [adapter["name"] for adapter in adapters if adapter["auto"]]
)
def adapters_with_exernal_addresses(adapters: list[Adapter]) -> list[str]:
"""Enable all interfaces with an external address."""
return [
adapter["name"]
for adapter in adapters
if _adapter_has_external_address(adapter)
]
def _adapter_has_external_address(adapter: Adapter) -> bool:
"""Adapter has a non-loopback and non-link-local address."""
return any(
_has_external_address(v4_config["address"]) for v4_config in adapter["ipv4"]
) or any(
_has_external_address(v6_config["address"]) for v6_config in adapter["ipv6"]
)
def _has_external_address(ip_str: str) -> bool:
return _ip_address_is_external(ip_address(ip_str))
def _ip_address_is_external(ip_addr: IPv4Address | IPv6Address) -> bool:
return (
not ip_addr.is_multicast
and not ip_addr.is_loopback
and not ip_addr.is_link_local
)
def _reset_enabled_adapters(adapters: list[Adapter]) -> None:
for adapter in adapters:
adapter["enabled"] = False
def _ifaddr_adapter_to_ha(
adapter: ifaddr.Adapter, next_hop_address: None | IPv4Address | IPv6Address
) -> Adapter:
"""Convert an ifaddr adapter to ha."""
ip_v4s: list[IPv4ConfiguredAddress] = []
ip_v6s: list[IPv6ConfiguredAddress] = []
default = False
auto = False
for ip_config in adapter.ips:
if ip_config.is_IPv6:
ip_addr = ip_address(ip_config.ip[0])
ip_v6s.append(_ip_v6_from_adapter(ip_config))
else:
ip_addr = ip_address(ip_config.ip)
ip_v4s.append(_ip_v4_from_adapter(ip_config))
if ip_addr == next_hop_address:
default = True
if _ip_address_is_external(ip_addr):
auto = True
return {
"name": adapter.nice_name,
"enabled": False,
"auto": auto,
"default": default,
"ipv4": ip_v4s,
"ipv6": ip_v6s,
}
def _ip_v6_from_adapter(ip_config: ifaddr.IP) -> IPv6ConfiguredAddress:
return {
"address": ip_config.ip[0],
"flowinfo": ip_config.ip[1],
"scope_id": ip_config.ip[2],
"network_prefix": ip_config.network_prefix,
}
def _ip_v4_from_adapter(ip_config: ifaddr.IP) -> IPv4ConfiguredAddress:
return {
"address": ip_config.ip,
"network_prefix": ip_config.network_prefix,
}
@callback
def async_get_source_ip(target_ip: str) -> str | None:
"""Return the source ip that will reach target_ip."""
test_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
test_sock.setblocking(False) # must be non-blocking for async
try:
test_sock.connect((target_ip, 1))
return cast(str, test_sock.getsockname()[0])
except Exception: # pylint: disable=broad-except
_LOGGER.debug(
"The system could not auto detect the source ip for %s on your operating system",
target_ip,
)
return None
finally:
test_sock.close()

View file

@ -2,16 +2,14 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
from collections.abc import Coroutine, Iterable from collections.abc import Coroutine
from contextlib import suppress from contextlib import suppress
import fnmatch import fnmatch
import ipaddress import ipaddress
from ipaddress import ip_address
import logging import logging
import socket import socket
from typing import Any, TypedDict, cast from typing import Any, TypedDict, cast
from pyroute2 import IPRoute
import voluptuous as vol import voluptuous as vol
from zeroconf import ( from zeroconf import (
InterfaceChoice, InterfaceChoice,
@ -23,6 +21,8 @@ from zeroconf import (
) )
from homeassistant import config_entries, util from homeassistant import config_entries, util
from homeassistant.components import network
from homeassistant.components.network.models import Adapter
from homeassistant.const import ( from homeassistant.const import (
EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_START,
EVENT_HOMEASSISTANT_STARTED, EVENT_HOMEASSISTANT_STARTED,
@ -34,7 +34,6 @@ from homeassistant.data_entry_flow import FlowResult
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.network import NoURLAvailableError, get_url from homeassistant.helpers.network import NoURLAvailableError, get_url
from homeassistant.loader import async_get_homekit, async_get_zeroconf, bind_hass from homeassistant.loader import async_get_homekit, async_get_zeroconf, bind_hass
from homeassistant.util.network import is_loopback
from .models import HaAsyncZeroconf, HaServiceBrowser, HaZeroconf from .models import HaAsyncZeroconf, HaServiceBrowser, HaZeroconf
from .usage import install_multiple_zeroconf_catcher from .usage import install_multiple_zeroconf_catcher
@ -69,11 +68,14 @@ MAX_NAME_LEN = 63
CONFIG_SCHEMA = vol.Schema( CONFIG_SCHEMA = vol.Schema(
{ {
DOMAIN: vol.Schema( DOMAIN: vol.All(
{ cv.deprecated(CONF_DEFAULT_INTERFACE),
vol.Optional(CONF_DEFAULT_INTERFACE): cv.boolean, vol.Schema(
vol.Optional(CONF_IPV6, default=DEFAULT_IPV6): cv.boolean, {
} vol.Optional(CONF_DEFAULT_INTERFACE): cv.boolean,
vol.Optional(CONF_IPV6, default=DEFAULT_IPV6): cv.boolean,
}
),
) )
}, },
extra=vol.ALLOW_EXTRA, extra=vol.ALLOW_EXTRA,
@ -132,49 +134,11 @@ async def _async_get_instance(hass: HomeAssistant, **zcargs: Any) -> HaAsyncZero
return aio_zc return aio_zc
def _get_ip_route(dst_ip: str) -> Any: def _async_use_default_interface(adapters: list[Adapter]) -> bool:
"""Get ip next hop.""" for adapter in adapters:
return IPRoute().route("get", dst=dst_ip) if adapter["enabled"] and not adapter["default"]:
return False
return True
def _first_ip_nexthop_from_route(routes: Iterable) -> None | str:
"""Find the first RTA_PREFSRC in the routes."""
_LOGGER.debug("Routes: %s", routes)
for route in routes:
for key, value in route["attrs"]:
if key == "RTA_PREFSRC":
return cast(str, value)
return None
async def async_detect_interfaces_setting(hass: HomeAssistant) -> InterfaceChoice:
"""Auto detect the interfaces setting when unset."""
routes = []
try:
routes = await hass.async_add_executor_job(_get_ip_route, MDNS_TARGET_IP)
except Exception as ex: # pylint: disable=broad-except
_LOGGER.debug(
"The system could not auto detect routing data on your operating system; Zeroconf will broadcast on all interfaces",
exc_info=ex,
)
return InterfaceChoice.All
if not (first_ip := _first_ip_nexthop_from_route(routes)):
_LOGGER.debug(
"The system could not auto detect the nexthop for %s on your operating system; Zeroconf will broadcast on all interfaces",
MDNS_TARGET_IP,
)
return InterfaceChoice.All
if is_loopback(ip_address(first_ip)):
_LOGGER.debug(
"The next hop for %s is %s; Zeroconf will broadcast on all interfaces",
MDNS_TARGET_IP,
first_ip,
)
return InterfaceChoice.All
return InterfaceChoice.Default
async def async_setup(hass: HomeAssistant, config: dict) -> bool: async def async_setup(hass: HomeAssistant, config: dict) -> bool:
@ -182,10 +146,18 @@ async def async_setup(hass: HomeAssistant, config: dict) -> bool:
zc_config = config.get(DOMAIN, {}) zc_config = config.get(DOMAIN, {})
zc_args: dict = {} zc_args: dict = {}
if CONF_DEFAULT_INTERFACE not in zc_config: adapters = await network.async_get_adapters(hass)
zc_args["interfaces"] = await async_detect_interfaces_setting(hass) if _async_use_default_interface(adapters):
elif zc_config[CONF_DEFAULT_INTERFACE]:
zc_args["interfaces"] = InterfaceChoice.Default zc_args["interfaces"] = InterfaceChoice.Default
else:
interfaces = zc_args["interfaces"] = []
for adapter in adapters:
if not adapter["enabled"]:
continue
if ipv4s := adapter["ipv4"]:
interfaces.append(ipv4s[0]["address"])
elif ipv6s := adapter["ipv6"]:
interfaces.append(ipv6s[0]["scope_id"])
if not zc_config.get(CONF_IPV6, DEFAULT_IPV6): if not zc_config.get(CONF_IPV6, DEFAULT_IPV6):
zc_args["ip_version"] = IPVersion.V4Only zc_args["ip_version"] = IPVersion.V4Only

View file

@ -2,8 +2,8 @@
"domain": "zeroconf", "domain": "zeroconf",
"name": "Zero-configuration networking (zeroconf)", "name": "Zero-configuration networking (zeroconf)",
"documentation": "https://www.home-assistant.io/integrations/zeroconf", "documentation": "https://www.home-assistant.io/integrations/zeroconf",
"requirements": ["zeroconf==0.31.0","pyroute2==0.5.18"], "requirements": ["zeroconf==0.31.0"],
"dependencies": ["api"], "dependencies": ["network", "api"],
"codeowners": ["@bdraco"], "codeowners": ["@bdraco"],
"quality_scale": "internal", "quality_scale": "internal",
"iot_class": "local_push" "iot_class": "local_push"

View file

@ -19,12 +19,12 @@ emoji==1.2.0
hass-nabucasa==0.43.0 hass-nabucasa==0.43.0
home-assistant-frontend==20210518.0 home-assistant-frontend==20210518.0
httpx==0.18.0 httpx==0.18.0
ifaddr==0.1.7
jinja2>=3.0.1 jinja2>=3.0.1
netdisco==2.8.3 netdisco==2.8.3
paho-mqtt==1.5.1 paho-mqtt==1.5.1
pillow==8.1.2 pillow==8.1.2
pip>=8.0.3,<20.3 pip>=8.0.3,<20.3
pyroute2==0.5.18
python-slugify==4.0.1 python-slugify==4.0.1
pyyaml==5.4.1 pyyaml==5.4.1
requests==2.25.1 requests==2.25.1

View file

@ -506,6 +506,17 @@ no_implicit_optional = true
warn_return_any = true warn_return_any = true
warn_unreachable = true warn_unreachable = true
[mypy-homeassistant.components.network.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.notify.*] [mypy-homeassistant.components.notify.*]
check_untyped_defs = true check_untyped_defs = true
disallow_incomplete_defs = true disallow_incomplete_defs = true

View file

@ -818,6 +818,9 @@ ibmiotf==0.3.4
# homeassistant.components.ping # homeassistant.components.ping
icmplib==2.1.1 icmplib==2.1.1
# homeassistant.components.network
ifaddr==0.1.7
# homeassistant.components.iglo # homeassistant.components.iglo
iglo==1.2.7 iglo==1.2.7
@ -1690,9 +1693,6 @@ pyrisco==0.3.1
# homeassistant.components.rituals_perfume_genie # homeassistant.components.rituals_perfume_genie
pyrituals==0.0.3 pyrituals==0.0.3
# homeassistant.components.zeroconf
pyroute2==0.5.18
# homeassistant.components.ruckus_unleashed # homeassistant.components.ruckus_unleashed
pyruckus==0.12 pyruckus==0.12

View file

@ -462,6 +462,9 @@ iaqualink==0.3.4
# homeassistant.components.ping # homeassistant.components.ping
icmplib==2.1.1 icmplib==2.1.1
# homeassistant.components.network
ifaddr==0.1.7
# homeassistant.components.influxdb # homeassistant.components.influxdb
influxdb-client==1.14.0 influxdb-client==1.14.0
@ -947,9 +950,6 @@ pyrisco==0.3.1
# homeassistant.components.rituals_perfume_genie # homeassistant.components.rituals_perfume_genie
pyrituals==0.0.3 pyrituals==0.0.3
# homeassistant.components.zeroconf
pyroute2==0.5.18
# homeassistant.components.ruckus_unleashed # homeassistant.components.ruckus_unleashed
pyruckus==0.12 pyruckus==0.12

View file

@ -136,6 +136,8 @@ IGNORE_VIOLATIONS = {
("demo", "openalpr_local"), ("demo", "openalpr_local"),
# Migration wizard from zwave to ozw. # Migration wizard from zwave to ozw.
"ozw", "ozw",
# Migration of settings from zeroconf to network
("network", "zeroconf"),
# This should become a helper method that integrations can submit data to # This should become a helper method that integrations can submit data to
("websocket_api", "lovelace"), ("websocket_api", "lovelace"),
("websocket_api", "shopping_list"), ("websocket_api", "shopping_list"),

View file

@ -0,0 +1 @@
"""Tests for the Network Configuration integration."""

View file

@ -0,0 +1,446 @@
"""Test the Network Configuration."""
from unittest.mock import Mock, patch
import ifaddr
from homeassistant.components import network
from homeassistant.components.network.const import (
ATTR_ADAPTERS,
ATTR_CONFIGURED_ADAPTERS,
STORAGE_KEY,
STORAGE_VERSION,
)
from homeassistant.setup import async_setup_component
_NO_LOOPBACK_IPADDR = "192.168.1.5"
_LOOPBACK_IPADDR = "127.0.0.1"
def _generate_mock_adapters():
mock_lo0 = Mock(spec=ifaddr.Adapter)
mock_lo0.nice_name = "lo0"
mock_lo0.ips = [ifaddr.IP("127.0.0.1", 8, "lo0")]
mock_eth0 = Mock(spec=ifaddr.Adapter)
mock_eth0.nice_name = "eth0"
mock_eth0.ips = [ifaddr.IP(("2001:db8::", 1, 1), 8, "eth0")]
mock_eth1 = Mock(spec=ifaddr.Adapter)
mock_eth1.nice_name = "eth1"
mock_eth1.ips = [ifaddr.IP("192.168.1.5", 23, "eth1")]
mock_vtun0 = Mock(spec=ifaddr.Adapter)
mock_vtun0.nice_name = "vtun0"
mock_vtun0.ips = [ifaddr.IP("169.254.3.2", 16, "vtun0")]
return [mock_eth0, mock_lo0, mock_eth1, mock_vtun0]
async def test_async_detect_interfaces_setting_non_loopback_route(hass, hass_storage):
"""Test without default interface config and the route returns a non-loopback address."""
with patch(
"homeassistant.components.network.util.socket.socket.getsockname",
return_value=[_NO_LOOPBACK_IPADDR],
), patch(
"homeassistant.components.network.util.ifaddr.get_adapters",
return_value=_generate_mock_adapters(),
):
assert await async_setup_component(hass, network.DOMAIN, {network.DOMAIN: {}})
await hass.async_block_till_done()
network_obj = hass.data[network.DOMAIN]
assert network_obj.configured_adapters == []
assert network_obj.adapters == [
{
"auto": False,
"default": False,
"enabled": False,
"ipv4": [],
"ipv6": [
{
"address": "2001:db8::",
"network_prefix": 8,
"flowinfo": 1,
"scope_id": 1,
}
],
"name": "eth0",
},
{
"auto": False,
"default": False,
"enabled": False,
"ipv4": [{"address": "127.0.0.1", "network_prefix": 8}],
"ipv6": [],
"name": "lo0",
},
{
"auto": True,
"default": True,
"enabled": True,
"ipv4": [{"address": "192.168.1.5", "network_prefix": 23}],
"ipv6": [],
"name": "eth1",
},
{
"auto": False,
"default": False,
"enabled": False,
"ipv4": [{"address": "169.254.3.2", "network_prefix": 16}],
"ipv6": [],
"name": "vtun0",
},
]
async def test_async_detect_interfaces_setting_loopback_route(hass, hass_storage):
"""Test without default interface config and the route returns a loopback address."""
with patch(
"homeassistant.components.network.util.socket.socket.getsockname",
return_value=[_LOOPBACK_IPADDR],
), patch(
"homeassistant.components.network.util.ifaddr.get_adapters",
return_value=_generate_mock_adapters(),
):
assert await async_setup_component(hass, network.DOMAIN, {network.DOMAIN: {}})
await hass.async_block_till_done()
network_obj = hass.data[network.DOMAIN]
assert network_obj.configured_adapters == []
assert network_obj.adapters == [
{
"auto": True,
"default": False,
"enabled": True,
"ipv4": [],
"ipv6": [
{
"address": "2001:db8::",
"network_prefix": 8,
"flowinfo": 1,
"scope_id": 1,
}
],
"name": "eth0",
},
{
"auto": False,
"default": True,
"enabled": False,
"ipv4": [{"address": "127.0.0.1", "network_prefix": 8}],
"ipv6": [],
"name": "lo0",
},
{
"auto": True,
"default": False,
"enabled": True,
"ipv4": [{"address": "192.168.1.5", "network_prefix": 23}],
"ipv6": [],
"name": "eth1",
},
{
"auto": False,
"default": False,
"enabled": False,
"ipv4": [{"address": "169.254.3.2", "network_prefix": 16}],
"ipv6": [],
"name": "vtun0",
},
]
async def test_async_detect_interfaces_setting_empty_route(hass, hass_storage):
"""Test without default interface config and the route returns nothing."""
with patch(
"homeassistant.components.network.util.socket.socket.getsockname",
return_value=[],
), patch(
"homeassistant.components.network.util.ifaddr.get_adapters",
return_value=_generate_mock_adapters(),
):
assert await async_setup_component(hass, network.DOMAIN, {network.DOMAIN: {}})
await hass.async_block_till_done()
network_obj = hass.data[network.DOMAIN]
assert network_obj.configured_adapters == []
assert network_obj.adapters == [
{
"auto": True,
"default": False,
"enabled": True,
"ipv4": [],
"ipv6": [
{
"address": "2001:db8::",
"network_prefix": 8,
"flowinfo": 1,
"scope_id": 1,
}
],
"name": "eth0",
},
{
"auto": False,
"default": False,
"enabled": False,
"ipv4": [{"address": "127.0.0.1", "network_prefix": 8}],
"ipv6": [],
"name": "lo0",
},
{
"auto": True,
"default": False,
"enabled": True,
"ipv4": [{"address": "192.168.1.5", "network_prefix": 23}],
"ipv6": [],
"name": "eth1",
},
{
"auto": False,
"default": False,
"enabled": False,
"ipv4": [{"address": "169.254.3.2", "network_prefix": 16}],
"ipv6": [],
"name": "vtun0",
},
]
async def test_async_detect_interfaces_setting_exception(hass, hass_storage):
"""Test without default interface config and the route throws an exception."""
with patch(
"homeassistant.components.network.util.socket.socket.getsockname",
side_effect=AttributeError,
), patch(
"homeassistant.components.network.util.ifaddr.get_adapters",
return_value=_generate_mock_adapters(),
):
assert await async_setup_component(hass, network.DOMAIN, {network.DOMAIN: {}})
await hass.async_block_till_done()
network_obj = hass.data[network.DOMAIN]
assert network_obj.configured_adapters == []
assert network_obj.adapters == [
{
"auto": True,
"default": False,
"enabled": True,
"ipv4": [],
"ipv6": [
{
"address": "2001:db8::",
"network_prefix": 8,
"flowinfo": 1,
"scope_id": 1,
}
],
"name": "eth0",
},
{
"auto": False,
"default": False,
"enabled": False,
"ipv4": [{"address": "127.0.0.1", "network_prefix": 8}],
"ipv6": [],
"name": "lo0",
},
{
"auto": True,
"default": False,
"enabled": True,
"ipv4": [{"address": "192.168.1.5", "network_prefix": 23}],
"ipv6": [],
"name": "eth1",
},
{
"auto": False,
"default": False,
"enabled": False,
"ipv4": [{"address": "169.254.3.2", "network_prefix": 16}],
"ipv6": [],
"name": "vtun0",
},
]
async def test_interfaces_configured_from_storage(hass, hass_storage):
"""Test settings from storage are preferred over auto configure."""
hass_storage[STORAGE_KEY] = {
"version": STORAGE_VERSION,
"key": STORAGE_KEY,
"data": {ATTR_CONFIGURED_ADAPTERS: ["eth0", "eth1", "vtun0"]},
}
with patch(
"homeassistant.components.network.util.socket.socket.getsockname",
return_value=[_NO_LOOPBACK_IPADDR],
), patch(
"homeassistant.components.network.util.ifaddr.get_adapters",
return_value=_generate_mock_adapters(),
):
assert await async_setup_component(hass, network.DOMAIN, {network.DOMAIN: {}})
await hass.async_block_till_done()
network_obj = hass.data[network.DOMAIN]
assert network_obj.configured_adapters == ["eth0", "eth1", "vtun0"]
assert network_obj.adapters == [
{
"auto": False,
"default": False,
"enabled": True,
"ipv4": [],
"ipv6": [
{
"address": "2001:db8::",
"network_prefix": 8,
"flowinfo": 1,
"scope_id": 1,
}
],
"name": "eth0",
},
{
"auto": False,
"default": False,
"enabled": False,
"ipv4": [{"address": "127.0.0.1", "network_prefix": 8}],
"ipv6": [],
"name": "lo0",
},
{
"auto": True,
"default": True,
"enabled": True,
"ipv4": [{"address": "192.168.1.5", "network_prefix": 23}],
"ipv6": [],
"name": "eth1",
},
{
"auto": False,
"default": False,
"enabled": True,
"ipv4": [{"address": "169.254.3.2", "network_prefix": 16}],
"ipv6": [],
"name": "vtun0",
},
]
async def test_interfaces_configured_from_storage_websocket_update(
hass, hass_ws_client, hass_storage
):
"""Test settings from storage can be updated via websocket api."""
hass_storage[STORAGE_KEY] = {
"version": STORAGE_VERSION,
"key": STORAGE_KEY,
"data": {ATTR_CONFIGURED_ADAPTERS: ["eth0", "eth1", "vtun0"]},
}
with patch(
"homeassistant.components.network.util.socket.socket.getsockname",
return_value=[_NO_LOOPBACK_IPADDR],
), patch(
"homeassistant.components.network.util.ifaddr.get_adapters",
return_value=_generate_mock_adapters(),
):
assert await async_setup_component(hass, network.DOMAIN, {network.DOMAIN: {}})
await hass.async_block_till_done()
network_obj = hass.data[network.DOMAIN]
assert network_obj.configured_adapters == ["eth0", "eth1", "vtun0"]
ws_client = await hass_ws_client(hass)
await ws_client.send_json({"id": 1, "type": "network"})
response = await ws_client.receive_json()
assert response["success"]
assert response["result"][ATTR_CONFIGURED_ADAPTERS] == ["eth0", "eth1", "vtun0"]
assert response["result"][ATTR_ADAPTERS] == [
{
"auto": False,
"default": False,
"enabled": True,
"ipv4": [],
"ipv6": [
{
"address": "2001:db8::",
"network_prefix": 8,
"flowinfo": 1,
"scope_id": 1,
}
],
"name": "eth0",
},
{
"auto": False,
"default": False,
"enabled": False,
"ipv4": [{"address": "127.0.0.1", "network_prefix": 8}],
"ipv6": [],
"name": "lo0",
},
{
"auto": True,
"default": True,
"enabled": True,
"ipv4": [{"address": "192.168.1.5", "network_prefix": 23}],
"ipv6": [],
"name": "eth1",
},
{
"auto": False,
"default": False,
"enabled": True,
"ipv4": [{"address": "169.254.3.2", "network_prefix": 16}],
"ipv6": [],
"name": "vtun0",
},
]
await ws_client.send_json(
{"id": 2, "type": "network/configure", "config": {ATTR_CONFIGURED_ADAPTERS: []}}
)
response = await ws_client.receive_json()
assert response["result"][ATTR_CONFIGURED_ADAPTERS] == []
await ws_client.send_json({"id": 3, "type": "network"})
response = await ws_client.receive_json()
assert response["result"][ATTR_CONFIGURED_ADAPTERS] == []
assert response["result"][ATTR_ADAPTERS] == [
{
"auto": False,
"default": False,
"enabled": False,
"ipv4": [],
"ipv6": [
{
"address": "2001:db8::",
"network_prefix": 8,
"flowinfo": 1,
"scope_id": 1,
}
],
"name": "eth0",
},
{
"auto": False,
"default": False,
"enabled": False,
"ipv4": [{"address": "127.0.0.1", "network_prefix": 8}],
"ipv6": [],
"name": "lo0",
},
{
"auto": True,
"default": True,
"enabled": True,
"ipv4": [{"address": "192.168.1.5", "network_prefix": 23}],
"ipv6": [],
"name": "eth1",
},
{
"auto": False,
"default": False,
"enabled": False,
"ipv4": [{"address": "169.254.3.2", "network_prefix": 16}],
"ipv6": [],
"name": "vtun0",
},
]

View file

@ -1,5 +1,5 @@
"""Test Zeroconf component setup process.""" """Test Zeroconf component setup process."""
from unittest.mock import patch from unittest.mock import call, patch
from zeroconf import InterfaceChoice, IPVersion, ServiceInfo, ServiceStateChange from zeroconf import InterfaceChoice, IPVersion, ServiceInfo, ServiceStateChange
@ -697,13 +697,27 @@ async def test_removed_ignored(hass, mock_zeroconf):
assert mock_service_info.mock_calls[1][1][0] == "_service.updated.local." assert mock_service_info.mock_calls[1][1][0] == "_service.updated.local."
async def test_async_detect_interfaces_setting_non_loopback_route(hass, mock_zeroconf): _ADAPTER_WITH_DEFAULT_ENABLED = [
{
"auto": True,
"default": True,
"enabled": True,
"ipv4": [{"address": "192.168.1.5", "network_prefix": 23}],
"ipv6": [],
"name": "eth1",
}
]
async def test_async_detect_interfaces_setting_non_loopback_route(hass):
"""Test without default interface config and the route returns a non-loopback address.""" """Test without default interface config and the route returns a non-loopback address."""
with patch.object(hass.config_entries.flow, "async_init"), patch.object( with patch(
"homeassistant.components.zeroconf.models.HaZeroconf"
) as mock_zc, patch.object(hass.config_entries.flow, "async_init"), patch.object(
zeroconf, "HaServiceBrowser", side_effect=service_update_mock zeroconf, "HaServiceBrowser", side_effect=service_update_mock
), patch( ), patch(
"homeassistant.components.zeroconf.IPRoute.route", "homeassistant.components.zeroconf.network.async_get_adapters",
return_value=_ROUTE_NO_LOOPBACK, return_value=_ADAPTER_WITH_DEFAULT_ENABLED,
), patch( ), patch(
"homeassistant.components.zeroconf.ServiceInfo", "homeassistant.components.zeroconf.ServiceInfo",
side_effect=get_service_info_mock, side_effect=get_service_info_mock,
@ -712,47 +726,53 @@ async def test_async_detect_interfaces_setting_non_loopback_route(hass, mock_zer
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done() await hass.async_block_till_done()
assert mock_zeroconf.called_with(interface_choice=InterfaceChoice.Default) assert mock_zc.mock_calls[0] == call(interfaces=InterfaceChoice.Default)
async def test_async_detect_interfaces_setting_loopback_route(hass, mock_zeroconf): _ADAPTERS_WITH_MANUAL_CONFIG = [
"""Test without default interface config and the route returns a loopback address.""" {
with patch.object(hass.config_entries.flow, "async_init"), patch.object( "auto": True,
zeroconf, "HaServiceBrowser", side_effect=service_update_mock "default": False,
), patch( "enabled": True,
"homeassistant.components.zeroconf.IPRoute.route", return_value=_ROUTE_LOOPBACK "ipv4": [],
), patch( "ipv6": [
"homeassistant.components.zeroconf.ServiceInfo", {
side_effect=get_service_info_mock, "address": "2001:db8::",
): "network_prefix": 8,
assert await async_setup_component(hass, zeroconf.DOMAIN, {zeroconf.DOMAIN: {}}) "flowinfo": 1,
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) "scope_id": 1,
await hass.async_block_till_done() }
],
assert mock_zeroconf.called_with(interface_choice=InterfaceChoice.All) "name": "eth0",
},
{
"auto": True,
"default": False,
"enabled": True,
"ipv4": [{"address": "192.168.1.5", "network_prefix": 23}],
"ipv6": [],
"name": "eth1",
},
{
"auto": False,
"default": False,
"enabled": False,
"ipv4": [{"address": "169.254.3.2", "network_prefix": 16}],
"ipv6": [],
"name": "vtun0",
},
]
async def test_async_detect_interfaces_setting_empty_route(hass, mock_zeroconf): async def test_async_detect_interfaces_setting_empty_route(hass):
"""Test without default interface config and the route returns nothing.""" """Test without default interface config and the route returns nothing."""
with patch.object(hass.config_entries.flow, "async_init"), patch.object( with patch(
zeroconf, "HaServiceBrowser", side_effect=service_update_mock "homeassistant.components.zeroconf.models.HaZeroconf"
), patch("homeassistant.components.zeroconf.IPRoute.route", return_value=[]), patch( ) as mock_zc, patch.object(hass.config_entries.flow, "async_init"), patch.object(
"homeassistant.components.zeroconf.ServiceInfo",
side_effect=get_service_info_mock,
):
assert await async_setup_component(hass, zeroconf.DOMAIN, {zeroconf.DOMAIN: {}})
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done()
assert mock_zeroconf.called_with(interface_choice=InterfaceChoice.All)
async def test_async_detect_interfaces_setting_exception(hass, mock_zeroconf):
"""Test without default interface config and the route throws an exception."""
with patch.object(hass.config_entries.flow, "async_init"), patch.object(
zeroconf, "HaServiceBrowser", side_effect=service_update_mock zeroconf, "HaServiceBrowser", side_effect=service_update_mock
), patch( ), patch(
"homeassistant.components.zeroconf.IPRoute.route", side_effect=AttributeError "homeassistant.components.zeroconf.network.async_get_adapters",
return_value=_ADAPTERS_WITH_MANUAL_CONFIG,
), patch( ), patch(
"homeassistant.components.zeroconf.ServiceInfo", "homeassistant.components.zeroconf.ServiceInfo",
side_effect=get_service_info_mock, side_effect=get_service_info_mock,
@ -761,4 +781,4 @@ async def test_async_detect_interfaces_setting_exception(hass, mock_zeroconf):
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done() await hass.async_block_till_done()
assert mock_zeroconf.called_with(interface_choice=InterfaceChoice.All) assert mock_zc.mock_calls[0] == call(interfaces=[1, "192.168.1.5"])

View file

@ -361,7 +361,7 @@ async def test_discovery_requirements_ssdp(hass):
) as mock_process: ) as mock_process:
await async_get_integration_with_requirements(hass, "ssdp_comp") await async_get_integration_with_requirements(hass, "ssdp_comp")
assert len(mock_process.mock_calls) == 3 assert len(mock_process.mock_calls) == 4
assert mock_process.mock_calls[0][1][2] == ssdp.requirements assert mock_process.mock_calls[0][1][2] == ssdp.requirements
# Ensure zeroconf is a dep for ssdp # Ensure zeroconf is a dep for ssdp
assert mock_process.mock_calls[1][1][1] == "zeroconf" assert mock_process.mock_calls[1][1][1] == "zeroconf"
@ -386,7 +386,7 @@ async def test_discovery_requirements_zeroconf(hass, partial_manifest):
) as mock_process: ) as mock_process:
await async_get_integration_with_requirements(hass, "comp") await async_get_integration_with_requirements(hass, "comp")
assert len(mock_process.mock_calls) == 2 # zeroconf also depends on http assert len(mock_process.mock_calls) == 3 # zeroconf also depends on http
assert mock_process.mock_calls[0][1][2] == zeroconf.requirements assert mock_process.mock_calls[0][1][2] == zeroconf.requirements