mirror of
https://github.com/home-assistant/core.git
synced 2025-07-23 05:07:41 +00:00
Allow Shelly CoAP to honour default network adapter (#110997)
* Allow Shelly CoAP to honor default network adapter * apply review comment * 1 more debug log line * adapt code to library changes * test * improve test * one more test
This commit is contained in:
parent
30c3174498
commit
4095de0566
@ -7,7 +7,7 @@ from typing import Any, Final
|
||||
|
||||
from aioshelly.block_device import BlockDevice, BlockUpdateType
|
||||
from aioshelly.common import ConnectionOptions
|
||||
from aioshelly.const import RPC_GENERATIONS
|
||||
from aioshelly.const import DEFAULT_COAP_PORT, RPC_GENERATIONS
|
||||
from aioshelly.exceptions import (
|
||||
DeviceConnectionError,
|
||||
FirmwareUnsupported,
|
||||
@ -37,7 +37,6 @@ from .const import (
|
||||
CONF_COAP_PORT,
|
||||
CONF_SLEEP_PERIOD,
|
||||
DATA_CONFIG_ENTRY,
|
||||
DEFAULT_COAP_PORT,
|
||||
DOMAIN,
|
||||
FIRMWARE_UNSUPPORTED_ISSUE_ID,
|
||||
LOGGER,
|
||||
|
@ -33,7 +33,6 @@ LOGGER: Logger = getLogger(__package__)
|
||||
|
||||
DATA_CONFIG_ENTRY: Final = "config_entry"
|
||||
CONF_COAP_PORT: Final = "coap_port"
|
||||
DEFAULT_COAP_PORT: Final = 5683
|
||||
FIRMWARE_PATTERN: Final = re.compile(r"^(\d{8})")
|
||||
|
||||
# max light transition time in milliseconds
|
||||
|
@ -3,7 +3,7 @@
|
||||
"name": "Shelly",
|
||||
"codeowners": ["@balloob", "@bieniu", "@thecode", "@chemelli74", "@bdraco"],
|
||||
"config_flow": true,
|
||||
"dependencies": ["bluetooth", "http"],
|
||||
"dependencies": ["bluetooth", "http", "network"],
|
||||
"documentation": "https://www.home-assistant.io/integrations/shelly",
|
||||
"integration_type": "device",
|
||||
"iot_class": "local_push",
|
||||
|
@ -3,12 +3,14 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
from ipaddress import IPv4Address
|
||||
from typing import Any, cast
|
||||
|
||||
from aiohttp.web import Request, WebSocketResponse
|
||||
from aioshelly.block_device import COAP, Block, BlockDevice
|
||||
from aioshelly.const import (
|
||||
BLOCK_GENERATIONS,
|
||||
DEFAULT_COAP_PORT,
|
||||
MODEL_1L,
|
||||
MODEL_DIMMER,
|
||||
MODEL_DIMMER_2,
|
||||
@ -19,6 +21,7 @@ from aioshelly.const import (
|
||||
)
|
||||
from aioshelly.rpc_device import RpcDevice, WsServer
|
||||
|
||||
from homeassistant.components import network
|
||||
from homeassistant.components.http import HomeAssistantView
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
|
||||
@ -36,7 +39,6 @@ from .const import (
|
||||
BASIC_INPUTS_EVENTS_TYPES,
|
||||
CONF_COAP_PORT,
|
||||
CONF_GEN,
|
||||
DEFAULT_COAP_PORT,
|
||||
DEVICES_WITHOUT_FIRMWARE_CHANGELOG,
|
||||
DOMAIN,
|
||||
FIRMWARE_UNSUPPORTED_ISSUE_ID,
|
||||
@ -221,12 +223,27 @@ def get_shbtn_input_triggers() -> list[tuple[str, str]]:
|
||||
async def get_coap_context(hass: HomeAssistant) -> COAP:
|
||||
"""Get CoAP context to be used in all Shelly Gen1 devices."""
|
||||
context = COAP()
|
||||
|
||||
adapters = await network.async_get_adapters(hass)
|
||||
LOGGER.debug("Network adapters: %s", adapters)
|
||||
|
||||
ipv4: list[IPv4Address] = []
|
||||
if not network.async_only_default_interface_enabled(adapters):
|
||||
for address in await network.async_get_enabled_source_ips(hass):
|
||||
if address.version == 4 and not (
|
||||
address.is_link_local
|
||||
or address.is_loopback
|
||||
or address.is_multicast
|
||||
or address.is_unspecified
|
||||
):
|
||||
ipv4.append(address)
|
||||
LOGGER.debug("Network IPv4 addresses: %s", ipv4)
|
||||
if DOMAIN in hass.data:
|
||||
port = hass.data[DOMAIN].get(CONF_COAP_PORT, DEFAULT_COAP_PORT)
|
||||
else:
|
||||
port = DEFAULT_COAP_PORT
|
||||
LOGGER.info("Starting CoAP context with UDP port %s", port)
|
||||
await context.initialize(port)
|
||||
await context.initialize(port, ipv4)
|
||||
|
||||
@callback
|
||||
def shutdown_listener(ev: Event) -> None:
|
||||
|
@ -1,7 +1,9 @@
|
||||
"""Test cases for the Shelly component."""
|
||||
|
||||
from unittest.mock import AsyncMock, Mock, patch
|
||||
from ipaddress import IPv4Address
|
||||
from unittest.mock import AsyncMock, Mock, call, patch
|
||||
|
||||
from aioshelly.block_device import COAP
|
||||
from aioshelly.exceptions import (
|
||||
DeviceConnectionError,
|
||||
FirmwareUnsupported,
|
||||
@ -49,6 +51,55 @@ async def test_custom_coap_port(
|
||||
assert "Starting CoAP context with UDP port 7632" in caplog.text
|
||||
|
||||
|
||||
async def test_ip_address_with_only_default_interface(
|
||||
hass: HomeAssistant, mock_block_device: Mock, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Test more local ip addresses with only the default interface.."""
|
||||
with patch(
|
||||
"homeassistant.components.network.async_only_default_interface_enabled",
|
||||
return_value=True,
|
||||
), patch(
|
||||
"homeassistant.components.network.async_get_enabled_source_ips",
|
||||
return_value=[IPv4Address("192.168.1.10"), IPv4Address("10.10.10.10")],
|
||||
), patch(
|
||||
"homeassistant.components.shelly.utils.COAP",
|
||||
autospec=COAP,
|
||||
) as mock_coap_init:
|
||||
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {"coap_port": 7632}})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await init_integration(hass, 1)
|
||||
assert "Starting CoAP context with UDP port 7632" in caplog.text
|
||||
# Make sure COAP.initialize is called with an empty list
|
||||
# when async_only_default_interface_enabled is True even if
|
||||
# async_get_enabled_source_ips returns more than one address
|
||||
assert mock_coap_init.mock_calls[1] == call().initialize(7632, [])
|
||||
|
||||
|
||||
async def test_ip_address_without_only_default_interface(
|
||||
hass: HomeAssistant, mock_block_device: Mock, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Test more local ip addresses without only the default interface.."""
|
||||
with patch(
|
||||
"homeassistant.components.network.async_only_default_interface_enabled",
|
||||
return_value=False,
|
||||
), patch(
|
||||
"homeassistant.components.network.async_get_enabled_source_ips",
|
||||
return_value=[IPv4Address("192.168.1.10"), IPv4Address("10.10.10.10")],
|
||||
), patch(
|
||||
"homeassistant.components.shelly.utils.COAP",
|
||||
autospec=COAP,
|
||||
) as mock_coap_init:
|
||||
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {"coap_port": 7632}})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await init_integration(hass, 1)
|
||||
assert "Starting CoAP context with UDP port 7632" in caplog.text
|
||||
assert mock_coap_init.mock_calls[1] == call().initialize(
|
||||
7632, [IPv4Address("192.168.1.10"), IPv4Address("10.10.10.10")]
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("gen", [1, 2, 3])
|
||||
async def test_shared_device_mac(
|
||||
hass: HomeAssistant,
|
||||
|
Loading…
x
Reference in New Issue
Block a user