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

Add forecast service to amberelectric (#144848)

Browse Source 
Co-authored-by: G Johansson <goran.johansson@shiftit.se>
This commit is contained in:
Myles Eftos 2025-07-16 00:32:59 +10:00 committed by GitHub
parent c058561162
commit 087a938a7d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
16 changed files with 879 additions and 217 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -2,11 +2,22 @@
import amberelectric
from homeassistant.components.sensor import ConfigType
from homeassistant.const import CONF_API_TOKEN
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from .const import CONF_SITE_ID, PLATFORMS
from .const import CONF_SITE_ID, DOMAIN, PLATFORMS
from .coordinator import AmberConfigEntry, AmberUpdateCoordinator
from .services import setup_services
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Amber component."""
setup_services(hass)
return True
async def async_setup_entry(hass: HomeAssistant, entry: AmberConfigEntry) -> bool:

12
homeassistant/components/amberelectric/const.py
View File

@ -1,14 +1,24 @@
"""Amber Electric Constants."""
import logging
from typing import Final
from homeassistant.const import Platform
DOMAIN = "amberelectric"
DOMAIN: Final = "amberelectric"
CONF_SITE_NAME = "site_name"
CONF_SITE_ID = "site_id"
ATTR_CONFIG_ENTRY_ID = "config_entry_id"
ATTR_CHANNEL_TYPE = "channel_type"
ATTRIBUTION = "Data provided by Amber Electric"
LOGGER = logging.getLogger(__package__)
PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
SERVICE_GET_FORECASTS = "get_forecasts"
GENERAL_CHANNEL = "general"
CONTROLLED_LOAD_CHANNEL = "controlled_load"
FEED_IN_CHANNEL = "feed_in"

25
homeassistant/components/amberelectric/coordinator.py
View File

@ -10,7 +10,6 @@ from amberelectric.models.actual_interval import ActualInterval
from amberelectric.models.channel import ChannelType
from amberelectric.models.current_interval import CurrentInterval
from amberelectric.models.forecast_interval import ForecastInterval
from amberelectric.models.price_descriptor import PriceDescriptor
from amberelectric.rest import ApiException
from homeassistant.config_entries import ConfigEntry
@ -18,6 +17,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import LOGGER
from .helpers import normalize_descriptor
type AmberConfigEntry = ConfigEntry[AmberUpdateCoordinator]
@ -49,27 +49,6 @@ def is_feed_in(interval: ActualInterval | CurrentInterval | ForecastInterval) ->
return interval.channel_type == ChannelType.FEEDIN
def normalize_descriptor(descriptor: PriceDescriptor | None) -> str | None:
"""Return the snake case versions of descriptor names. Returns None if the name is not recognized."""
if descriptor is None:
return None
if descriptor.value == "spike":
return "spike"
if descriptor.value == "high":
return "high"
if descriptor.value == "neutral":
return "neutral"
if descriptor.value == "low":
return "low"
if descriptor.value == "veryLow":
return "very_low"
if descriptor.value == "extremelyLow":
return "extremely_low"
if descriptor.value == "negative":
return "negative"
return None
class AmberUpdateCoordinator(DataUpdateCoordinator):
"""AmberUpdateCoordinator - In charge of downloading the data for a site, which all the sensors read."""
@ -103,7 +82,7 @@ class AmberUpdateCoordinator(DataUpdateCoordinator):
"grid": {},
}
try:
data = self._api.get_current_prices(self.site_id, next=48)
data = self._api.get_current_prices(self.site_id, next=288)
intervals = [interval.actual_instance for interval in data]
except ApiException as api_exception:
raise UpdateFailed("Missing price data, skipping update") from api_exception

25
homeassistant/components/amberelectric/helpers.py Normal file
View File

@ -0,0 +1,25 @@
"""Formatting helpers used to convert things."""
from amberelectric.models.price_descriptor import PriceDescriptor
DESCRIPTOR_MAP: dict[str, str] = {
PriceDescriptor.SPIKE: "spike",
PriceDescriptor.HIGH: "high",
PriceDescriptor.NEUTRAL: "neutral",
PriceDescriptor.LOW: "low",
PriceDescriptor.VERYLOW: "very_low",
PriceDescriptor.EXTREMELYLOW: "extremely_low",
PriceDescriptor.NEGATIVE: "negative",
}
def normalize_descriptor(descriptor: PriceDescriptor | None) -> str | None:
"""Return the snake case versions of descriptor names. Returns None if the name is not recognized."""
if descriptor in DESCRIPTOR_MAP:
return DESCRIPTOR_MAP[descriptor]
return None
def format_cents_to_dollars(cents: float) -> float:
"""Return a formatted conversion from cents to dollars."""
return round(cents / 100, 2)

5
homeassistant/components/amberelectric/icons.json
View File

@ -22,5 +22,10 @@
}
}
}
},
"services": {
"get_forecasts": {
"service": "mdi:transmission-tower"
}
}
}

8
homeassistant/components/amberelectric/sensor.py
View File

@ -23,16 +23,12 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import ATTRIBUTION
from .coordinator import AmberConfigEntry, AmberUpdateCoordinator, normalize_descriptor
from .coordinator import AmberConfigEntry, AmberUpdateCoordinator
from .helpers import format_cents_to_dollars, normalize_descriptor
UNIT = f"{CURRENCY_DOLLAR}/{UnitOfEnergy.KILO_WATT_HOUR}"
def format_cents_to_dollars(cents: float) -> float:
"""Return a formatted conversion from cents to dollars."""
return round(cents / 100, 2)
def friendly_channel_type(channel_type: str) -> str:
"""Return a human readable version of the channel type."""
if channel_type == "controlled_load":

121
homeassistant/components/amberelectric/services.py Normal file
View File

@ -0,0 +1,121 @@
"""Amber Electric Service class."""
from amberelectric.models.channel import ChannelType
import voluptuous as vol
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import (
HomeAssistant,
ServiceCall,
ServiceResponse,
SupportsResponse,
)
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers.selector import ConfigEntrySelector
from homeassistant.util.json import JsonValueType
from .const import (
ATTR_CHANNEL_TYPE,
ATTR_CONFIG_ENTRY_ID,
CONTROLLED_LOAD_CHANNEL,
DOMAIN,
FEED_IN_CHANNEL,
GENERAL_CHANNEL,
SERVICE_GET_FORECASTS,
)
from .coordinator import AmberConfigEntry
from .helpers import format_cents_to_dollars, normalize_descriptor
GET_FORECASTS_SCHEMA = vol.Schema(
{
ATTR_CONFIG_ENTRY_ID: ConfigEntrySelector({"integration": DOMAIN}),
ATTR_CHANNEL_TYPE: vol.In(
[GENERAL_CHANNEL, CONTROLLED_LOAD_CHANNEL, FEED_IN_CHANNEL]
),
}
)
def async_get_entry(hass: HomeAssistant, config_entry_id: str) -> AmberConfigEntry:
"""Get the Amber config entry."""
if not (entry := hass.config_entries.async_get_entry(config_entry_id)):
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="integration_not_found",
translation_placeholders={"target": config_entry_id},
)
if entry.state is not ConfigEntryState.LOADED:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="not_loaded",
translation_placeholders={"target": entry.title},
)
return entry
def get_forecasts(channel_type: str, data: dict) -> list[JsonValueType]:
"""Return an array of forecasts."""
results: list[JsonValueType] = []
if channel_type not in data["forecasts"]:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="channel_not_found",
translation_placeholders={"channel_type": channel_type},
)
intervals = data["forecasts"][channel_type]
for interval in intervals:
datum = {}
datum["duration"] = interval.duration
datum["date"] = interval.var_date.isoformat()
datum["nem_date"] = interval.nem_time.isoformat()
datum["per_kwh"] = format_cents_to_dollars(interval.per_kwh)
if interval.channel_type == ChannelType.FEEDIN:
datum["per_kwh"] = datum["per_kwh"] * -1
datum["spot_per_kwh"] = format_cents_to_dollars(interval.spot_per_kwh)
datum["start_time"] = interval.start_time.isoformat()
datum["end_time"] = interval.end_time.isoformat()
datum["renewables"] = round(interval.renewables)
datum["spike_status"] = interval.spike_status.value
datum["descriptor"] = normalize_descriptor(interval.descriptor)
if interval.range is not None:
datum["range_min"] = format_cents_to_dollars(interval.range.min)
datum["range_max"] = format_cents_to_dollars(interval.range.max)
if interval.advanced_price is not None:
multiplier = -1 if interval.channel_type == ChannelType.FEEDIN else 1
datum["advanced_price_low"] = multiplier * format_cents_to_dollars(
interval.advanced_price.low
)
datum["advanced_price_predicted"] = multiplier * format_cents_to_dollars(
interval.advanced_price.predicted
)
datum["advanced_price_high"] = multiplier * format_cents_to_dollars(
interval.advanced_price.high
)
results.append(datum)
return results
def setup_services(hass: HomeAssistant) -> None:
"""Set up the services for the Amber integration."""
async def handle_get_forecasts(call: ServiceCall) -> ServiceResponse:
channel_type = call.data[ATTR_CHANNEL_TYPE]
entry = async_get_entry(hass, call.data[ATTR_CONFIG_ENTRY_ID])
coordinator = entry.runtime_data
forecasts = get_forecasts(channel_type, coordinator.data)
return {"forecasts": forecasts}
hass.services.async_register(
DOMAIN,
SERVICE_GET_FORECASTS,
handle_get_forecasts,
GET_FORECASTS_SCHEMA,
supports_response=SupportsResponse.ONLY,
)

16
homeassistant/components/amberelectric/services.yaml Normal file
View File

@ -0,0 +1,16 @@
get_forecasts:
fields:
config_entry_id:
required: true
selector:
config_entry:
integration: amberelectric
channel_type:
required: true
selector:
select:
options:
- general
- controlled_load
- feed_in
translation_key: channel_type

58
homeassistant/components/amberelectric/strings.json
View File

@ -1,25 +1,61 @@
{
"config": {
"error": {
"invalid_api_token": "[%key:common::config_flow::error::invalid_api_key%]",
"no_site": "No site provided",
"unknown_error": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"site": {
"data": {
"site_id": "Site NMI",
"site_name": "Site name"
},
"description": "Select the NMI of the site you would like to add"
},
"user": {
"data": {
"api_token": "[%key:common::config_flow::data::api_token%]",
"site_id": "Site ID"
},
"description": "Go to {api_url} to generate an API key"
},
"site": {
"data": {
"site_id": "Site NMI",
"site_name": "Site Name"
},
"description": "Select the NMI of the site you would like to add"
}
}
},
"error": {
"invalid_api_token": "[%key:common::config_flow::error::invalid_api_key%]",
"no_site": "No site provided",
"unknown_error": "[%key:common::config_flow::error::unknown%]"
"services": {
"get_forecasts": {
"name": "Get price forecasts",
"description": "Retrieves price forecasts from Amber Electric for a site.",
"fields": {
"config_entry_id": {
"description": "The config entry of the site to get forecasts for.",
"name": "Config entry"
},
"channel_type": {
"name": "Channel type",
"description": "The channel to get forecasts for."
}
}
}
},
"exceptions": {
"integration_not_found": {
"message": "Config entry \"{target}\" not found in registry."
},
"not_loaded": {
"message": "{target} is not loaded."
},
"channel_not_found": {
"message": "There is no {channel_type} channel at this site."
}
},
"selector": {
"channel_type": {
"options": {
"general": "General",
"controlled_load": "Controlled load",
"feed_in": "Feed-in"
}
}
}
}

12
tests/components/amberelectric/__init__.py
View File

@ -1 +1,13 @@
"""Tests for the amberelectric integration."""
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Fixture for setting up the component."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()

179
tests/components/amberelectric/conftest.py
View File

@ -1,10 +1,59 @@
"""Provide common Amber fixtures."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
from collections.abc import AsyncGenerator, Generator
from unittest.mock import AsyncMock, Mock, patch
from amberelectric.models.interval import Interval
import pytest
from homeassistant.components.amberelectric.const import (
CONF_SITE_ID,
CONF_SITE_NAME,
DOMAIN,
)
from homeassistant.const import CONF_API_TOKEN
from .helpers import (
CONTROLLED_LOAD_CHANNEL,
FEED_IN_CHANNEL,
FORECASTS,
GENERAL_AND_CONTROLLED_SITE_ID,
GENERAL_AND_FEED_IN_SITE_ID,
GENERAL_CHANNEL,
GENERAL_CHANNEL_WITH_RANGE,
GENERAL_FORECASTS,
GENERAL_ONLY_SITE_ID,
)
from tests.common import MockConfigEntry
MOCK_API_TOKEN = "psk_0000000000000000"
def create_amber_config_entry(
site_id: str, entry_id: str, name: str
) -> MockConfigEntry:
"""Create an Amber config entry."""
return MockConfigEntry(
domain=DOMAIN,
data={
CONF_API_TOKEN: MOCK_API_TOKEN,
CONF_SITE_NAME: name,
CONF_SITE_ID: site_id,
},
entry_id=entry_id,
)
@pytest.fixture
def mock_amber_client() -> Generator[AsyncMock]:
"""Mock the Amber API client."""
with patch(
"homeassistant.components.amberelectric.amberelectric.AmberApi",
autospec=True,
) as mock_client:
yield mock_client
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
@ -13,3 +62,129 @@ def mock_setup_entry() -> Generator[AsyncMock]:
"homeassistant.components.amberelectric.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
async def general_channel_config_entry():
"""Generate the default Amber config entry."""
return create_amber_config_entry(GENERAL_ONLY_SITE_ID, GENERAL_ONLY_SITE_ID, "home")
@pytest.fixture
async def general_channel_and_controlled_load_config_entry():
"""Generate the default Amber config entry for site with controlled load."""
return create_amber_config_entry(
GENERAL_AND_CONTROLLED_SITE_ID, GENERAL_AND_CONTROLLED_SITE_ID, "home"
)
@pytest.fixture
async def general_channel_and_feed_in_config_entry():
"""Generate the default Amber config entry for site with feed in."""
return create_amber_config_entry(
GENERAL_AND_FEED_IN_SITE_ID, GENERAL_AND_FEED_IN_SITE_ID, "home"
)
@pytest.fixture
def general_channel_prices() -> list[Interval]:
"""List containing general channel prices."""
return GENERAL_CHANNEL
@pytest.fixture
def general_channel_prices_with_range() -> list[Interval]:
"""List containing general channel prices."""
return GENERAL_CHANNEL_WITH_RANGE
@pytest.fixture
def controlled_load_channel_prices() -> list[Interval]:
"""List containing controlled load channel prices."""
return CONTROLLED_LOAD_CHANNEL
@pytest.fixture
def feed_in_channel_prices() -> list[Interval]:
"""List containing feed in channel prices."""
return FEED_IN_CHANNEL
@pytest.fixture
def forecast_prices() -> list[Interval]:
"""List containing forecasts with advanced prices."""
return FORECASTS
@pytest.fixture
def general_forecast_prices() -> list[Interval]:
"""List containing forecasts with advanced prices."""
return GENERAL_FORECASTS
@pytest.fixture
def mock_amber_client_general_channel(
mock_amber_client: AsyncMock, general_channel_prices: list[Interval]
) -> Generator[AsyncMock]:
"""Fake general channel prices."""
client = mock_amber_client.return_value
client.get_current_prices.return_value = general_channel_prices
return mock_amber_client
@pytest.fixture
def mock_amber_client_general_channel_with_range(
mock_amber_client: AsyncMock, general_channel_prices_with_range: list[Interval]
) -> Generator[AsyncMock]:
"""Fake general channel prices with a range."""
client = mock_amber_client.return_value
client.get_current_prices.return_value = general_channel_prices_with_range
return mock_amber_client
@pytest.fixture
def mock_amber_client_general_and_controlled_load(
mock_amber_client: AsyncMock,
general_channel_prices: list[Interval],
controlled_load_channel_prices: list[Interval],
) -> Generator[AsyncMock]:
"""Fake general channel and controlled load channel prices."""
client = mock_amber_client.return_value
client.get_current_prices.return_value = (
general_channel_prices + controlled_load_channel_prices
)
return mock_amber_client
@pytest.fixture
async def mock_amber_client_general_and_feed_in(
mock_amber_client: AsyncMock,
general_channel_prices: list[Interval],
feed_in_channel_prices: list[Interval],
) -> AsyncGenerator[Mock]:
"""Set up general channel and feed in channel."""
client = mock_amber_client.return_value
client.get_current_prices.return_value = (
general_channel_prices + feed_in_channel_prices
)
return mock_amber_client
@pytest.fixture
async def mock_amber_client_forecasts(
mock_amber_client: AsyncMock, forecast_prices: list[Interval]
) -> AsyncGenerator[Mock]:
"""Set up general channel, controlled load and feed in channel."""
client = mock_amber_client.return_value
client.get_current_prices.return_value = forecast_prices
return mock_amber_client
@pytest.fixture
async def mock_amber_client_general_forecasts(
mock_amber_client: AsyncMock, general_forecast_prices: list[Interval]
) -> AsyncGenerator[Mock]:
"""Set up general channel only."""
client = mock_amber_client.return_value
client.get_current_prices.return_value = general_forecast_prices
return mock_amber_client

150
tests/components/amberelectric/helpers.py
View File

@ -3,11 +3,13 @@
from datetime import datetime, timedelta
from amberelectric.models.actual_interval import ActualInterval
from amberelectric.models.advanced_price import AdvancedPrice
from amberelectric.models.channel import ChannelType
from amberelectric.models.current_interval import CurrentInterval
from amberelectric.models.forecast_interval import ForecastInterval
from amberelectric.models.interval import Interval
from amberelectric.models.price_descriptor import PriceDescriptor
from amberelectric.models.range import Range
from amberelectric.models.spike_status import SpikeStatus
from dateutil import parser
@ -15,12 +17,16 @@ from dateutil import parser
def generate_actual_interval(channel_type: ChannelType, end_time: datetime) -> Interval:
"""Generate a mock actual interval."""
start_time = end_time - timedelta(minutes=30)
if channel_type == ChannelType.CONTROLLEDLOAD:
per_kwh = 4.4
if channel_type == ChannelType.FEEDIN:
per_kwh = 1.1
return Interval(
ActualInterval(
type="ActualInterval",
duration=30,
spot_per_kwh=1.0,
per_kwh=8.0,
per_kwh=per_kwh,
date=start_time.date(),
nem_time=end_time,
start_time=start_time,
@ -34,16 +40,23 @@ def generate_actual_interval(channel_type: ChannelType, end_time: datetime) -> I
def generate_current_interval(
channel_type: ChannelType, end_time: datetime
channel_type: ChannelType,
end_time: datetime,
range=False,
) -> Interval:
"""Generate a mock current price."""
start_time = end_time - timedelta(minutes=30)
return Interval(
per_kwh = 8.8
if channel_type == ChannelType.CONTROLLEDLOAD:
per_kwh = 4.4
if channel_type == ChannelType.FEEDIN:
per_kwh = 1.1
interval = Interval(
CurrentInterval(
type="CurrentInterval",
duration=30,
spot_per_kwh=1.0,
per_kwh=8.0,
per_kwh=per_kwh,
date=start_time.date(),
nem_time=end_time,
start_time=start_time,
@ -56,18 +69,28 @@ def generate_current_interval(
)
)
if range:
interval.actual_instance.range = Range(min=6.7, max=9.1)
return interval
def generate_forecast_interval(
channel_type: ChannelType, end_time: datetime
channel_type: ChannelType, end_time: datetime, range=False, advanced_price=False
) -> Interval:
"""Generate a mock forecast interval."""
start_time = end_time - timedelta(minutes=30)
return Interval(
per_kwh = 8.8
if channel_type == ChannelType.CONTROLLEDLOAD:
per_kwh = 4.4
if channel_type == ChannelType.FEEDIN:
per_kwh = 1.1
interval = Interval(
ForecastInterval(
type="ForecastInterval",
duration=30,
spot_per_kwh=1.1,
per_kwh=8.8,
per_kwh=per_kwh,
date=start_time.date(),
nem_time=end_time,
start_time=start_time,
@ -79,12 +102,20 @@ def generate_forecast_interval(
estimate=True,
)
)
if range:
interval.actual_instance.range = Range(min=6.7, max=9.1)
if advanced_price:
interval.actual_instance.advanced_price = AdvancedPrice(
low=6.7, predicted=9.0, high=10.2
)
return interval
GENERAL_ONLY_SITE_ID = "01FG2K6V5TB6X9W0EWPPMZD6MJ"
GENERAL_AND_CONTROLLED_SITE_ID = "01FG2MC8RF7GBC4KJXP3YFZ162"
GENERAL_AND_FEED_IN_SITE_ID = "01FG2MCD8KTRZR9MNNW84VP50S"
GENERAL_AND_CONTROLLED_FEED_IN_SITE_ID = "01FG2MCD8KTRZR9MNNW847S50S"
GENERAL_FOR_FAIL = "01JVCEYVSD5HGJG0KT7RNM91GG"
GENERAL_CHANNEL = [
generate_current_interval(
@ -101,6 +132,21 @@ GENERAL_CHANNEL = [
),
]
GENERAL_CHANNEL_WITH_RANGE = [
generate_current_interval(
ChannelType.GENERAL, parser.parse("2021-09-21T08:30:00+10:00"), range=True
),
generate_forecast_interval(
ChannelType.GENERAL, parser.parse("2021-09-21T09:00:00+10:00"), range=True
),
generate_forecast_interval(
ChannelType.GENERAL, parser.parse("2021-09-21T09:30:00+10:00"), range=True
),
generate_forecast_interval(
ChannelType.GENERAL, parser.parse("2021-09-21T10:00:00+10:00"), range=True
),
]
CONTROLLED_LOAD_CHANNEL = [
generate_current_interval(
ChannelType.CONTROLLEDLOAD, parser.parse("2021-09-21T08:30:00+10:00")
@ -131,3 +177,93 @@ FEED_IN_CHANNEL = [
ChannelType.FEEDIN, parser.parse("2021-09-21T10:00:00+10:00")
),
]
GENERAL_FORECASTS = [
generate_current_interval(
ChannelType.GENERAL, parser.parse("2021-09-21T08:30:00+10:00")
),
generate_forecast_interval(
ChannelType.GENERAL,
parser.parse("2021-09-21T09:00:00+10:00"),
range=True,
advanced_price=True,
),
generate_forecast_interval(
ChannelType.GENERAL,
parser.parse("2021-09-21T09:30:00+10:00"),
range=True,
advanced_price=True,
),
generate_forecast_interval(
ChannelType.GENERAL,
parser.parse("2021-09-21T10:00:00+10:00"),
range=True,
advanced_price=True,
),
]
FORECASTS = [
generate_current_interval(
ChannelType.GENERAL, parser.parse("2021-09-21T08:30:00+10:00")
),
generate_current_interval(
ChannelType.CONTROLLEDLOAD, parser.parse("2021-09-21T08:30:00+10:00")
),
generate_current_interval(
ChannelType.FEEDIN, parser.parse("2021-09-21T08:30:00+10:00")
),
generate_forecast_interval(
ChannelType.GENERAL,
parser.parse("2021-09-21T09:00:00+10:00"),
range=True,
advanced_price=True,
),
generate_forecast_interval(
ChannelType.GENERAL,
parser.parse("2021-09-21T09:30:00+10:00"),
range=True,
advanced_price=True,
),
generate_forecast_interval(
ChannelType.GENERAL,
parser.parse("2021-09-21T10:00:00+10:00"),
range=True,
advanced_price=True,
),
generate_forecast_interval(
ChannelType.CONTROLLEDLOAD,
parser.parse("2021-09-21T09:00:00+10:00"),
range=True,
advanced_price=True,
),
generate_forecast_interval(
ChannelType.CONTROLLEDLOAD,
parser.parse("2021-09-21T09:30:00+10:00"),
range=True,
advanced_price=True,
),
generate_forecast_interval(
ChannelType.CONTROLLEDLOAD,
parser.parse("2021-09-21T10:00:00+10:00"),
range=True,
advanced_price=True,
),
generate_forecast_interval(
ChannelType.FEEDIN,
parser.parse("2021-09-21T09:00:00+10:00"),
range=True,
advanced_price=True,
),
generate_forecast_interval(
ChannelType.FEEDIN,
parser.parse("2021-09-21T09:30:00+10:00"),
range=True,
advanced_price=True,
),
generate_forecast_interval(
ChannelType.FEEDIN,
parser.parse("2021-09-21T10:00:00+10:00"),
range=True,
advanced_price=True,
),
]

28
tests/components/amberelectric/test_coordinator.py
View File

@ -9,7 +9,6 @@ from unittest.mock import Mock, patch
from amberelectric import ApiException
from amberelectric.models.channel import Channel, ChannelType
from amberelectric.models.interval import Interval
from amberelectric.models.price_descriptor import PriceDescriptor
from amberelectric.models.site import Site
from amberelectric.models.site_status import SiteStatus
from amberelectric.models.spike_status import SpikeStatus
@ -17,10 +16,7 @@ from dateutil import parser
import pytest
from homeassistant.components.amberelectric.const import CONF_SITE_ID, CONF_SITE_NAME
from homeassistant.components.amberelectric.coordinator import (
AmberUpdateCoordinator,
normalize_descriptor,
)
from homeassistant.components.amberelectric.coordinator import AmberUpdateCoordinator
from homeassistant.const import CONF_API_TOKEN
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import UpdateFailed
@ -98,18 +94,6 @@ def mock_api_current_price() -> Generator:
yield instance
def test_normalize_descriptor() -> None:
"""Test normalizing descriptors works correctly."""
assert normalize_descriptor(None) is None
assert normalize_descriptor(PriceDescriptor.NEGATIVE) == "negative"
assert normalize_descriptor(PriceDescriptor.EXTREMELYLOW) == "extremely_low"
assert normalize_descriptor(PriceDescriptor.VERYLOW) == "very_low"
assert normalize_descriptor(PriceDescriptor.LOW) == "low"
assert normalize_descriptor(PriceDescriptor.NEUTRAL) == "neutral"
assert normalize_descriptor(PriceDescriptor.HIGH) == "high"
assert normalize_descriptor(PriceDescriptor.SPIKE) == "spike"
async def test_fetch_general_site(hass: HomeAssistant, current_price_api: Mock) -> None:
"""Test fetching a site with only a general channel."""
@ -120,7 +104,7 @@ async def test_fetch_general_site(hass: HomeAssistant, current_price_api: Mock)
result = await data_service._async_update_data()
current_price_api.get_current_prices.assert_called_with(
GENERAL_ONLY_SITE_ID, next=48
GENERAL_ONLY_SITE_ID, next=288
)
assert result["current"].get("general") == GENERAL_CHANNEL[0].actual_instance
@ -152,7 +136,7 @@ async def test_fetch_no_general_site(
await data_service._async_update_data()
current_price_api.get_current_prices.assert_called_with(
GENERAL_ONLY_SITE_ID, next=48
GENERAL_ONLY_SITE_ID, next=288
)
@ -166,7 +150,7 @@ async def test_fetch_api_error(hass: HomeAssistant, current_price_api: Mock) ->
result = await data_service._async_update_data()
current_price_api.get_current_prices.assert_called_with(
GENERAL_ONLY_SITE_ID, next=48
GENERAL_ONLY_SITE_ID, next=288
)
assert result["current"].get("general") == GENERAL_CHANNEL[0].actual_instance
@ -217,7 +201,7 @@ async def test_fetch_general_and_controlled_load_site(
result = await data_service._async_update_data()
current_price_api.get_current_prices.assert_called_with(
GENERAL_AND_CONTROLLED_SITE_ID, next=48
GENERAL_AND_CONTROLLED_SITE_ID, next=288
)
assert result["current"].get("general") == GENERAL_CHANNEL[0].actual_instance
@ -257,7 +241,7 @@ async def test_fetch_general_and_feed_in_site(
result = await data_service._async_update_data()
current_price_api.get_current_prices.assert_called_with(
GENERAL_AND_FEED_IN_SITE_ID, next=48
GENERAL_AND_FEED_IN_SITE_ID, next=288
)
assert result["current"].get("general") == GENERAL_CHANNEL[0].actual_instance

17
tests/components/amberelectric/test_helpers.py Normal file
View File

@ -0,0 +1,17 @@
"""Test formatters."""
from amberelectric.models.price_descriptor import PriceDescriptor
from homeassistant.components.amberelectric.helpers import normalize_descriptor
def test_normalize_descriptor() -> None:
"""Test normalizing descriptors works correctly."""
assert normalize_descriptor(None) is None
assert normalize_descriptor(PriceDescriptor.NEGATIVE) == "negative"
assert normalize_descriptor(PriceDescriptor.EXTREMELYLOW) == "extremely_low"
assert normalize_descriptor(PriceDescriptor.VERYLOW) == "very_low"
assert normalize_descriptor(PriceDescriptor.LOW) == "low"
assert normalize_descriptor(PriceDescriptor.NEUTRAL) == "neutral"
assert normalize_descriptor(PriceDescriptor.HIGH) == "high"
assert normalize_descriptor(PriceDescriptor.SPIKE) == "spike"

225
tests/components/amberelectric/test_sensor.py
View File

@ -1,119 +1,26 @@
"""Test the Amber Electric Sensors."""
from collections.abc import AsyncGenerator
from unittest.mock import Mock, patch
from amberelectric.models.current_interval import CurrentInterval
from amberelectric.models.interval import Interval
from amberelectric.models.range import Range
import pytest
from homeassistant.components.amberelectric.const import (
CONF_SITE_ID,
CONF_SITE_NAME,
DOMAIN,
)
from homeassistant.const import CONF_API_TOKEN
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .helpers import (
CONTROLLED_LOAD_CHANNEL,
FEED_IN_CHANNEL,
GENERAL_AND_CONTROLLED_SITE_ID,
GENERAL_AND_FEED_IN_SITE_ID,
GENERAL_CHANNEL,
GENERAL_ONLY_SITE_ID,
)
from tests.common import MockConfigEntry
MOCK_API_TOKEN = "psk_0000000000000000"
from . import MockConfigEntry, setup_integration
@pytest.fixture
async def setup_general(hass: HomeAssistant) -> AsyncGenerator[Mock]:
"""Set up general channel."""
MockConfigEntry(
domain="amberelectric",
data={
CONF_SITE_NAME: "mock_title",
CONF_API_TOKEN: MOCK_API_TOKEN,
CONF_SITE_ID: GENERAL_ONLY_SITE_ID,
},
).add_to_hass(hass)
instance = Mock()
with patch(
"amberelectric.AmberApi",
return_value=instance,
) as mock_update:
instance.get_current_prices = Mock(return_value=GENERAL_CHANNEL)
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
yield mock_update.return_value
@pytest.fixture
async def setup_general_and_controlled_load(
hass: HomeAssistant,
) -> AsyncGenerator[Mock]:
"""Set up general channel and controller load channel."""
MockConfigEntry(
domain="amberelectric",
data={
CONF_API_TOKEN: MOCK_API_TOKEN,
CONF_SITE_ID: GENERAL_AND_CONTROLLED_SITE_ID,
},
).add_to_hass(hass)
instance = Mock()
with patch(
"amberelectric.AmberApi",
return_value=instance,
) as mock_update:
instance.get_current_prices = Mock(
return_value=GENERAL_CHANNEL + CONTROLLED_LOAD_CHANNEL
)
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
yield mock_update.return_value
@pytest.fixture
async def setup_general_and_feed_in(hass: HomeAssistant) -> AsyncGenerator[Mock]:
"""Set up general channel and feed in channel."""
MockConfigEntry(
domain="amberelectric",
data={
CONF_API_TOKEN: MOCK_API_TOKEN,
CONF_SITE_ID: GENERAL_AND_FEED_IN_SITE_ID,
},
).add_to_hass(hass)
instance = Mock()
with patch(
"amberelectric.AmberApi",
return_value=instance,
) as mock_update:
instance.get_current_prices = Mock(
return_value=GENERAL_CHANNEL + FEED_IN_CHANNEL
)
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
yield mock_update.return_value
async def test_general_price_sensor(hass: HomeAssistant, setup_general: Mock) -> None:
@pytest.mark.usefixtures("mock_amber_client_general_channel")
async def test_general_price_sensor(
hass: HomeAssistant, general_channel_config_entry: MockConfigEntry
) -> None:
"""Test the General Price sensor."""
await setup_integration(hass, general_channel_config_entry)
assert len(hass.states.async_all()) == 6
price = hass.states.get("sensor.mock_title_general_price")
assert price
assert price.state == "0.08"
assert price.state == "0.09"
attributes = price.attributes
assert attributes["duration"] == 30
assert attributes["date"] == "2021-09-21"
assert attributes["per_kwh"] == 0.08
assert attributes["per_kwh"] == 0.09
assert attributes["nem_date"] == "2021-09-21T08:30:00+10:00"
assert attributes["spot_per_kwh"] == 0.01
assert attributes["start_time"] == "2021-09-21T08:00:00+10:00"
@ -126,32 +33,36 @@ async def test_general_price_sensor(hass: HomeAssistant, setup_general: Mock) ->
assert attributes.get("range_min") is None
assert attributes.get("range_max") is None
with_range: list[CurrentInterval] = GENERAL_CHANNEL
with_range[0].actual_instance.range = Range(min=7.8, max=12.4)
setup_general.get_current_price.return_value = with_range
config_entry = hass.config_entries.async_entries(DOMAIN)[0]
await hass.config_entries.async_reload(config_entry.entry_id)
await hass.async_block_till_done()
@pytest.mark.usefixtures("mock_amber_client_general_channel_with_range")
async def test_general_price_sensor_with_range(
hass: HomeAssistant, general_channel_config_entry: MockConfigEntry
) -> None:
"""Test the General Price sensor with a range."""
await setup_integration(hass, general_channel_config_entry)
assert len(hass.states.async_all()) == 6
price = hass.states.get("sensor.mock_title_general_price")
assert price
attributes = price.attributes
assert attributes.get("range_min") == 0.08
assert attributes.get("range_max") == 0.12
assert attributes.get("range_min") == 0.07
assert attributes.get("range_max") == 0.09
@pytest.mark.usefixtures("setup_general_and_controlled_load")
async def test_general_and_controlled_load_price_sensor(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("mock_amber_client_general_and_controlled_load")
async def test_general_and_controlled_load_price_sensor(
hass: HomeAssistant,
general_channel_and_controlled_load_config_entry: MockConfigEntry,
) -> None:
"""Test the Controlled Price sensor."""
await setup_integration(hass, general_channel_and_controlled_load_config_entry)
assert len(hass.states.async_all()) == 9
price = hass.states.get("sensor.mock_title_controlled_load_price")
assert price
assert price.state == "0.08"
assert price.state == "0.04"
attributes = price.attributes
assert attributes["duration"] == 30
assert attributes["date"] == "2021-09-21"
assert attributes["per_kwh"] == 0.08
assert attributes["per_kwh"] == 0.04
assert attributes["nem_date"] == "2021-09-21T08:30:00+10:00"
assert attributes["spot_per_kwh"] == 0.01
assert attributes["start_time"] == "2021-09-21T08:00:00+10:00"
@ -163,17 +74,20 @@ async def test_general_and_controlled_load_price_sensor(hass: HomeAssistant) ->
assert attributes["attribution"] == "Data provided by Amber Electric"
@pytest.mark.usefixtures("setup_general_and_feed_in")
async def test_general_and_feed_in_price_sensor(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("mock_amber_client_general_and_feed_in")
async def test_general_and_feed_in_price_sensor(
hass: HomeAssistant, general_channel_and_feed_in_config_entry: MockConfigEntry
) -> None:
"""Test the Feed In sensor."""
await setup_integration(hass, general_channel_and_feed_in_config_entry)
assert len(hass.states.async_all()) == 9
price = hass.states.get("sensor.mock_title_feed_in_price")
assert price
assert price.state == "-0.08"
assert price.state == "-0.01"
attributes = price.attributes
assert attributes["duration"] == 30
assert attributes["date"] == "2021-09-21"
assert attributes["per_kwh"] == -0.08
assert attributes["per_kwh"] == -0.01
assert attributes["nem_date"] == "2021-09-21T08:30:00+10:00"
assert attributes["spot_per_kwh"] == 0.01
assert attributes["start_time"] == "2021-09-21T08:00:00+10:00"
@ -185,10 +99,12 @@ async def test_general_and_feed_in_price_sensor(hass: HomeAssistant) -> None:
assert attributes["attribution"] == "Data provided by Amber Electric"
@pytest.mark.usefixtures("mock_amber_client_general_channel")
async def test_general_forecast_sensor(
hass: HomeAssistant, setup_general: Mock
hass: HomeAssistant, general_channel_config_entry: MockConfigEntry
) -> None:
"""Test the General Forecast sensor."""
await setup_integration(hass, general_channel_config_entry)
assert len(hass.states.async_all()) == 6
price = hass.states.get("sensor.mock_title_general_forecast")
assert price
@ -212,29 +128,33 @@ async def test_general_forecast_sensor(
assert first_forecast.get("range_min") is None
assert first_forecast.get("range_max") is None
with_range: list[Interval] = GENERAL_CHANNEL
with_range[1].actual_instance.range = Range(min=7.8, max=12.4)
setup_general.get_current_price.return_value = with_range
config_entry = hass.config_entries.async_entries(DOMAIN)[0]
await hass.config_entries.async_reload(config_entry.entry_id)
await hass.async_block_till_done()
@pytest.mark.usefixtures("mock_amber_client_general_channel_with_range")
async def test_general_forecast_sensor_with_range(
hass: HomeAssistant, general_channel_config_entry: MockConfigEntry
) -> None:
"""Test the General Forecast sensor with a range."""
await setup_integration(hass, general_channel_config_entry)
assert len(hass.states.async_all()) == 6
price = hass.states.get("sensor.mock_title_general_forecast")
assert price
attributes = price.attributes
first_forecast = attributes["forecasts"][0]
assert first_forecast.get("range_min") == 0.08
assert first_forecast.get("range_max") == 0.12
assert first_forecast.get("range_min") == 0.07
assert first_forecast.get("range_max") == 0.09
@pytest.mark.usefixtures("setup_general_and_controlled_load")
async def test_controlled_load_forecast_sensor(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("mock_amber_client_general_and_controlled_load")
async def test_controlled_load_forecast_sensor(
hass: HomeAssistant,
general_channel_and_controlled_load_config_entry: MockConfigEntry,
) -> None:
"""Test the Controlled Load Forecast sensor."""
await setup_integration(hass, general_channel_and_controlled_load_config_entry)
assert len(hass.states.async_all()) == 9
price = hass.states.get("sensor.mock_title_controlled_load_forecast")
assert price
assert price.state == "0.09"
assert price.state == "0.04"
attributes = price.attributes
assert attributes["channel_type"] == "controlledLoad"
assert attributes["attribution"] == "Data provided by Amber Electric"
@ -242,7 +162,7 @@ async def test_controlled_load_forecast_sensor(hass: HomeAssistant) -> None:
first_forecast = attributes["forecasts"][0]
assert first_forecast["duration"] == 30
assert first_forecast["date"] == "2021-09-21"
assert first_forecast["per_kwh"] == 0.09
assert first_forecast["per_kwh"] == 0.04
assert first_forecast["nem_date"] == "2021-09-21T09:00:00+10:00"
assert first_forecast["spot_per_kwh"] == 0.01
assert first_forecast["start_time"] == "2021-09-21T08:30:00+10:00"
@ -252,13 +172,16 @@ async def test_controlled_load_forecast_sensor(hass: HomeAssistant) -> None:
assert first_forecast["descriptor"] == "very_low"
@pytest.mark.usefixtures("setup_general_and_feed_in")
async def test_feed_in_forecast_sensor(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("mock_amber_client_general_and_feed_in")
async def test_feed_in_forecast_sensor(
hass: HomeAssistant, general_channel_and_feed_in_config_entry: MockConfigEntry
) -> None:
"""Test the Feed In Forecast sensor."""
await setup_integration(hass, general_channel_and_feed_in_config_entry)
assert len(hass.states.async_all()) == 9
price = hass.states.get("sensor.mock_title_feed_in_forecast")
assert price
assert price.state == "-0.09"
assert price.state == "-0.01"
attributes = price.attributes
assert attributes["channel_type"] == "feedIn"
assert attributes["attribution"] == "Data provided by Amber Electric"
@ -266,7 +189,7 @@ async def test_feed_in_forecast_sensor(hass: HomeAssistant) -> None:
first_forecast = attributes["forecasts"][0]
assert first_forecast["duration"] == 30
assert first_forecast["date"] == "2021-09-21"
assert first_forecast["per_kwh"] == -0.09
assert first_forecast["per_kwh"] == -0.01
assert first_forecast["nem_date"] == "2021-09-21T09:00:00+10:00"
assert first_forecast["spot_per_kwh"] == 0.01
assert first_forecast["start_time"] == "2021-09-21T08:30:00+10:00"
@ -276,38 +199,52 @@ async def test_feed_in_forecast_sensor(hass: HomeAssistant) -> None:
assert first_forecast["descriptor"] == "very_low"
@pytest.mark.usefixtures("setup_general")
def test_renewable_sensor(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("mock_amber_client_general_channel")
async def test_renewable_sensor(
hass: HomeAssistant, general_channel_config_entry: MockConfigEntry
) -> None:
"""Testing the creation of the Amber renewables sensor."""
await setup_integration(hass, general_channel_config_entry)
assert len(hass.states.async_all()) == 6
sensor = hass.states.get("sensor.mock_title_renewables")
assert sensor
assert sensor.state == "51"
@pytest.mark.usefixtures("setup_general")
def test_general_price_descriptor_descriptor_sensor(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("mock_amber_client_general_channel")
async def test_general_price_descriptor_descriptor_sensor(
hass: HomeAssistant, general_channel_config_entry: MockConfigEntry
) -> None:
"""Test the General Price Descriptor sensor."""
await setup_integration(hass, general_channel_config_entry)
assert len(hass.states.async_all()) == 6
price = hass.states.get("sensor.mock_title_general_price_descriptor")
assert price
assert price.state == "extremely_low"
@pytest.mark.usefixtures("setup_general_and_controlled_load")
def test_general_and_controlled_load_price_descriptor_sensor(
@pytest.mark.usefixtures("mock_amber_client_general_and_controlled_load")
async def test_general_and_controlled_load_price_descriptor_sensor(
hass: HomeAssistant,
general_channel_and_controlled_load_config_entry: MockConfigEntry,
) -> None:
"""Test the Controlled Price Descriptor sensor."""
await setup_integration(hass, general_channel_and_controlled_load_config_entry)
assert len(hass.states.async_all()) == 9
price = hass.states.get("sensor.mock_title_controlled_load_price_descriptor")
assert price
assert price.state == "extremely_low"
@pytest.mark.usefixtures("setup_general_and_feed_in")
def test_general_and_feed_in_price_descriptor_sensor(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("mock_amber_client_general_and_feed_in")
async def test_general_and_feed_in_price_descriptor_sensor(
hass: HomeAssistant, general_channel_and_feed_in_config_entry: MockConfigEntry
) -> None:
"""Test the Feed In Price Descriptor sensor."""
await setup_integration(hass, general_channel_and_feed_in_config_entry)
assert len(hass.states.async_all()) == 9
price = hass.states.get("sensor.mock_title_feed_in_price_descriptor")
assert price

202
tests/components/amberelectric/test_services.py Normal file
View File

@ -0,0 +1,202 @@
"""Test the Amber Service object."""
import re
import pytest
import voluptuous as vol
from homeassistant.components.amberelectric.const import DOMAIN, SERVICE_GET_FORECASTS
from homeassistant.components.amberelectric.services import (
ATTR_CHANNEL_TYPE,
ATTR_CONFIG_ENTRY_ID,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from . import setup_integration
from .helpers import (
GENERAL_AND_CONTROLLED_SITE_ID,
GENERAL_AND_FEED_IN_SITE_ID,
GENERAL_ONLY_SITE_ID,
)
from tests.common import MockConfigEntry
@pytest.mark.usefixtures("mock_amber_client_forecasts")
async def test_get_general_forecasts(
hass: HomeAssistant,
general_channel_config_entry: MockConfigEntry,
) -> None:
"""Test fetching general forecasts."""
await setup_integration(hass, general_channel_config_entry)
result = await hass.services.async_call(
DOMAIN,
SERVICE_GET_FORECASTS,
{ATTR_CONFIG_ENTRY_ID: GENERAL_ONLY_SITE_ID, ATTR_CHANNEL_TYPE: "general"},
blocking=True,
return_response=True,
)
assert len(result["forecasts"]) == 3
first = result["forecasts"][0]
assert first["duration"] == 30
assert first["date"] == "2021-09-21"
assert first["nem_date"] == "2021-09-21T09:00:00+10:00"
assert first["per_kwh"] == 0.09
assert first["spot_per_kwh"] == 0.01
assert first["start_time"] == "2021-09-21T08:30:00+10:00"
assert first["end_time"] == "2021-09-21T09:00:00+10:00"
assert first["renewables"] == 50
assert first["spike_status"] == "none"
assert first["descriptor"] == "very_low"
@pytest.mark.usefixtures("mock_amber_client_forecasts")
async def test_get_controlled_load_forecasts(
hass: HomeAssistant,
general_channel_and_controlled_load_config_entry: MockConfigEntry,
) -> None:
"""Test fetching general forecasts."""
await setup_integration(hass, general_channel_and_controlled_load_config_entry)
result = await hass.services.async_call(
DOMAIN,
SERVICE_GET_FORECASTS,
{
ATTR_CONFIG_ENTRY_ID: GENERAL_AND_CONTROLLED_SITE_ID,
ATTR_CHANNEL_TYPE: "controlled_load",
},
blocking=True,
return_response=True,
)
assert len(result["forecasts"]) == 3
first = result["forecasts"][0]
assert first["duration"] == 30
assert first["date"] == "2021-09-21"
assert first["nem_date"] == "2021-09-21T09:00:00+10:00"
assert first["per_kwh"] == 0.04
assert first["spot_per_kwh"] == 0.01
assert first["start_time"] == "2021-09-21T08:30:00+10:00"
assert first["end_time"] == "2021-09-21T09:00:00+10:00"
assert first["renewables"] == 50
assert first["spike_status"] == "none"
assert first["descriptor"] == "very_low"
@pytest.mark.usefixtures("mock_amber_client_forecasts")
async def test_get_feed_in_forecasts(
hass: HomeAssistant,
general_channel_and_feed_in_config_entry: MockConfigEntry,
) -> None:
"""Test fetching general forecasts."""
await setup_integration(hass, general_channel_and_feed_in_config_entry)
result = await hass.services.async_call(
DOMAIN,
SERVICE_GET_FORECASTS,
{
ATTR_CONFIG_ENTRY_ID: GENERAL_AND_FEED_IN_SITE_ID,
ATTR_CHANNEL_TYPE: "feed_in",
},
blocking=True,
return_response=True,
)
assert len(result["forecasts"]) == 3
first = result["forecasts"][0]
assert first["duration"] == 30
assert first["date"] == "2021-09-21"
assert first["nem_date"] == "2021-09-21T09:00:00+10:00"
assert first["per_kwh"] == -0.01
assert first["spot_per_kwh"] == 0.01
assert first["start_time"] == "2021-09-21T08:30:00+10:00"
assert first["end_time"] == "2021-09-21T09:00:00+10:00"
assert first["renewables"] == 50
assert first["spike_status"] == "none"
assert first["descriptor"] == "very_low"
@pytest.mark.usefixtures("mock_amber_client_forecasts")
async def test_incorrect_channel_type(
hass: HomeAssistant,
general_channel_config_entry: MockConfigEntry,
) -> None:
"""Test error when the channel type is incorrect."""
await setup_integration(hass, general_channel_config_entry)
with pytest.raises(
vol.error.MultipleInvalid,
match=re.escape(
"value must be one of ['controlled_load', 'feed_in', 'general'] for dictionary value @ data['channel_type']"
),
):
await hass.services.async_call(
DOMAIN,
SERVICE_GET_FORECASTS,
{
ATTR_CONFIG_ENTRY_ID: GENERAL_ONLY_SITE_ID,
ATTR_CHANNEL_TYPE: "incorrect",
},
blocking=True,
return_response=True,
)
@pytest.mark.usefixtures("mock_amber_client_general_forecasts")
async def test_unavailable_channel_type(
hass: HomeAssistant,
general_channel_config_entry: MockConfigEntry,
) -> None:
"""Test error when the channel type is not found."""
await setup_integration(hass, general_channel_config_entry)
with pytest.raises(
ServiceValidationError, match="There is no controlled_load channel at this site"
):
await hass.services.async_call(
DOMAIN,
SERVICE_GET_FORECASTS,
{
ATTR_CONFIG_ENTRY_ID: GENERAL_ONLY_SITE_ID,
ATTR_CHANNEL_TYPE: "controlled_load",
},
blocking=True,
return_response=True,
)
@pytest.mark.usefixtures("mock_amber_client_forecasts")
async def test_service_entry_availability(
hass: HomeAssistant,
general_channel_config_entry: MockConfigEntry,
) -> None:
"""Test the services without valid entry."""
general_channel_config_entry.add_to_hass(hass)
mock_config_entry2 = MockConfigEntry(domain=DOMAIN)
mock_config_entry2.add_to_hass(hass)
await hass.config_entries.async_setup(general_channel_config_entry.entry_id)
await hass.async_block_till_done()
with pytest.raises(ServiceValidationError, match="Mock Title is not loaded"):
await hass.services.async_call(
DOMAIN,
SERVICE_GET_FORECASTS,
{
ATTR_CONFIG_ENTRY_ID: mock_config_entry2.entry_id,
ATTR_CHANNEL_TYPE: "general",
},
blocking=True,
return_response=True,
)
with pytest.raises(
ServiceValidationError,
match='Config entry "bad-config_id" not found in registry',
):
await hass.services.async_call(
DOMAIN,
SERVICE_GET_FORECASTS,
{ATTR_CONFIG_ENTRY_ID: "bad-config_id", ATTR_CHANNEL_TYPE: "general"},
blocking=True,
return_response=True,
)