Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Prevent powerwall from switching addresses if its online (#82410) 

* Prevent powerwall from switching addresses if its online

If the wifi interface was discovered we would switch the
ip address in the entry to the wifi ip even if it was
connected via ethernet

* cover

* more cover
pull/82419/head
J. Nick Koston 2022-11-20 09:38:30 -06:00 committed by GitHub
parent 5c00b8b823
commit a1d9d7116c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 194 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
homeassistant/components/powerwall/__init__.py
View File

@ -18,7 +18,7 @@ from tesla_powerwall import (
from homeassistant.components import persistent_notification
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD, Platform
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
@ -221,6 +221,17 @@ def _fetch_powerwall_data(power_wall: Powerwall) -> PowerwallData:
)
@callback
def async_last_update_was_successful(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Return True if the last update was successful."""
return bool(
(domain_data := hass.data.get(DOMAIN))
and (entry_data := domain_data.get(entry.entry_id))
and (coordinator := entry_data.get(POWERWALL_COORDINATOR))
and coordinator.last_update_success
)
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

49
homeassistant/components/powerwall/config_flow.py
View File

@ -20,11 +20,18 @@ from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD
from homeassistant.data_entry_flow import FlowResult
from homeassistant.util.network import is_ip_address
from . import async_last_update_was_successful
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
ENTRY_FAILURE_STATES = {
config_entries.ConfigEntryState.SETUP_ERROR,
config_entries.ConfigEntryState.SETUP_RETRY,
}
def _login_and_fetch_site_info(
power_wall: Powerwall, password: str
) -> tuple[SiteInfo, str]:
@ -34,6 +41,17 @@ def _login_and_fetch_site_info(
return power_wall.get_site_info(), power_wall.get_gateway_din()
def _powerwall_is_reachable(ip_address: str, password: str) -> bool:
"""Check if the powerwall is reachable."""
try:
Powerwall(ip_address).login(password)
except AccessDeniedError:
return True
except PowerwallUnreachableError:
return False
return True
async def validate_input(
hass: core.HomeAssistant, data: dict[str, str]
) -> dict[str, str]:
@ -69,13 +87,31 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self.title: str | None = None
self.reauth_entry: config_entries.ConfigEntry | None = None
async def _async_powerwall_is_offline(
self, entry: config_entries.ConfigEntry
) -> bool:
"""Check if the power wall is offline.
We define offline by the config entry
is in a failure/retry state or the updates
are failing and the powerwall is unreachable
since device may be updating.
"""
ip_address = entry.data[CONF_IP_ADDRESS]
password = entry.data[CONF_PASSWORD]
return bool(
entry.state in ENTRY_FAILURE_STATES
or not async_last_update_was_successful(self.hass, entry)
) and not await self.hass.async_add_executor_job(
_powerwall_is_reachable, ip_address, password
)
async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult:
"""Handle dhcp discovery."""
self.ip_address = discovery_info.ip
gateway_din = discovery_info.hostname.upper()
# The hostname is the gateway_din (unique_id)
await self.async_set_unique_id(gateway_din)
self._abort_if_unique_id_configured(updates={CONF_IP_ADDRESS: self.ip_address})
for entry in self._async_current_entries(include_ignore=False):
if entry.data[CONF_IP_ADDRESS] == discovery_info.ip:
if entry.unique_id is not None and is_ip_address(entry.unique_id):
@ -86,6 +122,17 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self.hass.config_entries.async_reload(entry.entry_id)
)
return self.async_abort(reason="already_configured")
if entry.unique_id == gateway_din:
if await self._async_powerwall_is_offline(entry):
if self.hass.config_entries.async_update_entry(
entry, data={**entry.data, CONF_IP_ADDRESS: self.ip_address}
):
self.hass.async_create_task(
self.hass.config_entries.async_reload(entry.entry_id)
)
return self.async_abort(reason="already_configured")
# Still need to abort for ignored entries
self._abort_if_unique_id_configured()
self.context["title_placeholders"] = {
"name": gateway_din,
"ip_address": self.ip_address,

136
tests/components/powerwall/test_config_flow.py
View File

@ -1,6 +1,6 @@
"""Test the Powerwall config flow."""
from unittest.mock import patch
from unittest.mock import MagicMock, patch
from tesla_powerwall import (
AccessDeniedError,
@ -18,6 +18,7 @@ from .mocks import (
MOCK_GATEWAY_DIN,
_mock_powerwall_side_effect,
_mock_powerwall_site_name,
_mock_powerwall_with_fixtures,
)
from tests.common import MockConfigEntry
@ -351,7 +352,7 @@ async def test_dhcp_discovery_update_ip_address(hass):
unique_id=MOCK_GATEWAY_DIN,
)
entry.add_to_hass(hass)
mock_powerwall = await _mock_powerwall_site_name(hass, "Some site")
mock_powerwall = MagicMock(login=MagicMock(side_effect=PowerwallUnreachableError))
with patch(
"homeassistant.components.powerwall.config_flow.Powerwall",
@ -375,6 +376,70 @@ async def test_dhcp_discovery_update_ip_address(hass):
assert entry.data[CONF_IP_ADDRESS] == "1.1.1.1"
async def test_dhcp_discovery_does_not_update_ip_when_auth_fails(hass):
"""Test we do not switch to another interface when auth is failing."""
entry = MockConfigEntry(
domain=DOMAIN,
data=VALID_CONFIG,
unique_id=MOCK_GATEWAY_DIN,
)
entry.add_to_hass(hass)
mock_powerwall = MagicMock(login=MagicMock(side_effect=AccessDeniedError("any")))
with patch(
"homeassistant.components.powerwall.config_flow.Powerwall",
return_value=mock_powerwall,
), patch(
"homeassistant.components.powerwall.async_setup_entry",
return_value=True,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=dhcp.DhcpServiceInfo(
ip="1.1.1.1",
macaddress="AA:BB:CC:DD:EE:FF",
hostname=MOCK_GATEWAY_DIN.lower(),
),
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert entry.data[CONF_IP_ADDRESS] == "1.2.3.4"
async def test_dhcp_discovery_does_not_update_ip_when_auth_successful(hass):
"""Test we do not switch to another interface when auth is successful."""
entry = MockConfigEntry(
domain=DOMAIN,
data=VALID_CONFIG,
unique_id=MOCK_GATEWAY_DIN,
)
entry.add_to_hass(hass)
mock_powerwall = MagicMock(login=MagicMock(return_value=True))
with patch(
"homeassistant.components.powerwall.config_flow.Powerwall",
return_value=mock_powerwall,
), patch(
"homeassistant.components.powerwall.async_setup_entry",
return_value=True,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=dhcp.DhcpServiceInfo(
ip="1.1.1.1",
macaddress="AA:BB:CC:DD:EE:FF",
hostname=MOCK_GATEWAY_DIN.lower(),
),
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert entry.data[CONF_IP_ADDRESS] == "1.2.3.4"
async def test_dhcp_discovery_updates_unique_id(hass):
"""Test we can update the unique id from dhcp."""
entry = MockConfigEntry(
@ -406,3 +471,70 @@ async def test_dhcp_discovery_updates_unique_id(hass):
assert result["reason"] == "already_configured"
assert entry.data[CONF_IP_ADDRESS] == "1.2.3.4"
assert entry.unique_id == MOCK_GATEWAY_DIN
async def test_dhcp_discovery_updates_unique_id_when_entry_is_failed(hass):
"""Test we can update the unique id from dhcp in a failed state."""
entry = MockConfigEntry(
domain=DOMAIN,
data=VALID_CONFIG,
unique_id="1.2.3.4",
)
entry.add_to_hass(hass)
entry.state = config_entries.ConfigEntryState.SETUP_ERROR
mock_powerwall = await _mock_powerwall_site_name(hass, "Some site")
with patch(
"homeassistant.components.powerwall.config_flow.Powerwall",
return_value=mock_powerwall,
), patch(
"homeassistant.components.powerwall.async_setup_entry",
return_value=True,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=dhcp.DhcpServiceInfo(
ip="1.2.3.4",
macaddress="AA:BB:CC:DD:EE:FF",
hostname=MOCK_GATEWAY_DIN.lower(),
),
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert entry.data[CONF_IP_ADDRESS] == "1.2.3.4"
assert entry.unique_id == MOCK_GATEWAY_DIN
async def test_discovered_wifi_does_not_update_ip_if_is_still_online(hass) -> None:
"""Test a discovery does not update the ip unless the powerwall at the old ip is offline."""
entry = MockConfigEntry(
domain=DOMAIN,
data=VALID_CONFIG,
unique_id=MOCK_GATEWAY_DIN,
)
entry.add_to_hass(hass)
mock_powerwall = await _mock_powerwall_with_fixtures(hass)
with patch(
"homeassistant.components.powerwall.config_flow.Powerwall",
return_value=mock_powerwall,
), patch(
"homeassistant.components.powerwall.Powerwall", return_value=mock_powerwall
):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=dhcp.DhcpServiceInfo(
ip="1.2.3.5",
macaddress="AA:BB:CC:DD:EE:FF",
hostname=MOCK_GATEWAY_DIN.lower(),
),
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert entry.data[CONF_IP_ADDRESS] == "1.2.3.4"