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

Add config flow for enphase envoy (#48517)

Browse source 
Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
This commit is contained in:
J. Nick Koston 2021-04-04 22:11:06 -10:00 • committed by GitHub
parent 6204765835
commit 94fde73add
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
13 changed files with 719 additions and 92 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.coveragerc
View file

@ -247,6 +247,7 @@ omit =
homeassistant/components/enocean/light.py
homeassistant/components/enocean/sensor.py
homeassistant/components/enocean/switch.py
homeassistant/components/enphase_envoy/__init__.py
homeassistant/components/enphase_envoy/sensor.py
homeassistant/components/entur_public_transport/*
homeassistant/components/environment_canada/*

101
homeassistant/components/enphase_envoy/__init__.py
View file

@ -1 +1,100 @@
"""The enphase_envoy component."""
"""The Enphase Envoy integration."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
import async_timeout
from envoy_reader.envoy_reader import EnvoyReader
import httpx
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import COORDINATOR, DOMAIN, NAME, PLATFORMS, SENSORS
SCAN_INTERVAL = timedelta(seconds=60)
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Enphase Envoy from a config entry."""
config = entry.data
name = config[CONF_NAME]
envoy_reader = EnvoyReader(
config[CONF_HOST], config[CONF_USERNAME], config[CONF_PASSWORD]
)
try:
await envoy_reader.getData()
except httpx.HTTPStatusError as err:
_LOGGER.error("Authentication failure during setup: %s", err)
return
except (AttributeError, httpx.HTTPError) as err:
raise ConfigEntryNotReady from err
async def async_update_data():
"""Fetch data from API endpoint."""
data = {}
async with async_timeout.timeout(30):
try:
await envoy_reader.getData()
except httpx.HTTPError as err:
raise UpdateFailed(f"Error communicating with API: {err}") from err
for condition in SENSORS:
if condition != "inverters":
data[condition] = await getattr(envoy_reader, condition)()
else:
data[
"inverters_production"
] = await envoy_reader.inverters_production()
_LOGGER.debug("Retrieved data from API: %s", data)
return data
coordinator = DataUpdateCoordinator(
hass,
_LOGGER,
name="envoy {name}",
update_method=async_update_data,
update_interval=SCAN_INTERVAL,
)
envoy_reader.get_inverters = True
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
COORDINATOR: coordinator,
NAME: name,
}
for platform in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, platform)
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(entry, platform)
for platform in PLATFORMS
]
)
)
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

162
homeassistant/components/enphase_envoy/config_flow.py Normal file
View file

@ -0,0 +1,162 @@
"""Config flow for Enphase Envoy integration."""
from __future__ import annotations
import logging
from typing import Any
from envoy_reader.envoy_reader import EnvoyReader
import httpx
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import (
CONF_HOST,
CONF_IP_ADDRESS,
CONF_NAME,
CONF_PASSWORD,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
ENVOY = "Envoy"
CONF_SERIAL = "serial"
async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, Any]:
"""Validate the user input allows us to connect."""
envoy_reader = EnvoyReader(
data[CONF_HOST], data[CONF_USERNAME], data[CONF_PASSWORD], inverters=True
)
try:
await envoy_reader.getData()
except httpx.HTTPStatusError as err:
raise InvalidAuth from err
except (AttributeError, httpx.HTTPError) as err:
raise CannotConnect from err
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Enphase Envoy."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
def __init__(self):
"""Initialize an envoy flow."""
self.ip_address = None
self.name = None
self.username = None
self.serial = None
@callback
def _async_generate_schema(self):
"""Generate schema."""
schema = {}
if self.ip_address:
schema[vol.Required(CONF_HOST, default=self.ip_address)] = vol.In(
[self.ip_address]
)
else:
schema[vol.Required(CONF_HOST)] = str
schema[vol.Optional(CONF_USERNAME, default=self.username or "envoy")] = str
schema[vol.Optional(CONF_PASSWORD, default="")] = str
return vol.Schema(schema)
async def async_step_import(self, import_config):
"""Handle a flow import."""
self.ip_address = import_config[CONF_IP_ADDRESS]
self.username = import_config[CONF_USERNAME]
self.name = import_config[CONF_NAME]
return await self.async_step_user(
{
CONF_HOST: import_config[CONF_IP_ADDRESS],
CONF_USERNAME: import_config[CONF_USERNAME],
CONF_PASSWORD: import_config[CONF_PASSWORD],
}
)
@callback
def _async_current_hosts(self):
"""Return a set of hosts."""
return {
entry.data[CONF_HOST]
for entry in self._async_current_entries(include_ignore=False)
if CONF_HOST in entry.data
}
async def async_step_zeroconf(self, discovery_info):
"""Handle a flow initialized by zeroconf discovery."""
self.serial = discovery_info["properties"]["serialnum"]
await self.async_set_unique_id(self.serial)
self.ip_address = discovery_info[CONF_HOST]
self._abort_if_unique_id_configured({CONF_HOST: self.ip_address})
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
):
title = f"{ENVOY} {self.serial}" if entry.title == ENVOY else ENVOY
self.hass.config_entries.async_update_entry(
entry, title=title, unique_id=self.serial
)
self.hass.async_create_task(
self.hass.config_entries.async_reload(entry.entry_id)
)
return self.async_abort(reason="already_configured")
return await self.async_step_user()
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Handle the initial step."""
errors = {}
if user_input is not None:
if user_input[CONF_HOST] in self._async_current_hosts():
return self.async_abort(reason="already_configured")
try:
await validate_input(self.hass, user_input)
except CannotConnect:
errors["base"] = "cannot_connect"
except InvalidAuth:
errors["base"] = "invalid_auth"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
data = user_input.copy()
if self.serial:
data[CONF_NAME] = f"{ENVOY} {self.serial}"
else:
data[CONF_NAME] = self.name or ENVOY
return self.async_create_entry(title=data[CONF_NAME], data=data)
if self.serial:
self.context["title_placeholders"] = {
CONF_SERIAL: self.serial,
CONF_HOST: self.ip_address,
}
return self.async_show_form(
step_id="user",
data_schema=self._async_generate_schema(),
errors=errors,
)
class CannotConnect(HomeAssistantError):
"""Error to indicate we cannot connect."""
class InvalidAuth(HomeAssistantError):
"""Error to indicate there is invalid auth."""

30
homeassistant/components/enphase_envoy/const.py Normal file
View file

@ -0,0 +1,30 @@
"""The enphase_envoy component."""
from homeassistant.const import ENERGY_WATT_HOUR, POWER_WATT
DOMAIN = "enphase_envoy"
PLATFORMS = ["sensor"]
COORDINATOR = "coordinator"
NAME = "name"
SENSORS = {
"production": ("Current Energy Production", POWER_WATT),
"daily_production": ("Today's Energy Production", ENERGY_WATT_HOUR),
"seven_days_production": (
"Last Seven Days Energy Production",
ENERGY_WATT_HOUR,
),
"lifetime_production": ("Lifetime Energy Production", ENERGY_WATT_HOUR),
"consumption": ("Current Energy Consumption", POWER_WATT),
"daily_consumption": ("Today's Energy Consumption", ENERGY_WATT_HOUR),
"seven_days_consumption": (
"Last Seven Days Energy Consumption",
ENERGY_WATT_HOUR,
),
"lifetime_consumption": ("Lifetime Energy Consumption", ENERGY_WATT_HOUR),
"inverters": ("Inverter", POWER_WATT),
}

10
homeassistant/components/enphase_envoy/manifest.json
View file

@ -2,8 +2,12 @@
"domain": "enphase_envoy",
"name": "Enphase Envoy",
"documentation": "https://www.home-assistant.io/integrations/enphase_envoy",
"requirements": ["envoy_reader==0.18.3"],
"requirements": [
"envoy_reader==0.18.3"
],
"codeowners": [
"@gtdiehl"
]
}
],
"config_flow": true,
"zeroconf": [{ "type": "_enphase-envoy._tcp.local."}]
}

149
homeassistant/components/enphase_envoy/sensor.py
View file

@ -1,55 +1,27 @@
"""Support for Enphase Envoy solar energy monitor."""
from datetime import timedelta
import logging
import async_timeout
from envoy_reader.envoy_reader import EnvoyReader
import httpx
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import (
CONF_IP_ADDRESS,
CONF_MONITORED_CONDITIONS,
CONF_NAME,
CONF_PASSWORD,
CONF_USERNAME,
ENERGY_WATT_HOUR,
POWER_WATT,
)
from homeassistant.exceptions import PlatformNotReady
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
UpdateFailed,
)
from homeassistant.helpers.update_coordinator import CoordinatorEntity
_LOGGER = logging.getLogger(__name__)
SENSORS = {
"production": ("Envoy Current Energy Production", POWER_WATT),
"daily_production": ("Envoy Today's Energy Production", ENERGY_WATT_HOUR),
"seven_days_production": (
"Envoy Last Seven Days Energy Production",
ENERGY_WATT_HOUR,
),
"lifetime_production": ("Envoy Lifetime Energy Production", ENERGY_WATT_HOUR),
"consumption": ("Envoy Current Energy Consumption", POWER_WATT),
"daily_consumption": ("Envoy Today's Energy Consumption", ENERGY_WATT_HOUR),
"seven_days_consumption": (
"Envoy Last Seven Days Energy Consumption",
ENERGY_WATT_HOUR,
),
"lifetime_consumption": ("Envoy Lifetime Energy Consumption", ENERGY_WATT_HOUR),
"inverters": ("Envoy Inverter", POWER_WATT),
}
from .const import COORDINATOR, DOMAIN, NAME, SENSORS
ICON = "mdi:flash"
CONST_DEFAULT_HOST = "envoy"
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(seconds=60)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
@ -64,89 +36,59 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
)
async def async_setup_platform(
homeassistant, config, async_add_entities, discovery_info=None
):
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the Enphase Envoy sensor."""
ip_address = config[CONF_IP_ADDRESS]
monitored_conditions = config[CONF_MONITORED_CONDITIONS]
name = config[CONF_NAME]
username = config[CONF_USERNAME]
password = config[CONF_PASSWORD]
if "inverters" in monitored_conditions:
envoy_reader = EnvoyReader(ip_address, username, password, inverters=True)
else:
envoy_reader = EnvoyReader(ip_address, username, password)
try:
await envoy_reader.getData()
except httpx.HTTPStatusError as err:
_LOGGER.error("Authentication failure during setup: %s", err)
return
except httpx.HTTPError as err:
raise PlatformNotReady from err
async def async_update_data():
"""Fetch data from API endpoint."""
data = {}
async with async_timeout.timeout(30):
try:
await envoy_reader.getData()
except httpx.HTTPError as err:
raise UpdateFailed(f"Error communicating with API: {err}") from err
for condition in monitored_conditions:
if condition != "inverters":
data[condition] = await getattr(envoy_reader, condition)()
else:
data["inverters_production"] = await getattr(
envoy_reader, "inverters_production"
)()
_LOGGER.debug("Retrieved data from API: %s", data)
return data
coordinator = DataUpdateCoordinator(
homeassistant,
_LOGGER,
name="sensor",
update_method=async_update_data,
update_interval=SCAN_INTERVAL,
_LOGGER.warning(
"Loading enphase_envoy via platform config is deprecated; The configuration"
" has been migrated to a config entry and can be safely removed"
)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=config
)
)
await coordinator.async_refresh()
if coordinator.data is None:
raise PlatformNotReady
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up envoy sensor platform."""
data = hass.data[DOMAIN][config_entry.entry_id]
coordinator = data[COORDINATOR]
name = data[NAME]
entities = []
for condition in monitored_conditions:
for condition in SENSORS:
entity_name = ""
if (
condition == "inverters"
and coordinator.data.get("inverters_production") is not None
):
for inverter in coordinator.data["inverters_production"]:
entity_name = f"{name}{SENSORS[condition][0]} {inverter}"
entity_name = f"{name} {SENSORS[condition][0]} {inverter}"
split_name = entity_name.split(" ")
serial_number = split_name[-1]
entities.append(
Envoy(
condition,
entity_name,
name,
config_entry.unique_id,
serial_number,
SENSORS[condition][1],
coordinator,
)
)
elif condition != "inverters":
entity_name = f"{name}{SENSORS[condition][0]}"
data = coordinator.data.get(condition)
if isinstance(data, str) and "not available" in data:
continue
entity_name = f"{name} {SENSORS[condition][0]}"
entities.append(
Envoy(
condition,
entity_name,
name,
config_entry.unique_id,
None,
SENSORS[condition][1],
coordinator,
@ -159,11 +101,22 @@ async def async_setup_platform(
class Envoy(CoordinatorEntity, SensorEntity):
"""Envoy entity."""
def __init__(self, sensor_type, name, serial_number, unit, coordinator):
def __init__(
self,
sensor_type,
name,
device_name,
device_serial_number,
serial_number,
unit,
coordinator,
):
"""Initialize Envoy entity."""
self._type = sensor_type
self._name = name
self._serial_number = serial_number
self._device_name = device_name
self._device_serial_number = device_serial_number
self._unit_of_measurement = unit
super().__init__(coordinator)
@ -173,6 +126,14 @@ class Envoy(CoordinatorEntity, SensorEntity):
"""Return the name of the sensor."""
return self._name
@property
def unique_id(self):
"""Return the unique id of the sensor."""
if self._serial_number:
return self._serial_number
if self._device_serial_number:
return f"{self._device_serial_number}_{self._type}"
@property
def state(self):
"""Return the state of the sensor."""
@ -214,3 +175,15 @@ class Envoy(CoordinatorEntity, SensorEntity):
return {"last_reported": value}
return None
@property
def device_info(self):
"""Return the device_info of the device."""
if not self._device_serial_number:
return None
return {
"identifiers": {(DOMAIN, str(self._device_serial_number))},
"name": self._device_name,
"model": "Envoy",
"manufacturer": "Enphase",
}

22
homeassistant/components/enphase_envoy/strings.json Normal file
View file

@ -0,0 +1,22 @@
{
"config": {
"flow_title": "Envoy {serial} ({host})",
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
}
}

22
homeassistant/components/enphase_envoy/translations/en.json Normal file
View file

@ -0,0 +1,22 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured"
},
"error": {
"cannot_connect": "Failed to connect",
"invalid_auth": "Invalid authentication",
"unknown": "Unexpected error"
},
"flow_title": "Envoy {serial} ({host})",
"step": {
"user": {
"data": {
"host": "Host",
"password": "Password",
"username": "Username"
}
}
}
}
}

1
homeassistant/generated/config_flows.py
View file

@ -62,6 +62,7 @@ FLOWS = [
"elkm1",
"emulated_roku",
"enocean",
"enphase_envoy",
"epson",
"esphome",
"faa_delays",

5
homeassistant/generated/zeroconf.py
View file

@ -54,6 +54,11 @@ ZEROCONF = {
"domain": "elgato"
}
],
"_enphase-envoy._tcp.local.": [
{
"domain": "enphase_envoy"
}
],
"_esphomelib._tcp.local.": [
{
"domain": "esphome"

3
requirements_test_all.txt
View file

@ -296,6 +296,9 @@ emulated_roku==0.2.1
# homeassistant.components.enocean
enocean==0.50
# homeassistant.components.enphase_envoy
envoy_reader==0.18.3
# homeassistant.components.season
ephem==3.7.7.0

1
tests/components/enphase_envoy/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Tests for the Enphase Envoy integration."""

304
tests/components/enphase_envoy/test_config_flow.py Normal file
View file

@ -0,0 +1,304 @@
"""Test the Enphase Envoy config flow."""
from unittest.mock import MagicMock, patch
import httpx
from homeassistant import config_entries, setup
from homeassistant.components.enphase_envoy.const import DOMAIN
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def test_form(hass: HomeAssistant) -> None:
"""Test we get the form."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["errors"] == {}
with patch(
"homeassistant.components.enphase_envoy.config_flow.EnvoyReader.getData",
return_value=True,
), patch(
"homeassistant.components.enphase_envoy.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = 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 result2["type"] == "create_entry"
assert result2["title"] == "Envoy"
assert result2["data"] == {
"host": "1.1.1.1",
"name": "Envoy",
"username": "test-username",
"password": "test-password",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_invalid_auth(hass: HomeAssistant) -> None:
"""Test we handle invalid auth."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.enphase_envoy.config_flow.EnvoyReader.getData",
side_effect=httpx.HTTPStatusError(
"any", request=MagicMock(), response=MagicMock()
),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
"username": "test-username",
"password": "test-password",
},
)
assert result2["type"] == "form"
assert result2["errors"] == {"base": "invalid_auth"}
async def test_form_cannot_connect(hass: HomeAssistant) -> None:
"""Test we handle cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.enphase_envoy.config_flow.EnvoyReader.getData",
side_effect=httpx.HTTPError("any", request=MagicMock()),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
"username": "test-username",
"password": "test-password",
},
)
assert result2["type"] == "form"
assert result2["errors"] == {"base": "cannot_connect"}
async def test_form_unknown_error(hass: HomeAssistant) -> None:
"""Test we handle unknown error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.enphase_envoy.config_flow.EnvoyReader.getData",
side_effect=ValueError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
"username": "test-username",
"password": "test-password",
},
)
assert result2["type"] == "form"
assert result2["errors"] == {"base": "unknown"}
async def test_import(hass: HomeAssistant) -> None:
"""Test we can import from yaml."""
await setup.async_setup_component(hass, "persistent_notification", {})
with patch(
"homeassistant.components.enphase_envoy.config_flow.EnvoyReader.getData",
return_value=True,
), patch(
"homeassistant.components.enphase_envoy.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": "import"},
data={
"ip_address": "1.1.1.1",
"name": "Pool Envoy",
"username": "test-username",
"password": "test-password",
},
)
await hass.async_block_till_done()
assert result2["type"] == "create_entry"
assert result2["title"] == "Pool Envoy"
assert result2["data"] == {
"host": "1.1.1.1",
"name": "Pool Envoy",
"username": "test-username",
"password": "test-password",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_zeroconf(hass: HomeAssistant) -> None:
"""Test we can setup from zeroconf."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": "zeroconf"},
data={
"properties": {"serialnum": "1234"},
"host": "1.1.1.1",
},
)
await hass.async_block_till_done()
assert result["type"] == "form"
assert result["step_id"] == "user"
with patch(
"homeassistant.components.enphase_envoy.config_flow.EnvoyReader.getData",
return_value=True,
), patch(
"homeassistant.components.enphase_envoy.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = 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 result2["type"] == "create_entry"
assert result2["title"] == "Envoy 1234"
assert result2["result"].unique_id == "1234"
assert result2["data"] == {
"host": "1.1.1.1",
"name": "Envoy 1234",
"username": "test-username",
"password": "test-password",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_host_already_exists(hass: HomeAssistant) -> None:
"""Test host already exists."""
await setup.async_setup_component(hass, "persistent_notification", {})
config_entry = MockConfigEntry(
domain=DOMAIN,
data={
"host": "1.1.1.1",
"name": "Envoy",
"username": "test-username",
"password": "test-password",
},
title="Envoy",
)
config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["errors"] == {}
with patch(
"homeassistant.components.enphase_envoy.config_flow.EnvoyReader.getData",
return_value=True,
):
result2 = 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 result2["type"] == "abort"
assert result2["reason"] == "already_configured"
async def test_zeroconf_serial_already_exists(hass: HomeAssistant) -> None:
"""Test serial number already exists from zeroconf."""
await setup.async_setup_component(hass, "persistent_notification", {})
config_entry = MockConfigEntry(
domain=DOMAIN,
data={
"host": "1.1.1.1",
"name": "Envoy",
"username": "test-username",
"password": "test-password",
},
unique_id="1234",
title="Envoy",
)
config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": "zeroconf"},
data={
"properties": {"serialnum": "1234"},
"host": "1.1.1.1",
},
)
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
async def test_zeroconf_host_already_exists(hass: HomeAssistant) -> None:
"""Test hosts already exists from zeroconf."""
await setup.async_setup_component(hass, "persistent_notification", {})
config_entry = MockConfigEntry(
domain=DOMAIN,
data={
"host": "1.1.1.1",
"name": "Envoy",
"username": "test-username",
"password": "test-password",
},
title="Envoy",
)
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.enphase_envoy.config_flow.EnvoyReader.getData",
return_value=True,
), patch(
"homeassistant.components.enphase_envoy.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": "zeroconf"},
data={
"properties": {"serialnum": "1234"},
"host": "1.1.1.1",
},
)
await hass.async_block_till_done()
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
assert config_entry.unique_id == "1234"
assert config_entry.title == "Envoy 1234"
assert len(mock_setup_entry.mock_calls) == 1