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

Update Flick Electric API (#133475)

Browse Source
This commit is contained in:
Brynley McDonald 2025-01-01 02:28:24 +13:00 committed by GitHub
parent 4a9d545ffe
commit 9348569f90
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 1046 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
homeassistant/components/flick_electric/__init__.py
View File

@ -20,7 +20,8 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.helpers import aiohttp_client
from .const import CONF_TOKEN_EXPIRY, DOMAIN
from .const import CONF_ACCOUNT_ID, CONF_SUPPLY_NODE_REF, CONF_TOKEN_EXPIRY
from .coordinator import FlickConfigEntry, FlickElectricDataCoordinator
_LOGGER = logging.getLogger(__name__)
@ -29,24 +30,67 @@ CONF_ID_TOKEN = "id_token"
PLATFORMS = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: FlickConfigEntry) -> bool:
"""Set up Flick Electric from a config entry."""
auth = HassFlickAuth(hass, entry)
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = FlickAPI(auth)
coordinator = FlickElectricDataCoordinator(
hass, FlickAPI(auth), entry.data[CONF_SUPPLY_NODE_REF]
)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: FlickConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Migrate old entry."""
_LOGGER.debug(
"Migrating configuration from version %s.%s",
config_entry.version,
config_entry.minor_version,
)
if config_entry.version > 2:
return False
if config_entry.version == 1:
api = FlickAPI(HassFlickAuth(hass, config_entry))
accounts = await api.getCustomerAccounts()
active_accounts = [
account for account in accounts if account["status"] == "active"
]
# A single active account can be auto-migrated
if (len(active_accounts)) == 1:
account = active_accounts[0]
new_data = {**config_entry.data}
new_data[CONF_ACCOUNT_ID] = account["id"]
new_data[CONF_SUPPLY_NODE_REF] = account["main_consumer"]["supply_node_ref"]
hass.config_entries.async_update_entry(
config_entry,
title=account["address"],
unique_id=account["id"],
data=new_data,
version=2,
)
return True
config_entry.async_start_reauth(hass, data={**config_entry.data})
return False
return True
class HassFlickAuth(AbstractFlickAuth):

157
homeassistant/components/flick_electric/config_flow.py
View File

@ -1,14 +1,18 @@
"""Config Flow for Flick Electric integration."""
import asyncio
from collections.abc import Mapping
import logging
from typing import Any
from pyflick.authentication import AuthException, SimpleFlickAuth
from aiohttp import ClientResponseError
from pyflick import FlickAPI
from pyflick.authentication import AbstractFlickAuth, SimpleFlickAuth
from pyflick.const import DEFAULT_CLIENT_ID, DEFAULT_CLIENT_SECRET
from pyflick.types import APIException, AuthException, CustomerAccount
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlow, ConfigFlowResult
from homeassistant.const import (
CONF_CLIENT_ID,
CONF_CLIENT_SECRET,
@ -17,12 +21,18 @@ from homeassistant.const import (
)
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import aiohttp_client
from homeassistant.helpers.selector import (
SelectOptionDict,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
)
from .const import DOMAIN
from .const import CONF_ACCOUNT_ID, CONF_SUPPLY_NODE_REF, DOMAIN
_LOGGER = logging.getLogger(__name__)
DATA_SCHEMA = vol.Schema(
LOGIN_SCHEMA = vol.Schema(
{
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
@ -35,10 +45,13 @@ DATA_SCHEMA = vol.Schema(
class FlickConfigFlow(ConfigFlow, domain=DOMAIN):
"""Flick config flow."""
VERSION = 1
VERSION = 2
auth: AbstractFlickAuth
accounts: list[CustomerAccount]
data: dict[str, Any]
async def _validate_input(self, user_input):
auth = SimpleFlickAuth(
async def _validate_auth(self, user_input: Mapping[str, Any]) -> bool:
self.auth = SimpleFlickAuth(
username=user_input[CONF_USERNAME],
password=user_input[CONF_PASSWORD],
websession=aiohttp_client.async_get_clientsession(self.hass),
@ -48,22 +61,83 @@ class FlickConfigFlow(ConfigFlow, domain=DOMAIN):
try:
async with asyncio.timeout(60):
token = await auth.async_get_access_token()
except TimeoutError as err:
token = await self.auth.async_get_access_token()
except (TimeoutError, ClientResponseError) as err:
raise CannotConnect from err
except AuthException as err:
raise InvalidAuth from err
return token is not None
async def async_step_select_account(
self, user_input: Mapping[str, Any] | None = None
) -> ConfigFlowResult:
"""Ask user to select account."""
errors = {}
if user_input is not None and CONF_ACCOUNT_ID in user_input:
self.data[CONF_ACCOUNT_ID] = user_input[CONF_ACCOUNT_ID]
self.data[CONF_SUPPLY_NODE_REF] = self._get_supply_node_ref(
user_input[CONF_ACCOUNT_ID]
)
try:
# Ensure supply node is active
await FlickAPI(self.auth).getPricing(self.data[CONF_SUPPLY_NODE_REF])
except (APIException, ClientResponseError):
errors["base"] = "cannot_connect"
except AuthException:
# We should never get here as we have a valid token
return self.async_abort(reason="no_permissions")
else:
# Supply node is active
return await self._async_create_entry()
try:
self.accounts = await FlickAPI(self.auth).getCustomerAccounts()
except (APIException, ClientResponseError):
errors["base"] = "cannot_connect"
active_accounts = [a for a in self.accounts if a["status"] == "active"]
if len(active_accounts) == 0:
return self.async_abort(reason="no_accounts")
if len(active_accounts) == 1:
self.data[CONF_ACCOUNT_ID] = active_accounts[0]["id"]
self.data[CONF_SUPPLY_NODE_REF] = self._get_supply_node_ref(
active_accounts[0]["id"]
)
return await self._async_create_entry()
return self.async_show_form(
step_id="select_account",
data_schema=vol.Schema(
{
vol.Required(CONF_ACCOUNT_ID): SelectSelector(
SelectSelectorConfig(
options=[
SelectOptionDict(
value=account["id"], label=account["address"]
)
for account in active_accounts
],
mode=SelectSelectorMode.LIST,
)
)
}
),
errors=errors,
)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
self, user_input: Mapping[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle gathering login info."""
errors = {}
if user_input is not None:
try:
await self._validate_input(user_input)
await self._validate_auth(user_input)
except CannotConnect:
errors["base"] = "cannot_connect"
except InvalidAuth:
@ -72,20 +146,61 @@ class FlickConfigFlow(ConfigFlow, domain=DOMAIN):
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
await self.async_set_unique_id(
f"flick_electric_{user_input[CONF_USERNAME]}"
)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=f"Flick Electric: {user_input[CONF_USERNAME]}",
data=user_input,
)
self.data = dict(user_input)
return await self.async_step_select_account(user_input)
return self.async_show_form(
step_id="user", data_schema=DATA_SCHEMA, errors=errors
step_id="user", data_schema=LOGIN_SCHEMA, errors=errors
)
async def async_step_reauth(
self, user_input: Mapping[str, Any]
) -> ConfigFlowResult:
"""Handle re-authentication."""
self.data = {**user_input}
return await self.async_step_user(user_input)
async def _async_create_entry(self) -> ConfigFlowResult:
"""Create an entry for the flow."""
await self.async_set_unique_id(self.data[CONF_ACCOUNT_ID])
account = self._get_account(self.data[CONF_ACCOUNT_ID])
if self.source == SOURCE_REAUTH:
# Migration completed
if self._get_reauth_entry().version == 1:
self.hass.config_entries.async_update_entry(
self._get_reauth_entry(),
unique_id=self.unique_id,
data=self.data,
version=self.VERSION,
)
return self.async_update_reload_and_abort(
self._get_reauth_entry(),
unique_id=self.unique_id,
title=account["address"],
data=self.data,
)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=account["address"],
data=self.data,
)
def _get_account(self, account_id: str) -> CustomerAccount:
"""Get the account for the account ID."""
return next(a for a in self.accounts if a["id"] == account_id)
def _get_supply_node_ref(self, account_id: str) -> str:
"""Get the supply node ref for the account."""
return self._get_account(account_id)["main_consumer"][CONF_SUPPLY_NODE_REF]
class CannotConnect(HomeAssistantError):
"""Error to indicate we cannot connect."""

2
homeassistant/components/flick_electric/const.py
View File

@ -3,6 +3,8 @@
DOMAIN = "flick_electric"
CONF_TOKEN_EXPIRY = "expires"
CONF_ACCOUNT_ID = "account_id"
CONF_SUPPLY_NODE_REF = "supply_node_ref"
ATTR_START_AT = "start_at"
ATTR_END_AT = "end_at"

47
homeassistant/components/flick_electric/coordinator.py Normal file
View File

@ -0,0 +1,47 @@
"""Data Coordinator for Flick Electric."""
import asyncio
from datetime import timedelta
import logging
import aiohttp
from pyflick import FlickAPI, FlickPrice
from pyflick.types import APIException, AuthException
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=5)
type FlickConfigEntry = ConfigEntry[FlickElectricDataCoordinator]
class FlickElectricDataCoordinator(DataUpdateCoordinator[FlickPrice]):
"""Coordinator for flick power price."""
def __init__(
self, hass: HomeAssistant, api: FlickAPI, supply_node_ref: str
) -> None:
"""Initialize FlickElectricDataCoordinator."""
super().__init__(
hass,
_LOGGER,
name="Flick Electric",
update_interval=SCAN_INTERVAL,
)
self.supply_node_ref = supply_node_ref
self._api = api
async def _async_update_data(self) -> FlickPrice:
"""Fetch pricing data from Flick Electric."""
try:
async with asyncio.timeout(60):
return await self._api.getPricing(self.supply_node_ref)
except AuthException as err:
raise ConfigEntryAuthFailed from err
except (APIException, aiohttp.ClientResponseError) as err:
raise UpdateFailed from err

2
homeassistant/components/flick_electric/manifest.json
View File

@ -7,5 +7,5 @@
"integration_type": "service",
"iot_class": "cloud_polling",
"loggers": ["pyflick"],
"requirements": ["PyFlick==0.0.2"]
"requirements": ["PyFlick==1.1.2"]
}

64
homeassistant/components/flick_electric/sensor.py
View File

@ -1,74 +1,72 @@
"""Support for Flick Electric Pricing data."""
import asyncio
from datetime import timedelta
from decimal import Decimal
import logging
from typing import Any
from pyflick import FlickAPI, FlickPrice
from homeassistant.components.sensor import SensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CURRENCY_CENT, UnitOfEnergy
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util.dt import utcnow
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import ATTR_COMPONENTS, ATTR_END_AT, ATTR_START_AT, DOMAIN
from .const import ATTR_COMPONENTS, ATTR_END_AT, ATTR_START_AT
from .coordinator import FlickConfigEntry, FlickElectricDataCoordinator
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=5)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
hass: HomeAssistant,
entry: FlickConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Flick Sensor Setup."""
api: FlickAPI = hass.data[DOMAIN][entry.entry_id]
coordinator = entry.runtime_data
async_add_entities([FlickPricingSensor(api)], True)
async_add_entities([FlickPricingSensor(coordinator)])
class FlickPricingSensor(SensorEntity):
class FlickPricingSensor(CoordinatorEntity[FlickElectricDataCoordinator], SensorEntity):
"""Entity object for Flick Electric sensor."""
_attr_attribution = "Data provided by Flick Electric"
_attr_native_unit_of_measurement = f"{CURRENCY_CENT}/{UnitOfEnergy.KILO_WATT_HOUR}"
_attr_has_entity_name = True
_attr_translation_key = "power_price"
_attributes: dict[str, Any] = {}
def __init__(self, api: FlickAPI) -> None:
def __init__(self, coordinator: FlickElectricDataCoordinator) -> None:
"""Entity object for Flick Electric sensor."""
self._api: FlickAPI = api
self._price: FlickPrice = None
super().__init__(coordinator)
self._attr_unique_id = f"{coordinator.supply_node_ref}_pricing"
@property
def native_value(self):
def native_value(self) -> Decimal:
"""Return the state of the sensor."""
return self._price.price
# The API should return a unit price with quantity of 1.0 when no start/end time is provided
if self.coordinator.data.quantity != 1:
_LOGGER.warning(
"Unexpected quantity for unit price: %s", self.coordinator.data
)
return self.coordinator.data.cost
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return the state attributes."""
return self._attributes
components: dict[str, Decimal] = {}
async def async_update(self) -> None:
"""Get the Flick Pricing data from the web service."""
if self._price and self._price.end_at >= utcnow():
return # Power price data is still valid
async with asyncio.timeout(60):
self._price = await self._api.getPricing()
_LOGGER.debug("Pricing data: %s", self._price)
self._attributes[ATTR_START_AT] = self._price.start_at
self._attributes[ATTR_END_AT] = self._price.end_at
for component in self._price.components:
for component in self.coordinator.data.components:
if component.charge_setter not in ATTR_COMPONENTS:
_LOGGER.warning("Found unknown component: %s", component.charge_setter)
continue
self._attributes[component.charge_setter] = float(component.value)
components[component.charge_setter] = component.value
return {
ATTR_START_AT: self.coordinator.data.start_at,
ATTR_END_AT: self.coordinator.data.end_at,
**components,
}

11
homeassistant/components/flick_electric/strings.json
View File

@ -9,6 +9,12 @@
"client_id": "Client ID (optional)",
"client_secret": "Client Secret (optional)"
}
},
"select_account": {
"title": "Select account",
"data": {
"account_id": "Account"
}
}
},
"error": {
@ -17,7 +23,10 @@
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"no_permissions": "Cannot get pricing for this account. Please check user permissions.",
"no_accounts": "No services are active on this Flick account"
}
},
"entity": {

2
requirements_all.txt
View File

@ -48,7 +48,7 @@ ProgettiHWSW==0.1.3
PyChromecast==14.0.5
# homeassistant.components.flick_electric
PyFlick==0.0.2
PyFlick==1.1.2
# homeassistant.components.flume
PyFlume==0.6.5

2
requirements_test_all.txt
View File

@ -45,7 +45,7 @@ ProgettiHWSW==0.1.3
PyChromecast==14.0.5
# homeassistant.components.flick_electric
PyFlick==0.0.2
PyFlick==1.1.2
# homeassistant.components.flume
PyFlume==0.6.5

50
tests/components/flick_electric/__init__.py
View File

@ -1 +1,51 @@
"""Tests for the Flick Electric integration."""
from pyflick.types import FlickPrice
from homeassistant.components.flick_electric.const import (
CONF_ACCOUNT_ID,
CONF_SUPPLY_NODE_REF,
)
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
CONF = {
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_ACCOUNT_ID: "1234",
CONF_SUPPLY_NODE_REF: "123",
}
def _mock_flick_price():
return FlickPrice(
{
"cost": "0.25",
"quantity": "1.0",
"status": "final",
"start_at": "2024-01-01T00:00:00Z",
"end_at": "2024-01-01T00:00:00Z",
"type": "flat",
"components": [
{
"charge_method": "kwh",
"charge_setter": "network",
"value": "1.00",
"single_unit_price": "1.00",
"quantity": "1.0",
"unit_code": "NZD",
"charge_per": "kwh",
"flow_direction": "import",
},
{
"charge_method": "kwh",
"charge_setter": "nonsupported",
"value": "1.00",
"single_unit_price": "1.00",
"quantity": "1.0",
"unit_code": "NZD",
"charge_per": "kwh",
"flow_direction": "import",
},
],
}
)

594
tests/components/flick_electric/test_config_flow.py
View File

@ -3,29 +3,37 @@
from unittest.mock import patch
from pyflick.authentication import AuthException
from pyflick.types import APIException
from homeassistant import config_entries
from homeassistant.components.flick_electric.const import DOMAIN
from homeassistant.components.flick_electric.const import (
CONF_ACCOUNT_ID,
CONF_SUPPLY_NODE_REF,
DOMAIN,
)
from homeassistant.config_entries import ConfigFlowResult
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
from . import CONF, _mock_flick_price
CONF = {CONF_USERNAME: "test-username", CONF_PASSWORD: "test-password"}
from tests.common import MockConfigEntry
async def _flow_submit(hass: HomeAssistant) -> ConfigFlowResult:
return await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data=CONF,
data={
CONF_USERNAME: CONF[CONF_USERNAME],
CONF_PASSWORD: CONF[CONF_PASSWORD],
},
)
async def test_form(hass: HomeAssistant) -> None:
"""Test we get the form."""
"""Test we get the form with only one, with no account picker."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
@ -38,6 +46,21 @@ async def test_form(hass: HomeAssistant) -> None:
"homeassistant.components.flick_electric.config_flow.SimpleFlickAuth.async_get_access_token",
return_value="123456789abcdef",
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getCustomerAccounts",
return_value=[
{
"id": "1234",
"status": "active",
"address": "123 Fake St",
"main_consumer": {"supply_node_ref": "123"},
}
],
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getPricing",
return_value=_mock_flick_price(),
),
patch(
"homeassistant.components.flick_electric.async_setup_entry",
return_value=True,
@ -45,29 +68,293 @@ async def test_form(hass: HomeAssistant) -> None:
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
CONF,
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert result2["title"] == "Flick Electric: test-username"
assert result2["title"] == "123 Fake St"
assert result2["data"] == CONF
assert result2["result"].unique_id == "1234"
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_duplicate_login(hass: HomeAssistant) -> None:
"""Test uniqueness of username."""
async def test_form_multi_account(hass: HomeAssistant) -> None:
"""Test the form when multiple accounts are available."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
with (
patch(
"homeassistant.components.flick_electric.config_flow.SimpleFlickAuth.async_get_access_token",
return_value="123456789abcdef",
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getCustomerAccounts",
return_value=[
{
"id": "1234",
"status": "active",
"address": "123 Fake St",
"main_consumer": {"supply_node_ref": "123"},
},
{
"id": "5678",
"status": "active",
"address": "456 Fake St",
"main_consumer": {"supply_node_ref": "456"},
},
],
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getPricing",
return_value=_mock_flick_price(),
),
patch(
"homeassistant.components.flick_electric.async_setup_entry",
return_value=True,
) as mock_setup_entry,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "select_account"
assert len(mock_setup_entry.mock_calls) == 0
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{"account_id": "5678"},
)
await hass.async_block_till_done()
assert result3["type"] is FlowResultType.CREATE_ENTRY
assert result3["title"] == "456 Fake St"
assert result3["data"] == {
**CONF,
CONF_SUPPLY_NODE_REF: "456",
CONF_ACCOUNT_ID: "5678",
}
assert result3["result"].unique_id == "5678"
assert len(mock_setup_entry.mock_calls) == 1
async def test_reauth_token(hass: HomeAssistant) -> None:
"""Test reauth flow when username/password is wrong."""
entry = MockConfigEntry(
domain=DOMAIN,
data=CONF,
title="Flick Electric: test-username",
unique_id="flick_electric_test-username",
data={**CONF},
title="123 Fake St",
unique_id="1234",
version=2,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.flick_electric.config_flow.SimpleFlickAuth.async_get_access_token",
return_value="123456789abcdef",
with (
patch(
"homeassistant.components.flick_electric.config_flow.SimpleFlickAuth.async_get_access_token",
side_effect=AuthException,
),
):
result = await entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "invalid_auth"}
assert result["step_id"] == "user"
with (
patch(
"homeassistant.components.flick_electric.config_flow.SimpleFlickAuth.async_get_access_token",
return_value="123456789abcdef",
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getCustomerAccounts",
return_value=[
{
"id": "1234",
"status": "active",
"address": "123 Fake St",
"main_consumer": {"supply_node_ref": "123"},
},
],
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getPricing",
return_value=_mock_flick_price(),
),
patch(
"homeassistant.config_entries.ConfigEntries.async_update_entry",
return_value=True,
) as mock_update_entry,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "reauth_successful"
assert len(mock_update_entry.mock_calls) > 0
async def test_form_reauth_migrate(hass: HomeAssistant) -> None:
"""Test reauth flow for v1 with single account."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
title="123 Fake St",
unique_id="test-username",
version=1,
)
entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.flick_electric.config_flow.SimpleFlickAuth.async_get_access_token",
return_value="123456789abcdef",
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getCustomerAccounts",
return_value=[
{
"id": "1234",
"status": "active",
"address": "123 Fake St",
"main_consumer": {"supply_node_ref": "123"},
},
],
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getPricing",
return_value=_mock_flick_price(),
),
):
result = await entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert entry.version == 2
assert entry.unique_id == "1234"
assert entry.data == CONF
async def test_form_reauth_migrate_multi_account(hass: HomeAssistant) -> None:
"""Test the form when multiple accounts are available."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
title="123 Fake St",
unique_id="test-username",
version=1,
)
entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.flick_electric.config_flow.SimpleFlickAuth.async_get_access_token",
return_value="123456789abcdef",
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getCustomerAccounts",
return_value=[
{
"id": "1234",
"status": "active",
"address": "123 Fake St",
"main_consumer": {"supply_node_ref": "123"},
},
{
"id": "5678",
"status": "active",
"address": "456 Fake St",
"main_consumer": {"supply_node_ref": "456"},
},
],
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getPricing",
return_value=_mock_flick_price(),
),
):
result = await entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "select_account"
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"account_id": "5678"},
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "reauth_successful"
assert entry.version == 2
assert entry.unique_id == "5678"
assert entry.data == {
**CONF,
CONF_ACCOUNT_ID: "5678",
CONF_SUPPLY_NODE_REF: "456",
}
async def test_form_duplicate_account(hass: HomeAssistant) -> None:
"""Test uniqueness for account_id."""
entry = MockConfigEntry(
domain=DOMAIN,
data={**CONF, CONF_ACCOUNT_ID: "1234", CONF_SUPPLY_NODE_REF: "123"},
title="123 Fake St",
unique_id="1234",
version=2,
)
entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.flick_electric.config_flow.SimpleFlickAuth.async_get_access_token",
return_value="123456789abcdef",
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getCustomerAccounts",
return_value=[
{
"id": "1234",
"status": "active",
"address": "123 Fake St",
"main_consumer": {"supply_node_ref": "123"},
}
],
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getPricing",
return_value=_mock_flick_price(),
),
):
result = await _flow_submit(hass)
@ -109,3 +396,280 @@ async def test_form_generic_exception(hass: HomeAssistant) -> None:
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "unknown"}
async def test_form_select_account_cannot_connect(hass: HomeAssistant) -> None:
"""Test we handle connection errors for select account."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
with (
patch(
"homeassistant.components.flick_electric.config_flow.SimpleFlickAuth.async_get_access_token",
return_value="123456789abcdef",
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getCustomerAccounts",
return_value=[
{
"id": "1234",
"status": "active",
"address": "123 Fake St",
"main_consumer": {"supply_node_ref": "123"},
},
{
"id": "5678",
"status": "active",
"address": "456 Fake St",
"main_consumer": {"supply_node_ref": "456"},
},
],
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getPricing",
side_effect=APIException,
),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "select_account"
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{"account_id": "5678"},
)
assert result3["type"] is FlowResultType.FORM
assert result3["step_id"] == "select_account"
assert result3["errors"] == {"base": "cannot_connect"}
async def test_form_select_account_invalid_auth(hass: HomeAssistant) -> None:
"""Test we handle auth errors for select account."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
with (
patch(
"homeassistant.components.flick_electric.config_flow.SimpleFlickAuth.async_get_access_token",
return_value="123456789abcdef",
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getCustomerAccounts",
return_value=[
{
"id": "1234",
"status": "active",
"address": "123 Fake St",
"main_consumer": {"supply_node_ref": "123"},
},
{
"id": "5678",
"status": "active",
"address": "456 Fake St",
"main_consumer": {"supply_node_ref": "456"},
},
],
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getPricing",
side_effect=AuthException,
),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "select_account"
with (
patch(
"homeassistant.components.flick_electric.config_flow.SimpleFlickAuth.async_get_access_token",
side_effect=AuthException,
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getCustomerAccounts",
side_effect=AuthException,
),
):
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{"account_id": "5678"},
)
assert result3["type"] is FlowResultType.ABORT
assert result3["reason"] == "no_permissions"
async def test_form_select_account_failed_to_connect(hass: HomeAssistant) -> None:
"""Test we handle connection errors for select account."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
with (
patch(
"homeassistant.components.flick_electric.config_flow.SimpleFlickAuth.async_get_access_token",
return_value="123456789abcdef",
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getCustomerAccounts",
return_value=[
{
"id": "1234",
"status": "active",
"address": "123 Fake St",
"main_consumer": {"supply_node_ref": "123"},
},
{
"id": "5678",
"status": "active",
"address": "456 Fake St",
"main_consumer": {"supply_node_ref": "456"},
},
],
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getPricing",
side_effect=AuthException,
),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "select_account"
with (
patch(
"homeassistant.components.flick_electric.config_flow.SimpleFlickAuth.async_get_access_token",
return_value="123456789abcdef",
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getCustomerAccounts",
side_effect=APIException,
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getPricing",
side_effect=APIException,
),
):
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{"account_id": "5678"},
)
assert result3["type"] is FlowResultType.FORM
assert result3["errors"] == {"base": "cannot_connect"}
with (
patch(
"homeassistant.components.flick_electric.config_flow.SimpleFlickAuth.async_get_access_token",
return_value="123456789abcdef",
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getCustomerAccounts",
return_value=[
{
"id": "1234",
"status": "active",
"address": "123 Fake St",
"main_consumer": {"supply_node_ref": "123"},
},
{
"id": "5678",
"status": "active",
"address": "456 Fake St",
"main_consumer": {"supply_node_ref": "456"},
},
],
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getPricing",
return_value=_mock_flick_price(),
),
patch(
"homeassistant.components.flick_electric.async_setup_entry",
return_value=True,
) as mock_setup_entry,
):
result4 = await hass.config_entries.flow.async_configure(
result3["flow_id"],
{"account_id": "5678"},
)
assert result4["type"] is FlowResultType.CREATE_ENTRY
assert result4["title"] == "456 Fake St"
assert result4["data"] == {
**CONF,
CONF_SUPPLY_NODE_REF: "456",
CONF_ACCOUNT_ID: "5678",
}
assert result4["result"].unique_id == "5678"
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_select_account_no_accounts(hass: HomeAssistant) -> None:
"""Test we handle connection errors for select account."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
with (
patch(
"homeassistant.components.flick_electric.config_flow.SimpleFlickAuth.async_get_access_token",
return_value="123456789abcdef",
),
patch(
"homeassistant.components.flick_electric.config_flow.FlickAPI.getCustomerAccounts",
return_value=[
{
"id": "1234",
"status": "closed",
"address": "123 Fake St",
"main_consumer": {"supply_node_ref": "123"},
},
],
),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "no_accounts"

135
tests/components/flick_electric/test_init.py Normal file
View File

@ -0,0 +1,135 @@
"""Test the Flick Electric config flow."""
from unittest.mock import patch
from pyflick.authentication import AuthException
from homeassistant.components.flick_electric.const import CONF_ACCOUNT_ID, DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from . import CONF, _mock_flick_price
from tests.common import MockConfigEntry
async def test_init_auth_failure_triggers_auth(hass: HomeAssistant) -> None:
"""Test reauth flow is triggered when username/password is wrong."""
with (
patch(
"homeassistant.components.flick_electric.HassFlickAuth.async_get_access_token",
side_effect=AuthException,
),
):
entry = MockConfigEntry(
domain=DOMAIN,
data={**CONF},
title="123 Fake St",
unique_id="1234",
version=2,
)
entry.add_to_hass(hass)
# Ensure setup fails
assert not await hass.config_entries.async_setup(entry.entry_id)
assert entry.state is ConfigEntryState.SETUP_ERROR
# Ensure reauth flow is triggered
await hass.async_block_till_done()
assert len(hass.config_entries.flow.async_progress()) == 1
async def test_init_migration_single_account(hass: HomeAssistant) -> None:
"""Test migration with single account."""
with (
patch(
"homeassistant.components.flick_electric.HassFlickAuth.async_get_access_token",
return_value="123456789abcdef",
),
patch(
"homeassistant.components.flick_electric.FlickAPI.getCustomerAccounts",
return_value=[
{
"id": "1234",
"status": "active",
"address": "123 Fake St",
"main_consumer": {"supply_node_ref": "123"},
}
],
),
patch(
"homeassistant.components.flick_electric.FlickAPI.getPricing",
return_value=_mock_flick_price(),
),
):
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_USERNAME: CONF[CONF_USERNAME],
CONF_PASSWORD: CONF[CONF_PASSWORD],
},
title=CONF_USERNAME,
unique_id=CONF_USERNAME,
version=1,
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert len(hass.config_entries.flow.async_progress()) == 0
assert entry.state is ConfigEntryState.LOADED
assert entry.version == 2
assert entry.unique_id == CONF[CONF_ACCOUNT_ID]
assert entry.data == CONF
async def test_init_migration_multi_account_reauth(hass: HomeAssistant) -> None:
"""Test migration triggers reauth with multiple accounts."""
with (
patch(
"homeassistant.components.flick_electric.HassFlickAuth.async_get_access_token",
return_value="123456789abcdef",
),
patch(
"homeassistant.components.flick_electric.FlickAPI.getCustomerAccounts",
return_value=[
{
"id": "1234",
"status": "active",
"address": "123 Fake St",
"main_consumer": {"supply_node_ref": "123"},
},
{
"id": "5678",
"status": "active",
"address": "456 Fake St",
"main_consumer": {"supply_node_ref": "456"},
},
],
),
patch(
"homeassistant.components.flick_electric.FlickAPI.getPricing",
return_value=_mock_flick_price(),
),
):
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_USERNAME: CONF[CONF_USERNAME],
CONF_PASSWORD: CONF[CONF_PASSWORD],
},
title=CONF_USERNAME,
unique_id=CONF_USERNAME,
version=1,
)
entry.add_to_hass(hass)
# ensure setup fails
assert not await hass.config_entries.async_setup(entry.entry_id)
assert entry.state is ConfigEntryState.MIGRATION_ERROR
await hass.async_block_till_done()
# Ensure reauth flow is triggered
await hass.async_block_till_done()
assert len(hass.config_entries.flow.async_progress()) == 1