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

Clean up srp_energy (#86822) 

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
pull/91313/head^2
Brig Lamoreaux 2023-04-12 14:47:37 -07:00 committed by GitHub
parent 7446ff478f
commit ff1fd86b91
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 430 additions and 390 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
homeassistant/components/srp_energy/__init__.py
View File

@ -1,34 +1,31 @@
"""The SRP Energy integration."""
import logging
from srpenergy.client import SrpEnergyClient
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ID, CONF_PASSWORD, CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from .const import SRP_ENERGY_DOMAIN
_LOGGER = logging.getLogger(__name__)
from .const import DOMAIN, LOGGER
PLATFORMS = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up the SRP Energy component from a config entry."""
# Store an SrpEnergyClient object for your srp_energy to access
try:
srp_energy_client = SrpEnergyClient(
entry.data.get(CONF_ID),
entry.data.get(CONF_USERNAME),
entry.data.get(CONF_PASSWORD),
)
hass.data[SRP_ENERGY_DOMAIN] = srp_energy_client
except Exception as ex:
_LOGGER.error("Unable to connect to Srp Energy: %s", str(ex))
raise ConfigEntryNotReady from ex
api_account_id: str = entry.data[CONF_ID]
api_username: str = entry.data[CONF_USERNAME]
api_password: str = entry.data[CONF_PASSWORD]
LOGGER.debug("Configuring client using account_id %s", api_account_id)
api_instance = SrpEnergyClient(
api_account_id,
api_username,
api_password,
)
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = api_instance
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
@ -37,7 +34,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
# unload srp client
hass.data[SRP_ENERGY_DOMAIN] = None
# Remove config entry
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

98
homeassistant/components/srp_energy/config_flow.py
View File

@ -1,65 +1,85 @@
"""Config flow for SRP Energy."""
import logging
from __future__ import annotations
from typing import Any
from srpenergy.client import SrpEnergyClient
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_ID, CONF_NAME, CONF_PASSWORD, CONF_USERNAME
from homeassistant.const import CONF_ID, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
from homeassistant.exceptions import HomeAssistantError
from .const import CONF_IS_TOU, DEFAULT_NAME, SRP_ENERGY_DOMAIN
_LOGGER = logging.getLogger(__name__)
from .const import CONF_IS_TOU, DEFAULT_NAME, DOMAIN, LOGGER
class ConfigFlow(config_entries.ConfigFlow, domain=SRP_ENERGY_DOMAIN):
"""Handle a config flow for SRP Energy."""
async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, Any]:
"""Validate the user input allows us to connect.
Data has the keys from DATA_SCHEMA with values provided by the user.
"""
srp_client = SrpEnergyClient(
data[CONF_ID],
data[CONF_USERNAME],
data[CONF_PASSWORD],
)
is_valid = await hass.async_add_executor_job(srp_client.validate)
LOGGER.debug("Is user input valid: %s", is_valid)
if not is_valid:
raise InvalidAuth
return is_valid
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle an SRP Energy config flow."""
VERSION = 1
config = {
vol.Required(CONF_ID): str,
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): str,
vol.Optional(CONF_IS_TOU, default=False): bool,
}
async def async_step_user(self, user_input=None):
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initialized by the user."""
errors = {}
default_title: str = DEFAULT_NAME
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
if user_input is not None:
if self.hass.config.location_name:
default_title = self.hass.config.location_name
if user_input:
try:
srp_client = SrpEnergyClient(
user_input[CONF_ID],
user_input[CONF_USERNAME],
user_input[CONF_PASSWORD],
)
is_valid = await self.hass.async_add_executor_job(srp_client.validate)
if is_valid:
return self.async_create_entry(
title=user_input[CONF_NAME], data=user_input
)
errors["base"] = "invalid_auth"
await validate_input(self.hass, user_input)
except ValueError:
# Thrown when the account id is malformed
errors["base"] = "invalid_account"
except InvalidAuth:
errors["base"] = "invalid_auth"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
LOGGER.exception("Unexpected exception")
return self.async_abort(reason="unknown")
else:
return self.async_create_entry(title=default_title, data=user_input)
return self.async_show_form(
step_id="user", data_schema=vol.Schema(self.config), errors=errors
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_ID): str,
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
vol.Optional(CONF_IS_TOU, default=False): bool,
}
),
errors=errors or {},
)
async def async_step_import(self, import_config):
"""Import from config."""
# Validate config values
return await self.async_step_user(user_input=import_config)
class InvalidAuth(HomeAssistantError):
"""Error to indicate there is invalid auth."""

11
homeassistant/components/srp_energy/const.py
View File

@ -1,13 +1,16 @@
"""Constants for the SRP Energy integration."""
from datetime import timedelta
import logging
SRP_ENERGY_DOMAIN = "srp_energy"
DEFAULT_NAME = "SRP Energy"
LOGGER = logging.getLogger(__package__)
DOMAIN = "srp_energy"
DEFAULT_NAME = "Home"
CONF_IS_TOU = "is_tou"
PHOENIX_TIME_ZONE = "America/Phoenix"
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=1440)
SENSOR_NAME = "Usage"
SENSOR_NAME = "Energy Usage"
SENSOR_TYPE = "usage"

72
homeassistant/components/srp_energy/sensor.py
View File

@ -1,6 +1,7 @@
"""Support for SRP Energy Sensor."""
from datetime import datetime, timedelta
import logging
from __future__ import annotations
from datetime import timedelta
import async_timeout
from requests.exceptions import ConnectionError as ConnectError, HTTPError, Timeout
@ -14,28 +15,29 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import UnitOfEnergy
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util import dt as dt_util
from .const import (
CONF_IS_TOU,
DEFAULT_NAME,
DOMAIN,
LOGGER,
MIN_TIME_BETWEEN_UPDATES,
PHOENIX_TIME_ZONE,
SENSOR_NAME,
SENSOR_TYPE,
SRP_ENERGY_DOMAIN,
)
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the SRP Energy Usage sensor."""
# API object stored here by __init__.py
is_time_of_use = False
api = hass.data[SRP_ENERGY_DOMAIN]
if entry and entry.data:
is_time_of_use = entry.data["is_tou"]
api = hass.data[DOMAIN][entry.entry_id]
is_time_of_use = entry.data[CONF_IS_TOU]
async def async_update_data():
"""Fetch data from API endpoint.
@ -43,10 +45,13 @@ async def async_setup_entry(
This is the place to pre-process the data to lookup tables
so entities can quickly look up their data.
"""
LOGGER.debug("async_update_data enter")
try:
# Fetch srp_energy data
start_date = datetime.now() + timedelta(days=-1)
end_date = datetime.now()
phx_time_zone = dt_util.get_time_zone(PHOENIX_TIME_ZONE)
end_date = dt_util.now(phx_time_zone)
start_date = end_date - timedelta(days=1)
async with async_timeout.timeout(10):
hourly_usage = await hass.async_add_executor_job(
api.usage,
@ -55,9 +60,22 @@ async def async_setup_entry(
is_time_of_use,
)
LOGGER.debug(
"async_update_data: Received %s record(s) from %s to %s",
len(hourly_usage) if hourly_usage else "None",
start_date,
end_date,
)
previous_daily_usage = 0.0
for _, _, _, kwh, _ in hourly_usage:
previous_daily_usage += float(kwh)
LOGGER.debug(
"async_update_data: previous_daily_usage %s",
previous_daily_usage,
)
return previous_daily_usage
except TimeoutError as timeout_err:
raise UpdateFailed("Timeout communicating with API") from timeout_err
@ -66,7 +84,7 @@ async def async_setup_entry(
coordinator = DataUpdateCoordinator(
hass,
_LOGGER,
LOGGER,
name="sensor",
update_method=async_update_data,
update_interval=MIN_TIME_BETWEEN_UPDATES,
@ -85,7 +103,7 @@ class SrpEntity(SensorEntity):
_attr_icon = "mdi:flash"
_attr_should_poll = False
def __init__(self, coordinator):
def __init__(self, coordinator) -> None:
"""Initialize the SrpEntity class."""
self._name = SENSOR_NAME
self.type = SENSOR_TYPE
@ -94,46 +112,32 @@ class SrpEntity(SensorEntity):
self._state = None
@property
def name(self):
def name(self) -> str:
"""Return the name of the sensor."""
return f"{DEFAULT_NAME} {self._name}"
@property
def unique_id(self):
"""Return sensor unique_id."""
return self.type
@property
def native_value(self):
def native_value(self) -> StateType:
"""Return the state of the device."""
if self._state:
return f"{self._state:.2f}"
return None
return self.coordinator.data
@property
def native_unit_of_measurement(self):
def native_unit_of_measurement(self) -> str:
"""Return the unit of measurement of this entity, if any."""
return self._unit_of_measurement
@property
def usage(self):
"""Return entity state."""
if self.coordinator.data:
return f"{self.coordinator.data:.2f}"
return None
@property
def available(self):
def available(self) -> bool:
"""Return if entity is available."""
return self.coordinator.last_update_success
@property
def device_class(self):
def device_class(self) -> SensorDeviceClass:
"""Return the device class."""
return SensorDeviceClass.ENERGY
@property
def state_class(self):
def state_class(self) -> SensorStateClass:
"""Return the state class."""
return SensorStateClass.TOTAL_INCREASING

123
tests/components/srp_energy/__init__.py
View File

@ -1,54 +1,81 @@
"""Tests for the SRP Energy integration."""
from unittest.mock import patch
from homeassistant import config_entries
from homeassistant.components import srp_energy
from homeassistant.const import CONF_ID, CONF_NAME, CONF_PASSWORD, CONF_USERNAME
from homeassistant.components.srp_energy.const import CONF_IS_TOU
from homeassistant.const import CONF_ID, CONF_PASSWORD, CONF_USERNAME
from tests.common import MockConfigEntry
ACCNT_ID = "123456789"
ACCNT_IS_TOU = False
ACCNT_USERNAME = "abba"
ACCNT_PASSWORD = "ana"
ACCNT_NAME = "Home"
ENTRY_OPTIONS = {}
ENTRY_CONFIG = {
CONF_NAME: "Test",
CONF_ID: "123456789",
CONF_USERNAME: "abba",
CONF_PASSWORD: "ana",
srp_energy.const.CONF_IS_TOU: False,
TEST_USER_INPUT = {
CONF_ID: ACCNT_ID,
CONF_USERNAME: ACCNT_USERNAME,
CONF_PASSWORD: ACCNT_PASSWORD,
CONF_IS_TOU: ACCNT_IS_TOU,
}
async def init_integration(
hass,
config=None,
options=None,
entry_id="1",
source=config_entries.SOURCE_USER,
side_effect=None,
usage=None,
):
"""Set up the srp_energy integration in Home Assistant."""
if not config:
config = ENTRY_CONFIG
if not options:
options = ENTRY_OPTIONS
config_entry = MockConfigEntry(
domain=srp_energy.SRP_ENERGY_DOMAIN,
source=source,
data=config,
options=options,
entry_id=entry_id,
)
with patch("srpenergy.client.SrpEnergyClient"), patch(
"homeassistant.components.srp_energy.SrpEnergyClient", side_effect=side_effect
), patch("srpenergy.client.SrpEnergyClient.usage", return_value=usage), patch(
"homeassistant.components.srp_energy.SrpEnergyClient.usage", return_value=usage
):
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return config_entry
MOCK_USAGE = [
("7/31/2022", "00:00 AM", "2022-07-31T00:00:00", "1.2", "0.19"),
("7/31/2022", "01:00 AM", "2022-07-31T01:00:00", "1.3", "0.20"),
("7/31/2022", "02:00 AM", "2022-07-31T02:00:00", "1.1", "0.17"),
("7/31/2022", "03:00 AM", "2022-07-31T03:00:00", "1.2", "0.18"),
("7/31/2022", "04:00 AM", "2022-07-31T04:00:00", "0.8", "0.13"),
("7/31/2022", "05:00 AM", "2022-07-31T05:00:00", "0.9", "0.14"),
("7/31/2022", "06:00 AM", "2022-07-31T06:00:00", "1.6", "0.24"),
("7/31/2022", "07:00 AM", "2022-07-31T07:00:00", "3.7", "0.53"),
("7/31/2022", "08:00 AM", "2022-07-31T08:00:00", "1.0", "0.16"),
("7/31/2022", "09:00 AM", "2022-07-31T09:00:00", "0.7", "0.12"),
("7/31/2022", "10:00 AM", "2022-07-31T10:00:00", "1.9", "0.28"),
("7/31/2022", "11:00 AM", "2022-07-31T11:00:00", "4.3", "0.61"),
("7/31/2022", "12:00 PM", "2022-07-31T12:00:00", "2.0", "0.29"),
("7/31/2022", "01:00 PM", "2022-07-31T13:00:00", "3.9", "0.55"),
("7/31/2022", "02:00 PM", "2022-07-31T14:00:00", "5.3", "0.75"),
("7/31/2022", "03:00 PM", "2022-07-31T15:00:00", "5.0", "0.70"),
("7/31/2022", "04:00 PM", "2022-07-31T16:00:00", "2.2", "0.31"),
("7/31/2022", "05:00 PM", "2022-07-31T17:00:00", "2.6", "0.37"),
("7/31/2022", "06:00 PM", "2022-07-31T18:00:00", "4.5", "0.64"),
("7/31/2022", "07:00 PM", "2022-07-31T19:00:00", "2.5", "0.35"),
("7/31/2022", "08:00 PM", "2022-07-31T20:00:00", "2.9", "0.42"),
("7/31/2022", "09:00 PM", "2022-07-31T21:00:00", "2.2", "0.32"),
("7/31/2022", "10:00 PM", "2022-07-31T22:00:00", "2.1", "0.30"),
("7/31/2022", "11:00 PM", "2022-07-31T23:00:00", "2.0", "0.28"),
("8/01/2022", "00:00 AM", "2022-08-01T00:00:00", "1.8", "0.26"),
("8/01/2022", "01:00 AM", "2022-08-01T01:00:00", "1.7", "0.26"),
("8/01/2022", "02:00 AM", "2022-08-01T02:00:00", "1.7", "0.26"),
("8/01/2022", "03:00 AM", "2022-08-01T03:00:00", "0.8", "0.14"),
("8/01/2022", "04:00 AM", "2022-08-01T04:00:00", "1.2", "0.19"),
("8/01/2022", "05:00 AM", "2022-08-01T05:00:00", "1.6", "0.23"),
("8/01/2022", "06:00 AM", "2022-08-01T06:00:00", "1.2", "0.18"),
("8/01/2022", "07:00 AM", "2022-08-01T07:00:00", "3.1", "0.44"),
("8/01/2022", "08:00 AM", "2022-08-01T08:00:00", "2.5", "0.35"),
("8/01/2022", "09:00 AM", "2022-08-01T09:00:00", "3.3", "0.47"),
("8/01/2022", "10:00 AM", "2022-08-01T10:00:00", "2.6", "0.37"),
("8/01/2022", "11:00 AM", "2022-08-01T11:00:00", "0.8", "0.13"),
("8/01/2022", "12:00 PM", "2022-08-01T12:00:00", "0.6", "0.11"),
("8/01/2022", "01:00 PM", "2022-08-01T13:00:00", "6.4", "0.9"),
("8/01/2022", "02:00 PM", "2022-08-01T14:00:00", "3.6", "0.52"),
("8/01/2022", "03:00 PM", "2022-08-01T15:00:00", "5.5", "0.79"),
("8/01/2022", "04:00 PM", "2022-08-01T16:00:00", "3", "0.43"),
("8/01/2022", "05:00 PM", "2022-08-01T17:00:00", "5", "0.71"),
("8/01/2022", "06:00 PM", "2022-08-01T18:00:00", "4.4", "0.63"),
("8/01/2022", "07:00 PM", "2022-08-01T19:00:00", "3.8", "0.54"),
("8/01/2022", "08:00 PM", "2022-08-01T20:00:00", "3.6", "0.51"),
("8/01/2022", "09:00 PM", "2022-08-01T21:00:00", "2.9", "0.4"),
("8/01/2022", "10:00 PM", "2022-08-01T22:00:00", "3.4", "0.49"),
("8/01/2022", "11:00 PM", "2022-08-01T23:00:00", "2.9", "0.41"),
("8/02/2022", "00:00 AM", "2022-08-02T00:00:00", "2", "0.3"),
("8/02/2022", "01:00 AM", "2022-08-02T01:00:00", "2", "0.29"),
("8/02/2022", "02:00 AM", "2022-08-02T02:00:00", "1.9", "0.28"),
("8/02/2022", "03:00 AM", "2022-08-02T03:00:00", "1.8", "0.27"),
("8/02/2022", "04:00 AM", "2022-08-02T04:00:00", "1.8", "0.26"),
("8/02/2022", "05:00 AM", "2022-08-02T05:00:00", "1.6", "0.23"),
("8/02/2022", "06:00 AM", "2022-08-02T06:00:00", "0.8", "0.14"),
("8/02/2022", "07:00 AM", "2022-08-02T07:00:00", "4", "0.56"),
("8/02/2022", "08:00 AM", "2022-08-02T08:00:00", "2.4", "0.34"),
("8/02/2022", "09:00 AM", "2022-08-02T09:00:00", "4.1", "0.58"),
("8/02/2022", "10:00 AM", "2022-08-02T10:00:00", "2.6", "0.37"),
("8/02/2022", "11:00 AM", "2022-08-02T11:00:00", "0.5", "0.1"),
("8/02/2022", "00:00 AM", "2022-08-02T12:00:00", "1", "0.16"),
]

91
tests/components/srp_energy/conftest.py Normal file
View File

@ -0,0 +1,91 @@
"""Fixtures for Srp Energy integration tests."""
from __future__ import annotations
from collections.abc import Generator
import datetime as dt
from unittest.mock import MagicMock, patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from homeassistant.components.srp_energy.const import DOMAIN, PHOENIX_TIME_ZONE
from homeassistant.core import HomeAssistant
import homeassistant.util.dt as dt_util
from . import MOCK_USAGE, TEST_USER_INPUT
from tests.common import MockConfigEntry
@pytest.fixture(name="setup_hass_config", autouse=True)
def fixture_setup_hass_config(hass: HomeAssistant) -> None:
"""Set up things to be run when tests are started."""
hass.config.latitude = 33.27
hass.config.longitude = 112
hass.config.set_time_zone(PHOENIX_TIME_ZONE)
@pytest.fixture(name="hass_tz_info")
def fixture_hass_tz_info(hass: HomeAssistant, setup_hass_config) -> dt.tzinfo | None:
"""Return timezone info for the hass timezone."""
return dt_util.get_time_zone(hass.config.time_zone)
@pytest.fixture(name="test_date")
def fixture_test_date(hass: HomeAssistant, hass_tz_info) -> dt.datetime | None:
"""Return test datetime for the hass timezone."""
test_date = dt.datetime(2022, 8, 2, 0, 0, 0, 0, tzinfo=hass_tz_info)
return test_date
@pytest.fixture(name="mock_config_entry")
def fixture_mock_config_entry() -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
domain=DOMAIN,
data=TEST_USER_INPUT,
)
@pytest.fixture(name="mock_srp_energy")
def fixture_mock_srp_energy() -> Generator[None, MagicMock, None]:
"""Return a mocked SrpEnergyClient client."""
with patch(
"homeassistant.components.srp_energy.SrpEnergyClient", autospec=True
) as srp_energy_mock:
client = srp_energy_mock.return_value
client.validate.return_value = True
client.usage.return_value = MOCK_USAGE
yield client
@pytest.fixture(name="mock_srp_energy_config_flow")
def fixture_mock_srp_energy_config_flow() -> Generator[None, MagicMock, None]:
"""Return a mocked config_flow SrpEnergyClient client."""
with patch(
"homeassistant.components.srp_energy.config_flow.SrpEnergyClient", autospec=True
) as srp_energy_mock:
client = srp_energy_mock.return_value
client.validate.return_value = True
client.usage.return_value = MOCK_USAGE
yield client
@pytest.fixture
async def init_integration(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
test_date: dt.datetime,
mock_config_entry: MockConfigEntry,
mock_srp_energy,
mock_srp_energy_config_flow,
) -> MockConfigEntry:
"""Set up the Srp Energy integration for testing."""
freezer.move_to(test_date)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
return mock_config_entry

198
tests/components/srp_energy/test_config_flow.py
View File

@ -1,120 +1,126 @@
"""Test the SRP Energy config flow."""
from unittest.mock import patch
from unittest.mock import MagicMock, patch
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.srp_energy.const import CONF_IS_TOU, SRP_ENERGY_DOMAIN
from homeassistant import config_entries
from homeassistant.components.srp_energy.const import CONF_IS_TOU, DOMAIN
from homeassistant.const import CONF_ID, CONF_PASSWORD, CONF_SOURCE, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from . import ENTRY_CONFIG, init_integration
from . import ACCNT_ID, ACCNT_IS_TOU, ACCNT_PASSWORD, ACCNT_USERNAME, TEST_USER_INPUT
from tests.common import MockConfigEntry
async def test_form(hass: HomeAssistant) -> None:
"""Test user config."""
# First get the form
async def test_show_form(
hass: HomeAssistant, mock_srp_energy_config_flow: MagicMock, capsys
) -> None:
"""Test show configuration form."""
result = await hass.config_entries.flow.async_init(
SRP_ENERGY_DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN, context={CONF_SOURCE: config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
# Fill submit form data for config entry
with patch(
"homeassistant.components.srp_energy.config_flow.SrpEnergyClient"
), patch(
"homeassistant.components.srp_energy.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=ENTRY_CONFIG,
flow_id=result["flow_id"], user_input=TEST_USER_INPUT
)
assert result["type"] == "create_entry"
assert result["title"] == "Test"
assert result["data"][CONF_IS_TOU] is False
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_invalid_auth(hass: HomeAssistant) -> None:
"""Test user config with invalid auth."""
result = await hass.config_entries.flow.async_init(
SRP_ENERGY_DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.srp_energy.config_flow.SrpEnergyClient.validate",
return_value=False,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=ENTRY_CONFIG,
)
assert result["errors"]["base"] == "invalid_auth"
async def test_form_value_error(hass: HomeAssistant) -> None:
"""Test user config that throws a value error."""
result = await hass.config_entries.flow.async_init(
SRP_ENERGY_DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.srp_energy.config_flow.SrpEnergyClient",
side_effect=ValueError(),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=ENTRY_CONFIG,
)
assert result["errors"]["base"] == "invalid_account"
async def test_form_unknown_exception(hass: HomeAssistant) -> None:
"""Test user config that throws an unknown exception."""
result = await hass.config_entries.flow.async_init(
SRP_ENERGY_DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.srp_energy.config_flow.SrpEnergyClient",
side_effect=Exception(),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=ENTRY_CONFIG,
)
assert result["errors"]["base"] == "unknown"
async def test_config(hass: HomeAssistant) -> None:
"""Test handling of configuration imported."""
with patch(
"homeassistant.components.srp_energy.config_flow.SrpEnergyClient"
), patch(
"homeassistant.components.srp_energy.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
SRP_ENERGY_DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=ENTRY_CONFIG,
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "test home"
assert "data" in result
assert result["data"][CONF_ID] == ACCNT_ID
assert result["data"][CONF_USERNAME] == ACCNT_USERNAME
assert result["data"][CONF_PASSWORD] == ACCNT_PASSWORD
assert result["data"][CONF_IS_TOU] == ACCNT_IS_TOU
captured = capsys.readouterr()
assert "myaccount.srpnet.com" not in captured.err
assert len(mock_setup_entry.mock_calls) == 1
async def test_integration_already_configured(hass: HomeAssistant) -> None:
"""Test integration is already configured."""
await init_integration(hass)
async def test_form_invalid_account(
hass: HomeAssistant,
mock_srp_energy_config_flow: MagicMock,
) -> None:
"""Test flow to handle invalid account error."""
mock_srp_energy_config_flow.validate.side_effect = ValueError
result = await hass.config_entries.flow.async_init(
SRP_ENERGY_DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN, context={CONF_SOURCE: config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
result = await hass.config_entries.flow.async_configure(
flow_id=result["flow_id"], user_input=TEST_USER_INPUT
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": "invalid_account"}
async def test_form_invalid_auth(
hass: HomeAssistant,
mock_srp_energy_config_flow: MagicMock,
) -> None:
"""Test flow to handle invalid authentication error."""
mock_srp_energy_config_flow.validate.return_value = False
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: config_entries.SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
flow_id=result["flow_id"], user_input=TEST_USER_INPUT
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": "invalid_auth"}
async def test_form_unknown_error(
hass: HomeAssistant,
mock_srp_energy_config_flow: MagicMock,
) -> None:
"""Test flow to handle invalid authentication error."""
mock_srp_energy_config_flow.validate.side_effect = Exception
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: config_entries.SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
flow_id=result["flow_id"], user_input=TEST_USER_INPUT
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "unknown"
async def test_flow_entry_already_configured(
hass: HomeAssistant, init_integration: MockConfigEntry
) -> None:
"""Test user input for config_entry that already exists."""
user_input = {
CONF_ID: init_integration.data[CONF_ID],
CONF_USERNAME: "abba2",
CONF_PASSWORD: "ana2",
CONF_IS_TOU: False,
}
assert user_input[CONF_ID] == ACCNT_ID
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: config_entries.SOURCE_USER}, data=user_input
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "single_instance_allowed"

30
tests/components/srp_energy/test_init.py
View File

@ -1,28 +1,16 @@
"""Tests for Srp Energy component Init."""
from homeassistant import config_entries
from homeassistant.components import srp_energy
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from . import init_integration
async def test_setup_entry(hass: HomeAssistant, init_integration) -> None:
"""Test setup entry."""
assert init_integration.state == ConfigEntryState.LOADED
async def test_setup_entry(hass: HomeAssistant) -> None:
"""Test setup entry fails if deCONZ is not available."""
config_entry = await init_integration(hass)
assert config_entry.state == config_entries.ConfigEntryState.LOADED
assert hass.data[srp_energy.SRP_ENERGY_DOMAIN]
async def test_unload_entry(hass: HomeAssistant) -> None:
async def test_unload_entry(hass: HomeAssistant, init_integration) -> None:
"""Test being able to unload an entry."""
config_entry = await init_integration(hass)
assert hass.data[srp_energy.SRP_ENERGY_DOMAIN]
assert init_integration.state == ConfigEntryState.LOADED
assert await srp_energy.async_unload_entry(hass, config_entry)
assert not hass.data[srp_energy.SRP_ENERGY_DOMAIN]
async def test_async_setup_entry_with_exception(hass: HomeAssistant) -> None:
"""Test exception when SrpClient can't load."""
await init_integration(hass, side_effect=Exception())
assert srp_energy.SRP_ENERGY_DOMAIN not in hass.data
assert await hass.config_entries.async_unload(init_integration.entry_id)
await hass.async_block_till_done()

156
tests/components/srp_energy/test_sensor.py
View File

@ -1,135 +1,39 @@
"""Tests for the srp_energy sensor platform."""
from unittest.mock import MagicMock
from homeassistant.components.sensor import SensorDeviceClass, SensorStateClass
from homeassistant.components.srp_energy.const import (
DEFAULT_NAME,
SENSOR_NAME,
SENSOR_TYPE,
SRP_ENERGY_DOMAIN,
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import (
ATTR_ATTRIBUTION,
ATTR_DEVICE_CLASS,
ATTR_ICON,
ATTR_UNIT_OF_MEASUREMENT,
UnitOfEnergy,
)
from homeassistant.components.srp_energy.sensor import SrpEntity, async_setup_entry
from homeassistant.const import UnitOfEnergy
from homeassistant.core import HomeAssistant
async def test_async_setup_entry(hass: HomeAssistant) -> None:
"""Test the sensor."""
fake_async_add_entities = MagicMock()
fake_srp_energy_client = MagicMock()
fake_srp_energy_client.usage.return_value = [{1, 2, 3, 1.999, 4}]
fake_config = MagicMock(
data={
"name": "SRP Energy",
"is_tou": False,
"id": "0123456789",
"username": "testuser@example.com",
"password": "mypassword",
}
async def test_loading_sensors(hass: HomeAssistant, init_integration) -> None:
"""Test the srp energy sensors."""
# Validate the Config Entry was initialized
assert init_integration.state == ConfigEntryState.LOADED
# Check sensors were loaded
assert len(hass.states.async_all()) == 1
async def test_srp_entity(hass: HomeAssistant, init_integration) -> None:
"""Test the SrpEntity."""
usage_state = hass.states.get("sensor.home_energy_usage")
assert usage_state.state == "150.8"
# Validate attributions
assert (
usage_state.attributes.get("state_class") is SensorStateClass.TOTAL_INCREASING
)
hass.data[SRP_ENERGY_DOMAIN] = fake_srp_energy_client
await async_setup_entry(hass, fake_config, fake_async_add_entities)
async def test_async_setup_entry_timeout_error(hass: HomeAssistant) -> None:
"""Test fetching usage data. Failed the first time because was too get response."""
fake_async_add_entities = MagicMock()
fake_srp_energy_client = MagicMock()
fake_srp_energy_client.usage.return_value = [{1, 2, 3, 1.999, 4}]
fake_config = MagicMock(
data={
"name": "SRP Energy",
"is_tou": False,
"id": "0123456789",
"username": "testuser@example.com",
"password": "mypassword",
}
assert usage_state.attributes.get(ATTR_ATTRIBUTION) == "Powered by SRP Energy"
assert (
usage_state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== UnitOfEnergy.KILO_WATT_HOUR
)
hass.data[SRP_ENERGY_DOMAIN] = fake_srp_energy_client
fake_srp_energy_client.usage.side_effect = TimeoutError()
await async_setup_entry(hass, fake_config, fake_async_add_entities)
assert not fake_async_add_entities.call_args[0][0][
0
].coordinator.last_update_success
async def test_async_setup_entry_connect_error(hass: HomeAssistant) -> None:
"""Test fetching usage data. Failed the first time because was too get response."""
fake_async_add_entities = MagicMock()
fake_srp_energy_client = MagicMock()
fake_srp_energy_client.usage.return_value = [{1, 2, 3, 1.999, 4}]
fake_config = MagicMock(
data={
"name": "SRP Energy",
"is_tou": False,
"id": "0123456789",
"username": "testuser@example.com",
"password": "mypassword",
}
)
hass.data[SRP_ENERGY_DOMAIN] = fake_srp_energy_client
fake_srp_energy_client.usage.side_effect = ValueError()
await async_setup_entry(hass, fake_config, fake_async_add_entities)
assert not fake_async_add_entities.call_args[0][0][
0
].coordinator.last_update_success
async def test_srp_entity(hass: HomeAssistant) -> None:
"""Test the SrpEntity."""
fake_coordinator = MagicMock(data=1.99999999999)
srp_entity = SrpEntity(fake_coordinator)
srp_entity.hass = hass
assert srp_entity is not None
assert srp_entity.name == f"{DEFAULT_NAME} {SENSOR_NAME}"
assert srp_entity.unique_id == SENSOR_TYPE
assert srp_entity.state is None
assert srp_entity.unit_of_measurement == UnitOfEnergy.KILO_WATT_HOUR
assert srp_entity.icon == "mdi:flash"
assert srp_entity.usage == "2.00"
assert srp_entity.should_poll is False
assert srp_entity.attribution == "Powered by SRP Energy"
assert srp_entity.available is not None
assert srp_entity.device_class is SensorDeviceClass.ENERGY
assert srp_entity.state_class is SensorStateClass.TOTAL_INCREASING
await srp_entity.async_added_to_hass()
assert srp_entity.state is not None
assert fake_coordinator.async_add_listener.called
assert not fake_coordinator.async_add_listener.data.called
async def test_srp_entity_no_data(hass: HomeAssistant) -> None:
"""Test the SrpEntity."""
fake_coordinator = MagicMock(data=False)
srp_entity = SrpEntity(fake_coordinator)
srp_entity.hass = hass
assert srp_entity.extra_state_attributes is None
async def test_srp_entity_no_coord_data(hass: HomeAssistant) -> None:
"""Test the SrpEntity."""
fake_coordinator = MagicMock(data=False)
srp_entity = SrpEntity(fake_coordinator)
srp_entity.hass = hass
assert srp_entity.usage is None
async def test_srp_entity_async_update(hass: HomeAssistant) -> None:
"""Test the SrpEntity."""
async def async_magic():
pass
MagicMock.__await__ = lambda x: async_magic().__await__()
fake_coordinator = MagicMock(data=False)
srp_entity = SrpEntity(fake_coordinator)
srp_entity.hass = hass
await srp_entity.async_update()
assert fake_coordinator.async_request_refresh.called
assert usage_state.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.ENERGY
assert usage_state.attributes.get(ATTR_ICON) == "mdi:flash"