jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-14 16:57:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

SMA add DHCP discovery (#135843)

Browse Source 
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Erwin Douna 2025-04-09 16:24:30 +02:00 committed by GitHub
parent 170e6bdcab
commit b058b2574f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 318 additions and 105 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
homeassistant/components/sma/__init__.py
View File

@ -10,7 +10,9 @@ import pysma
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_CONNECTIONS,
CONF_HOST,
CONF_MAC,
CONF_PASSWORD,
CONF_SCAN_INTERVAL,
CONF_SSL,
@ -19,6 +21,7 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
@ -75,6 +78,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
serial_number=sma_device_info["serial"],
)
# Add the MAC address to connections, if it comes via DHCP
if CONF_MAC in entry.data:
device_info[ATTR_CONNECTIONS] = {
(dr.CONNECTION_NETWORK_MAC, entry.data[CONF_MAC])
}
# Define the coordinator
async def async_update_data():
"""Update the used SMA sensors."""

133
homeassistant/components/sma/config_flow.py
View File

@ -7,26 +7,43 @@ from typing import Any
import pysma
import voluptuous as vol
from yarl import URL
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_SSL, CONF_VERIFY_SSL
from homeassistant.const import (
CONF_HOST,
CONF_MAC,
CONF_NAME,
CONF_PASSWORD,
CONF_SSL,
CONF_VERIFY_SSL,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
from .const import CONF_GROUP, DOMAIN, GROUPS
_LOGGER = logging.getLogger(__name__)
async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, Any]:
async def validate_input(
hass: HomeAssistant,
user_input: dict[str, Any],
data: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Validate the user input allows us to connect."""
session = async_get_clientsession(hass, verify_ssl=data[CONF_VERIFY_SSL])
session = async_get_clientsession(hass, verify_ssl=user_input[CONF_VERIFY_SSL])
protocol = "https" if data[CONF_SSL] else "http"
url = f"{protocol}://{data[CONF_HOST]}"
protocol = "https" if user_input[CONF_SSL] else "http"
host = data[CONF_HOST] if data is not None else user_input[CONF_HOST]
url = URL.build(scheme=protocol, host=host)
sma = pysma.SMA(session, url, data[CONF_PASSWORD], group=data[CONF_GROUP])
sma = pysma.SMA(
session, str(url), user_input[CONF_PASSWORD], group=user_input[CONF_GROUP]
)
# new_session raises SmaAuthenticationException on failure
await sma.new_session()
@ -51,34 +68,53 @@ class SmaConfigFlow(ConfigFlow, domain=DOMAIN):
CONF_GROUP: GROUPS[0],
CONF_PASSWORD: vol.UNDEFINED,
}
self._discovery_data: dict[str, Any] = {}
async def _handle_user_input(
self, user_input: dict[str, Any], discovery: bool = False
) -> tuple[dict[str, str], dict[str, str]]:
"""Handle the user input."""
errors: dict[str, str] = {}
device_info: dict[str, str] = {}
if not discovery:
self._data[CONF_HOST] = user_input[CONF_HOST]
self._data[CONF_SSL] = user_input[CONF_SSL]
self._data[CONF_VERIFY_SSL] = user_input[CONF_VERIFY_SSL]
self._data[CONF_GROUP] = user_input[CONF_GROUP]
self._data[CONF_PASSWORD] = user_input[CONF_PASSWORD]
try:
device_info = await validate_input(
self.hass, user_input=user_input, data=self._data
)
except pysma.exceptions.SmaConnectionException:
errors["base"] = "cannot_connect"
except pysma.exceptions.SmaAuthenticationException:
errors["base"] = "invalid_auth"
except pysma.exceptions.SmaReadException:
errors["base"] = "cannot_retrieve_device_info"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
return errors, device_info
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""First step in config flow."""
errors = {}
errors: dict[str, str] = {}
if user_input is not None:
self._data[CONF_HOST] = user_input[CONF_HOST]
self._data[CONF_SSL] = user_input[CONF_SSL]
self._data[CONF_VERIFY_SSL] = user_input[CONF_VERIFY_SSL]
self._data[CONF_GROUP] = user_input[CONF_GROUP]
self._data[CONF_PASSWORD] = user_input[CONF_PASSWORD]
try:
device_info = await validate_input(self.hass, user_input)
except pysma.exceptions.SmaConnectionException:
errors["base"] = "cannot_connect"
except pysma.exceptions.SmaAuthenticationException:
errors["base"] = "invalid_auth"
except pysma.exceptions.SmaReadException:
errors["base"] = "cannot_retrieve_device_info"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
errors, device_info = await self._handle_user_input(user_input=user_input)
if not errors:
await self.async_set_unique_id(str(device_info["serial"]))
await self.async_set_unique_id(
str(device_info["serial"]), raise_on_progress=False
)
self._abort_if_unique_id_configured(updates=self._data)
return self.async_create_entry(
title=self._data[CONF_HOST], data=self._data
)
@ -100,3 +136,50 @@ class SmaConfigFlow(ConfigFlow, domain=DOMAIN):
),
errors=errors,
)
async def async_step_dhcp(
self, discovery_info: DhcpServiceInfo
) -> ConfigFlowResult:
"""Handle DHCP discovery."""
self._discovery_data[CONF_HOST] = discovery_info.ip
self._discovery_data[CONF_MAC] = format_mac(discovery_info.macaddress)
self._discovery_data[CONF_NAME] = discovery_info.hostname
self._data[CONF_HOST] = discovery_info.ip
self._data[CONF_MAC] = format_mac(self._discovery_data[CONF_MAC])
await self.async_set_unique_id(discovery_info.hostname.replace("SMA", ""))
self._abort_if_unique_id_configured()
return await self.async_step_discovery_confirm()
async def async_step_discovery_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Confirm discovery."""
errors: dict[str, str] = {}
if user_input is not None:
errors, device_info = await self._handle_user_input(
user_input=user_input, discovery=True
)
if not errors:
return self.async_create_entry(
title=self._data[CONF_HOST], data=self._data
)
return self.async_show_form(
step_id="discovery_confirm",
data_schema=vol.Schema(
{
vol.Optional(CONF_SSL, default=self._data[CONF_SSL]): cv.boolean,
vol.Optional(
CONF_VERIFY_SSL, default=self._data[CONF_VERIFY_SSL]
): cv.boolean,
vol.Optional(CONF_GROUP, default=self._data[CONF_GROUP]): vol.In(
GROUPS
),
vol.Required(CONF_PASSWORD): cv.string,
}
),
errors=errors,
)

7
homeassistant/components/sma/manifest.json
View File

@ -3,6 +3,13 @@
"name": "SMA Solar",
"codeowners": ["@kellerza", "@rklomp", "@erwindouna"],
"config_flow": true,
"dhcp": [
{
"hostname": "sma*",
"macaddress": "0015BB*"
},
{ "registered_devices": true }
],
"documentation": "https://www.home-assistant.io/integrations/sma",
"iot_class": "local_polling",
"loggers": ["pysma"],

9
homeassistant/generated/dhcp.py generated
View File

@ -613,6 +613,15 @@ DHCP: Final[list[dict[str, str | bool]]] = [
"domain": "sleepiq",
"macaddress": "64DBA0*",
},
{
"domain": "sma",
"hostname": "sma*",
"macaddress": "0015BB*",
},
{
"domain": "sma",
"registered_devices": True,
},
{
"domain": "smartthings",
"hostname": "st*",

40
tests/components/sma/__init__.py
View File

@ -1,7 +1,17 @@
"""Tests for the sma integration."""
import unittest
from unittest.mock import patch
from homeassistant.components.sma.const import CONF_GROUP
from homeassistant.const import (
CONF_HOST,
CONF_MAC,
CONF_PASSWORD,
CONF_SSL,
CONF_VERIFY_SSL,
)
MOCK_DEVICE = {
"manufacturer": "SMA",
"name": "SMA Device Name",
@ -10,15 +20,33 @@ MOCK_DEVICE = {
}
MOCK_USER_INPUT = {
"host": "1.1.1.1",
"ssl": True,
"verify_ssl": False,
"group": "user",
"password": "password",
CONF_HOST: "1.1.1.1",
CONF_SSL: True,
CONF_VERIFY_SSL: False,
CONF_GROUP: "user",
CONF_PASSWORD: "password",
}
MOCK_DHCP_DISCOVERY_INPUT = {
# CONF_HOST: "1.1.1.2",
CONF_SSL: True,
CONF_VERIFY_SSL: False,
CONF_GROUP: "user",
CONF_PASSWORD: "password",
}
MOCK_DHCP_DISCOVERY = {
CONF_HOST: "1.1.1.1",
CONF_SSL: True,
CONF_VERIFY_SSL: False,
CONF_GROUP: "user",
CONF_PASSWORD: "password",
CONF_MAC: "00:15:bb:00:ab:cd",
}
def _patch_async_setup_entry(return_value=True):
def _patch_async_setup_entry(return_value=True) -> unittest.mock._patch:
"""Patch async_setup_entry."""
return patch(
"homeassistant.components.sma.async_setup_entry",
return_value=return_value,

6
tests/components/sma/conftest.py
View File

@ -17,9 +17,9 @@ from tests.common import MockConfigEntry
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
def mock_config_entry(hass: HomeAssistant) -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
entry = MockConfigEntry(
domain=DOMAIN,
title=MOCK_DEVICE["name"],
unique_id=str(MOCK_DEVICE["serial"]),
@ -27,6 +27,8 @@ def mock_config_entry() -> MockConfigEntry:
source=config_entries.SOURCE_IMPORT,
minor_version=2,
)
entry.add_to_hass(hass)
return entry
@pytest.fixture

219
tests/components/sma/test_config_flow.py
View File

@ -7,13 +7,35 @@ from pysma.exceptions import (
SmaConnectionException,
SmaReadException,
)
import pytest
from homeassistant.components.sma.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.config_entries import SOURCE_DHCP, SOURCE_USER
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
from . import MOCK_DEVICE, MOCK_USER_INPUT, _patch_async_setup_entry
from . import (
MOCK_DEVICE,
MOCK_DHCP_DISCOVERY,
MOCK_DHCP_DISCOVERY_INPUT,
MOCK_USER_INPUT,
_patch_async_setup_entry,
)
from tests.conftest import MockConfigEntry
DHCP_DISCOVERY = DhcpServiceInfo(
ip="1.1.1.1",
hostname="SMA123456",
macaddress="0015BB00abcd",
)
DHCP_DISCOVERY_DUPLICATE = DhcpServiceInfo(
ip="1.1.1.1",
hostname="SMA123456789",
macaddress="0015BB00abcd",
)
async def test_form(hass: HomeAssistant) -> None:
@ -43,14 +65,27 @@ async def test_form(hass: HomeAssistant) -> None:
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_cannot_connect(hass: HomeAssistant) -> None:
@pytest.mark.parametrize(
("exception", "error"),
[
(SmaConnectionException, "cannot_connect"),
(SmaAuthenticationException, "invalid_auth"),
(SmaReadException, "cannot_retrieve_device_info"),
(Exception, "unknown"),
],
)
async def test_form_exceptions(
hass: HomeAssistant, exception: Exception, error: str
) -> None:
"""Test we handle cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
with (
patch("pysma.SMA.new_session", side_effect=SmaConnectionException),
patch(
"homeassistant.components.sma.pysma.SMA.new_session", side_effect=exception
),
_patch_async_setup_entry() as mock_setup_entry,
):
result = await hass.config_entries.flow.async_configure(
@ -59,83 +94,27 @@ async def test_form_cannot_connect(hass: HomeAssistant) -> None:
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "cannot_connect"}
assert result["errors"] == {"base": error}
assert len(mock_setup_entry.mock_calls) == 0
async def test_form_invalid_auth(hass: HomeAssistant) -> None:
"""Test we handle invalid auth error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
with (
patch("pysma.SMA.new_session", side_effect=SmaAuthenticationException),
_patch_async_setup_entry() as mock_setup_entry,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_USER_INPUT,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "invalid_auth"}
assert len(mock_setup_entry.mock_calls) == 0
async def test_form_cannot_retrieve_device_info(hass: HomeAssistant) -> None:
"""Test we handle cannot retrieve device info error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
with (
patch("pysma.SMA.new_session", return_value=True),
patch("pysma.SMA.read", side_effect=SmaReadException),
_patch_async_setup_entry() as mock_setup_entry,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_USER_INPUT,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "cannot_retrieve_device_info"}
assert len(mock_setup_entry.mock_calls) == 0
async def test_form_unexpected_exception(hass: HomeAssistant) -> None:
"""Test we handle unexpected exception."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
with (
patch("pysma.SMA.new_session", side_effect=Exception),
_patch_async_setup_entry() as mock_setup_entry,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_USER_INPUT,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "unknown"}
assert len(mock_setup_entry.mock_calls) == 0
async def test_form_already_configured(hass: HomeAssistant, mock_config_entry) -> None:
async def test_form_already_configured(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test starting a flow by user when already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
with (
patch("pysma.SMA.new_session", return_value=True),
patch("pysma.SMA.device_info", return_value=MOCK_DEVICE),
patch("pysma.SMA.close_session", return_value=True),
patch("homeassistant.components.sma.pysma.SMA.new_session", return_value=True),
patch(
"homeassistant.components.sma.pysma.SMA.device_info",
return_value=MOCK_DEVICE,
),
patch(
"homeassistant.components.sma.pysma.SMA.close_session", return_value=True
),
_patch_async_setup_entry() as mock_setup_entry,
):
result = await hass.config_entries.flow.async_configure(
@ -146,3 +125,99 @@ async def test_form_already_configured(hass: HomeAssistant, mock_config_entry) -
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert len(mock_setup_entry.mock_calls) == 0
async def test_dhcp_discovery(hass: HomeAssistant) -> None:
"""Test we can setup from dhcp discovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_DHCP},
data=DHCP_DISCOVERY,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "discovery_confirm"
with (
patch("homeassistant.components.sma.pysma.SMA.new_session", return_value=True),
patch(
"homeassistant.components.sma.pysma.SMA.device_info",
return_value=MOCK_DEVICE,
),
_patch_async_setup_entry() as mock_setup_entry,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_DHCP_DISCOVERY_INPUT,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == MOCK_DHCP_DISCOVERY["host"]
assert result["data"] == MOCK_DHCP_DISCOVERY
assert result["result"].unique_id == DHCP_DISCOVERY.hostname.replace("SMA", "")
assert len(mock_setup_entry.mock_calls) == 1
async def test_dhcp_already_configured(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test starting a flow by dhcp when already configured."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_DHCP}, data=DHCP_DISCOVERY_DUPLICATE
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
("exception", "error"),
[
(SmaConnectionException, "cannot_connect"),
(SmaAuthenticationException, "invalid_auth"),
(SmaReadException, "cannot_retrieve_device_info"),
(Exception, "unknown"),
],
)
async def test_dhcp_exceptions(
hass: HomeAssistant, exception: Exception, error: str
) -> None:
"""Test we handle cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_DHCP},
data=DHCP_DISCOVERY,
)
with patch(
"homeassistant.components.sma.pysma.SMA.new_session", side_effect=exception
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_DHCP_DISCOVERY_INPUT,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": error}
with (
patch("homeassistant.components.sma.pysma.SMA.new_session", return_value=True),
patch(
"homeassistant.components.sma.pysma.SMA.device_info",
return_value=MOCK_DEVICE,
),
patch(
"homeassistant.components.sma.pysma.SMA.close_session", return_value=True
),
_patch_async_setup_entry(),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_DHCP_DISCOVERY_INPUT,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == MOCK_DHCP_DISCOVERY["host"]
assert result["data"] == MOCK_DHCP_DISCOVERY
assert result["result"].unique_id == DHCP_DISCOVERY.hostname.replace("SMA", "")