jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 13:17:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add missing Opower tests (#147934)

Browse Source
This commit is contained in:
tronikos 2025-07-02 04:46:40 -07:00 committed by GitHub
parent f77e6cc8fc
commit bbe03dcab7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 668 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
tests/components/opower/conftest.py
View File

@ -1,5 +1,11 @@
"""Fixtures for the Opower integration tests."""
from collections.abc import Generator
from datetime import date
from unittest.mock import AsyncMock, Mock, patch
from opower import Account, Forecast, MeterType, ReadResolution, UnitOfMeasure
from opower.utilities.pge import PGE
import pytest
from homeassistant.components.opower.const import DOMAIN
@ -22,3 +28,76 @@ def mock_config_entry(hass: HomeAssistant) -> MockConfigEntry:
)
config_entry.add_to_hass(hass)
return config_entry
@pytest.fixture
def mock_opower_api() -> Generator[AsyncMock]:
"""Mock Opower API."""
with patch(
"homeassistant.components.opower.coordinator.Opower", autospec=True
) as mock_api:
api = mock_api.return_value
api.utility = PGE
api.async_get_accounts.return_value = [
Account(
customer=Mock(),
uuid="111111-uuid",
utility_account_id="111111",
id="111111",
meter_type=MeterType.ELEC,
read_resolution=ReadResolution.HOUR,
),
Account(
customer=Mock(),
uuid="222222-uuid",
utility_account_id="222222",
id="222222",
meter_type=MeterType.GAS,
read_resolution=ReadResolution.DAY,
),
]
api.async_get_forecast.return_value = [
Forecast(
account=Account(
customer=Mock(),
uuid="111111-uuid",
utility_account_id="111111",
id="111111",
meter_type=MeterType.ELEC,
read_resolution=ReadResolution.HOUR,
),
usage_to_date=100,
cost_to_date=20.0,
forecasted_usage=200,
forecasted_cost=40.0,
typical_usage=180,
typical_cost=36.0,
unit_of_measure=UnitOfMeasure.KWH,
start_date=date(2023, 1, 1),
end_date=date(2023, 1, 31),
current_date=date(2023, 1, 15),
),
Forecast(
account=Account(
customer=Mock(),
uuid="222222-uuid",
utility_account_id="222222",
id="222222",
meter_type=MeterType.GAS,
read_resolution=ReadResolution.DAY,
),
usage_to_date=50,
cost_to_date=15.0,
forecasted_usage=100,
forecasted_cost=30.0,
typical_usage=90,
typical_cost=27.0,
unit_of_measure=UnitOfMeasure.CCF,
start_date=date(2023, 1, 1),
end_date=date(2023, 1, 31),
current_date=date(2023, 1, 15),
),
]
api.async_get_cost_reads.return_value = []
yield api

177
tests/components/opower/snapshots/test_coordinator.ambr Normal file
View File

@ -0,0 +1,177 @@
# serializer version: 1
# name: test_coordinator_first_run
defaultdict({
'opower:pge_elec_111111_energy_compensation': list([
dict({
'end': 1672592400.0,
'start': 1672588800.0,
'state': 0.0,
'sum': 0.0,
}),
dict({
'end': 1672596000.0,
'start': 1672592400.0,
'state': 0.1,
'sum': 0.1,
}),
]),
'opower:pge_elec_111111_energy_consumption': list([
dict({
'end': 1672592400.0,
'start': 1672588800.0,
'state': 1.5,
'sum': 1.5,
}),
dict({
'end': 1672596000.0,
'start': 1672592400.0,
'state': 0.0,
'sum': 1.5,
}),
]),
'opower:pge_elec_111111_energy_cost': list([
dict({
'end': 1672592400.0,
'start': 1672588800.0,
'state': 0.5,
'sum': 0.5,
}),
dict({
'end': 1672596000.0,
'start': 1672592400.0,
'state': 0.0,
'sum': 0.5,
}),
]),
'opower:pge_elec_111111_energy_return': list([
dict({
'end': 1672592400.0,
'start': 1672588800.0,
'state': 0.0,
'sum': 0.0,
}),
dict({
'end': 1672596000.0,
'start': 1672592400.0,
'state': 0.5,
'sum': 0.5,
}),
]),
})
# ---
# name: test_coordinator_migration
defaultdict({
'opower:pge_elec_111111_energy_consumption': list([
dict({
'end': 1672592400.0,
'start': 1672588800.0,
'state': 1.5,
'sum': 1.5,
}),
dict({
'end': 1672596000.0,
'start': 1672592400.0,
'state': 0.0,
'sum': 1.5,
}),
]),
'opower:pge_elec_111111_energy_return': list([
dict({
'end': 1672592400.0,
'start': 1672588800.0,
'state': 0.0,
'sum': 0.0,
}),
dict({
'end': 1672596000.0,
'start': 1672592400.0,
'state': 0.5,
'sum': 0.5,
}),
]),
})
# ---
# name: test_coordinator_subsequent_run
defaultdict({
'opower:pge_elec_111111_energy_compensation': list([
dict({
'end': 1672592400.0,
'start': 1672588800.0,
'state': 0.0,
'sum': 0.0,
}),
dict({
'end': 1672596000.0,
'start': 1672592400.0,
'state': 0.1,
'sum': 0.1,
}),
dict({
'end': 1672599600.0,
'start': 1672596000.0,
'state': 0.0,
'sum': 0.1,
}),
]),
'opower:pge_elec_111111_energy_consumption': list([
dict({
'end': 1672592400.0,
'start': 1672588800.0,
'state': 1.5,
'sum': 1.5,
}),
dict({
'end': 1672596000.0,
'start': 1672592400.0,
'state': 0.0,
'sum': 1.5,
}),
dict({
'end': 1672599600.0,
'start': 1672596000.0,
'state': 2.0,
'sum': 3.5,
}),
]),
'opower:pge_elec_111111_energy_cost': list([
dict({
'end': 1672592400.0,
'start': 1672588800.0,
'state': 0.5,
'sum': 0.5,
}),
dict({
'end': 1672596000.0,
'start': 1672592400.0,
'state': 0.0,
'sum': 0.5,
}),
dict({
'end': 1672599600.0,
'start': 1672596000.0,
'state': 0.7,
'sum': 1.2,
}),
]),
'opower:pge_elec_111111_energy_return': list([
dict({
'end': 1672592400.0,
'start': 1672588800.0,
'state': 0.0,
'sum': 0.0,
}),
dict({
'end': 1672596000.0,
'start': 1672592400.0,
'state': 0.5,
'sum': 0.5,
}),
dict({
'end': 1672599600.0,
'start': 1672596000.0,
'state': 0.0,
'sum': 0.5,
}),
]),
})
# ---

236
tests/components/opower/test_coordinator.py Normal file
View File

@ -0,0 +1,236 @@
"""Tests for the Opower coordinator."""
from datetime import datetime
from unittest.mock import AsyncMock
from opower import CostRead
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.opower.const import DOMAIN
from homeassistant.components.opower.coordinator import OpowerCoordinator
from homeassistant.components.recorder import Recorder
from homeassistant.components.recorder.models import StatisticData, StatisticMetaData
from homeassistant.components.recorder.statistics import (
async_add_external_statistics,
get_last_statistics,
statistics_during_period,
)
from homeassistant.const import UnitOfEnergy
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from homeassistant.util import dt as dt_util
from tests.common import MockConfigEntry
from tests.components.recorder.common import async_wait_recording_done
async def test_coordinator_first_run(
recorder_mock: Recorder,
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_opower_api: AsyncMock,
snapshot: SnapshotAssertion,
) -> None:
"""Test the coordinator on its first run with no existing statistics."""
mock_opower_api.async_get_cost_reads.return_value = [
CostRead(
start_time=dt_util.as_utc(datetime(2023, 1, 1, 8)),
end_time=dt_util.as_utc(datetime(2023, 1, 1, 9)),
consumption=1.5,
provided_cost=0.5,
),
CostRead(
start_time=dt_util.as_utc(datetime(2023, 1, 1, 9)),
end_time=dt_util.as_utc(datetime(2023, 1, 1, 10)),
consumption=-0.5, # Grid return
provided_cost=-0.1, # Compensation
),
]
coordinator = OpowerCoordinator(hass, mock_config_entry)
await coordinator._async_update_data()
await async_wait_recording_done(hass)
# Check stats for electric account '111111'
stats = await hass.async_add_executor_job(
statistics_during_period,
hass,
dt_util.utc_from_timestamp(0),
None,
{
"opower:pge_elec_111111_energy_consumption",
"opower:pge_elec_111111_energy_return",
"opower:pge_elec_111111_energy_cost",
"opower:pge_elec_111111_energy_compensation",
},
"hour",
None,
{"state", "sum"},
)
assert stats == snapshot
async def test_coordinator_subsequent_run(
recorder_mock: Recorder,
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_opower_api: AsyncMock,
snapshot: SnapshotAssertion,
) -> None:
"""Test the coordinator correctly updates statistics on subsequent runs."""
# First run
mock_opower_api.async_get_cost_reads.return_value = [
CostRead(
start_time=dt_util.as_utc(datetime(2023, 1, 1, 8)),
end_time=dt_util.as_utc(datetime(2023, 1, 1, 9)),
consumption=1.5,
provided_cost=0.5,
),
CostRead(
start_time=dt_util.as_utc(datetime(2023, 1, 1, 9)),
end_time=dt_util.as_utc(datetime(2023, 1, 1, 10)),
consumption=-0.5,
provided_cost=-0.1,
),
]
coordinator = OpowerCoordinator(hass, mock_config_entry)
await coordinator._async_update_data()
await async_wait_recording_done(hass)
# Second run with updated data for one hour and new data for the next hour
mock_opower_api.async_get_cost_reads.return_value = [
CostRead(
start_time=dt_util.as_utc(datetime(2023, 1, 1, 9)), # Updated data
end_time=dt_util.as_utc(datetime(2023, 1, 1, 10)),
consumption=-1.0, # Was -0.5
provided_cost=-0.2, # Was -0.1
),
CostRead(
start_time=dt_util.as_utc(datetime(2023, 1, 1, 10)), # New data
end_time=dt_util.as_utc(datetime(2023, 1, 1, 11)),
consumption=2.0,
provided_cost=0.7,
),
]
await coordinator._async_update_data()
await async_wait_recording_done(hass)
# Check all stats
stats = await hass.async_add_executor_job(
statistics_during_period,
hass,
dt_util.utc_from_timestamp(0),
None,
{
"opower:pge_elec_111111_energy_consumption",
"opower:pge_elec_111111_energy_return",
"opower:pge_elec_111111_energy_cost",
"opower:pge_elec_111111_energy_compensation",
},
"hour",
None,
{"state", "sum"},
)
assert stats == snapshot
async def test_coordinator_subsequent_run_no_energy_data(
recorder_mock: Recorder,
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_opower_api: AsyncMock,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test the coordinator handles no recent usage/cost data."""
# First run
mock_opower_api.async_get_cost_reads.return_value = [
CostRead(
start_time=dt_util.as_utc(datetime(2023, 1, 1, 8)),
end_time=dt_util.as_utc(datetime(2023, 1, 1, 9)),
consumption=1.5,
provided_cost=0.5,
),
]
coordinator = OpowerCoordinator(hass, mock_config_entry)
await coordinator._async_update_data()
await async_wait_recording_done(hass)
# Second run with no data
mock_opower_api.async_get_cost_reads.return_value = []
coordinator = OpowerCoordinator(hass, mock_config_entry)
await coordinator._async_update_data()
assert "No recent usage/cost data. Skipping update" in caplog.text
# Verify no new stats were added by checking the sum remains 1.5
statistic_id = "opower:pge_elec_111111_energy_consumption"
stats = await hass.async_add_executor_job(
get_last_statistics, hass, 1, statistic_id, True, {"sum"}
)
assert stats[statistic_id][0]["sum"] == 1.5
async def test_coordinator_migration(
recorder_mock: Recorder,
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_opower_api: AsyncMock,
snapshot: SnapshotAssertion,
) -> None:
"""Test the one-time migration for return-to-grid statistics."""
# Setup: Create old-style consumption data with negative values
statistic_id = "opower:pge_elec_111111_energy_consumption"
metadata = StatisticMetaData(
has_sum=True,
name="Opower pge elec 111111 consumption",
source=DOMAIN,
statistic_id=statistic_id,
unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
)
statistics_to_add = [
StatisticData(
start=dt_util.as_utc(datetime(2023, 1, 1, 8)),
state=1.5,
sum=1.5,
),
StatisticData(
start=dt_util.as_utc(datetime(2023, 1, 1, 9)),
state=-0.5, # This should be migrated
sum=1.0,
),
]
async_add_external_statistics(hass, metadata, statistics_to_add)
await async_wait_recording_done(hass)
# When the coordinator runs, it should trigger the migration
# Don't need new cost reads for this test
mock_opower_api.async_get_cost_reads.return_value = []
coordinator = OpowerCoordinator(hass, mock_config_entry)
await coordinator._async_update_data()
await async_wait_recording_done(hass)
# Check that the stats have been migrated
stats = await hass.async_add_executor_job(
statistics_during_period,
hass,
dt_util.utc_from_timestamp(0),
None,
{
"opower:pge_elec_111111_energy_consumption",
"opower:pge_elec_111111_energy_return",
},
"hour",
None,
{"state", "sum"},
)
assert stats == snapshot
# Check that an issue was created
issue_registry = ir.async_get(hass)
issue = issue_registry.async_get_issue(DOMAIN, "return_to_grid_migration_111111")
assert issue is not None
assert issue.severity == ir.IssueSeverity.WARNING

116
tests/components/opower/test_init.py Normal file
View File

@ -0,0 +1,116 @@
"""Tests for the Opower integration."""
from unittest.mock import AsyncMock
from opower.exceptions import ApiException, CannotConnect, InvalidAuth
import pytest
from homeassistant.components.opower.const import DOMAIN
from homeassistant.components.recorder import Recorder
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def test_setup_unload_entry(
recorder_mock: Recorder,
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_opower_api: AsyncMock,
) -> None:
"""Test successful setup and unload of a config entry."""
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
mock_opower_api.async_login.assert_awaited_once()
mock_opower_api.async_get_forecast.assert_awaited_once()
mock_opower_api.async_get_accounts.assert_awaited_once()
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
assert not hass.data.get(DOMAIN)
@pytest.mark.parametrize(
("login_side_effect", "expected_state"),
[
(
CannotConnect(),
ConfigEntryState.SETUP_RETRY,
),
(
InvalidAuth(),
ConfigEntryState.SETUP_ERROR,
),
],
)
async def test_login_error(
recorder_mock: Recorder,
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_opower_api: AsyncMock,
login_side_effect: Exception,
expected_state: ConfigEntryState,
) -> None:
"""Test for login error."""
mock_opower_api.async_login.side_effect = login_side_effect
assert not await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is expected_state
async def test_get_forecast_error(
recorder_mock: Recorder,
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_opower_api: AsyncMock,
) -> None:
"""Test for API error when getting forecast."""
mock_opower_api.async_get_forecast.side_effect = ApiException(
message="forecast error", url=""
)
assert not await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_get_accounts_error(
recorder_mock: Recorder,
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_opower_api: AsyncMock,
) -> None:
"""Test for API error when getting accounts."""
mock_opower_api.async_get_accounts.side_effect = ApiException(
message="accounts error", url=""
)
assert not await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_get_cost_reads_error(
recorder_mock: Recorder,
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_opower_api: AsyncMock,
) -> None:
"""Test for API error when getting cost reads."""
mock_opower_api.async_get_cost_reads.side_effect = ApiException(
message="cost reads error", url=""
)
assert not await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY

60
tests/components/opower/test_sensor.py Normal file
View File

@ -0,0 +1,60 @@
"""Tests for the Opower sensor platform."""
from unittest.mock import AsyncMock
import pytest
from homeassistant.components.recorder import Recorder
from homeassistant.const import ATTR_UNIT_OF_MEASUREMENT, UnitOfEnergy, UnitOfVolume
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry
async def test_sensors(
recorder_mock: Recorder,
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_opower_api: AsyncMock,
) -> None:
"""Test the creation and values of Opower sensors."""
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entity_registry = er.async_get(hass)
# Check electric sensors
entry = entity_registry.async_get("sensor.current_bill_electric_usage_to_date")
assert entry
assert entry.unique_id == "pge_111111_elec_usage_to_date"
state = hass.states.get("sensor.current_bill_electric_usage_to_date")
assert state
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == UnitOfEnergy.KILO_WATT_HOUR
assert state.state == "100"
entry = entity_registry.async_get("sensor.current_bill_electric_cost_to_date")
assert entry
assert entry.unique_id == "pge_111111_elec_cost_to_date"
state = hass.states.get("sensor.current_bill_electric_cost_to_date")
assert state
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == "USD"
assert state.state == "20.0"
# Check gas sensors
entry = entity_registry.async_get("sensor.current_bill_gas_usage_to_date")
assert entry
assert entry.unique_id == "pge_222222_gas_usage_to_date"
state = hass.states.get("sensor.current_bill_gas_usage_to_date")
assert state
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == UnitOfVolume.CUBIC_METERS
# Convert 50 CCF to m³
assert float(state.state) == pytest.approx(50 * 2.83168, abs=1e-3)
entry = entity_registry.async_get("sensor.current_bill_gas_cost_to_date")
assert entry
assert entry.unique_id == "pge_222222_gas_cost_to_date"
state = hass.states.get("sensor.current_bill_gas_cost_to_date")
assert state
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == "USD"
assert state.state == "15.0"