Compare commits

..

1 Commits

Author SHA1 Message Date
Erik
b4a117e6b1 Adjust device_registry.async_setup 2026-04-08 08:24:49 +02:00
63 changed files with 273 additions and 3577 deletions

2
CODEOWNERS generated
View File

@@ -355,8 +355,6 @@ build.json @home-assistant/supervisor
/tests/components/deluge/ @tkdrob
/homeassistant/components/demo/ @home-assistant/core
/tests/components/demo/ @home-assistant/core
/homeassistant/components/denon_rs232/ @balloob
/tests/components/denon_rs232/ @balloob
/homeassistant/components/denonavr/ @ol-iver @starkillerOG
/tests/components/denonavr/ @ol-iver @starkillerOG
/homeassistant/components/derivative/ @afaucogney @karwosts

View File

@@ -1,5 +1,5 @@
{
"domain": "denon",
"name": "Denon",
"integrations": ["denon", "denonavr", "denon_rs232", "heos"]
"integrations": ["denon", "denonavr", "heos"]
}

View File

@@ -54,7 +54,7 @@
"message": "Storage account {account_name} not found"
},
"cannot_connect": {
"message": "Cannot connect to storage account {account_name}"
"message": "Can not connect to storage account {account_name}"
},
"container_not_found": {
"message": "Storage container {container_name} not found"

View File

@@ -16,7 +16,6 @@ PLATFORMS: list[Platform] = [
Platform.BUTTON,
Platform.LIGHT,
Platform.SELECT,
Platform.SENSOR,
]

View File

@@ -4,11 +4,7 @@ from __future__ import annotations
from pycasperglow import GlowState
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.const import EntityCategory
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
@@ -25,12 +21,7 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the binary sensor platform for Casper Glow."""
async_add_entities(
[
CasperGlowPausedBinarySensor(entry.runtime_data),
CasperGlowChargingBinarySensor(entry.runtime_data),
]
)
async_add_entities([CasperGlowPausedBinarySensor(entry.runtime_data)])
class CasperGlowPausedBinarySensor(CasperGlowEntity, BinarySensorEntity):
@@ -55,34 +46,6 @@ class CasperGlowPausedBinarySensor(CasperGlowEntity, BinarySensorEntity):
@callback
def _async_handle_state_update(self, state: GlowState) -> None:
"""Handle a state update from the device."""
if state.is_paused is not None and state.is_paused != self._attr_is_on:
if state.is_paused is not None:
self._attr_is_on = state.is_paused
self.async_write_ha_state()
class CasperGlowChargingBinarySensor(CasperGlowEntity, BinarySensorEntity):
"""Binary sensor indicating whether the Casper Glow is charging."""
_attr_device_class = BinarySensorDeviceClass.BATTERY_CHARGING
_attr_entity_category = EntityCategory.DIAGNOSTIC
def __init__(self, coordinator: CasperGlowCoordinator) -> None:
"""Initialize the charging binary sensor."""
super().__init__(coordinator)
self._attr_unique_id = f"{format_mac(coordinator.device.address)}_charging"
if coordinator.device.state.is_charging is not None:
self._attr_is_on = coordinator.device.state.is_charging
async def async_added_to_hass(self) -> None:
"""Register state update callback when entity is added."""
await super().async_added_to_hass()
self.async_on_remove(
self._device.register_callback(self._async_handle_state_update)
)
@callback
def _async_handle_state_update(self, state: GlowState) -> None:
"""Handle a state update from the device."""
if state.is_charging is not None and state.is_charging != self._attr_is_on:
self._attr_is_on = state.is_charging
self.async_write_ha_state()
self.async_write_ha_state()

View File

@@ -53,15 +53,15 @@ rules:
docs-use-cases: todo
dynamic-devices: todo
entity-category: done
entity-device-class: done
entity-device-class:
status: exempt
comment: No applicable device classes for binary_sensor, button, light, or select entities.
entity-disabled-by-default: todo
entity-translations: done
exception-translations: done
icon-translations: done
reconfiguration-flow: todo
repair-issues:
status: exempt
comment: Integration does not register repair issues.
repair-issues: todo
stale-devices: todo
# Platinum

View File

@@ -1,61 +0,0 @@
"""Casper Glow integration sensor platform."""
from __future__ import annotations
from pycasperglow import GlowState
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorStateClass,
)
from homeassistant.const import PERCENTAGE, EntityCategory
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import CasperGlowConfigEntry, CasperGlowCoordinator
from .entity import CasperGlowEntity
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant,
entry: CasperGlowConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the sensor platform for Casper Glow."""
async_add_entities([CasperGlowBatterySensor(entry.runtime_data)])
class CasperGlowBatterySensor(CasperGlowEntity, SensorEntity):
"""Sensor entity for Casper Glow battery level."""
_attr_device_class = SensorDeviceClass.BATTERY
_attr_native_unit_of_measurement = PERCENTAGE
_attr_state_class = SensorStateClass.MEASUREMENT
_attr_entity_category = EntityCategory.DIAGNOSTIC
def __init__(self, coordinator: CasperGlowCoordinator) -> None:
"""Initialize the battery sensor."""
super().__init__(coordinator)
self._attr_unique_id = f"{format_mac(coordinator.device.address)}_battery"
if coordinator.device.state.battery_level is not None:
self._attr_native_value = coordinator.device.state.battery_level.percentage
async def async_added_to_hass(self) -> None:
"""Register state update callback when entity is added."""
await super().async_added_to_hass()
self.async_on_remove(
self._device.register_callback(self._async_handle_state_update)
)
@callback
def _async_handle_state_update(self, state: GlowState) -> None:
"""Handle a state update from the device."""
if state.battery_level is not None:
new_value = state.battery_level.percentage
if new_value != self._attr_native_value:
self._attr_native_value = new_value
self.async_write_ha_state()

View File

@@ -1,54 +0,0 @@
"""The Denon RS232 integration."""
from __future__ import annotations
from denon_rs232 import DenonReceiver, ReceiverState
from denon_rs232.models import MODELS
from homeassistant.const import CONF_DEVICE, CONF_MODEL, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from .const import LOGGER, DenonRS232ConfigEntry
PLATFORMS = [Platform.MEDIA_PLAYER]
async def async_setup_entry(hass: HomeAssistant, entry: DenonRS232ConfigEntry) -> bool:
"""Set up Denon RS232 from a config entry."""
port = entry.data[CONF_DEVICE]
model = MODELS[entry.data[CONF_MODEL]]
receiver = DenonReceiver(port, model=model)
try:
await receiver.connect()
await receiver.query_state()
except (ConnectionError, OSError) as err:
LOGGER.error("Error connecting to Denon receiver at %s: %s", port, err)
if receiver.connected:
await receiver.disconnect()
raise ConfigEntryNotReady from err
entry.runtime_data = receiver
@callback
def _on_disconnect(state: ReceiverState | None) -> None:
if state is None:
LOGGER.warning("Denon receiver disconnected, reloading config entry")
hass.config_entries.async_schedule_reload(entry.entry_id)
entry.async_on_unload(receiver.subscribe(_on_disconnect))
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: DenonRS232ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
await entry.runtime_data.disconnect()
return unload_ok

View File

@@ -1,166 +0,0 @@
"""Config flow for the Denon RS232 integration."""
from __future__ import annotations
from typing import Any
from denon_rs232 import DenonReceiver
from denon_rs232.models import MODELS
import voluptuous as vol
from homeassistant.components.usb import human_readable_device_name, scan_serial_ports
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_DEVICE, CONF_MODEL
from homeassistant.helpers.selector import (
SelectOptionDict,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
)
from .const import DOMAIN, LOGGER
OPTION_PICK_MANUAL = "manual"
async def _async_attempt_connect(port: str, model_key: str) -> str | None:
"""Attempt to connect to the receiver at the given port.
Returns None on success, error on failure.
"""
model = MODELS[model_key]
receiver = DenonReceiver(port, model=model)
try:
await receiver.connect()
except (
# When the port contains invalid connection data
ValueError,
# If it is a remote port, and we cannot connect
ConnectionError,
OSError,
TimeoutError,
):
return "cannot_connect"
except Exception: # noqa: BLE001
LOGGER.exception("Unexpected exception")
return "unknown"
else:
await receiver.disconnect()
return None
class DenonRS232ConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Denon RS232."""
VERSION = 1
_model: str
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is not None:
if user_input[CONF_DEVICE] == OPTION_PICK_MANUAL:
self._model = user_input[CONF_MODEL]
return await self.async_step_manual()
self._async_abort_entries_match({CONF_DEVICE: user_input[CONF_DEVICE]})
error = await _async_attempt_connect(
user_input[CONF_DEVICE], user_input[CONF_MODEL]
)
if not error:
return self.async_create_entry(
title=MODELS[user_input[CONF_MODEL]].name,
data={
CONF_DEVICE: user_input[CONF_DEVICE],
CONF_MODEL: user_input[CONF_MODEL],
},
)
errors["base"] = error
ports = await self.hass.async_add_executor_job(get_ports)
port_options = [
SelectOptionDict(value=device, label=name) for device, name in ports.items()
]
port_options.append(
SelectOptionDict(value=OPTION_PICK_MANUAL, label=OPTION_PICK_MANUAL)
)
if user_input is None and port_options:
user_input = {CONF_DEVICE: port_options[0]["value"]}
return self.async_show_form(
step_id="user",
data_schema=self.add_suggested_values_to_schema(
vol.Schema(
{
vol.Required(CONF_MODEL): SelectSelector(
SelectSelectorConfig(
options=[
SelectOptionDict(value=key, label=model.name)
for key, model in MODELS.items()
],
mode=SelectSelectorMode.DROPDOWN,
translation_key="model",
)
),
vol.Required(CONF_DEVICE): SelectSelector(
SelectSelectorConfig(
options=port_options,
mode=SelectSelectorMode.DROPDOWN,
translation_key="device",
)
),
}
),
user_input or {},
),
errors=errors,
)
async def async_step_manual(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a manual port selection."""
errors: dict[str, str] = {}
if user_input is not None:
self._async_abort_entries_match({CONF_DEVICE: user_input[CONF_DEVICE]})
error = await _async_attempt_connect(user_input[CONF_DEVICE], self._model)
if not error:
return self.async_create_entry(
title=MODELS[self._model].name,
data={
CONF_DEVICE: user_input[CONF_DEVICE],
CONF_MODEL: self._model,
},
)
errors["base"] = error
return self.async_show_form(
step_id="manual",
data_schema=self.add_suggested_values_to_schema(
vol.Schema({vol.Required(CONF_DEVICE): str}),
user_input or {},
),
errors=errors,
)
def get_ports() -> dict[str, str]:
"""Get available serial ports keyed by their device path."""
return {
port.device: human_readable_device_name(
port.device,
port.serial_number,
port.manufacturer,
port.description,
port.vid,
port.pid,
)
for port in scan_serial_ports()
}

View File

@@ -1,12 +0,0 @@
"""Constants for the Denon RS232 integration."""
import logging
from denon_rs232 import DenonReceiver
from homeassistant.config_entries import ConfigEntry
LOGGER = logging.getLogger(__package__)
DOMAIN = "denon_rs232"
type DenonRS232ConfigEntry = ConfigEntry[DenonReceiver]

View File

@@ -1,13 +0,0 @@
{
"domain": "denon_rs232",
"name": "Denon RS232",
"codeowners": ["@balloob"],
"config_flow": true,
"dependencies": ["usb"],
"documentation": "https://www.home-assistant.io/integrations/denon_rs232",
"integration_type": "hub",
"iot_class": "local_push",
"loggers": ["denon_rs232"],
"quality_scale": "bronze",
"requirements": ["denon-rs232==4.0.0"]
}

View File

@@ -1,234 +0,0 @@
"""Media player platform for the Denon RS232 integration."""
from __future__ import annotations
from typing import Literal, cast
from denon_rs232 import (
MIN_VOLUME_DB,
VOLUME_DB_RANGE,
DenonReceiver,
InputSource,
MainPlayer,
ReceiverState,
ZonePlayer,
)
from homeassistant.components.media_player import (
MediaPlayerDeviceClass,
MediaPlayerEntity,
MediaPlayerEntityFeature,
MediaPlayerState,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN, DenonRS232ConfigEntry
INPUT_SOURCE_DENON_TO_HA: dict[InputSource, str] = {
InputSource.PHONO: "phono",
InputSource.CD: "cd",
InputSource.TUNER: "tuner",
InputSource.DVD: "dvd",
InputSource.VDP: "vdp",
InputSource.TV: "tv",
InputSource.DBS_SAT: "dbs_sat",
InputSource.VCR_1: "vcr_1",
InputSource.VCR_2: "vcr_2",
InputSource.VCR_3: "vcr_3",
InputSource.V_AUX: "v_aux",
InputSource.CDR_TAPE1: "cdr_tape1",
InputSource.MD_TAPE2: "md_tape2",
InputSource.HDP: "hdp",
InputSource.DVR: "dvr",
InputSource.TV_CBL: "tv_cbl",
InputSource.SAT: "sat",
InputSource.NET_USB: "net_usb",
InputSource.DOCK: "dock",
InputSource.IPOD: "ipod",
InputSource.BD: "bd",
InputSource.SAT_CBL: "sat_cbl",
InputSource.MPLAY: "mplay",
InputSource.GAME: "game",
InputSource.AUX1: "aux1",
InputSource.AUX2: "aux2",
InputSource.NET: "net",
InputSource.BT: "bt",
InputSource.USB_IPOD: "usb_ipod",
InputSource.EIGHT_K: "eight_k",
InputSource.PANDORA: "pandora",
InputSource.SIRIUSXM: "siriusxm",
InputSource.SPOTIFY: "spotify",
InputSource.FLICKR: "flickr",
InputSource.IRADIO: "iradio",
InputSource.SERVER: "server",
InputSource.FAVORITES: "favorites",
InputSource.LASTFM: "lastfm",
InputSource.XM: "xm",
InputSource.SIRIUS: "sirius",
InputSource.HDRADIO: "hdradio",
InputSource.DAB: "dab",
}
async def async_setup_entry(
hass: HomeAssistant,
config_entry: DenonRS232ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Denon RS232 media player."""
receiver = config_entry.runtime_data
entities = [DenonRS232MediaPlayer(receiver, receiver.main, config_entry, "main")]
if receiver.zone_2.power is not None:
entities.append(
DenonRS232MediaPlayer(receiver, receiver.zone_2, config_entry, "zone_2")
)
if receiver.zone_3.power is not None:
entities.append(
DenonRS232MediaPlayer(receiver, receiver.zone_3, config_entry, "zone_3")
)
async_add_entities(entities)
class DenonRS232MediaPlayer(MediaPlayerEntity):
"""Representation of a Denon receiver controlled over RS232."""
_attr_device_class = MediaPlayerDeviceClass.RECEIVER
_attr_has_entity_name = True
_attr_translation_key = "receiver"
_attr_should_poll = False
_volume_min = MIN_VOLUME_DB
_volume_range = VOLUME_DB_RANGE
def __init__(
self,
receiver: DenonReceiver,
player: MainPlayer | ZonePlayer,
config_entry: DenonRS232ConfigEntry,
zone: Literal["main", "zone_2", "zone_3"],
) -> None:
"""Initialize the media player."""
self._receiver = receiver
self._player = player
self._is_main = zone == "main"
model = receiver.model
assert model is not None # We always set this
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, config_entry.entry_id)},
manufacturer="Denon",
model_id=model.name,
)
self._attr_unique_id = f"{config_entry.entry_id}_{zone}"
self._attr_source_list = sorted(
INPUT_SOURCE_DENON_TO_HA[source] for source in model.input_sources
)
self._attr_supported_features = (
MediaPlayerEntityFeature.TURN_ON
| MediaPlayerEntityFeature.TURN_OFF
| MediaPlayerEntityFeature.VOLUME_SET
| MediaPlayerEntityFeature.VOLUME_STEP
| MediaPlayerEntityFeature.SELECT_SOURCE
)
if zone == "main":
self._attr_name = None
self._attr_supported_features |= MediaPlayerEntityFeature.VOLUME_MUTE
else:
self._attr_name = "Zone 2" if zone == "zone_2" else "Zone 3"
self._async_update_from_player()
async def async_added_to_hass(self) -> None:
"""Subscribe to receiver state updates."""
self.async_on_remove(self._receiver.subscribe(self._async_on_state_update))
@callback
def _async_on_state_update(self, state: ReceiverState | None) -> None:
"""Handle a state update from the receiver."""
if state is None:
self._attr_available = False
else:
self._attr_available = True
self._async_update_from_player()
self.async_write_ha_state()
@callback
def _async_update_from_player(self) -> None:
"""Update entity attributes from the shared player object."""
if self._player.power is None:
self._attr_state = None
else:
self._attr_state = (
MediaPlayerState.ON if self._player.power else MediaPlayerState.OFF
)
source = self._player.input_source
self._attr_source = INPUT_SOURCE_DENON_TO_HA.get(source) if source else None
volume_min = self._player.volume_min
volume_max = self._player.volume_max
if volume_min is not None:
self._volume_min = volume_min
if volume_max is not None and volume_max > volume_min:
self._volume_range = volume_max - volume_min
volume = self._player.volume
if volume is not None:
self._attr_volume_level = (volume - self._volume_min) / self._volume_range
else:
self._attr_volume_level = None
if self._is_main:
self._attr_is_volume_muted = cast(MainPlayer, self._player).mute
async def async_turn_on(self) -> None:
"""Turn the receiver on."""
await self._player.power_on()
async def async_turn_off(self) -> None:
"""Turn the receiver off."""
await self._player.power_standby()
async def async_set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1."""
db = volume * self._volume_range + self._volume_min
await self._player.set_volume(db)
async def async_volume_up(self) -> None:
"""Volume up."""
await self._player.volume_up()
async def async_volume_down(self) -> None:
"""Volume down."""
await self._player.volume_down()
async def async_mute_volume(self, mute: bool) -> None:
"""Mute or unmute."""
player = cast(MainPlayer, self._player)
if mute:
await player.mute_on()
else:
await player.mute_off()
async def async_select_source(self, source: str) -> None:
"""Select input source."""
input_source = next(
(
input_source
for input_source, ha_source in INPUT_SOURCE_DENON_TO_HA.items()
if ha_source == source
),
None,
)
if input_source is None:
raise HomeAssistantError("Invalid source")
await self._player.select_input_source(input_source)

View File

@@ -1,64 +0,0 @@
rules:
# Bronze
action-setup: done
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions: done
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup: done
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions: todo
config-entry-unloading: todo
docs-configuration-parameters: todo
docs-installation-parameters: todo
entity-unavailable: done
integration-owner: done
log-when-unavailable: todo
parallel-updates: todo
reauthentication-flow: todo
test-coverage: todo
# Gold
devices: done
diagnostics: todo
discovery-update-info: todo
discovery: todo
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
docs-supported-functions: todo
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices:
status: exempt
comment: "The integration does not create dynamic devices."
entity-category: todo
entity-device-class: todo
entity-disabled-by-default: todo
entity-translations: todo
exception-translations: todo
icon-translations: todo
reconfiguration-flow: todo
repair-issues: todo
stale-devices:
status: exempt
comment: "The integration does not create devices that can become stale."
# Platinum
async-dependency: done
inject-websession: todo
strict-typing: todo

View File

@@ -1,97 +0,0 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"manual": {
"data": {
"device": "[%key:common::config_flow::data::port%]"
},
"data_description": {
"device": "[%key:component::denon_rs232::config::step::user::data_description::device%]"
}
},
"user": {
"data": {
"device": "[%key:common::config_flow::data::port%]",
"model": "Receiver model"
},
"data_description": {
"device": "Serial port path to connect to",
"model": "Determines available features"
}
}
}
},
"entity": {
"media_player": {
"receiver": {
"state_attributes": {
"source": {
"state": {
"aux1": "Aux 1",
"aux2": "Aux 2",
"bd": "BD Player",
"bt": "Bluetooth",
"cd": "CD",
"cdr_tape1": "CDR/Tape 1",
"dab": "DAB",
"dbs_sat": "DBS/Sat",
"dock": "Dock",
"dvd": "DVD",
"dvr": "DVR",
"eight_k": "8K",
"favorites": "Favorites",
"flickr": "Flickr",
"game": "Game",
"hdp": "HDP",
"hdradio": "HD Radio",
"ipod": "iPod",
"iradio": "Internet Radio",
"lastfm": "Last.fm",
"md_tape2": "MD/Tape 2",
"mplay": "Media Player",
"net": "HEOS Music",
"net_usb": "Network/USB",
"pandora": "Pandora",
"phono": "Phono",
"sat": "Sat",
"sat_cbl": "Satellite/Cable",
"server": "Server",
"sirius": "Sirius",
"siriusxm": "SiriusXM",
"spotify": "Spotify",
"tuner": "Tuner",
"tv": "TV Audio",
"tv_cbl": "TV/Cable",
"usb_ipod": "USB/iPod",
"v_aux": "V. Aux",
"vcr_1": "VCR 1",
"vcr_2": "VCR 2",
"vcr_3": "VCR 3",
"vdp": "VDP",
"xm": "XM"
}
}
}
}
}
},
"selector": {
"device": {
"options": {
"manual": "Enter manually"
}
},
"model": {
"options": {
"other": "Other"
}
}
}
}

View File

@@ -2,26 +2,14 @@
from __future__ import annotations
from types import MappingProxyType
from homeassistant.config_entries import ConfigSubentry
from homeassistant.const import CONF_API_KEY, Platform
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError
from .const import (
CONF_AZIMUTH,
CONF_DAMPING,
CONF_DAMPING_EVENING,
CONF_DAMPING_MORNING,
CONF_DECLINATION,
CONF_MODULES_POWER,
DEFAULT_AZIMUTH,
DEFAULT_DAMPING,
DEFAULT_DECLINATION,
DEFAULT_MODULES_POWER,
DOMAIN,
SUBENTRY_TYPE_PLANE,
)
from .coordinator import ForecastSolarConfigEntry, ForecastSolarDataUpdateCoordinator
@@ -37,41 +25,14 @@ async def async_migrate_entry(
new_options = entry.options.copy()
new_options |= {
CONF_MODULES_POWER: new_options.pop("modules power"),
CONF_DAMPING_MORNING: new_options.get(CONF_DAMPING, DEFAULT_DAMPING),
CONF_DAMPING_EVENING: new_options.pop(CONF_DAMPING, DEFAULT_DAMPING),
CONF_DAMPING_MORNING: new_options.get(CONF_DAMPING, 0.0),
CONF_DAMPING_EVENING: new_options.pop(CONF_DAMPING, 0.0),
}
hass.config_entries.async_update_entry(
entry, data=entry.data, options=new_options, version=2
)
if entry.version == 2:
# Migrate the main plane from options to a subentry
declination = entry.options.get(CONF_DECLINATION, DEFAULT_DECLINATION)
azimuth = entry.options.get(CONF_AZIMUTH, DEFAULT_AZIMUTH)
modules_power = entry.options.get(CONF_MODULES_POWER, DEFAULT_MODULES_POWER)
subentry = ConfigSubentry(
data=MappingProxyType(
{
CONF_DECLINATION: declination,
CONF_AZIMUTH: azimuth,
CONF_MODULES_POWER: modules_power,
}
),
subentry_type=SUBENTRY_TYPE_PLANE,
title=f"{declination}° / {azimuth}° / {modules_power}W",
unique_id=None,
)
hass.config_entries.async_add_subentry(entry, subentry)
new_options = dict(entry.options)
new_options.pop(CONF_DECLINATION, None)
new_options.pop(CONF_AZIMUTH, None)
new_options.pop(CONF_MODULES_POWER, None)
hass.config_entries.async_update_entry(entry, options=new_options, version=3)
return True
@@ -79,19 +40,6 @@ async def async_setup_entry(
hass: HomeAssistant, entry: ForecastSolarConfigEntry
) -> bool:
"""Set up Forecast.Solar from a config entry."""
plane_subentries = entry.get_subentries_of_type(SUBENTRY_TYPE_PLANE)
if not plane_subentries:
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="no_plane",
)
if len(plane_subentries) > 1 and not entry.options.get(CONF_API_KEY):
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="api_key_required",
)
coordinator = ForecastSolarDataUpdateCoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
@@ -99,18 +47,9 @@ async def async_setup_entry(
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(_async_update_listener))
return True
async def _async_update_listener(
hass: HomeAssistant, entry: ForecastSolarConfigEntry
) -> None:
"""Handle config entry updates (options or subentry changes)."""
hass.config_entries.async_schedule_reload(entry.entry_id)
async def async_unload_entry(
hass: HomeAssistant, entry: ForecastSolarConfigEntry
) -> bool:

View File

@@ -11,13 +11,11 @@ from homeassistant.config_entries import (
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
ConfigSubentryFlow,
OptionsFlow,
SubentryFlowResult,
OptionsFlowWithReload,
)
from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME
from homeassistant.core import callback
from homeassistant.helpers import config_validation as cv, selector
from homeassistant.helpers import config_validation as cv
from .const import (
CONF_AZIMUTH,
@@ -26,51 +24,16 @@ from .const import (
CONF_DECLINATION,
CONF_INVERTER_SIZE,
CONF_MODULES_POWER,
DEFAULT_AZIMUTH,
DEFAULT_DAMPING,
DEFAULT_DECLINATION,
DEFAULT_MODULES_POWER,
DOMAIN,
MAX_PLANES,
SUBENTRY_TYPE_PLANE,
)
RE_API_KEY = re.compile(r"^[a-zA-Z0-9]{16}$")
PLANE_SCHEMA = vol.Schema(
{
vol.Required(CONF_DECLINATION): vol.All(
selector.NumberSelector(
selector.NumberSelectorConfig(
min=0, max=90, step=1, mode=selector.NumberSelectorMode.BOX
),
),
vol.Coerce(int),
),
vol.Required(CONF_AZIMUTH): vol.All(
selector.NumberSelector(
selector.NumberSelectorConfig(
min=0, max=360, step=1, mode=selector.NumberSelectorMode.BOX
),
),
vol.Coerce(int),
),
vol.Required(CONF_MODULES_POWER): vol.All(
selector.NumberSelector(
selector.NumberSelectorConfig(
min=1, step=1, mode=selector.NumberSelectorMode.BOX
),
),
vol.Coerce(int),
),
}
)
class ForecastSolarFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Forecast.Solar."""
VERSION = 3
VERSION = 2
@staticmethod
@callback
@@ -80,14 +43,6 @@ class ForecastSolarFlowHandler(ConfigFlow, domain=DOMAIN):
"""Get the options flow for this handler."""
return ForecastSolarOptionFlowHandler()
@classmethod
@callback
def async_get_supported_subentry_types(
cls, config_entry: ConfigEntry
) -> dict[str, type[ConfigSubentryFlow]]:
"""Return subentries supported by this handler."""
return {SUBENTRY_TYPE_PLANE: PlaneSubentryFlowHandler}
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
@@ -99,112 +54,94 @@ class ForecastSolarFlowHandler(ConfigFlow, domain=DOMAIN):
CONF_LATITUDE: user_input[CONF_LATITUDE],
CONF_LONGITUDE: user_input[CONF_LONGITUDE],
},
subentries=[
{
"subentry_type": SUBENTRY_TYPE_PLANE,
"data": {
CONF_DECLINATION: user_input[CONF_DECLINATION],
CONF_AZIMUTH: user_input[CONF_AZIMUTH],
CONF_MODULES_POWER: user_input[CONF_MODULES_POWER],
},
"title": f"{user_input[CONF_DECLINATION]}° / {user_input[CONF_AZIMUTH]}° / {user_input[CONF_MODULES_POWER]}W",
"unique_id": None,
},
],
options={
CONF_AZIMUTH: user_input[CONF_AZIMUTH],
CONF_DECLINATION: user_input[CONF_DECLINATION],
CONF_MODULES_POWER: user_input[CONF_MODULES_POWER],
},
)
return self.async_show_form(
step_id="user",
data_schema=self.add_suggested_values_to_schema(
vol.Schema(
{
vol.Required(CONF_NAME): str,
vol.Required(CONF_LATITUDE): cv.latitude,
vol.Required(CONF_LONGITUDE): cv.longitude,
}
).extend(PLANE_SCHEMA.schema),
data_schema=vol.Schema(
{
CONF_NAME: self.hass.config.location_name,
CONF_LATITUDE: self.hass.config.latitude,
CONF_LONGITUDE: self.hass.config.longitude,
CONF_DECLINATION: DEFAULT_DECLINATION,
CONF_AZIMUTH: DEFAULT_AZIMUTH,
CONF_MODULES_POWER: DEFAULT_MODULES_POWER,
},
vol.Required(
CONF_NAME, default=self.hass.config.location_name
): str,
vol.Required(
CONF_LATITUDE, default=self.hass.config.latitude
): cv.latitude,
vol.Required(
CONF_LONGITUDE, default=self.hass.config.longitude
): cv.longitude,
vol.Required(CONF_DECLINATION, default=25): vol.All(
vol.Coerce(int), vol.Range(min=0, max=90)
),
vol.Required(CONF_AZIMUTH, default=180): vol.All(
vol.Coerce(int), vol.Range(min=0, max=360)
),
vol.Required(CONF_MODULES_POWER): vol.All(
vol.Coerce(int), vol.Range(min=1)
),
}
),
)
class ForecastSolarOptionFlowHandler(OptionsFlow):
class ForecastSolarOptionFlowHandler(OptionsFlowWithReload):
"""Handle options."""
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Manage the options."""
errors: dict[str, str] = {}
planes_count = len(
self.config_entry.get_subentries_of_type(SUBENTRY_TYPE_PLANE)
)
errors = {}
if user_input is not None:
api_key = user_input.get(CONF_API_KEY)
if planes_count > 1 and not api_key:
errors[CONF_API_KEY] = "api_key_required"
elif api_key and RE_API_KEY.match(api_key) is None:
if (api_key := user_input.get(CONF_API_KEY)) and RE_API_KEY.match(
api_key
) is None:
errors[CONF_API_KEY] = "invalid_api_key"
else:
return self.async_create_entry(
title="", data=user_input | {CONF_API_KEY: api_key or None}
)
suggested_api_key = self.config_entry.options.get(CONF_API_KEY, "")
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(
{
vol.Required(
vol.Optional(
CONF_API_KEY,
default=suggested_api_key,
)
if planes_count > 1
else vol.Optional(
CONF_API_KEY,
description={"suggested_value": suggested_api_key},
description={
"suggested_value": self.config_entry.options.get(
CONF_API_KEY, ""
)
},
): str,
vol.Required(
CONF_DECLINATION,
default=self.config_entry.options[CONF_DECLINATION],
): vol.All(vol.Coerce(int), vol.Range(min=0, max=90)),
vol.Required(
CONF_AZIMUTH,
default=self.config_entry.options.get(CONF_AZIMUTH),
): vol.All(vol.Coerce(int), vol.Range(min=-0, max=360)),
vol.Required(
CONF_MODULES_POWER,
default=self.config_entry.options[CONF_MODULES_POWER],
): vol.All(vol.Coerce(int), vol.Range(min=1)),
vol.Optional(
CONF_DAMPING_MORNING,
default=self.config_entry.options.get(
CONF_DAMPING_MORNING, DEFAULT_DAMPING
CONF_DAMPING_MORNING, 0.0
),
): vol.All(
selector.NumberSelector(
selector.NumberSelectorConfig(
min=0,
max=1,
step=0.01,
mode=selector.NumberSelectorMode.BOX,
),
),
vol.Coerce(float),
),
): vol.Coerce(float),
vol.Optional(
CONF_DAMPING_EVENING,
default=self.config_entry.options.get(
CONF_DAMPING_EVENING, DEFAULT_DAMPING
CONF_DAMPING_EVENING, 0.0
),
): vol.All(
selector.NumberSelector(
selector.NumberSelectorConfig(
min=0,
max=1,
step=0.01,
mode=selector.NumberSelectorMode.BOX,
),
),
vol.Coerce(float),
),
): vol.Coerce(float),
vol.Optional(
CONF_INVERTER_SIZE,
description={
@@ -212,89 +149,8 @@ class ForecastSolarOptionFlowHandler(OptionsFlow):
CONF_INVERTER_SIZE
)
},
): vol.All(
selector.NumberSelector(
selector.NumberSelectorConfig(
min=1,
step=1,
mode=selector.NumberSelectorMode.BOX,
),
),
vol.Coerce(int),
),
): vol.Coerce(int),
}
),
errors=errors,
)
class PlaneSubentryFlowHandler(ConfigSubentryFlow):
"""Handle a subentry flow for adding/editing a plane."""
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Handle the user step to add a new plane."""
entry = self._get_entry()
planes_count = len(entry.get_subentries_of_type(SUBENTRY_TYPE_PLANE))
if planes_count >= MAX_PLANES:
return self.async_abort(reason="max_planes")
if planes_count >= 1 and not entry.options.get(CONF_API_KEY):
return self.async_abort(reason="api_key_required")
if user_input is not None:
return self.async_create_entry(
title=f"{user_input[CONF_DECLINATION]}° / {user_input[CONF_AZIMUTH]}° / {user_input[CONF_MODULES_POWER]}W",
data={
CONF_DECLINATION: user_input[CONF_DECLINATION],
CONF_AZIMUTH: user_input[CONF_AZIMUTH],
CONF_MODULES_POWER: user_input[CONF_MODULES_POWER],
},
)
return self.async_show_form(
step_id="user",
data_schema=self.add_suggested_values_to_schema(
PLANE_SCHEMA,
{
CONF_DECLINATION: DEFAULT_DECLINATION,
CONF_AZIMUTH: DEFAULT_AZIMUTH,
CONF_MODULES_POWER: DEFAULT_MODULES_POWER,
},
),
)
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Handle reconfiguration of an existing plane."""
subentry = self._get_reconfigure_subentry()
if user_input is not None:
entry = self._get_entry()
if self._async_update(
entry,
subentry,
data={
CONF_DECLINATION: user_input[CONF_DECLINATION],
CONF_AZIMUTH: user_input[CONF_AZIMUTH],
CONF_MODULES_POWER: user_input[CONF_MODULES_POWER],
},
title=f"{user_input[CONF_DECLINATION]}° / {user_input[CONF_AZIMUTH]}° / {user_input[CONF_MODULES_POWER]}W",
):
if not entry.update_listeners:
self.hass.config_entries.async_schedule_reload(entry.entry_id)
return self.async_abort(reason="reconfigure_successful")
return self.async_show_form(
step_id="reconfigure",
data_schema=self.add_suggested_values_to_schema(
PLANE_SCHEMA,
{
CONF_DECLINATION: subentry.data[CONF_DECLINATION],
CONF_AZIMUTH: subentry.data[CONF_AZIMUTH],
CONF_MODULES_POWER: subentry.data[CONF_MODULES_POWER],
},
),
)

View File

@@ -14,9 +14,3 @@ CONF_DAMPING = "damping"
CONF_DAMPING_MORNING = "damping_morning"
CONF_DAMPING_EVENING = "damping_evening"
CONF_INVERTER_SIZE = "inverter_size"
DEFAULT_DECLINATION = 25
DEFAULT_AZIMUTH = 180
DEFAULT_MODULES_POWER = 10000
DEFAULT_DAMPING = 0.0
MAX_PLANES = 4
SUBENTRY_TYPE_PLANE = "plane"

View File

@@ -4,7 +4,7 @@ from __future__ import annotations
from datetime import timedelta
from forecast_solar import Estimate, ForecastSolar, ForecastSolarConnectionError, Plane
from forecast_solar import Estimate, ForecastSolar, ForecastSolarConnectionError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE
@@ -19,10 +19,8 @@ from .const import (
CONF_DECLINATION,
CONF_INVERTER_SIZE,
CONF_MODULES_POWER,
DEFAULT_DAMPING,
DOMAIN,
LOGGER,
SUBENTRY_TYPE_PLANE,
)
type ForecastSolarConfigEntry = ConfigEntry[ForecastSolarDataUpdateCoordinator]
@@ -32,7 +30,6 @@ class ForecastSolarDataUpdateCoordinator(DataUpdateCoordinator[Estimate]):
"""The Forecast.Solar Data Update Coordinator."""
config_entry: ForecastSolarConfigEntry
forecast: ForecastSolar
def __init__(self, hass: HomeAssistant, entry: ForecastSolarConfigEntry) -> None:
"""Initialize the Forecast.Solar coordinator."""
@@ -46,34 +43,17 @@ class ForecastSolarDataUpdateCoordinator(DataUpdateCoordinator[Estimate]):
) is not None and inverter_size > 0:
inverter_size = inverter_size / 1000
# Build the list of planes from subentries.
plane_subentries = entry.get_subentries_of_type(SUBENTRY_TYPE_PLANE)
# The first plane subentry is the main plane
main_plane = plane_subentries[0]
# Additional planes
planes: list[Plane] = [
Plane(
declination=subentry.data[CONF_DECLINATION],
azimuth=(subentry.data[CONF_AZIMUTH] - 180),
kwp=(subentry.data[CONF_MODULES_POWER] / 1000),
)
for subentry in plane_subentries[1:]
]
self.forecast = ForecastSolar(
api_key=api_key,
session=async_get_clientsession(hass),
latitude=entry.data[CONF_LATITUDE],
longitude=entry.data[CONF_LONGITUDE],
declination=main_plane.data[CONF_DECLINATION],
azimuth=(main_plane.data[CONF_AZIMUTH] - 180),
kwp=(main_plane.data[CONF_MODULES_POWER] / 1000),
damping_morning=entry.options.get(CONF_DAMPING_MORNING, DEFAULT_DAMPING),
damping_evening=entry.options.get(CONF_DAMPING_EVENING, DEFAULT_DAMPING),
declination=entry.options[CONF_DECLINATION],
azimuth=(entry.options[CONF_AZIMUTH] - 180),
kwp=(entry.options[CONF_MODULES_POWER] / 1000),
damping_morning=entry.options.get(CONF_DAMPING_MORNING, 0.0),
damping_evening=entry.options.get(CONF_DAMPING_EVENING, 0.0),
inverter=inverter_size,
planes=planes,
)
# Free account have a resolution of 1 hour, using that as the default

View File

@@ -28,13 +28,6 @@ async def async_get_config_entry_diagnostics(
"title": entry.title,
"data": async_redact_data(entry.data, TO_REDACT),
"options": async_redact_data(entry.options, TO_REDACT),
"subentries": [
{
"data": dict(subentry.data),
"title": subentry.title,
}
for subentry in entry.subentries.values()
],
},
"data": {
"energy_production_today": coordinator.data.energy_production_today,

View File

@@ -14,37 +14,6 @@
}
}
},
"config_subentries": {
"plane": {
"abort": {
"api_key_required": "An API key is required to add more than one plane. You can configure it in the integration options.",
"max_planes": "You can add a maximum of 4 planes.",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]"
},
"entry_type": "Plane",
"initiate_flow": {
"user": "Add plane"
},
"step": {
"reconfigure": {
"data": {
"azimuth": "[%key:component::forecast_solar::config::step::user::data::azimuth%]",
"declination": "[%key:component::forecast_solar::config::step::user::data::declination%]",
"modules_power": "[%key:component::forecast_solar::config::step::user::data::modules_power%]"
},
"description": "Edit the solar plane configuration."
},
"user": {
"data": {
"azimuth": "[%key:component::forecast_solar::config::step::user::data::azimuth%]",
"declination": "[%key:component::forecast_solar::config::step::user::data::declination%]",
"modules_power": "[%key:component::forecast_solar::config::step::user::data::modules_power%]"
},
"description": "Add a solar plane. Multiple planes are supported with a Forecast.Solar API subscription."
}
}
}
},
"entity": {
"sensor": {
"energy_current_hour": {
@@ -82,26 +51,20 @@
}
}
},
"exceptions": {
"api_key_required": {
"message": "An API key is required when more than one plane is configured"
},
"no_plane": {
"message": "No plane configured, cannot set up Forecast.Solar"
}
},
"options": {
"error": {
"api_key_required": "An API key is required to add more than one plane. You can configure it in the integration options.",
"invalid_api_key": "[%key:common::config_flow::error::invalid_api_key%]"
},
"step": {
"init": {
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]",
"azimuth": "[%key:component::forecast_solar::config::step::user::data::azimuth%]",
"damping_evening": "Damping factor: adjusts the results in the evening",
"damping_morning": "Damping factor: adjusts the results in the morning",
"inverter_size": "Inverter size (Watt)"
"declination": "[%key:component::forecast_solar::config::step::user::data::declination%]",
"inverter_size": "Inverter size (Watt)",
"modules_power": "[%key:component::forecast_solar::config::step::user::data::modules_power%]"
},
"description": "These values allow tweaking the Forecast.Solar result. Please refer to the documentation if a field is unclear."
}

View File

@@ -72,7 +72,7 @@
"cold_tea": {
"fix_flow": {
"abort": {
"not_tea_time": "Cannot reheat the tea at this time"
"not_tea_time": "Can not re-heat the tea at this time"
},
"step": {}
},

View File

@@ -34,7 +34,6 @@ from .const import (
EVENT_TYPE_OFF,
EVENT_TYPE_ON,
MANUFACTURER,
NETATMO_ALIM_STATUS_ONLINE,
NETATMO_CREATE_CAMERA,
SERVICE_SET_CAMERA_LIGHT,
SERVICE_SET_PERSON_AWAY,
@@ -175,16 +174,18 @@ class NetatmoCamera(NetatmoModuleEntity, Camera):
self._monitoring = False
elif event_type in [EVENT_TYPE_CONNECTION, EVENT_TYPE_ON]:
_LOGGER.debug(
"Camera %s has received %s event, turning on and enabling streaming if applicable",
"Camera %s has received %s event, turning on and enabling streaming",
data["camera_id"],
event_type,
)
if self.device_type != "NDB":
self._attr_is_streaming = True
self._attr_is_streaming = True
self._monitoring = True
elif event_type == EVENT_TYPE_LIGHT_MODE:
if data.get("sub_type"):
self._light_state = data["sub_type"]
self._attr_extra_state_attributes.update(
{"light_state": self._light_state}
)
else:
_LOGGER.debug(
"Camera %s has received light mode event without sub_type",
@@ -224,20 +225,6 @@ class NetatmoCamera(NetatmoModuleEntity, Camera):
supported_features |= CameraEntityFeature.STREAM
return supported_features
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return entity specific state attributes."""
return {
"id": self.device.entity_id,
"monitoring": self._monitoring,
"sd_status": self.device.sd_status,
"alim_status": self.device.alim_status,
"is_local": self.device.is_local,
"vpn_url": self.device.vpn_url,
"local_url": self.device.local_url,
"light_state": self._light_state,
}
async def async_turn_off(self) -> None:
"""Turn off camera."""
await self.device.async_monitoring_off()
@@ -261,10 +248,7 @@ class NetatmoCamera(NetatmoModuleEntity, Camera):
self._attr_is_on = self.device.alim_status is not None
self._attr_available = self.device.alim_status is not None
if self.device_type == "NDB":
self._monitoring = self.device.alim_status == NETATMO_ALIM_STATUS_ONLINE
elif self.device.monitoring is not None:
self._monitoring = self.device.monitoring
if self.device.monitoring is not None:
self._attr_is_streaming = self.device.monitoring
self._attr_motion_detection_enabled = self.device.monitoring
@@ -272,6 +256,19 @@ class NetatmoCamera(NetatmoModuleEntity, Camera):
self.process_events(self.device.events)
)
self._attr_extra_state_attributes.update(
{
"id": self.device.entity_id,
"monitoring": self._monitoring,
"sd_status": self.device.sd_status,
"alim_status": self.device.alim_status,
"is_local": self.device.is_local,
"vpn_url": self.device.vpn_url,
"local_url": self.device.local_url,
"light_state": self._light_state,
}
)
def process_events(self, event_list: list[NaEvent]) -> dict:
"""Add meta data to events."""
events = {}

View File

@@ -215,15 +215,5 @@ WEBHOOK_ACTIVATION = "webhook_activation"
WEBHOOK_DEACTIVATION = "webhook_deactivation"
WEBHOOK_NACAMERA_CONNECTION = "NACamera-connection"
WEBHOOK_NOCAMERA_CONNECTION = "NOC-connection"
WEBHOOK_NDB_CONNECTION = "NDB-connection"
WEBHOOK_PUSH_TYPE = "push_type"
CAMERA_CONNECTION_WEBHOOKS = [
WEBHOOK_NACAMERA_CONNECTION,
WEBHOOK_NOCAMERA_CONNECTION,
WEBHOOK_NDB_CONNECTION,
]
# Alimentation status (alim_status) for cameras and door bells (NDB).
# For NDB there is no monitoring attribute in status but only alim_status.
# 2 = Full power/online for NDB (and also Correct power adapter for NACamera).
NETATMO_ALIM_STATUS_ONLINE = 2
CAMERA_CONNECTION_WEBHOOKS = [WEBHOOK_NACAMERA_CONNECTION, WEBHOOK_NOCAMERA_CONNECTION]

View File

@@ -17,7 +17,6 @@ from opendisplay import (
from homeassistant.components.bluetooth import async_ble_device_from_address
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv, device_registry as dr
@@ -28,20 +27,15 @@ if TYPE_CHECKING:
from opendisplay.models import FirmwareVersion
from .const import DOMAIN
from .coordinator import OpenDisplayCoordinator
from .services import async_setup_services
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
_BASE_PLATFORMS: list[Platform] = []
_FLEX_PLATFORMS = [Platform.SENSOR]
@dataclass
class OpenDisplayRuntimeData:
"""Runtime data for an OpenDisplay config entry."""
coordinator: OpenDisplayCoordinator
firmware: FirmwareVersion
device_config: GlobalConfig
is_flex: bool
@@ -83,8 +77,13 @@ async def async_setup_entry(hass: HomeAssistant, entry: OpenDisplayConfigEntry)
if TYPE_CHECKING:
assert device_config is not None
coordinator = OpenDisplayCoordinator(hass, address)
entry.runtime_data = OpenDisplayRuntimeData(
firmware=fw,
device_config=device_config,
is_flex=is_flex,
)
# Will be moved to DeviceInfo object in entity.py once entities are added
manufacturer = device_config.manufacturer
display = device_config.displays[0]
color_scheme_enum = display.color_scheme_enum
@@ -98,16 +97,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: OpenDisplayConfigEntry)
if display.screen_diagonal_inches is not None
else f"{display.pixel_width}x{display.pixel_height}"
)
dr.async_get(hass).async_get_or_create(
config_entry_id=entry.entry_id,
connections={(CONNECTION_BLUETOOTH, address)},
manufacturer=manufacturer.manufacturer_name,
model=f"{size} {color_scheme}",
sw_version=f"{fw['major']}.{fw['minor']}",
hw_version=(
f"{manufacturer.board_type_name or manufacturer.board_type}"
f" rev. {manufacturer.board_revision}"
)
hw_version=f"{manufacturer.board_type_name or manufacturer.board_type} rev. {manufacturer.board_revision}"
if is_flex
else None,
configuration_url="https://opendisplay.org/firmware/config/"
@@ -115,18 +112,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: OpenDisplayConfigEntry)
else None,
)
entry.runtime_data = OpenDisplayRuntimeData(
coordinator=coordinator,
firmware=fw,
device_config=device_config,
is_flex=is_flex,
)
await hass.config_entries.async_forward_entry_setups(
entry, _FLEX_PLATFORMS if is_flex else _BASE_PLATFORMS
)
entry.async_on_unload(coordinator.async_start())
return True
@@ -139,6 +124,4 @@ async def async_unload_entry(
with contextlib.suppress(asyncio.CancelledError):
await task
return await hass.config_entries.async_unload_platforms(
entry, _FLEX_PLATFORMS if entry.runtime_data.is_flex else _BASE_PLATFORMS
)
return True

View File

@@ -1,86 +0,0 @@
"""Passive BLE coordinator for OpenDisplay devices."""
from __future__ import annotations
from dataclasses import dataclass
import logging
from opendisplay import MANUFACTURER_ID, parse_advertisement
from opendisplay.models.advertisement import AdvertisementData
from homeassistant.components.bluetooth import (
BluetoothChange,
BluetoothScanningMode,
BluetoothServiceInfoBleak,
)
from homeassistant.components.bluetooth.passive_update_coordinator import (
PassiveBluetoothDataUpdateCoordinator,
)
from homeassistant.core import HomeAssistant, callback
_LOGGER: logging.Logger = logging.getLogger(__package__)
@dataclass
class OpenDisplayUpdate:
"""Parsed advertisement data for one OpenDisplay device."""
address: str
advertisement: AdvertisementData
class OpenDisplayCoordinator(PassiveBluetoothDataUpdateCoordinator):
"""Coordinator for passive BLE advertisement updates from an OpenDisplay device."""
def __init__(self, hass: HomeAssistant, address: str) -> None:
"""Initialize the coordinator."""
super().__init__(
hass,
_LOGGER,
address,
BluetoothScanningMode.PASSIVE,
connectable=True,
)
self.data: OpenDisplayUpdate | None = None
@callback
def _async_handle_unavailable(
self, service_info: BluetoothServiceInfoBleak
) -> None:
"""Handle the device going unavailable."""
if self._available:
_LOGGER.info("%s: Device is unavailable", service_info.address)
super()._async_handle_unavailable(service_info)
@callback
def _async_handle_bluetooth_event(
self,
service_info: BluetoothServiceInfoBleak,
change: BluetoothChange,
) -> None:
"""Handle a Bluetooth advertisement event."""
if not self._available:
_LOGGER.info("%s: Device is available again", service_info.address)
if MANUFACTURER_ID not in service_info.manufacturer_data:
super()._async_handle_bluetooth_event(service_info, change)
return
try:
advertisement = parse_advertisement(
service_info.manufacturer_data[MANUFACTURER_ID]
)
except ValueError as err:
_LOGGER.debug(
"%s: Failed to parse advertisement data: %s",
service_info.address,
err,
exc_info=True,
)
else:
self.data = OpenDisplayUpdate(
address=service_info.address,
advertisement=advertisement,
)
super()._async_handle_bluetooth_event(service_info, change)

View File

@@ -1,31 +0,0 @@
"""Base entity for OpenDisplay devices."""
from __future__ import annotations
from homeassistant.components.bluetooth.passive_update_coordinator import (
PassiveBluetoothCoordinatorEntity,
)
from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH, DeviceInfo
from homeassistant.helpers.entity import EntityDescription
from .coordinator import OpenDisplayCoordinator
class OpenDisplayEntity(PassiveBluetoothCoordinatorEntity[OpenDisplayCoordinator]):
"""Base class for all OpenDisplay entities."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: OpenDisplayCoordinator,
description: EntityDescription,
) -> None:
"""Initialize the entity."""
super().__init__(coordinator)
self.entity_description = description
self._attr_unique_id = f"{coordinator.address}-{description.key}"
self._attr_device_info = DeviceInfo(
connections={(CONNECTION_BLUETOOTH, coordinator.address)},
)

View File

@@ -6,7 +6,9 @@ rules:
comment: |
The `opendisplay` integration is a `local_push` integration that does not perform periodic polling.
brands: done
common-modules: done
common-modules:
status: exempt
comment: Integration does not currently use entities or a DataUpdateCoordinator.
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
@@ -14,9 +16,15 @@ rules:
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup: done
entity-unique-id: done
has-entity-name: done
entity-event-setup:
status: exempt
comment: Integration does not currently provide any entities.
entity-unique-id:
status: exempt
comment: Integration does not currently provide any entities.
has-entity-name:
status: exempt
comment: Integration does not currently provide any entities.
runtime-data: done
test-before-configure: done
test-before-setup: done
@@ -29,10 +37,16 @@ rules:
status: exempt
comment: Integration has no options flow.
docs-installation-parameters: done
entity-unavailable: done
entity-unavailable:
status: exempt
comment: Integration does not currently provide any entities.
integration-owner: done
log-when-unavailable: done
parallel-updates: done
log-when-unavailable:
status: exempt
comment: Integration does not currently implement any entities or background polling.
parallel-updates:
status: exempt
comment: Integration does not provide any entities.
reauthentication-flow:
status: exempt
comment: Devices do not require authentication.
@@ -45,7 +59,9 @@ rules:
status: exempt
comment: The device's BLE MAC address is both its unique identifier and does not change.
discovery: done
docs-data-update: todo
docs-data-update:
status: exempt
comment: Integration does not poll or push data to entities.
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
@@ -55,10 +71,18 @@ rules:
dynamic-devices:
status: exempt
comment: Only one device per config entry. New devices are set up as new entries.
entity-category: done
entity-device-class: done
entity-disabled-by-default: done
entity-translations: done
entity-category:
status: exempt
comment: Integration does not provide any entities.
entity-device-class:
status: exempt
comment: Integration does not provide any entities.
entity-disabled-by-default:
status: exempt
comment: Integration does not provide any entities.
entity-translations:
status: exempt
comment: Integration does not provide any entities.
exception-translations: done
icon-translations: done
reconfiguration-flow:

View File

@@ -1,106 +0,0 @@
"""Sensor platform for OpenDisplay devices."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from opendisplay import voltage_to_percent
from opendisplay.models.advertisement import AdvertisementData
from opendisplay.models.enums import CapacityEstimator, PowerMode
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import (
PERCENTAGE,
EntityCategory,
UnitOfElectricPotential,
UnitOfTemperature,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import OpenDisplayConfigEntry
from .entity import OpenDisplayEntity
PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True)
class OpenDisplaySensorEntityDescription(SensorEntityDescription):
"""Describes an OpenDisplay sensor entity."""
value_fn: Callable[[AdvertisementData], float | int | None]
_TEMPERATURE_DESCRIPTION = OpenDisplaySensorEntityDescription(
key="temperature",
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda adv: adv.temperature_c,
)
_BATTERY_POWER_MODES = {PowerMode.BATTERY, PowerMode.SOLAR}
_BATTERY_VOLTAGE_DESCRIPTION = OpenDisplaySensorEntityDescription(
key="battery_voltage",
translation_key="battery_voltage",
device_class=SensorDeviceClass.VOLTAGE,
native_unit_of_measurement=UnitOfElectricPotential.MILLIVOLT,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda adv: adv.battery_mv,
)
async def async_setup_entry(
hass: HomeAssistant,
entry: OpenDisplayConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up OpenDisplay sensor entities."""
coordinator = entry.runtime_data.coordinator
power_config = entry.runtime_data.device_config.power
descriptions: list[OpenDisplaySensorEntityDescription] = [_TEMPERATURE_DESCRIPTION]
if power_config.power_mode_enum in _BATTERY_POWER_MODES:
capacity_estimator = power_config.capacity_estimator or CapacityEstimator.LI_ION
descriptions += [
_BATTERY_VOLTAGE_DESCRIPTION,
OpenDisplaySensorEntityDescription(
key="battery",
device_class=SensorDeviceClass.BATTERY,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda adv: voltage_to_percent(
adv.battery_mv, capacity_estimator
),
),
]
async_add_entities(
OpenDisplaySensorEntity(coordinator, description)
for description in descriptions
)
class OpenDisplaySensorEntity(OpenDisplayEntity, SensorEntity):
"""A sensor entity for an OpenDisplay device."""
entity_description: OpenDisplaySensorEntityDescription
@property
def native_value(self) -> float | int | None:
"""Return the sensor value."""
if self.coordinator.data is None:
return None
return self.entity_description.value_fn(self.coordinator.data.advertisement)

View File

@@ -27,13 +27,6 @@
}
}
},
"entity": {
"sensor": {
"battery_voltage": {
"name": "Battery voltage"
}
}
},
"exceptions": {
"device_not_found": {
"message": "Could not find Bluetooth device with address `{address}`."

View File

@@ -346,7 +346,7 @@
},
"exceptions": {
"cannot_connect": {
"message": "Value cannot be set because the device is not connected"
"message": "Value can not be set because the device is not connected"
},
"write_rejected": {
"message": "The device rejected the value for {entity}: {value}"

View File

@@ -122,7 +122,7 @@
},
"exceptions": {
"cannot_connect": {
"message": "Cannot connect to Rehlko servers."
"message": "Can not connect to Rehlko servers."
},
"invalid_auth": {
"message": "Authentication failed for email {email}."

View File

@@ -19,6 +19,7 @@ is_option_selected:
required: true
selector:
state:
attribute: options
hide_states:
- unavailable
- unknown

View File

@@ -61,9 +61,7 @@ rules:
dynamic-devices: todo
entity-category: done
entity-device-class: done
entity-disabled-by-default:
status: exempt
comment: No noisy or non-essential entities to disable.
entity-disabled-by-default: todo
entity-translations: done
exception-translations: todo
icon-translations: todo

View File

@@ -30,8 +30,4 @@ class TeslaUserImplementation(AuthImplementation):
@property
def extra_authorize_data(self) -> dict[str, Any]:
"""Extra data that needs to be appended to the authorize url."""
return {
"prompt": "login",
"prompt_missing_scopes": "true",
"scope": " ".join(SCOPES),
}
return {"prompt": "login", "scope": " ".join(SCOPES)}

View File

@@ -50,7 +50,7 @@
"title": "[%key:common::config_flow::title::oauth2_pick_implementation%]"
},
"reauth_confirm": {
"description": "The {name} integration needs to re-authenticate your account. Reauthentication refreshes the Tesla API permissions granted to Home Assistant, including any newly enabled scopes.",
"description": "The {name} integration needs to re-authenticate your account",
"title": "[%key:common::config_flow::title::reauth%]"
},
"registration_complete": {
@@ -60,7 +60,7 @@
"data_description": {
"qr_code": "Scan this QR code with your phone to set up the virtual key."
},
"description": "To enable command signing, you must open the Tesla app, select your vehicle, and then visit the following URL to set up a virtual key. You must repeat this process for each vehicle.\n\n{virtual_key_url}\n\nIf you later enable additional Tesla API permissions, reauthenticate the integration to refresh the granted scopes.",
"description": "To enable command signing, you must open the Tesla app, select your vehicle, and then visit the following URL to set up a virtual key. You must repeat this process for each vehicle.\n\n{virtual_key_url}",
"title": "Command signing"
}
}

View File

@@ -6,10 +6,10 @@
"no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]"
},
"error": {
"bulb_time_out": "Cannot connect to the bulb. Maybe the bulb is offline or a wrong IP was entered. Please turn on the light and try again!",
"bulb_time_out": "Can not connect to the bulb. Maybe the bulb is offline or a wrong IP was entered. Please turn on the light and try again!",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"no_ip": "Not a valid IP address.",
"no_wiz_light": "The bulb cannot be connected via WiZ integration.",
"no_wiz_light": "The bulb cannot be connected via WiZ Platform integration.",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"flow_title": "{name} ({host})",
@@ -26,7 +26,7 @@
"data": {
"host": "[%key:common::config_flow::data::ip%]"
},
"description": "If you leave the IP address empty, discovery will be used to find devices."
"description": "If you leave the IP Address empty, discovery will be used to find devices."
}
}
},

View File

@@ -142,7 +142,6 @@ FLOWS = {
"deconz",
"decora_wifi",
"deluge",
"denon_rs232",
"denonavr",
"devialet",
"devolo_home_control",

View File

@@ -1323,12 +1323,6 @@
"iot_class": "local_push",
"name": "Denon AVR Network Receivers"
},
"denon_rs232": {
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_push",
"name": "Denon RS232"
},
"heos": {
"integration_type": "hub",
"config_flow": true,

View File

@@ -772,11 +772,11 @@ class DeviceRegistry(BaseRegistry[dict[str, list[dict[str, Any]]]]):
devices: ActiveDeviceRegistryItems
deleted_devices: DeviceRegistryItems[DeletedDeviceEntry]
_device_data: dict[str, DeviceEntry]
_loaded_event: asyncio.Event | None = None
def __init__(self, hass: HomeAssistant) -> None:
"""Initialize the device registry."""
self.hass = hass
self._loaded_event = asyncio.Event()
self._store = DeviceRegistryStore(
hass,
STORAGE_VERSION_MAJOR,
@@ -786,11 +786,6 @@ class DeviceRegistry(BaseRegistry[dict[str, list[dict[str, Any]]]]):
serialize_in_event_loop=False,
)
@callback
def async_setup(self) -> None:
"""Set up the registry."""
self._loaded_event = asyncio.Event()
@callback
def async_get(self, device_id: str) -> DeviceEntry | None:
"""Get device.
@@ -1470,9 +1465,6 @@ class DeviceRegistry(BaseRegistry[dict[str, list[dict[str, Any]]]]):
async def _async_load(self) -> None:
"""Load the device registry."""
assert self._loaded_event is not None
assert not self._loaded_event.is_set()
async_setup_cleanup(self.hass, self)
data = await self._store.async_load()
@@ -1573,12 +1565,8 @@ class DeviceRegistry(BaseRegistry[dict[str, list[dict[str, Any]]]]):
self._loaded_event.set()
async def async_wait_loaded(self) -> None:
"""Wait until the device registry is fully loaded.
Will only wait if the registry had already been set up.
"""
if self._loaded_event is not None:
await self._loaded_event.wait()
"""Wait until the device registry is fully loaded."""
await self._loaded_event.wait()
@callback
def _data_to_save(self) -> dict[str, Any]:
@@ -1723,13 +1711,14 @@ class DeviceRegistry(BaseRegistry[dict[str, list[dict[str, Any]]]]):
@singleton(DATA_REGISTRY)
def async_get(hass: HomeAssistant) -> DeviceRegistry:
"""Get device registry."""
return DeviceRegistry(hass)
return hass.data[DATA_REGISTRY]
def async_setup(hass: HomeAssistant) -> None:
"""Set up device registry."""
assert DATA_REGISTRY not in hass.data
async_get(hass).async_setup()
if DATA_REGISTRY in hass.data:
raise RuntimeError("Device registry is already set up")
hass.data[DATA_REGISTRY] = DeviceRegistry(hass)
async def async_load(hass: HomeAssistant, *, load_empty: bool = False) -> None:

3
requirements_all.txt generated
View File

@@ -803,9 +803,6 @@ deluge-client==1.10.2
# homeassistant.components.lametric
demetriek==1.3.0
# homeassistant.components.denon_rs232
denon-rs232==4.0.0
# homeassistant.components.denonavr
denonavr==1.3.2

View File

@@ -715,9 +715,6 @@ deluge-client==1.10.2
# homeassistant.components.lametric
demetriek==1.3.0
# homeassistant.components.denon_rs232
denon-rs232==4.0.0
# homeassistant.components.denonavr
denonavr==1.3.2

View File

@@ -1,55 +1,4 @@
# serializer version: 1
# name: test_entities[binary_sensor.jar_charging-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.jar_charging',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Charging',
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.BATTERY_CHARGING: 'battery_charging'>,
'original_icon': None,
'original_name': 'Charging',
'platform': 'casper_glow',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'aa:bb:cc:dd:ee:ff_charging',
'unit_of_measurement': None,
})
# ---
# name: test_entities[binary_sensor.jar_charging-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'battery_charging',
'friendly_name': 'Jar Charging',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.jar_charging',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_entities[binary_sensor.jar_dimming_paused-entry]
EntityRegistryEntrySnapshot({
'aliases': list([

View File

@@ -1,56 +0,0 @@
# serializer version: 1
# name: test_entities[sensor.jar_battery-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.jar_battery',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Battery',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.BATTERY: 'battery'>,
'original_icon': None,
'original_name': 'Battery',
'platform': 'casper_glow',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'aa:bb:cc:dd:ee:ff_battery',
'unit_of_measurement': '%',
})
# ---
# name: test_entities[sensor.jar_battery-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'battery',
'friendly_name': 'Jar Battery',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'sensor.jar_battery',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---

View File

@@ -1,6 +1,5 @@
"""Test the Casper Glow binary sensor platform."""
from collections.abc import Callable
from unittest.mock import MagicMock, patch
from pycasperglow import GlowState
@@ -15,8 +14,7 @@ from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
PAUSED_ENTITY_ID = "binary_sensor.jar_dimming_paused"
CHARGING_ENTITY_ID = "binary_sensor.jar_charging"
ENTITY_ID = "binary_sensor.jar_dimming_paused"
async def test_entities(
@@ -39,31 +37,31 @@ async def test_entities(
[(True, STATE_ON), (False, STATE_OFF)],
ids=["paused", "not-paused"],
)
async def test_paused_state_update(
async def test_binary_sensor_state_update(
hass: HomeAssistant,
mock_casper_glow: MagicMock,
mock_config_entry: MockConfigEntry,
fire_callbacks: Callable[[GlowState], None],
is_paused: bool,
expected_state: str,
) -> None:
"""Test that the paused binary sensor reflects is_paused state changes."""
"""Test that the binary sensor reflects is_paused state changes."""
with patch(
"homeassistant.components.casper_glow.PLATFORMS", [Platform.BINARY_SENSOR]
):
await setup_integration(hass, mock_config_entry)
fire_callbacks(GlowState(is_paused=is_paused))
state = hass.states.get(PAUSED_ENTITY_ID)
cb = mock_casper_glow.register_callback.call_args[0][0]
cb(GlowState(is_paused=is_paused))
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == expected_state
async def test_paused_ignores_none_state(
async def test_binary_sensor_ignores_none_paused_state(
hass: HomeAssistant,
mock_casper_glow: MagicMock,
mock_config_entry: MockConfigEntry,
fire_callbacks: Callable[[GlowState], None],
) -> None:
"""Test that a callback with is_paused=None does not overwrite the state."""
with patch(
@@ -71,64 +69,16 @@ async def test_paused_ignores_none_state(
):
await setup_integration(hass, mock_config_entry)
cb = mock_casper_glow.register_callback.call_args[0][0]
# Set a known value first
fire_callbacks(GlowState(is_paused=True))
state = hass.states.get(PAUSED_ENTITY_ID)
cb(GlowState(is_paused=True))
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == STATE_ON
# Callback with no is_paused data — state should remain unchanged
fire_callbacks(GlowState(is_on=True))
state = hass.states.get(PAUSED_ENTITY_ID)
assert state is not None
assert state.state == STATE_ON
@pytest.mark.parametrize(
("is_charging", "expected_state"),
[(True, STATE_ON), (False, STATE_OFF)],
ids=["charging", "not-charging"],
)
async def test_charging_state_update(
hass: HomeAssistant,
mock_casper_glow: MagicMock,
mock_config_entry: MockConfigEntry,
fire_callbacks: Callable[[GlowState], None],
is_charging: bool,
expected_state: str,
) -> None:
"""Test that the charging binary sensor reflects is_charging state changes."""
with patch(
"homeassistant.components.casper_glow.PLATFORMS", [Platform.BINARY_SENSOR]
):
await setup_integration(hass, mock_config_entry)
fire_callbacks(GlowState(is_charging=is_charging))
state = hass.states.get(CHARGING_ENTITY_ID)
assert state is not None
assert state.state == expected_state
async def test_charging_ignores_none_state(
hass: HomeAssistant,
mock_casper_glow: MagicMock,
mock_config_entry: MockConfigEntry,
fire_callbacks: Callable[[GlowState], None],
) -> None:
"""Test that a callback with is_charging=None does not overwrite the state."""
with patch(
"homeassistant.components.casper_glow.PLATFORMS", [Platform.BINARY_SENSOR]
):
await setup_integration(hass, mock_config_entry)
# Set a known value first
fire_callbacks(GlowState(is_charging=True))
state = hass.states.get(CHARGING_ENTITY_ID)
assert state is not None
assert state.state == STATE_ON
# Callback with no is_charging data — state should remain unchanged
fire_callbacks(GlowState(is_on=True))
state = hass.states.get(CHARGING_ENTITY_ID)
cb(GlowState(is_on=True))
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == STATE_ON

View File

@@ -153,7 +153,7 @@ async def test_select_ignores_remaining_time_updates(
fire_callbacks: Callable[[GlowState], None],
) -> None:
"""Test that callbacks with only remaining time do not change the select state."""
fire_callbacks(GlowState(dimming_time_remaining_ms=2_640_000))
fire_callbacks(GlowState(dimming_time_remaining_ms=44))
state = hass.states.get(ENTITY_ID)
assert state is not None

View File

@@ -1,72 +0,0 @@
"""Test the Casper Glow sensor platform."""
from collections.abc import Callable
from unittest.mock import MagicMock, patch
from pycasperglow import BatteryLevel, GlowState
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
BATTERY_ENTITY_ID = "sensor.jar_battery"
async def test_entities(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_casper_glow: MagicMock,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test all sensor entities match the snapshot."""
with patch("homeassistant.components.casper_glow.PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.parametrize(
("battery_level", "expected_state"),
[
(BatteryLevel.PCT_75, "75"),
(BatteryLevel.PCT_50, "50"),
],
)
async def test_battery_state(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_casper_glow: MagicMock,
battery_level: BatteryLevel,
expected_state: str,
) -> None:
"""Test that the battery sensor reflects device state at setup."""
mock_casper_glow.state = GlowState(battery_level=battery_level)
with patch("homeassistant.components.casper_glow.PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_config_entry)
state = hass.states.get(BATTERY_ENTITY_ID)
assert state is not None
assert state.state == expected_state
async def test_battery_state_updated_via_callback(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_casper_glow: MagicMock,
fire_callbacks: Callable[[GlowState], None],
) -> None:
"""Test battery sensor updates when a device callback fires."""
with patch("homeassistant.components.casper_glow.PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_config_entry)
fire_callbacks(GlowState(battery_level=BatteryLevel.PCT_50))
state = hass.states.get(BATTERY_ENTITY_ID)
assert state is not None
assert state.state == "50"

View File

@@ -1,4 +0,0 @@
"""Tests for the Denon RS232 integration."""
MOCK_DEVICE = "/dev/ttyUSB0"
MOCK_MODEL = "avr_3805"

View File

@@ -1,165 +0,0 @@
"""Test fixtures for the Denon RS232 integration."""
from __future__ import annotations
from typing import Literal
from unittest.mock import AsyncMock, patch
from denon_rs232 import (
DenonReceiver,
DigitalInputMode,
InputSource,
MainZoneState,
ReceiverState,
TunerBand,
TunerMode,
ZoneState,
)
from denon_rs232.models import MODELS
import pytest
from homeassistant.components.denon_rs232.const import DOMAIN
from homeassistant.const import CONF_DEVICE, CONF_MODEL
from homeassistant.core import HomeAssistant
from . import MOCK_DEVICE, MOCK_MODEL
from tests.common import MockConfigEntry
ZoneName = Literal["main", "zone_2", "zone_3"]
class MockState(ReceiverState):
"""Receiver state with helpers for zone-oriented tests."""
def get_zone(self, zone: ZoneName) -> ZoneState:
"""Return the requested zone state."""
if zone == "main":
return self.main_zone
return getattr(self, zone)
class MockReceiver(DenonReceiver):
"""Receiver test double built on the real receiver/player objects."""
def __init__(self, state: MockState) -> None:
"""Initialize the mock receiver."""
super().__init__(MOCK_DEVICE, model=MODELS[MOCK_MODEL])
self._connected = True
self._load_state(state)
self._send_command = AsyncMock()
self._query = AsyncMock()
self.connect = AsyncMock(side_effect=self._mock_connect)
self.query_state = AsyncMock()
self.disconnect = AsyncMock(side_effect=self._mock_disconnect)
def get_zone(self, zone: ZoneName):
"""Return the matching live player object."""
if zone == "main":
return self.main
if zone == "zone_2":
return self.zone_2
return self.zone_3
def mock_state(self, state: MockState | None) -> None:
"""Push a state update through the receiver."""
self._connected = state is not None
if state is not None:
self._load_state(state)
self._notify_subscribers()
async def _mock_connect(self) -> None:
"""Pretend to open the serial connection."""
self._connected = True
async def _mock_disconnect(self) -> None:
"""Pretend to close the serial connection."""
self._connected = False
def _load_state(self, state: MockState) -> None:
"""Swap in a new state object and rebind the live players to it."""
self._state = state
self.main._state = state.main_zone
self.zone_2._state = state.zone_2
self.zone_3._state = state.zone_3
def _default_state() -> MockState:
"""Return a ReceiverState with typical defaults."""
return MockState(
power=True,
main_zone=MainZoneState(
power=True,
volume=-30.0,
volume_min=-80,
volume_max=10,
mute=False,
input_source=InputSource.CD,
surround_mode="STEREO",
digital_input=DigitalInputMode.AUTO,
tuner_band=TunerBand.FM,
tuner_mode=TunerMode.AUTO,
),
zone_2=ZoneState(
power=True,
input_source=InputSource.TUNER,
volume=-20.0,
),
zone_3=ZoneState(
power=False,
input_source=InputSource.CD,
volume=-35.0,
),
)
@pytest.fixture
def initial_receiver_state(request: pytest.FixtureRequest) -> MockState:
"""Return the initial receiver state for a test."""
state = _default_state()
if getattr(request, "param", None) == "main_only":
state.zone_2 = ZoneState()
state.zone_3 = ZoneState()
return state
@pytest.fixture
def mock_receiver(initial_receiver_state: MockState) -> MockReceiver:
"""Create a mock DenonReceiver."""
return MockReceiver(initial_receiver_state)
@pytest.fixture
def mock_config_entry(hass: HomeAssistant) -> MockConfigEntry:
"""Create a mock config entry."""
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL},
title=MODELS[MOCK_MODEL].name,
)
entry.add_to_hass(hass)
return entry
@pytest.fixture(autouse=True)
async def mock_usb_component(hass: HomeAssistant) -> None:
"""Mock the USB component to prevent setup failures."""
hass.config.components.add("usb")
@pytest.fixture
async def init_components(
hass: HomeAssistant,
mock_usb_component: None,
mock_receiver: MockReceiver,
mock_config_entry: MockConfigEntry,
) -> None:
"""Initialize the Denon component."""
with patch(
"homeassistant.components.denon_rs232.DenonReceiver",
return_value=mock_receiver,
):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()

View File

@@ -1,93 +0,0 @@
# serializer version: 1
# name: test_entities_created
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'receiver',
'friendly_name': 'AVR-3805 / AVC-3890',
'is_volume_muted': False,
'last_non_buffering_state': <MediaPlayerState.ON: 'on'>,
'source': 'cd',
'source_list': list([
'cd',
'cdr_tape1',
'dbs_sat',
'dvd',
'phono',
'tuner',
'tv',
'v_aux',
'vcr_1',
'vcr_2',
'vdp',
]),
'supported_features': <MediaPlayerEntityFeature: 3468>,
'volume_level': 0.5555555555555556,
}),
'context': <ANY>,
'entity_id': 'media_player.avr_3805_avc_3890',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_entities_created.1
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'receiver',
'friendly_name': 'AVR-3805 / AVC-3890 Zone 2',
'last_non_buffering_state': <MediaPlayerState.ON: 'on'>,
'source': 'tuner',
'source_list': list([
'cd',
'cdr_tape1',
'dbs_sat',
'dvd',
'phono',
'tuner',
'tv',
'v_aux',
'vcr_1',
'vcr_2',
'vdp',
]),
'supported_features': <MediaPlayerEntityFeature: 3460>,
'volume_level': 0.6666666666666666,
}),
'context': <ANY>,
'entity_id': 'media_player.avr_3805_avc_3890_zone_2',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_entities_created.2
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'receiver',
'friendly_name': 'AVR-3805 / AVC-3890 Zone 3',
'last_non_buffering_state': <MediaPlayerState.OFF: 'off'>,
'source_list': list([
'cd',
'cdr_tape1',
'dbs_sat',
'dvd',
'phono',
'tuner',
'tv',
'v_aux',
'vcr_1',
'vcr_2',
'vdp',
]),
'supported_features': <MediaPlayerEntityFeature: 3460>,
}),
'context': <ANY>,
'entity_id': 'media_player.avr_3805_avc_3890_zone_3',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---

View File

@@ -1,262 +0,0 @@
"""Tests for the Denon RS232 config flow."""
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from homeassistant.components.denon_rs232.config_flow import OPTION_PICK_MANUAL
from homeassistant.components.denon_rs232.const import DOMAIN
from homeassistant.components.usb import USBDevice
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_DEVICE, CONF_MODEL
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from . import MOCK_DEVICE, MOCK_MODEL
from tests.common import MockConfigEntry, get_schema_suggested_value
@pytest.fixture(autouse=True)
def mock_list_serial_ports() -> Generator[list[USBDevice]]:
"""Mock discovered serial ports."""
ports = [
USBDevice(
device=MOCK_DEVICE,
vid="123",
pid="456",
serial_number="mock-serial",
manufacturer="mock-manuf",
description=None,
)
]
with patch(
"homeassistant.components.denon_rs232.config_flow.scan_serial_ports",
return_value=ports,
):
yield ports
@pytest.fixture
def mock_async_setup_entry(mock_receiver: MagicMock) -> Generator[AsyncMock]:
"""Prevent config-entry creation tests from setting up the integration."""
with patch(
"homeassistant.components.denon_rs232.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry
async def test_user_form_creates_entry(
hass: HomeAssistant,
mock_receiver: MagicMock,
mock_async_setup_entry: AsyncMock,
) -> None:
"""Test successful config flow creates an entry."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
with patch(
"homeassistant.components.denon_rs232.config_flow.DenonReceiver",
return_value=mock_receiver,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "AVR-3805 / AVC-3890"
assert result["data"] == {CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL}
mock_async_setup_entry.assert_awaited_once()
mock_receiver.connect.assert_awaited_once()
mock_receiver.disconnect.assert_awaited_once()
@pytest.mark.parametrize(
("exception", "error"),
[
(ConnectionError("No response"), "cannot_connect"),
(OSError("No such device"), "cannot_connect"),
(RuntimeError("boom"), "unknown"),
],
)
async def test_user_form_error(
hass: HomeAssistant,
exception: Exception,
error: str,
mock_receiver: MagicMock,
) -> None:
"""Test the user step reports connection and unexpected errors."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
mock_receiver.connect.side_effect = exception
with patch(
"homeassistant.components.denon_rs232.config_flow.DenonReceiver",
return_value=mock_receiver,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": error}
mock_receiver.connect.side_effect = None
with patch(
"homeassistant.components.denon_rs232.config_flow.DenonReceiver",
return_value=mock_receiver,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_user_duplicate_port_aborts(hass: HomeAssistant) -> None:
"""Test we abort if the same port is already configured."""
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL},
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_manual_form_creates_entry(
hass: HomeAssistant,
mock_receiver: MagicMock,
mock_async_setup_entry: AsyncMock,
) -> None:
"""Test creating entry with manual user input."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: OPTION_PICK_MANUAL, CONF_MODEL: MOCK_MODEL},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "manual"
with patch(
"homeassistant.components.denon_rs232.config_flow.DenonReceiver",
return_value=mock_receiver,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: MOCK_DEVICE},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "AVR-3805 / AVC-3890"
assert result["data"] == {CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL}
mock_async_setup_entry.assert_awaited_once()
mock_receiver.connect.assert_awaited_once()
mock_receiver.disconnect.assert_awaited_once()
@pytest.mark.parametrize(
("exception", "error"),
[
(ValueError("Invalid port"), "cannot_connect"),
(ConnectionError("No response"), "cannot_connect"),
(OSError("No such device"), "cannot_connect"),
(RuntimeError("boom"), "unknown"),
],
)
async def test_manual_form_error_handling(
hass: HomeAssistant,
exception: Exception,
error: str,
mock_receiver: MagicMock,
) -> None:
"""Test the manual step reports connection and unexpected errors."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: OPTION_PICK_MANUAL, CONF_MODEL: MOCK_MODEL},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "manual"
mock_receiver.connect.side_effect = exception
with patch(
"homeassistant.components.denon_rs232.config_flow.DenonReceiver",
return_value=mock_receiver,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: MOCK_DEVICE},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "manual"
assert result["errors"] == {"base": error}
assert (
get_schema_suggested_value(result["data_schema"].schema, CONF_DEVICE)
== MOCK_DEVICE
)
mock_receiver.connect.side_effect = None
with patch(
"homeassistant.components.denon_rs232.config_flow.DenonReceiver",
return_value=mock_receiver,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: MOCK_DEVICE},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_manual_duplicate_port_aborts(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test we abort if the same port is already configured."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: OPTION_PICK_MANUAL, CONF_MODEL: MOCK_MODEL},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: MOCK_DEVICE},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"

View File

@@ -1,319 +0,0 @@
"""Tests for the Denon RS232 media player platform."""
from __future__ import annotations
from pathlib import Path
from typing import Literal
from unittest.mock import call
from denon_rs232 import InputSource
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.denon_rs232.media_player import INPUT_SOURCE_DENON_TO_HA
from homeassistant.components.media_player import (
ATTR_INPUT_SOURCE,
ATTR_INPUT_SOURCE_LIST,
ATTR_MEDIA_VOLUME_LEVEL,
ATTR_MEDIA_VOLUME_MUTED,
DOMAIN as MP_DOMAIN,
SERVICE_SELECT_SOURCE,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
SERVICE_VOLUME_DOWN,
SERVICE_VOLUME_MUTE,
SERVICE_VOLUME_SET,
SERVICE_VOLUME_UP,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.util.json import load_json
from .conftest import MockReceiver, MockState, _default_state
ZoneName = Literal["main", "zone_2", "zone_3"]
MAIN_ENTITY_ID = "media_player.avr_3805_avc_3890"
ZONE_2_ENTITY_ID = "media_player.avr_3805_avc_3890_zone_2"
ZONE_3_ENTITY_ID = "media_player.avr_3805_avc_3890_zone_3"
STRINGS_PATH = Path("homeassistant/components/denon_rs232/strings.json")
@pytest.fixture(autouse=True)
async def auto_init_components(init_components) -> None:
"""Set up the component."""
async def test_entities_created(
hass: HomeAssistant, mock_receiver: MockReceiver, snapshot: SnapshotAssertion
) -> None:
"""Test media player entities are created through config entry setup."""
assert hass.states.get(MAIN_ENTITY_ID) == snapshot
assert hass.states.get(ZONE_2_ENTITY_ID) == snapshot
assert hass.states.get(ZONE_3_ENTITY_ID) == snapshot
mock_receiver.query_state.assert_awaited_once()
@pytest.mark.parametrize("initial_receiver_state", ["main_only"], indirect=True)
async def test_only_active_zones_are_created(
hass: HomeAssistant, initial_receiver_state: MockState
) -> None:
"""Test setup only creates entities for zones with queried power state."""
assert hass.states.get(MAIN_ENTITY_ID) is not None
assert hass.states.get(ZONE_2_ENTITY_ID) is None
assert hass.states.get(ZONE_3_ENTITY_ID) is None
@pytest.mark.parametrize(
("zone", "entity_id", "initial_entity_state"),
[
("main", MAIN_ENTITY_ID, STATE_ON),
("zone_2", ZONE_2_ENTITY_ID, STATE_ON),
("zone_3", ZONE_3_ENTITY_ID, STATE_OFF),
],
)
async def test_zone_state_updates(
hass: HomeAssistant,
mock_receiver: MockReceiver,
zone: ZoneName,
entity_id: str,
initial_entity_state: str,
) -> None:
"""Test each zone updates from receiver pushes and disconnects."""
assert hass.states.get(entity_id).state == initial_entity_state
state = _default_state()
state.get_zone(zone).power = initial_entity_state != STATE_ON
mock_receiver.mock_state(state)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state != initial_entity_state
mock_receiver.mock_state(None)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
@pytest.mark.parametrize(
("zone", "entity_id", "power_on_command", "power_off_command"),
[
("main", MAIN_ENTITY_ID, ("ZM", "ON"), ("ZM", "OFF")),
("zone_2", ZONE_2_ENTITY_ID, ("Z2", "ON"), ("Z2", "OFF")),
("zone_3", ZONE_3_ENTITY_ID, ("Z1", "ON"), ("Z1", "OFF")),
],
)
async def test_power_controls(
hass: HomeAssistant,
mock_receiver: MockReceiver,
zone: ZoneName,
entity_id: str,
power_on_command: tuple[str, str],
power_off_command: tuple[str, str],
) -> None:
"""Test power services send the right commands for each zone."""
await hass.services.async_call(
MP_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
assert mock_receiver._send_command.await_args == call(*power_on_command)
await hass.services.async_call(
MP_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
assert mock_receiver._send_command.await_args == call(*power_off_command)
@pytest.mark.parametrize(
(
"zone",
"entity_id",
"initial_volume_level",
"set_command",
"volume_up_command",
"volume_down_command",
),
[
(
"main",
MAIN_ENTITY_ID,
50.0 / 90.0,
("MV", "45"),
("MV", "UP"),
("MV", "DOWN"),
),
(
"zone_2",
ZONE_2_ENTITY_ID,
60.0 / 90.0,
("Z2", "45"),
("Z2", "UP"),
("Z2", "DOWN"),
),
],
)
async def test_volume_controls(
hass: HomeAssistant,
mock_receiver: MockReceiver,
zone: ZoneName,
entity_id: str,
initial_volume_level: float,
set_command: tuple[str, str],
volume_up_command: tuple[str, str],
volume_down_command: tuple[str, str],
) -> None:
"""Test volume state and controls for each zone."""
state = hass.states.get(entity_id)
assert abs(state.attributes[ATTR_MEDIA_VOLUME_LEVEL] - initial_volume_level) < 0.001
await hass.services.async_call(
MP_DOMAIN,
SERVICE_VOLUME_SET,
{ATTR_ENTITY_ID: entity_id, ATTR_MEDIA_VOLUME_LEVEL: 0.5},
blocking=True,
)
assert mock_receiver._send_command.await_args == call(*set_command)
await hass.services.async_call(
MP_DOMAIN,
SERVICE_VOLUME_UP,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
assert mock_receiver._send_command.await_args == call(*volume_up_command)
await hass.services.async_call(
MP_DOMAIN,
SERVICE_VOLUME_DOWN,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
assert mock_receiver._send_command.await_args == call(*volume_down_command)
async def test_main_mute_controls(
hass: HomeAssistant, mock_receiver: MockReceiver
) -> None:
"""Test mute state and controls for the main zone."""
state = hass.states.get(MAIN_ENTITY_ID)
assert state.attributes[ATTR_MEDIA_VOLUME_MUTED] is False
await hass.services.async_call(
MP_DOMAIN,
SERVICE_VOLUME_MUTE,
{ATTR_ENTITY_ID: MAIN_ENTITY_ID, ATTR_MEDIA_VOLUME_MUTED: True},
blocking=True,
)
assert mock_receiver._send_command.await_args == call("MU", "ON")
await hass.services.async_call(
MP_DOMAIN,
SERVICE_VOLUME_MUTE,
{ATTR_ENTITY_ID: MAIN_ENTITY_ID, ATTR_MEDIA_VOLUME_MUTED: False},
blocking=True,
)
assert mock_receiver._send_command.await_args == call("MU", "OFF")
@pytest.mark.parametrize(
(
"zone",
"entity_id",
"initial_source",
"updated_source",
"expected_source",
"select_source_command",
),
[
("main", MAIN_ENTITY_ID, "cd", InputSource.NET, "net", ("SI", "NET")),
(
"zone_2",
ZONE_2_ENTITY_ID,
"tuner",
InputSource.BT,
"bt",
("Z2", "BT"),
),
("zone_3", ZONE_3_ENTITY_ID, None, InputSource.DVD, "dvd", ("Z1", "DVD")),
],
)
async def test_source_state_and_controls(
hass: HomeAssistant,
mock_receiver: MockReceiver,
zone: ZoneName,
entity_id: str,
initial_source: str | None,
updated_source: InputSource,
expected_source: str,
select_source_command: tuple[str, str],
) -> None:
"""Test source state and selection for each zone."""
entity_state = hass.states.get(entity_id)
assert entity_state.attributes.get(ATTR_INPUT_SOURCE) == initial_source
source_list = entity_state.attributes[ATTR_INPUT_SOURCE_LIST]
assert "cd" in source_list
assert "dvd" in source_list
assert "tuner" in source_list
assert source_list == sorted(source_list)
state = _default_state()
zone_state = state.get_zone(zone)
zone_state.power = True
zone_state.input_source = updated_source
mock_receiver.mock_state(state)
await hass.async_block_till_done()
assert hass.states.get(entity_id).attributes[ATTR_INPUT_SOURCE] == expected_source
await hass.services.async_call(
MP_DOMAIN,
SERVICE_SELECT_SOURCE,
{ATTR_ENTITY_ID: entity_id, ATTR_INPUT_SOURCE: expected_source},
blocking=True,
)
assert mock_receiver._send_command.await_args == call(*select_source_command)
async def test_main_invalid_source_raises(
hass: HomeAssistant,
) -> None:
"""Test invalid main-zone sources raise an error."""
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
MP_DOMAIN,
SERVICE_SELECT_SOURCE,
{
ATTR_ENTITY_ID: MAIN_ENTITY_ID,
ATTR_INPUT_SOURCE: "NONEXISTENT",
},
blocking=True,
)
def test_input_source_translation_keys_cover_all_enum_members() -> None:
"""Test all input sources have a declared translation key."""
assert set(INPUT_SOURCE_DENON_TO_HA) == set(InputSource)
strings = load_json(STRINGS_PATH)
assert set(INPUT_SOURCE_DENON_TO_HA.values()) == set(
strings["entity"]["media_player"]["receiver"]["state_attributes"]["source"][
"state"
]
)

View File

@@ -2,7 +2,6 @@
from collections.abc import Generator
from datetime import datetime, timedelta
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
from forecast_solar import models
@@ -16,9 +15,7 @@ from homeassistant.components.forecast_solar.const import (
CONF_INVERTER_SIZE,
CONF_MODULES_POWER,
DOMAIN,
SUBENTRY_TYPE_PLANE,
)
from homeassistant.config_entries import ConfigSubentryData
from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
@@ -36,44 +33,26 @@ def mock_setup_entry() -> Generator[AsyncMock]:
@pytest.fixture
def api_key_present() -> bool:
"""Return whether an API key should be present in the config entry options."""
return True
@pytest.fixture
def mock_config_entry(api_key_present: bool) -> MockConfigEntry:
def mock_config_entry() -> MockConfigEntry:
"""Return the default mocked config entry."""
options: dict[str, Any] = {
CONF_DAMPING_MORNING: 0.5,
CONF_DAMPING_EVENING: 0.5,
CONF_INVERTER_SIZE: 2000,
}
if api_key_present:
options[CONF_API_KEY] = "abcdef1234567890"
return MockConfigEntry(
title="Green House",
unique_id="unique",
version=3,
version=2,
domain=DOMAIN,
data={
CONF_LATITUDE: 52.42,
CONF_LONGITUDE: 4.42,
},
options=options,
subentries_data=[
ConfigSubentryData(
data={
CONF_DECLINATION: 30,
CONF_AZIMUTH: 190,
CONF_MODULES_POWER: 5100,
},
subentry_id="mock_plane_id",
subentry_type=SUBENTRY_TYPE_PLANE,
title="30° / 190° / 5100W",
unique_id=None,
),
],
options={
CONF_API_KEY: "abcdef12345",
CONF_DECLINATION: 30,
CONF_AZIMUTH: 190,
CONF_MODULES_POWER: 5100,
CONF_DAMPING_MORNING: 0.5,
CONF_DAMPING_EVENING: 0.5,
CONF_INVERTER_SIZE: 2000,
},
)

View File

@@ -32,20 +32,13 @@
}),
'options': dict({
'api_key': '**REDACTED**',
'azimuth': 190,
'damping_evening': 0.5,
'damping_morning': 0.5,
'declination': 30,
'inverter_size': 2000,
'modules_power': 5100,
}),
'subentries': list([
dict({
'data': dict({
'azimuth': 190,
'declination': 30,
'modules_power': 5100,
}),
'title': '30° / 190° / 5100W',
}),
]),
'title': 'Green House',
}),
})

View File

@@ -0,0 +1,32 @@
# serializer version: 1
# name: test_migration
ConfigEntrySnapshot({
'data': dict({
'latitude': 52.42,
'longitude': 4.42,
}),
'disabled_by': None,
'discovery_keys': dict({
}),
'domain': 'forecast_solar',
'entry_id': <ANY>,
'minor_version': 1,
'options': dict({
'api_key': 'abcdef12345',
'azimuth': 190,
'damping_evening': 0.5,
'damping_morning': 0.5,
'declination': 30,
'inverter_size': 2000,
'modules_power': 5100,
}),
'pref_disable_new_entities': False,
'pref_disable_polling': False,
'source': 'user',
'subentries': list([
]),
'title': 'Green House',
'unique_id': 'unique',
'version': 2,
})
# ---

View File

@@ -12,13 +12,8 @@ from homeassistant.components.forecast_solar.const import (
CONF_INVERTER_SIZE,
CONF_MODULES_POWER,
DOMAIN,
SUBENTRY_TYPE_PLANE,
)
from homeassistant.config_entries import (
SOURCE_RECONFIGURE,
SOURCE_USER,
ConfigSubentryData,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
@@ -56,19 +51,11 @@ async def test_user_flow(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> No
CONF_LATITUDE: 52.42,
CONF_LONGITUDE: 4.42,
}
assert config_entry.options == {}
# Verify a plane subentry was created
plane_subentries = config_entry.get_subentries_of_type(SUBENTRY_TYPE_PLANE)
assert len(plane_subentries) == 1
subentry = plane_subentries[0]
assert subentry.subentry_type == SUBENTRY_TYPE_PLANE
assert subentry.data == {
CONF_DECLINATION: 42,
assert config_entry.options == {
CONF_AZIMUTH: 142,
CONF_DECLINATION: 42,
CONF_MODULES_POWER: 4242,
}
assert subentry.title == "42° / 142° / 4242W"
assert len(mock_setup_entry.mock_calls) == 1
@@ -92,11 +79,15 @@ async def test_options_flow_invalid_api(
result["flow_id"],
user_input={
CONF_API_KEY: "solarPOWER!",
CONF_DECLINATION: 21,
CONF_AZIMUTH: 22,
CONF_MODULES_POWER: 2122,
CONF_DAMPING_MORNING: 0.25,
CONF_DAMPING_EVENING: 0.25,
CONF_INVERTER_SIZE: 2000,
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {CONF_API_KEY: "invalid_api_key"}
@@ -106,15 +97,22 @@ async def test_options_flow_invalid_api(
result["flow_id"],
user_input={
CONF_API_KEY: "SolarForecast150",
CONF_DECLINATION: 21,
CONF_AZIMUTH: 22,
CONF_MODULES_POWER: 2122,
CONF_DAMPING_MORNING: 0.25,
CONF_DAMPING_EVENING: 0.25,
CONF_INVERTER_SIZE: 2000,
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {
CONF_API_KEY: "SolarForecast150",
CONF_DECLINATION: 21,
CONF_AZIMUTH: 22,
CONF_MODULES_POWER: 2122,
CONF_DAMPING_MORNING: 0.25,
CONF_DAMPING_EVENING: 0.25,
CONF_INVERTER_SIZE: 2000,
@@ -141,15 +139,22 @@ async def test_options_flow(
result["flow_id"],
user_input={
CONF_API_KEY: "SolarForecast150",
CONF_DECLINATION: 21,
CONF_AZIMUTH: 22,
CONF_MODULES_POWER: 2122,
CONF_DAMPING_MORNING: 0.25,
CONF_DAMPING_EVENING: 0.25,
CONF_INVERTER_SIZE: 2000,
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {
CONF_API_KEY: "SolarForecast150",
CONF_DECLINATION: 21,
CONF_AZIMUTH: 22,
CONF_MODULES_POWER: 2122,
CONF_DAMPING_MORNING: 0.25,
CONF_DAMPING_EVENING: 0.25,
CONF_INVERTER_SIZE: 2000,
@@ -175,293 +180,23 @@ async def test_options_flow_without_key(
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_DECLINATION: 21,
CONF_AZIMUTH: 22,
CONF_MODULES_POWER: 2122,
CONF_DAMPING_MORNING: 0.25,
CONF_DAMPING_EVENING: 0.25,
CONF_INVERTER_SIZE: 2000,
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {
CONF_API_KEY: None,
CONF_DECLINATION: 21,
CONF_AZIMUTH: 22,
CONF_MODULES_POWER: 2122,
CONF_DAMPING_MORNING: 0.25,
CONF_DAMPING_EVENING: 0.25,
CONF_INVERTER_SIZE: 2000,
}
@pytest.mark.usefixtures("mock_setup_entry")
async def test_options_flow_required_api_key(
hass: HomeAssistant,
) -> None:
"""Test config flow options requires API key when multiple planes are present."""
mock_config_entry = MockConfigEntry(
title="Green House",
unique_id="unique",
version=3,
domain=DOMAIN,
data={
CONF_LATITUDE: 52.42,
CONF_LONGITUDE: 4.42,
},
options={
CONF_DAMPING_MORNING: 0.5,
CONF_DAMPING_EVENING: 0.5,
CONF_INVERTER_SIZE: 2000,
CONF_API_KEY: "abcdef1234567890",
},
subentries_data=[
ConfigSubentryData(
data={
CONF_DECLINATION: 30,
CONF_AZIMUTH: 190,
CONF_MODULES_POWER: 5100,
},
subentry_id="mock_plane_id",
subentry_type=SUBENTRY_TYPE_PLANE,
title="30° / 190° / 5100W",
unique_id=None,
),
ConfigSubentryData(
data={
CONF_DECLINATION: 45,
CONF_AZIMUTH: 270,
CONF_MODULES_POWER: 3000,
},
subentry_id="second_plane_id",
subentry_type=SUBENTRY_TYPE_PLANE,
title="45° / 270° / 3000W",
unique_id=None,
),
],
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
# Try to save with an empty API key
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_API_KEY: "",
CONF_DAMPING_MORNING: 0.25,
CONF_DAMPING_EVENING: 0.25,
CONF_INVERTER_SIZE: 2000,
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {CONF_API_KEY: "api_key_required"}
# Now provide an API key
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_API_KEY: "SolarForecast150",
CONF_DAMPING_MORNING: 0.25,
CONF_DAMPING_EVENING: 0.25,
CONF_INVERTER_SIZE: 2000,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {
CONF_API_KEY: "SolarForecast150",
CONF_DAMPING_MORNING: 0.25,
CONF_DAMPING_EVENING: 0.25,
CONF_INVERTER_SIZE: 2000,
}
@pytest.mark.usefixtures("mock_setup_entry")
async def test_subentry_flow_add_plane(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test adding a plane via subentry flow."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, SUBENTRY_TYPE_PLANE),
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
user_input={
CONF_DECLINATION: 45,
CONF_AZIMUTH: 270,
CONF_MODULES_POWER: 3000,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "45° / 270° / 3000W"
assert result["data"] == {
CONF_DECLINATION: 45,
CONF_AZIMUTH: 270,
CONF_MODULES_POWER: 3000,
}
assert len(mock_config_entry.get_subentries_of_type(SUBENTRY_TYPE_PLANE)) == 2
@pytest.mark.usefixtures("mock_forecast_solar")
async def test_subentry_flow_reconfigure_plane(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test reconfiguring a plane via subentry flow."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
# Get the existing plane subentry id
subentry_id = mock_config_entry.get_subentries_of_type(SUBENTRY_TYPE_PLANE)[
0
].subentry_id
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, SUBENTRY_TYPE_PLANE),
context={"source": SOURCE_RECONFIGURE, "subentry_id": subentry_id},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
user_input={
CONF_DECLINATION: 50,
CONF_AZIMUTH: 200,
CONF_MODULES_POWER: 6000,
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
plane_subentries = mock_config_entry.get_subentries_of_type(SUBENTRY_TYPE_PLANE)
assert len(plane_subentries) == 1
subentry = plane_subentries[0]
assert subentry.data == {
CONF_DECLINATION: 50,
CONF_AZIMUTH: 200,
CONF_MODULES_POWER: 6000,
}
assert subentry.title == "50° / 200° / 6000W"
@pytest.mark.parametrize("api_key_present", [False])
@pytest.mark.usefixtures("mock_setup_entry")
async def test_subentry_flow_no_api_key(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test that adding more than one plane without API key is not allowed."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, SUBENTRY_TYPE_PLANE),
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "api_key_required"
@pytest.mark.usefixtures("mock_setup_entry")
async def test_subentry_flow_max_planes(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test that adding more than 4 planes is not allowed."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
# mock_config_entry already has 1 plane subentry; add 3 more to reach the limit
for i in range(3):
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, SUBENTRY_TYPE_PLANE),
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
user_input={
CONF_DECLINATION: 10 * (i + 1),
CONF_AZIMUTH: 90 * (i + 1),
CONF_MODULES_POWER: 1000 * (i + 1),
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert len(mock_config_entry.get_subentries_of_type(SUBENTRY_TYPE_PLANE)) == 4
# Attempt to add a 5th plane should be aborted
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, SUBENTRY_TYPE_PLANE),
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "max_planes"
async def test_subentry_flow_reconfigure_plane_not_loaded(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test reconfiguring a plane via subentry flow when entry is not loaded."""
mock_config_entry.add_to_hass(hass)
# Entry is not loaded, so it has no update listeners
# Get the existing plane subentry id
subentry_id = mock_config_entry.get_subentries_of_type(SUBENTRY_TYPE_PLANE)[
0
].subentry_id
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, SUBENTRY_TYPE_PLANE),
context={"source": SOURCE_RECONFIGURE, "subentry_id": subentry_id},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
user_input={
CONF_DECLINATION: 50,
CONF_AZIMUTH: 200,
CONF_MODULES_POWER: 6000,
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
plane_subentries = mock_config_entry.get_subentries_of_type(SUBENTRY_TYPE_PLANE)
assert len(plane_subentries) == 1
subentry = plane_subentries[0]
assert subentry.data == {
CONF_DECLINATION: 50,
CONF_AZIMUTH: 200,
CONF_MODULES_POWER: 6000,
}
assert subentry.title == "50° / 200° / 6000W"

View File

@@ -65,8 +65,3 @@ async def test_energy_solar_forecast_filters_midnight_utc_zeros(
"2021-06-27T15:00:00+00:00": 292,
}
}
async def test_energy_solar_forecast_invalid_id(hass: HomeAssistant) -> None:
"""Test the Forecast.Solar energy platform with invalid config entry ID."""
assert await energy.async_get_solar_forecast(hass, "invalid_id") is None

View File

@@ -2,20 +2,17 @@
from unittest.mock import MagicMock, patch
from forecast_solar import ForecastSolarConnectionError, Plane
from forecast_solar import ForecastSolarConnectionError
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.forecast_solar.const import (
CONF_AZIMUTH,
CONF_DAMPING,
CONF_DAMPING_EVENING,
CONF_DAMPING_MORNING,
CONF_DECLINATION,
CONF_INVERTER_SIZE,
CONF_MODULES_POWER,
DOMAIN,
SUBENTRY_TYPE_PLANE,
)
from homeassistant.config_entries import ConfigEntryState, ConfigSubentryData
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
@@ -58,16 +55,12 @@ async def test_config_entry_not_ready(
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_migration_from_v1(
hass: HomeAssistant,
mock_forecast_solar: MagicMock,
) -> None:
"""Test config entry migration from version 1."""
async def test_migration(hass: HomeAssistant, snapshot: SnapshotAssertion) -> None:
"""Test config entry version 1 -> 2 migration."""
mock_config_entry = MockConfigEntry(
title="Green House",
unique_id="unique",
domain=DOMAIN,
version=1,
data={
CONF_LATITUDE: 52.42,
CONF_LONGITUDE: 4.42,
@@ -85,215 +78,4 @@ async def test_migration_from_v1(
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entry = hass.config_entries.async_get_entry(mock_config_entry.entry_id)
assert entry.version == 3
assert entry.options == {
CONF_API_KEY: "abcdef12345",
"damping_morning": 0.5,
"damping_evening": 0.5,
CONF_INVERTER_SIZE: 2000,
}
plane_subentries = entry.get_subentries_of_type(SUBENTRY_TYPE_PLANE)
assert len(plane_subentries) == 1
subentry = plane_subentries[0]
assert subentry.subentry_type == SUBENTRY_TYPE_PLANE
assert subentry.data == {
CONF_DECLINATION: 30,
CONF_AZIMUTH: 190,
CONF_MODULES_POWER: 5100,
}
assert subentry.title == "30° / 190° / 5100W"
async def test_migration_from_v2(
hass: HomeAssistant,
mock_forecast_solar: MagicMock,
) -> None:
"""Test config entry migration from version 2."""
mock_config_entry = MockConfigEntry(
title="Green House",
unique_id="unique",
domain=DOMAIN,
version=2,
data={
CONF_LATITUDE: 52.42,
CONF_LONGITUDE: 4.42,
},
options={
CONF_API_KEY: "abcdef12345",
CONF_DECLINATION: 30,
CONF_AZIMUTH: 190,
CONF_MODULES_POWER: 5100,
CONF_INVERTER_SIZE: 2000,
},
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entry = hass.config_entries.async_get_entry(mock_config_entry.entry_id)
assert entry.version == 3
assert entry.options == {
CONF_API_KEY: "abcdef12345",
CONF_INVERTER_SIZE: 2000,
}
plane_subentries = entry.get_subentries_of_type(SUBENTRY_TYPE_PLANE)
assert len(plane_subentries) == 1
subentry = plane_subentries[0]
assert subentry.subentry_type == SUBENTRY_TYPE_PLANE
assert subentry.data == {
CONF_DECLINATION: 30,
CONF_AZIMUTH: 190,
CONF_MODULES_POWER: 5100,
}
assert subentry.title == "30° / 190° / 5100W"
async def test_setup_entry_no_planes(
hass: HomeAssistant,
mock_forecast_solar: MagicMock,
) -> None:
"""Test setup fails when all plane subentries have been removed."""
mock_config_entry = MockConfigEntry(
title="Green House",
unique_id="unique",
version=3,
domain=DOMAIN,
data={
CONF_LATITUDE: 52.42,
CONF_LONGITUDE: 4.42,
},
options={
CONF_API_KEY: "abcdef1234567890",
},
)
mock_config_entry.add_to_hass(hass)
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_ERROR
async def test_setup_entry_multiple_planes_no_api_key(
hass: HomeAssistant,
mock_forecast_solar: MagicMock,
) -> None:
"""Test setup fails when multiple planes are configured without an API key."""
mock_config_entry = MockConfigEntry(
title="Green House",
unique_id="unique",
version=3,
domain=DOMAIN,
data={
CONF_LATITUDE: 52.42,
CONF_LONGITUDE: 4.42,
},
options={},
subentries_data=[
ConfigSubentryData(
data={
CONF_DECLINATION: 30,
CONF_AZIMUTH: 190,
CONF_MODULES_POWER: 5100,
},
subentry_id="plane_1",
subentry_type=SUBENTRY_TYPE_PLANE,
title="30° / 190° / 5100W",
unique_id=None,
),
ConfigSubentryData(
data={
CONF_DECLINATION: 45,
CONF_AZIMUTH: 90,
CONF_MODULES_POWER: 3000,
},
subentry_id="plane_2",
subentry_type=SUBENTRY_TYPE_PLANE,
title="45° / 90° / 3000W",
unique_id=None,
),
],
)
mock_config_entry.add_to_hass(hass)
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_ERROR
async def test_coordinator_multi_plane_initialization(
hass: HomeAssistant,
mock_forecast_solar: MagicMock,
) -> None:
"""Test the Forecast.Solar coordinator multi-plane initialization."""
options = {
CONF_API_KEY: "abcdef1234567890",
CONF_DAMPING_MORNING: 0.5,
CONF_DAMPING_EVENING: 0.5,
CONF_INVERTER_SIZE: 2000,
}
mock_config_entry = MockConfigEntry(
title="Green House",
unique_id="unique",
version=3,
domain=DOMAIN,
data={
CONF_LATITUDE: 52.42,
CONF_LONGITUDE: 4.42,
},
options=options,
subentries_data=[
ConfigSubentryData(
data={
CONF_DECLINATION: 30,
CONF_AZIMUTH: 190,
CONF_MODULES_POWER: 5100,
},
subentry_id="plane_1",
subentry_type=SUBENTRY_TYPE_PLANE,
title="30° / 190° / 5100W",
unique_id=None,
),
ConfigSubentryData(
data={
CONF_DECLINATION: 45,
CONF_AZIMUTH: 270,
CONF_MODULES_POWER: 3000,
},
subentry_id="plane_2",
subentry_type=SUBENTRY_TYPE_PLANE,
title="45° / 270° / 3000W",
unique_id=None,
),
],
)
mock_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.forecast_solar.coordinator.ForecastSolar",
return_value=mock_forecast_solar,
) as forecast_solar_mock:
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
forecast_solar_mock.assert_called_once()
_, kwargs = forecast_solar_mock.call_args
assert kwargs["latitude"] == 52.42
assert kwargs["longitude"] == 4.42
assert kwargs["api_key"] == "abcdef1234567890"
# Main plane (plane_1)
assert kwargs["declination"] == 30
assert kwargs["azimuth"] == 10 # 190 - 180
assert kwargs["kwp"] == 5.1 # 5100 / 1000
# Additional planes (plane_2)
planes = kwargs["planes"]
assert len(planes) == 1
assert isinstance(planes[0], Plane)
assert planes[0].declination == 45
assert planes[0].azimuth == 90 # 270 - 180
assert planes[0].kwp == 3.0 # 3000 / 1000
assert hass.config_entries.async_get_entry(mock_config_entry.entry_id) == snapshot

View File

@@ -49,7 +49,7 @@
'is_local': False,
'light_state': None,
'local_url': None,
'monitoring': True,
'monitoring': None,
'motion_detection': True,
'sd_status': 4,
'supported_features': <CameraEntityFeature: 3>,
@@ -113,7 +113,7 @@
'is_local': True,
'light_state': None,
'local_url': 'http://192.168.0.123/678460a0d47e5618699fb31169e2b47d',
'monitoring': True,
'monitoring': None,
'motion_detection': True,
'sd_status': 4,
'supported_features': <CameraEntityFeature: 3>,
@@ -177,7 +177,7 @@
'is_local': None,
'light_state': None,
'local_url': None,
'monitoring': True,
'monitoring': None,
'sd_status': 4,
'supported_features': <CameraEntityFeature: 1>,
'vpn_url': 'https://prodvpn-eu-6.netatmo.net/10.20.30.40/1111111111111/2222222222222,,',

View File

@@ -54,11 +54,10 @@ async def test_entity(
@pytest.mark.parametrize(
("camera_type", "camera_id", "camera_entity", "expected_state"),
("camera_type", "camera_id", "camera_entity"),
[
("NACamera", "12:34:56:00:f1:62", "camera.hall", "streaming"),
("NOC", "12:34:56:10:b9:0e", "camera.front", "streaming"),
("NDB", "12:34:56:10:f1:66", "camera.netatmo_doorbell", "idle"),
("NACamera", "12:34:56:00:f1:62", "camera.hall"),
("NOC", "12:34:56:10:b9:0e", "camera.front"),
],
)
async def test_setup_component_with_webhook(
@@ -68,7 +67,6 @@ async def test_setup_component_with_webhook(
camera_type: str,
camera_id: str,
camera_entity: str,
expected_state: str,
) -> None:
"""Test setup with webhook."""
with selected_platforms([Platform.CAMERA]):
@@ -80,8 +78,7 @@ async def test_setup_component_with_webhook(
await hass.async_block_till_done()
# Test on/off camera events
assert hass.states.get(camera_entity).state == expected_state
assert hass.states.get(camera_entity).attributes.get("monitoring") is True
assert hass.states.get(camera_entity).state == "streaming"
response = {
"event_type": "off",
"device_id": camera_id,
@@ -92,7 +89,6 @@ async def test_setup_component_with_webhook(
await simulate_webhook(hass, webhook_id, response)
assert hass.states.get(camera_entity).state == "idle"
assert hass.states.get(camera_entity).attributes.get("monitoring") is False
response = {
"event_type": "on",
@@ -103,8 +99,7 @@ async def test_setup_component_with_webhook(
}
await simulate_webhook(hass, webhook_id, response)
assert hass.states.get(camera_entity).state == expected_state
assert hass.states.get(camera_entity).attributes.get("monitoring") is True
assert hass.states.get(camera_entity).state == "streaming"
# Test turn_on/turn_off services
with patch("pyatmo.home.Home.async_set_state") as mock_set_state:
@@ -446,11 +441,10 @@ async def test_service_set_camera_light_invalid_type(
@pytest.mark.parametrize(
("camera_type", "camera_id", "camera_entity", "expected_state"),
("camera_type", "camera_id", "camera_entity"),
[
("NACamera", "12:34:56:00:f1:62", "camera.hall", "streaming"),
("NOC", "12:34:56:10:b9:0e", "camera.front", "streaming"),
("NDB", "12:34:56:10:f1:66", "camera.netatmo_doorbell", "idle"),
("NACamera", "12:34:56:00:f1:62", "camera.hall"),
("NOC", "12:34:56:10:b9:0e", "camera.front"),
],
)
async def test_camera_reconnect_webhook(
@@ -459,7 +453,6 @@ async def test_camera_reconnect_webhook(
camera_type: str,
camera_id: str,
camera_entity: str,
expected_state: str,
) -> None:
"""Test webhook event on camera reconnect."""
fake_post_hits = 0
@@ -518,8 +511,7 @@ async def test_camera_reconnect_webhook(
assert fake_post_hits >= calls
# Real camera disconnect
assert hass.states.get(camera_entity).state == expected_state
assert hass.states.get(camera_entity).attributes.get("monitoring") is True
assert hass.states.get(camera_entity).state == "streaming"
response = {
"event_type": "disconnection",
"device_id": camera_id,
@@ -530,7 +522,6 @@ async def test_camera_reconnect_webhook(
await simulate_webhook(hass, webhook_id, response)
assert hass.states.get(camera_entity).state == "idle"
assert hass.states.get(camera_entity).attributes.get("monitoring") is False
response = {
"event_type": "connection",
@@ -541,8 +532,7 @@ async def test_camera_reconnect_webhook(
}
await simulate_webhook(hass, webhook_id, response)
assert hass.states.get(camera_entity).state == expected_state
assert hass.states.get(camera_entity).attributes.get("monitoring") is True
assert hass.states.get(camera_entity).state == "streaming"
@pytest.mark.parametrize(

View File

@@ -1,230 +0,0 @@
# serializer version: 1
# name: test_sensor_entities_battery_device[sensor.opendisplay_1234_battery-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.opendisplay_1234_battery',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Battery',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.BATTERY: 'battery'>,
'original_icon': None,
'original_name': 'Battery',
'platform': 'opendisplay',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'AA:BB:CC:DD:EE:FF-battery',
'unit_of_measurement': '%',
})
# ---
# name: test_sensor_entities_battery_device[sensor.opendisplay_1234_battery-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'battery',
'friendly_name': 'OpenDisplay 1234 Battery',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'sensor.opendisplay_1234_battery',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '39',
})
# ---
# name: test_sensor_entities_battery_device[sensor.opendisplay_1234_battery_voltage-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.opendisplay_1234_battery_voltage',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Battery voltage',
'options': dict({
'sensor': dict({
'suggested_display_precision': 0,
}),
}),
'original_device_class': <SensorDeviceClass.VOLTAGE: 'voltage'>,
'original_icon': None,
'original_name': 'Battery voltage',
'platform': 'opendisplay',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'battery_voltage',
'unique_id': 'AA:BB:CC:DD:EE:FF-battery_voltage',
'unit_of_measurement': <UnitOfElectricPotential.MILLIVOLT: 'mV'>,
})
# ---
# name: test_sensor_entities_battery_device[sensor.opendisplay_1234_battery_voltage-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'voltage',
'friendly_name': 'OpenDisplay 1234 Battery voltage',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfElectricPotential.MILLIVOLT: 'mV'>,
}),
'context': <ANY>,
'entity_id': 'sensor.opendisplay_1234_battery_voltage',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '3700',
})
# ---
# name: test_sensor_entities_battery_device[sensor.opendisplay_1234_temperature-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.opendisplay_1234_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Temperature',
'options': dict({
'sensor': dict({
'suggested_display_precision': 1,
}),
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': None,
'original_name': 'Temperature',
'platform': 'opendisplay',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'AA:BB:CC:DD:EE:FF-temperature',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_sensor_entities_battery_device[sensor.opendisplay_1234_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'temperature',
'friendly_name': 'OpenDisplay 1234 Temperature',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.opendisplay_1234_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '25.0',
})
# ---
# name: test_sensor_entities_usb_device[sensor.opendisplay_1234_temperature-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.opendisplay_1234_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Temperature',
'options': dict({
'sensor': dict({
'suggested_display_precision': 1,
}),
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': None,
'original_name': 'Temperature',
'platform': 'opendisplay',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'AA:BB:CC:DD:EE:FF-temperature',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_sensor_entities_usb_device[sensor.opendisplay_1234_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'temperature',
'friendly_name': 'OpenDisplay 1234 Temperature',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.opendisplay_1234_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '25.0',
})
# ---

View File

@@ -1,226 +0,0 @@
"""Test the OpenDisplay sensor platform."""
from copy import deepcopy
from datetime import timedelta
import time
from unittest.mock import MagicMock
from habluetooth import CONNECTABLE_FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS
from opendisplay import voltage_to_percent
from opendisplay.models.config import PowerOption
from opendisplay.models.enums import CapacityEstimator, PowerMode
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.bluetooth.const import UNAVAILABLE_TRACK_SECONDS
from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.util import dt as dt_util
from . import DEVICE_CONFIG, TEST_ADDRESS, VALID_SERVICE_INFO, make_service_info
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
from tests.components.bluetooth import (
inject_bluetooth_service_info,
patch_all_discovered_devices,
patch_bluetooth_time,
)
pytestmark = pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def _setup_entry(hass: HomeAssistant, mock_config_entry: MockConfigEntry) -> None:
"""Set up the integration and wait for entities to be created."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
async def test_sensors_before_data(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test that sensors are created but unavailable before data arrives."""
await _setup_entry(hass, mock_config_entry)
# All sensors exist but coordinator has no data yet
assert hass.states.get("sensor.opendisplay_1234_temperature") is not None
assert (
hass.states.get("sensor.opendisplay_1234_temperature").state
== STATE_UNAVAILABLE
)
async def test_sensor_entities_usb_device(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test sensor entities for a USB-powered Flex device."""
await _setup_entry(hass, mock_config_entry)
inject_bluetooth_service_info(hass, VALID_SERVICE_INFO)
await hass.async_block_till_done()
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_sensor_entities_battery_device(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_opendisplay_device: MagicMock,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test sensor entities for a battery-powered Flex device with LI_ION chemistry."""
device_config = deepcopy(DEVICE_CONFIG)
power = device_config.power
device_config.power = PowerOption(
power_mode=PowerMode.BATTERY,
battery_capacity_mah=power.battery_capacity_mah,
sleep_timeout_ms=power.sleep_timeout_ms,
tx_power=power.tx_power,
sleep_flags=power.sleep_flags,
battery_sense_pin=power.battery_sense_pin,
battery_sense_enable_pin=power.battery_sense_enable_pin,
battery_sense_flags=power.battery_sense_flags,
capacity_estimator=1, # LI_ION
voltage_scaling_factor=power.voltage_scaling_factor,
deep_sleep_current_ua=power.deep_sleep_current_ua,
deep_sleep_time_seconds=power.deep_sleep_time_seconds,
reserved=power.reserved,
)
mock_opendisplay_device.config = device_config
await _setup_entry(hass, mock_config_entry)
inject_bluetooth_service_info(hass, VALID_SERVICE_INFO)
await hass.async_block_till_done()
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_battery_sensors_not_created_for_usb_devices(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test battery sensors are not created for USB-powered devices."""
await _setup_entry(hass, mock_config_entry)
inject_bluetooth_service_info(hass, VALID_SERVICE_INFO)
await hass.async_block_till_done()
assert entity_registry.async_get("sensor.opendisplay_1234_battery") is None
assert entity_registry.async_get("sensor.opendisplay_1234_battery_voltage") is None
async def test_no_sensors_for_non_flex_devices(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_opendisplay_device: MagicMock,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that no sensor entities are created for non-Flex devices."""
mock_opendisplay_device.is_flex = False
await _setup_entry(hass, mock_config_entry)
assert entity_registry.async_get("sensor.opendisplay_1234_temperature") is None
assert entity_registry.async_get("sensor.opendisplay_1234_battery") is None
assert entity_registry.async_get("sensor.opendisplay_1234_battery_voltage") is None
async def test_coordinator_ignores_unknown_manufacturer(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test that advertisements from an unknown manufacturer ID are ignored."""
await _setup_entry(hass, mock_config_entry)
unknown_service_info = make_service_info(
address=TEST_ADDRESS,
manufacturer_data={0x9999: b"\x00" * 14},
)
inject_bluetooth_service_info(hass, unknown_service_info)
await hass.async_block_till_done()
# Coordinator has no data; device is visible but no OpenDisplay data parsed
assert hass.states.get("sensor.opendisplay_1234_temperature").state == STATE_UNKNOWN
async def test_sensor_goes_unavailable_when_device_disappears(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test that sensors become unavailable when the device stops advertising."""
start_monotonic = time.monotonic()
await _setup_entry(hass, mock_config_entry)
inject_bluetooth_service_info(hass, VALID_SERVICE_INFO)
await hass.async_block_till_done()
assert (
hass.states.get("sensor.opendisplay_1234_temperature").state
!= STATE_UNAVAILABLE
)
# Must exceed both the connectable stale threshold (195s) and the
# unavailability polling interval (300s) to trigger the callback.
advance = (
CONNECTABLE_FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS
+ UNAVAILABLE_TRACK_SECONDS
+ 1
)
monotonic_now = start_monotonic + advance
with (
patch_bluetooth_time(monotonic_now),
patch_all_discovered_devices([]),
):
async_fire_time_changed(
hass,
dt_util.utcnow() + timedelta(seconds=advance),
)
await hass.async_block_till_done()
assert (
hass.states.get("sensor.opendisplay_1234_temperature").state
== STATE_UNAVAILABLE
)
async def test_battery_sensor_defaults_to_liion_when_capacity_estimator_unset(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_opendisplay_device: MagicMock,
) -> None:
"""Test battery % sensor uses LI_ION when capacity_estimator is 0 (not configured)."""
device_config = deepcopy(DEVICE_CONFIG)
power = device_config.power
device_config.power = PowerOption(
power_mode=PowerMode.BATTERY,
battery_capacity_mah=power.battery_capacity_mah,
sleep_timeout_ms=power.sleep_timeout_ms,
tx_power=power.tx_power,
sleep_flags=power.sleep_flags,
battery_sense_pin=power.battery_sense_pin,
battery_sense_enable_pin=power.battery_sense_enable_pin,
battery_sense_flags=power.battery_sense_flags,
capacity_estimator=0, # not configured — defaults to LI_ION in sensor.py
voltage_scaling_factor=power.voltage_scaling_factor,
deep_sleep_current_ua=power.deep_sleep_current_ua,
deep_sleep_time_seconds=power.deep_sleep_time_seconds,
reserved=power.reserved,
)
mock_opendisplay_device.config = device_config
await _setup_entry(hass, mock_config_entry)
inject_bluetooth_service_info(hass, VALID_SERVICE_INFO)
await hass.async_block_till_done()
battery_state = hass.states.get("sensor.opendisplay_1234_battery")
assert battery_state is not None
# capacity_estimator=0 should fall back to LI_ION, producing the same value as explicit LI_ION
expected = voltage_to_percent(3700, CapacityEstimator.LI_ION)
assert battery_state.state == str(expected)

View File

@@ -7,7 +7,7 @@ import pytest
@pytest.fixture(autouse=True)
def patch_getaddrinfo():
"""Patch getaddrinfo to avoid DNS lookups in SNMP tests."""
with patch.object(socket, "getaddrinfo"):
def patch_gethostbyname():
"""Patch gethostbyname to avoid DNS lookups in SNMP tests."""
with patch.object(socket, "gethostbyname"):
yield

View File

@@ -233,7 +233,6 @@ async def test_full_flow_with_domain_registration(
assert parsed_query["client_id"][0] == "user_client_id"
assert parsed_query["redirect_uri"][0] == REDIRECT
assert parsed_query["state"][0] == state
assert parsed_query["prompt_missing_scopes"][0] == "true"
assert parsed_query["scope"][0] == " ".join(SCOPES)
assert "code_challenge" not in parsed_query