evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add additional zeroconf discovery coverage and logging to enphase_envoy (#114405)

Browse source 
* add debug info to zeroconf for enphase_envoy

* Implement review feedback, lost space

Co-authored-by: Charles Garwood <cgarwood@newdealmultimedia.com>

* review feedback textual changes.

* implement review feedbackw.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Add some more zeroconf tests and valid jwt

* review feedback assert abort reason and keyerror for serialnumber

* Review feedback config flow test ends with abort or create_entry

* Review feedback optimize resource usage

* Cover new code in test.

* Use caplog for debug COV

---------

Co-authored-by: Charles Garwood <cgarwood@newdealmultimedia.com>
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Arie Catsman 2024-04-23 14:26:53 +02:00 • committed by GitHub
parent b8918d7d17
commit 2c651e190f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 261 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

18
homeassistant/components/enphase_envoy/config_flow.py
View file

@ -89,6 +89,14 @@ class EnphaseConfigFlow(ConfigFlow, domain=DOMAIN):
self, discovery_info: zeroconf.ZeroconfServiceInfo
) -> ConfigFlowResult:
"""Handle a flow initialized by zeroconf discovery."""
if _LOGGER.isEnabledFor(logging.DEBUG):
current_hosts = self._async_current_hosts()
_LOGGER.debug(
"Zeroconf ip %s processing %s, current hosts: %s",
discovery_info.ip_address.version,
discovery_info.host,
current_hosts,
)
if discovery_info.ip_address.version != 4:
return self.async_abort(reason="not_ipv4_address")
serial = discovery_info.properties["serialnum"]
@ -96,17 +104,27 @@ class EnphaseConfigFlow(ConfigFlow, domain=DOMAIN):
await self.async_set_unique_id(serial)
self.ip_address = discovery_info.host
self._abort_if_unique_id_configured({CONF_HOST: self.ip_address})
_LOGGER.debug(
"Zeroconf ip %s, fw %s, no existing entry with serial %s",
self.ip_address,
self.protovers,
serial,
)
for entry in self._async_current_entries(include_ignore=False):
if (
entry.unique_id is None
and CONF_HOST in entry.data
and entry.data[CONF_HOST] == self.ip_address
):
_LOGGER.debug(
"Zeroconf update envoy with this ip and blank serial in unique_id",
)
title = f"{ENVOY} {serial}" if entry.title == ENVOY else ENVOY
return self.async_update_reload_and_abort(
entry, title=title, unique_id=serial, reason="already_configured"
)
_LOGGER.debug("Zeroconf ip %s to step user", self.ip_address)
return await self.async_step_user()
async def async_step_reauth(

6
tests/components/enphase_envoy/conftest.py
View file

@ -2,6 +2,7 @@
from unittest.mock import AsyncMock, Mock, patch
import jwt
from pyenphase import (
Envoy,
EnvoyData,
@ -368,7 +369,10 @@ def mock_authenticate():
@pytest.fixture(name="mock_auth")
def mock_auth(serial_number):
"""Define a mocked EnvoyAuth fixture."""
return EnvoyTokenAuth("127.0.0.1", token="abc", envoy_serial=serial_number)
token = jwt.encode(
payload={"name": "envoy", "exp": 1907837780}, key="secret", algorithm="HS256"
)
return EnvoyTokenAuth("127.0.0.1", token=token, envoy_serial=serial_number)
@pytest.fixture(name="mock_setup")

239
tests/components/enphase_envoy/test_config_flow.py
View file

@ -1,6 +1,7 @@
"""Test the Enphase Envoy config flow."""
from ipaddress import ip_address
import logging
from unittest.mock import AsyncMock
from pyenphase import EnvoyAuthenticationError, EnvoyError
@ -13,6 +14,10 @@ from homeassistant.components.enphase_envoy.const import DOMAIN, PLATFORMS
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
_LOGGER = logging.getLogger(__name__)
async def test_form(hass: HomeAssistant, config, setup_enphase_envoy) -> None:
"""Test we get the form."""
@ -324,9 +329,13 @@ async def test_form_host_already_exists(
async def test_zeroconf_serial_already_exists(
hass: HomeAssistant, config_entry, setup_enphase_envoy
hass: HomeAssistant,
config_entry,
setup_enphase_envoy,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test serial number already exists from zeroconf."""
_LOGGER.setLevel(logging.DEBUG)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
@ -345,6 +354,7 @@ async def test_zeroconf_serial_already_exists(
assert result["reason"] == "already_configured"
assert config_entry.data["host"] == "4.4.4.4"
assert "Zeroconf ip 4 processing 4.4.4.4, current hosts: {'1.1.1.1'}" in caplog.text
async def test_zeroconf_serial_already_exists_ignores_ipv6(
@ -397,6 +407,233 @@ async def test_zeroconf_host_already_exists(
assert config_entry.title == "Envoy 1234"
async def test_zero_conf_while_form(
hass: HomeAssistant, config_entry, setup_enphase_envoy
) -> None:
"""Test zeroconf while form is active."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("1.1.1.1"),
ip_addresses=[ip_address("1.1.1.1")],
hostname="mock_hostname",
name="mock_name",
port=None,
properties={"serialnum": "1234", "protovers": "7.0.1"},
type="mock_type",
),
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert config_entry.data["host"] == "1.1.1.1"
assert config_entry.unique_id == "1234"
assert config_entry.title == "Envoy 1234"
async def test_zero_conf_second_envoy_while_form(
hass: HomeAssistant, config_entry, setup_enphase_envoy
) -> None:
"""Test zeroconf while form is active."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
result2 = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("4.4.4.4"),
ip_addresses=[ip_address("4.4.4.4")],
hostname="mock_hostname",
name="mock_name",
port=None,
properties={"serialnum": "4321", "protovers": "7.0.1"},
type="mock_type",
),
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.FORM
assert config_entry.data["host"] == "1.1.1.1"
assert config_entry.unique_id == "1234"
assert config_entry.title == "Envoy 1234"
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
"host": "4.4.4.4",
"username": "test-username",
"password": "test-password",
},
)
await hass.async_block_till_done()
assert result3["type"] is FlowResultType.CREATE_ENTRY
assert result3["title"] == "Envoy 4321"
assert result3["result"].unique_id == "4321"
result4 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
"username": "test-username",
"password": "test-password",
},
)
await hass.async_block_till_done()
assert result4["type"] is FlowResultType.ABORT
async def test_zero_conf_malformed_serial_property(
hass: HomeAssistant, config_entry, setup_enphase_envoy
) -> None:
"""Test malformed zeroconf properties."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
with pytest.raises(KeyError) as ex:
await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("1.1.1.1"),
ip_addresses=[ip_address("1.1.1.1")],
hostname="mock_hostname",
name="mock_name",
port=None,
properties={"serilnum": "1234", "protovers": "7.1.2"},
type="mock_type",
),
)
await hass.async_block_till_done()
assert "serialnum" in str(ex.value)
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
"username": "test-username",
"password": "test-password",
},
)
await hass.async_block_till_done()
assert result3["type"] is FlowResultType.ABORT
async def test_zero_conf_malformed_serial(
hass: HomeAssistant, config_entry, setup_enphase_envoy
) -> None:
"""Test malformed zeroconf properties."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
result2 = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("1.1.1.1"),
ip_addresses=[ip_address("1.1.1.1")],
hostname="mock_hostname",
name="mock_name",
port=None,
properties={"serialnum": "12%4", "protovers": "7.1.2"},
type="mock_type",
),
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.FORM
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
"host": "1.1.1.1",
"username": "test-username",
"password": "test-password",
},
)
await hass.async_block_till_done()
assert result3["type"] is FlowResultType.CREATE_ENTRY
assert result3["title"] == "Envoy 12%4"
async def test_zero_conf_malformed_fw_property(
hass: HomeAssistant, config_entry, setup_enphase_envoy
) -> None:
"""Test malformed zeroconf property."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("1.1.1.1"),
ip_addresses=[ip_address("1.1.1.1")],
hostname="mock_hostname",
name="mock_name",
port=None,
properties={"serialnum": "1234", "protvers": "7.1.2"},
type="mock_type",
),
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert config_entry.data["host"] == "1.1.1.1"
assert config_entry.unique_id == "1234"
assert config_entry.title == "Envoy 1234"
async def test_zero_conf_old_blank_entry(
hass: HomeAssistant, setup_enphase_envoy
) -> None:
"""Test re-using old blank entry."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
"host": "1.1.1.1",
"username": "",
"password": "",
"name": "unknown",
},
unique_id=None,
title="Envoy",
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("1.1.1.1"),
ip_addresses=[ip_address("1.1.1.1"), ip_address("1.1.1.2")],
hostname="mock_hostname",
name="mock_name",
port=None,
properties={"serialnum": "1234", "protovers": "7.1.2"},
type="mock_type",
),
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert entry.data["host"] == "1.1.1.1"
assert entry.unique_id == "1234"
assert entry.title == "Envoy 1234"
async def test_reauth(hass: HomeAssistant, config_entry, setup_enphase_envoy) -> None:
"""Test we reauth auth."""
result = await hass.config_entries.flow.async_init(