Compare commits

..

1 Commits

Author SHA1 Message Date
Ariel Ebersberger 308b6bae5a Use is for IntentResponseType identity check in conversation
`IntentResponseType` is a plain `enum.Enum`, so `==` and `is` are mathematically equivalent. The matching test assertion is adjusted from `len(mock_process.mock_calls) == 2` to `== 1` because the previous count included an incidental `call().response_type.__eq__(...)` entry that `==` recorded; `is` bypasses `__eq__`.
2026-05-21 12:42:23 +02:00
82 changed files with 182 additions and 5045 deletions
Generated
-2
View File
@@ -466,8 +466,6 @@ CLAUDE.md @home-assistant/core
/tests/components/electrasmart/ @jafar-atili
/homeassistant/components/electric_kiwi/ @mikey0000
/tests/components/electric_kiwi/ @mikey0000
/homeassistant/components/electrolux/ @electrolux-oss
/tests/components/electrolux/ @electrolux-oss
/homeassistant/components/elevenlabs/ @sorgfresser
/tests/components/elevenlabs/ @sorgfresser
/homeassistant/components/elgato/ @frenck
@@ -1447,7 +1447,7 @@ class DefaultAgent(ConversationEntity):
response = await self._async_process_intent_result(result, user_input, chat_log)
if (
response.response_type == intent.IntentResponseType.ERROR
response.response_type is intent.IntentResponseType.ERROR
and response.error_code
not in (
intent.IntentResponseErrorCode.FAILED_TO_HANDLE,
+1 -1
View File
@@ -26,12 +26,12 @@ async def async_setup_entry(
async_add_entities(EkeyEvent(data) for data in entry.data["webhooks"])
# pylint: disable-next=home-assistant-missing-has-entity-name
class EkeyEvent(EventEntity):
"""Ekey Event."""
_attr_device_class = EventDeviceClass.BUTTON
_attr_event_types = ["event happened"]
_attr_has_entity_name = True
def __init__(
self,
@@ -1,220 +0,0 @@
"""The Electrolux integration."""
from asyncio import CancelledError
from collections.abc import Awaitable, Callable
import logging
from electrolux_group_developer_sdk.auth.token_manager import TokenManager
from electrolux_group_developer_sdk.client.appliance_client import ApplianceClient
from electrolux_group_developer_sdk.client.appliances.appliance_data import (
ApplianceData,
)
from electrolux_group_developer_sdk.client.bad_credentials_exception import (
BadCredentialsException,
)
from electrolux_group_developer_sdk.client.client_exception import (
ApplianceClientException,
)
from electrolux_group_developer_sdk.client.failed_connection_exception import (
FailedConnectionException,
)
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_API_KEY, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.dispatcher import async_dispatcher_send
from .const import CONF_REFRESH_TOKEN, DOMAIN, NEW_APPLIANCE_SIGNAL, USER_AGENT
from .coordinator import (
ElectroluxConfigEntry,
ElectroluxData,
ElectroluxDataUpdateCoordinator,
)
_LOGGER: logging.Logger = logging.getLogger(__name__)
PLATFORMS = [
Platform.SENSOR,
]
async def async_setup_entry(hass: HomeAssistant, entry: ElectroluxConfigEntry) -> bool:
"""Set up Electrolux integration entry."""
token_manager = create_token_manager(hass, entry)
client = ApplianceClient(
token_manager=token_manager, external_user_agent=USER_AGENT
)
try:
await client.test_connection()
except BadCredentialsException as e:
raise ConfigEntryAuthFailed("Bad credentials detected.") from e
except FailedConnectionException as e:
raise ConfigEntryNotReady("Connection with client failed.") from e
try:
appliances = await fetch_appliance_data(client)
except ApplianceClientException as e:
raise ConfigEntryNotReady from e
coordinators: dict[str, ElectroluxDataUpdateCoordinator] = {}
on_livestream_opening_callback_list: list[Callable[[], Awaitable[None]]] = []
async def check_for_new_devices_callback() -> None:
"""Trigger _check_for_new_devices asynchronously."""
await _check_for_new_devices(
hass, entry, client, on_livestream_opening_callback_list
)
on_livestream_opening_callback_list.append(check_for_new_devices_callback)
for appliance in appliances:
appliance_id = appliance.appliance.applianceId
coordinator = ElectroluxDataUpdateCoordinator(
hass, entry, client=client, appliance_id=appliance_id
)
await coordinator.async_config_entry_first_refresh()
# Subscribe this coordinator to its appliance events
coordinator.add_client_listener()
coordinators[appliance_id] = coordinator
# Device state is refreshed whenever the SSE connection opens.
on_livestream_opening_callback_list.append(coordinator.async_refresh)
sse_task = entry.async_create_background_task(
hass,
client.start_event_stream(on_livestream_opening_callback_list),
"electrolux event listener",
)
entry.runtime_data = ElectroluxData(
client=client,
appliances=appliances,
coordinators=coordinators,
sse_task=sse_task,
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ElectroluxConfigEntry) -> bool:
"""Unload a config entry."""
# Remove SSE listeners
coordinators = entry.runtime_data.coordinators
for coordinator in coordinators.values():
coordinator.remove_client_listeners()
# Cancel SSE task
sse_task = entry.runtime_data.sse_task
sse_task.cancel()
try:
await sse_task
except CancelledError:
_LOGGER.info("SSE stream cancelled for entry %s", entry.entry_id)
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
def create_token_manager(
hass: HomeAssistant,
entry: ElectroluxConfigEntry,
) -> TokenManager:
"""Create a token manager for the Electrolux integration."""
def save_tokens(new_access: str, new_refresh: str, new_api_key: str) -> None:
hass.config_entries.async_update_entry(
entry,
data={
**entry.data,
CONF_API_KEY: new_api_key,
CONF_ACCESS_TOKEN: new_access,
CONF_REFRESH_TOKEN: new_refresh,
},
)
api_key = entry.data.get(CONF_API_KEY)
refresh_token = entry.data.get(CONF_REFRESH_TOKEN)
access_token = entry.data.get(CONF_ACCESS_TOKEN)
if access_token and refresh_token and api_key:
return TokenManager(access_token, refresh_token, api_key, save_tokens)
raise ConfigEntryAuthFailed("Missing access token, refresh token or API key")
async def _check_for_new_devices(
hass: HomeAssistant,
entry: ElectroluxConfigEntry,
client: ApplianceClient,
on_livestream_opening_callback_list: list[Callable[[], Awaitable[None]]],
) -> None:
"""Fetch appliances from API and trigger discovery for any new ones."""
_LOGGER.info("Checking for new devices")
coordinators = entry.runtime_data.coordinators
appliances = await fetch_appliance_data(client)
entry.runtime_data.appliances = appliances
existing_ids = set(coordinators.keys())
for appliance in appliances:
appliance_id = appliance.appliance.applianceId
# Detect NEW appliances
if appliance_id not in existing_ids:
# Create coordinator for appliance
coordinator = ElectroluxDataUpdateCoordinator(
hass, entry, client=client, appliance_id=appliance_id
)
await coordinator.async_refresh()
coordinator.add_client_listener()
coordinators[appliance_id] = coordinator
on_livestream_opening_callback_list.append(coordinator.async_refresh)
# Notify all platforms
async_dispatcher_send(
hass, f"{NEW_APPLIANCE_SIGNAL}_{entry.entry_id}", appliance
)
# Detect MISSING appliances
discovered_ids = {appliance.appliance.applianceId for appliance in appliances}
missing_ids = existing_ids - discovered_ids
device_registry = dr.async_get(hass)
for missing_id in missing_ids:
_LOGGER.warning("Appliance %s no longer found, removing", missing_id)
# Remove coordinator
coordinator = coordinators.pop(missing_id)
coordinator.remove_client_listeners()
on_livestream_opening_callback_list.remove(coordinator.async_refresh)
device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, missing_id)}
)
if device_entry:
device_registry.async_update_device(
device_entry.id, remove_config_entry_id=entry.entry_id
)
async def fetch_appliance_data(client: ApplianceClient) -> list[ApplianceData]:
"""Helper method to retrieve all the appliances data from the Electrolux APIs."""
try:
appliances = await client.get_appliance_data()
except ApplianceClientException as e:
_LOGGER.warning("Failed to get appliances: %s", e)
raise
# Filter out appliances where details or state is None
return [
appliance
for appliance in appliances
if appliance.details is not None and appliance.state is not None
]
@@ -1,99 +0,0 @@
"""Config flow for Electrolux integration."""
from collections.abc import Mapping
import logging
from typing import Any
from electrolux_group_developer_sdk.auth.invalid_credentials_exception import (
InvalidCredentialsException,
)
from electrolux_group_developer_sdk.auth.token_manager import TokenManager
from electrolux_group_developer_sdk.client.appliance_client import ApplianceClient
from electrolux_group_developer_sdk.client.bad_credentials_exception import (
BadCredentialsException,
)
from electrolux_group_developer_sdk.client.failed_connection_exception import (
FailedConnectionException,
)
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_API_KEY
from .const import CONF_REFRESH_TOKEN, DOMAIN, USER_AGENT
_LOGGER: logging.Logger = logging.getLogger(__name__)
class ElectroluxConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for the Electrolux integration."""
VERSION = 1
MINOR_VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step of the config flow."""
errors: dict[str, str] = {}
if user_input is not None:
token_manager: TokenManager
email: str
try:
token_manager = await _authenticate_user(user_input)
client = ApplianceClient(
token_manager=token_manager, external_user_agent=USER_AGENT
)
email = (await client.get_user_email()).email
except InvalidCredentialsException, BadCredentialsException:
errors["base"] = "invalid_auth"
except FailedConnectionException:
errors["base"] = "cannot_connect"
else:
await self.async_set_unique_id(token_manager.get_user_id())
self._abort_if_unique_id_configured()
if not errors:
return self.async_create_entry(
title=f"Electrolux for {email}",
data=user_input,
)
return self._show_form(step_id="user", errors=errors)
def _show_form(self, step_id: str, errors: dict[str, str]) -> ConfigFlowResult:
return self.async_show_form(
step_id=step_id,
data_schema=vol.Schema(
{
vol.Required(CONF_API_KEY): str,
vol.Required(CONF_ACCESS_TOKEN): str,
vol.Required(CONF_REFRESH_TOKEN): str,
}
),
errors=errors,
description_placeholders={
"portal_link": "https://developer.electrolux.one/generateToken"
},
)
async def _authenticate_user(user_input: Mapping[str, Any]) -> TokenManager:
token_manager = TokenManager(
access_token=user_input[CONF_ACCESS_TOKEN],
refresh_token=user_input[CONF_REFRESH_TOKEN],
api_key=user_input[CONF_API_KEY],
)
token_manager.ensure_credentials()
appliance_client = ApplianceClient(
token_manager=token_manager, external_user_agent=USER_AGENT
)
# Test a connection in the config flow
await appliance_client.test_connection()
return token_manager
@@ -1,11 +0,0 @@
"""Constants for Electrolux integration."""
from homeassistant.const import __version__ as HA_VERSION
DOMAIN = "electrolux"
CONF_REFRESH_TOKEN = "refresh_token"
NEW_APPLIANCE_SIGNAL = "electrolux_new_appliance"
USER_AGENT = f"HomeAssistant/{HA_VERSION}"
@@ -1,96 +0,0 @@
"""Electrolux coordinator class."""
from __future__ import annotations
from asyncio import Task
from dataclasses import dataclass
import logging
from electrolux_group_developer_sdk.client.appliance_client import (
ApplianceClient,
apply_sse_update,
)
from electrolux_group_developer_sdk.client.appliances.appliance_data import (
ApplianceData,
)
from electrolux_group_developer_sdk.client.client_exception import (
ApplianceClientException,
)
from electrolux_group_developer_sdk.client.dto.appliance_state import ApplianceState
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
_LOGGER: logging.Logger = logging.getLogger(__name__)
@dataclass(kw_only=True, slots=True)
class ElectroluxData:
"""Electrolux data type."""
client: ApplianceClient
appliances: list[ApplianceData]
coordinators: dict[str, ElectroluxDataUpdateCoordinator]
sse_task: Task
type ElectroluxConfigEntry = ConfigEntry[ElectroluxData]
class ElectroluxDataUpdateCoordinator(DataUpdateCoordinator[ApplianceState]):
"""Class for fetching appliance data from the API."""
def __init__(
self,
hass: HomeAssistant,
config_entry: ElectroluxConfigEntry,
client: ApplianceClient,
appliance_id: str,
) -> None:
"""Initialize."""
self.client = client
self._appliance_id = appliance_id
super().__init__(
hass,
_LOGGER,
config_entry=config_entry,
name=f"{DOMAIN}_{config_entry.entry_id}_{appliance_id}",
update_interval=None,
always_update=False,
)
async def _async_update_data(self) -> ApplianceState:
"""Return the current appliance state (SSE keeps it updated)."""
try:
appliance_state = await self.client.get_appliance_state(self._appliance_id)
except ValueError as exception:
raise UpdateFailed(exception) from exception
except ApplianceClientException as exception:
raise UpdateFailed(exception) from exception
else:
return appliance_state
def add_client_listener(self) -> None:
"""Register an SSE listener to the appliance client for appliance state updates."""
self.client.add_listener(self._appliance_id, self.callback_handle_event)
def remove_client_listeners(self) -> None:
"""Remove all SSE listeners."""
self.client.remove_all_listeners_by_appliance_id(self._appliance_id)
def callback_handle_event(self, event: dict) -> None:
"""Handle an incoming SSE event. Event will look like: {"userId": "...", "applianceId": "...", "property": "timeToEnd", "value": 720}."""
current_state = self.data
if not current_state:
return
updated_state = apply_sse_update(
current_state,
event,
)
self.async_set_updated_data(updated_state)
@@ -1,80 +0,0 @@
"""Base entity for Electrolux integration."""
from abc import abstractmethod
import logging
from typing import TYPE_CHECKING
from electrolux_group_developer_sdk.client.appliances.appliance_data import (
ApplianceData,
)
from homeassistant.core import callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import ElectroluxDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
class ElectroluxBaseEntity[T: ApplianceData](
CoordinatorEntity[ElectroluxDataUpdateCoordinator]
):
"""Base class for Electrolux entities."""
_attr_has_entity_name = True
def __init__(
self,
appliance_data: T,
coordinator: ElectroluxDataUpdateCoordinator,
unique_id_suffix: str,
) -> None:
"""Initialize the base device."""
super().__init__(coordinator)
appliance_name = appliance_data.appliance.applianceName
appliance_id = appliance_data.appliance.applianceId
if TYPE_CHECKING:
assert appliance_data.details
assert appliance_data.state
appliance_info = appliance_data.details.applianceInfo
self._appliance_data = appliance_data
self._attr_unique_id = f"{appliance_id}_{unique_id_suffix}"
self._appliance_id = appliance_id
self._appliance_capabilities = appliance_data.details.capabilities
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, appliance_id)},
name=appliance_name,
manufacturer=appliance_info.brand,
model=appliance_info.model,
serial_number=appliance_info.serialNumber,
)
async def async_added_to_hass(self) -> None:
"""When entity is added to HA."""
await super().async_added_to_hass()
self._handle_coordinator_update()
@abstractmethod
def _update_attr_state(self) -> bool:
"""Update entity-specific attributes. Returns True if any attributes were changed, otherwise False."""
@callback
def _handle_coordinator_update(self) -> None:
"""When the coordinator updates."""
appliance_state = self.coordinator.data
if not appliance_state:
_LOGGER.warning("Appliance %s not found in update", self._appliance_id)
return
# Update state
self._appliance_data.update_state(appliance_state)
state_changed = self._update_attr_state()
if state_changed:
self.async_write_ha_state()
@@ -1,49 +0,0 @@
"""Contains entity helper methods."""
from collections.abc import Callable
from electrolux_group_developer_sdk.client.appliances.appliance_data import (
ApplianceData,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import NEW_APPLIANCE_SIGNAL
from .coordinator import ElectroluxConfigEntry, ElectroluxDataUpdateCoordinator
from .entity import ElectroluxBaseEntity
async def async_setup_entities_helper(
hass: HomeAssistant,
entry: ElectroluxConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
build_entities_fn: Callable[
[ApplianceData, dict[str, ElectroluxDataUpdateCoordinator]],
list[ElectroluxBaseEntity],
],
) -> None:
"""Provide async_setup_entry helper."""
appliances: list[ApplianceData] = entry.runtime_data.appliances
coordinators = entry.runtime_data.coordinators
entities: list[ElectroluxBaseEntity] = []
for appliance_data in appliances:
entities.extend(build_entities_fn(appliance_data, coordinators))
async_add_entities(entities)
# Listen for new/removed appliances
async def _new_appliance(appliance_data: ApplianceData):
new_entities = build_entities_fn(appliance_data, coordinators)
if new_entities:
async_add_entities(new_entities)
entry.async_on_unload(
async_dispatcher_connect(
hass, f"{NEW_APPLIANCE_SIGNAL}_{entry.entry_id}", _new_appliance
)
)
@@ -1,18 +0,0 @@
{
"entity": {
"sensor": {
"appliance_state": {
"default": "mdi:information-outline"
},
"food_probe_state": {
"default": "mdi:thermometer-probe"
},
"food_probe_temperature": {
"default": "mdi:thermometer-probe"
},
"remote_control": {
"default": "mdi:remote"
}
}
}
}
@@ -1,11 +0,0 @@
{
"domain": "electrolux",
"name": "Electrolux",
"codeowners": ["@electrolux-oss"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/electrolux",
"integration_type": "hub",
"iot_class": "cloud_push",
"quality_scale": "bronze",
"requirements": ["electrolux-group-developer-sdk==0.5.0"]
}
@@ -1,70 +0,0 @@
rules:
# Bronze
action-setup:
status: exempt
comment: |
No actions are implemented currently.
appropriate-polling:
status: exempt
comment: |
Polling is only performed on infrequent events (when the livestream of events is opened, in order to sync),
otherwise the integration works via push
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: |
No actions are implemented currently.
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: done
docs-configuration-parameters: todo
docs-installation-parameters: todo
entity-unavailable: todo
integration-owner: done
log-when-unavailable: todo
parallel-updates: todo
reauthentication-flow: todo
test-coverage: todo
# Gold
devices: todo
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: todo
entity-category: todo
entity-device-class: todo
entity-disabled-by-default: todo
entity-translations: todo
exception-translations: todo
icon-translations: done
reconfiguration-flow: todo
repair-issues: todo
stale-devices: todo
# Platinum
async-dependency: done
inject-websession: todo
strict-typing: todo
@@ -1,290 +0,0 @@
"""Sensor entity for Electrolux Integration."""
from collections.abc import Callable
from dataclasses import dataclass
import logging
from typing import cast
from electrolux_group_developer_sdk.client.appliances.appliance_data import (
ApplianceData,
)
from electrolux_group_developer_sdk.client.appliances.cr_appliance import CRAppliance
from electrolux_group_developer_sdk.client.appliances.ov_appliance import OVAppliance
from electrolux_group_developer_sdk.feature_constants import (
APPLIANCE_STATE,
DISPLAY_FOOD_PROBE_TEMPERATURE_C,
DISPLAY_FOOD_PROBE_TEMPERATURE_F,
DISPLAY_TEMPERATURE_C,
DISPLAY_TEMPERATURE_F,
FOOD_PROBE_STATE,
REMOTE_CONTROL,
)
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
StateType,
)
from homeassistant.const import UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util.unit_conversion import TemperatureConverter
from .coordinator import ElectroluxConfigEntry, ElectroluxDataUpdateCoordinator
from .entity import ElectroluxBaseEntity
from .entity_helper import async_setup_entities_helper
_LOGGER = logging.getLogger(__name__)
ELECTROLUX_TO_HA_TEMPERATURE_UNIT = {
"CELSIUS": UnitOfTemperature.CELSIUS,
"FAHRENHEIT": UnitOfTemperature.FAHRENHEIT,
}
@dataclass(frozen=True, kw_only=True)
class ElectroluxSensorDescription(SensorEntityDescription):
"""Custom sensor description for Electrolux sensors."""
value_fn: Callable[..., StateType]
exists_fn: Callable[[ApplianceData], bool] = lambda *args: True
feature_name: str | None = None
known_values: set[str] | None = None
OVEN_ELECTROLUX_SENSORS: tuple[ElectroluxSensorDescription, ...] = (
ElectroluxSensorDescription(
key="appliance_state",
translation_key="appliance_state",
value_fn=lambda appliance: appliance.get_current_appliance_state(),
device_class=SensorDeviceClass.ENUM,
feature_name=APPLIANCE_STATE,
exists_fn=lambda appliance: appliance.is_feature_supported(APPLIANCE_STATE),
known_values={
"alarm",
"delayed_start",
"end_of_cycle",
"idle",
"off",
"paused",
"ready_to_start",
"running",
},
),
ElectroluxSensorDescription(
key="food_probe_state",
translation_key="food_probe_state",
value_fn=lambda appliance: appliance.get_current_food_probe_insertion_state(),
device_class=SensorDeviceClass.ENUM,
feature_name=FOOD_PROBE_STATE,
exists_fn=lambda appliance: appliance.is_feature_supported(FOOD_PROBE_STATE),
known_values={
"inserted",
"not_inserted",
},
),
ElectroluxSensorDescription(
key="remote_control",
translation_key="remote_control",
value_fn=lambda appliance: appliance.get_current_remote_control(),
device_class=SensorDeviceClass.ENUM,
feature_name=REMOTE_CONTROL,
exists_fn=lambda appliance: appliance.is_feature_supported(REMOTE_CONTROL),
known_values={
"disabled",
"enabled",
"not_safety_relevant_enabled",
"temporary_locked",
},
),
)
OVEN_TEMPERATURE_ELECTROLUX_SENSORS: tuple[ElectroluxSensorDescription, ...] = (
ElectroluxSensorDescription(
key="food_probe_temperature",
translation_key="food_probe_temperature",
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda appliance, temp_unit=None: (
appliance.get_current_display_food_probe_temperature_f()
if temp_unit == UnitOfTemperature.FAHRENHEIT
else appliance.get_current_display_food_probe_temperature_c()
),
exists_fn=lambda appliance: appliance.is_feature_supported(
[DISPLAY_FOOD_PROBE_TEMPERATURE_F, DISPLAY_FOOD_PROBE_TEMPERATURE_C]
),
),
ElectroluxSensorDescription(
key="display_temperature",
translation_key="display_temperature",
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda appliance, temp_unit=None: (
appliance.get_current_display_temperature_f()
if temp_unit == UnitOfTemperature.FAHRENHEIT
else appliance.get_current_display_temperature_c()
),
exists_fn=lambda appliance: appliance.is_feature_supported(
[DISPLAY_TEMPERATURE_C, DISPLAY_TEMPERATURE_F]
),
),
)
def build_entities_for_appliance(
appliance_data: ApplianceData,
coordinators: dict[str, ElectroluxDataUpdateCoordinator],
) -> list[ElectroluxBaseEntity]:
"""Return all entities for a single appliance."""
appliance = appliance_data.appliance
coordinator = coordinators[appliance.applianceId]
entities: list[ElectroluxBaseEntity] = []
if isinstance(appliance_data, OVAppliance):
entities.extend(
ElectroluxSensor(appliance_data, coordinator, description)
for description in OVEN_ELECTROLUX_SENSORS
if description.exists_fn(appliance_data)
)
entities.extend(
ElectroluxTemperatureSensor(appliance_data, coordinator, description)
for description in OVEN_TEMPERATURE_ELECTROLUX_SENSORS
if description.exists_fn(appliance_data)
)
return entities
async def async_setup_entry(
hass: HomeAssistant,
entry: ElectroluxConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set sensor for Electrolux Integration."""
await async_setup_entities_helper(
hass, entry, async_add_entities, build_entities_for_appliance
)
class ElectroluxSensor(ElectroluxBaseEntity[ApplianceData], SensorEntity):
"""Representation of a generic sensor for Electrolux appliances."""
entity_description: ElectroluxSensorDescription
def __init__(
self,
appliance_data: ApplianceData,
coordinator: ElectroluxDataUpdateCoordinator,
description: ElectroluxSensorDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(appliance_data, coordinator, description.key)
if (
description.feature_name is not None
and description.known_values is not None
):
options = appliance_data.get_feature_state_string_options(
description.feature_name
)
snake_case_options = [
snake_case_option
for option in options
if (snake_case_option := _convert_to_snake_case(option))
in description.known_values
]
if len(snake_case_options) > 0:
self._attr_options = snake_case_options
self.entity_description = description
def _update_attr_state(self) -> bool:
new_value = self._get_value()
if isinstance(new_value, str):
new_value = _convert_to_snake_case(new_value)
if self.entity_description.known_values:
new_value = _map_to_known_value(
self.entity_description.known_values,
self.entity_description.key,
new_value,
)
if self._attr_native_value != new_value:
self._attr_native_value = new_value
return True
return False
def _get_value(self) -> StateType:
return self.entity_description.value_fn(self._appliance_data)
class ElectroluxTemperatureSensor(ElectroluxSensor):
"""Representation of a temperature sensor for Electrolux appliances."""
def __init__(
self,
appliance_data: ApplianceData,
coordinator: ElectroluxDataUpdateCoordinator,
description: ElectroluxSensorDescription,
) -> None:
"""Initialize the sensor."""
self._appliance = cast(OVAppliance | CRAppliance, appliance_data)
self._attr_native_unit_of_measurement = UnitOfTemperature.CELSIUS
super().__init__(appliance_data, coordinator, description)
def _get_value(self) -> StateType:
temp_unit = self._get_temperature_unit()
temp_value: float | None = cast(
float | None,
self.entity_description.value_fn(self._appliance_data, temp_unit=temp_unit),
)
if temp_value is None:
return None
return TemperatureConverter.convert(
temp_value, temp_unit, UnitOfTemperature.CELSIUS
)
def _get_temperature_unit(self) -> UnitOfTemperature:
temp_unit = self._appliance.get_current_temperature_unit()
if temp_unit is not None:
temp_unit = temp_unit.upper()
return ELECTROLUX_TO_HA_TEMPERATURE_UNIT.get(
temp_unit, UnitOfTemperature.CELSIUS
)
def _convert_to_snake_case(x: str) -> str:
"""Converts a string to snake case."""
lower_case = x.lower()
return "".join([_convert_char_to_snake_case(char) for char in lower_case])
def _convert_char_to_snake_case(char: str) -> str:
if char.isspace():
return "_"
return char
def _map_to_known_value(
known_values: set[str], entity_key: str, value: str
) -> str | None:
"""Return provided value if it is known, otherwise log warn message and return None."""
if value not in known_values:
_LOGGER.warning(
"An unknown value %s was reported for a sensor of the Electrolux integration. "
"Please report it for the integration, and include the following information: "
'entity key="%s", reported value="%s"',
value,
entity_key,
value,
)
return None
return value
@@ -1,66 +0,0 @@
{
"config": {
"abort": {
"already_configured": "This Electrolux account is already configured."
},
"error": {
"cannot_connect": "Unable to connect to the Electrolux API. Please check credentials and try again.",
"invalid_auth": "Authentication failed. Please check your credentials."
},
"step": {
"user": {
"data": {
"access_token": "[%key:common::config_flow::data::access_token%]",
"api_key": "[%key:common::config_flow::data::api_key%]",
"refresh_token": "Refresh token"
},
"data_description": {
"access_token": "The access token from Electrolux Group for Developer.",
"api_key": "Your Electrolux Group for Developer API key.",
"refresh_token": "The refresh token used to renew your access token."
},
"description": "Please go to the [developer portal]({portal_link}) to generate new access and refresh tokens, then paste them below.",
"title": "Configure your Electrolux Group account"
}
}
},
"entity": {
"sensor": {
"appliance_state": {
"name": "Appliance state",
"state": {
"alarm": "Alarm",
"delayed_start": "Delayed start",
"end_of_cycle": "Cycle ended",
"idle": "[%key:common::state::idle%]",
"off": "[%key:common::state::off%]",
"paused": "[%key:common::state::paused%]",
"ready_to_start": "Ready to start",
"running": "Running"
}
},
"display_temperature": {
"name": "Current temperature"
},
"food_probe_state": {
"name": "Food probe state",
"state": {
"inserted": "Inserted",
"not_inserted": "Not inserted"
}
},
"food_probe_temperature": {
"name": "Food probe temperature"
},
"remote_control": {
"name": "Remote control",
"state": {
"disabled": "[%key:common::state::disabled%]",
"enabled": "[%key:common::state::enabled%]",
"not_safety_relevant_enabled": "Not safety relevant enabled",
"temporary_locked": "Temporarily locked"
}
}
}
}
}
+6 -2
View File
@@ -509,12 +509,14 @@ FITBIT_RESOURCE_BATTERY = FitbitSensorEntityDescription(
icon="mdi:battery",
scope=FitbitScope.DEVICE,
entity_category=EntityCategory.DIAGNOSTIC,
has_entity_name=True,
)
FITBIT_RESOURCE_BATTERY_LEVEL = FitbitSensorEntityDescription(
key="devices/battery_level",
translation_key="battery_level",
scope=FitbitScope.DEVICE,
entity_category=EntityCategory.DIAGNOSTIC,
has_entity_name=True,
device_class=SensorDeviceClass.BATTERY,
native_unit_of_measurement=PERCENTAGE,
)
@@ -647,12 +649,13 @@ class FitbitSensor(SensorEntity):
self.async_schedule_update_ha_state(force_refresh=True)
# has_entity_name=True is supplied by the FITBIT_RESOURCE_BATTERY description
# pylint: disable-next=home-assistant-missing-has-entity-name
class FitbitBatterySensor(CoordinatorEntity[FitbitDeviceCoordinator], SensorEntity):
"""Implementation of a Fitbit battery sensor."""
entity_description: FitbitSensorEntityDescription
_attr_attribution = ATTRIBUTION
_attr_has_entity_name = True
def __init__(
self,
@@ -706,13 +709,14 @@ class FitbitBatterySensor(CoordinatorEntity[FitbitDeviceCoordinator], SensorEnti
self.async_write_ha_state()
# has_entity_name=True is supplied by the FITBIT_RESOURCE_BATTERY_LEVEL description
# pylint: disable-next=home-assistant-missing-has-entity-name
class FitbitBatteryLevelSensor(
CoordinatorEntity[FitbitDeviceCoordinator], SensorEntity
):
"""Implementation of a Fitbit battery level sensor."""
entity_description: FitbitSensorEntityDescription
_attr_has_entity_name = True
_attr_attribution = ATTRIBUTION
def __init__(
@@ -180,24 +180,27 @@ async def async_setup_entry(
class HomematicipCloudConnectionSensor(HomematicipGenericEntity, BinarySensorEntity):
"""Representation of the HomematicIP cloud connection sensor."""
_attr_translation_key = "cloud_connection"
def __init__(self, hap: HomematicipHAP) -> None:
"""Initialize the cloud connection sensor."""
super().__init__(hap, hap.home, feature_id="cloud_connection")
@property
def name(self) -> str:
"""Return the name cloud connection entity."""
name = "Cloud Connection"
# Add a prefix to the name if the homematic ip home has a name.
return name if not self._home.name else f"{self._home.name} {name}"
@property
def device_info(self) -> DeviceInfo:
"""Return device specific attributes."""
# Merges into the existing HAP device registered in __init__.py.
# Name must match __init__.py logic for has_entity_name to work.
label = self._home.label or ""
# Adds a sensor to the existing HAP device
return DeviceInfo(
identifiers={
# Serial numbers of Homematic IP device
(DOMAIN, self._home.id)
},
name=label,
}
)
@property
@@ -576,7 +579,6 @@ class HomematicipPluggableMainsFailureSurveillanceSensor(
class HomematicipSecurityZoneSensorGroup(HomematicipGenericEntity, BinarySensorEntity):
"""Representation of the HomematicIP security zone sensor group."""
_attr_has_entity_name = False
_attr_device_class = BinarySensorDeviceClass.SAFETY
def __init__(
@@ -74,7 +74,6 @@ class HomematicipHeatingGroup(HomematicipGenericEntity, ClimateEntity):
basically enabled in the hmip app.
"""
_attr_has_entity_name = False
_attr_supported_features = (
ClimateEntityFeature.PRESET_MODE | ClimateEntityFeature.TARGET_TEMPERATURE
)
@@ -320,7 +320,6 @@ class HomematicipGarageDoorModule(HomematicipGenericEntity, CoverEntity):
class HomematicipCoverShutterGroup(HomematicipGenericEntity, CoverEntity):
"""Representation of the HomematicIP cover shutter group."""
_attr_has_entity_name = False
_attr_device_class = CoverDeviceClass.SHUTTER
def __init__(self, hap: HomematicipHAP, device, post: str = "ShutterGroup") -> None:
@@ -74,7 +74,6 @@ GROUP_ATTRIBUTES = {
class HomematicipGenericEntity(Entity):
"""Representation of the HomematicIP generic entity."""
_attr_has_entity_name = True
_attr_should_poll = False
def __init__(
@@ -113,14 +112,6 @@ class HomematicipGenericEntity(Entity):
# Marker showing that the HmIP device hase been removed.
self.hmip_device_removed = False
# Compute entity name based on has_entity_name mode.
if not self._attr_has_entity_name:
# Legacy mode (groups, special entities): compose the full name
# including device/group label and home prefix.
self._attr_name = self._compute_legacy_name()
else:
self._setup_entity_name()
@property
def device_info(self) -> DeviceInfo | None:
"""Return device specific attributes."""
@@ -129,14 +120,6 @@ class HomematicipGenericEntity(Entity):
device_id = str(self._device.id)
home_id = str(self._device.homeId)
# Include the home name in the device name so that the
# previous "{home} {device}" naming is preserved after
# switching to has_entity_name=True.
device_name = self._device.label
home_name = getattr(self._home, "name", None)
if device_name and home_name:
device_name = f"{home_name} {device_name}"
return DeviceInfo(
identifiers={
# Serial numbers of Homematic IP device
@@ -144,7 +127,7 @@ class HomematicipGenericEntity(Entity):
},
manufacturer=self._device.oem,
model=self._device.modelType,
name=device_name,
name=self._device.label,
sw_version=self._device.firmwareVersion,
# Link to the homematic ip access point.
via_device=(DOMAIN, home_id),
@@ -217,93 +200,38 @@ class HomematicipGenericEntity(Entity):
self.async_remove(force_remove=True), eager_start=False
)
def _compute_legacy_name(self) -> str:
"""Compute the full legacy name for entities without has_entity_name.
@property
def name(self) -> str:
"""Return the name of the generic entity."""
Used by group entities and other special cases where has_entity_name
is False. Includes device/group label, post suffix, and home prefix.
"""
name = self._device.label or ""
if self._post:
name = f"{name} {self._post}" if name else self._post
name = ""
# Try to get a label from a channel.
functional_channels = getattr(self._device, "functionalChannels", None)
if functional_channels and self.functional_channel:
if self._is_multi_channel:
label = getattr(self.functional_channel, "label", None)
if label:
name = str(label)
elif len(functional_channels) > 1:
label = getattr(functional_channels[1], "label", None)
if label:
name = str(label)
# Use device label, if name is not defined by channel label.
if not name:
name = self._device.label or ""
if self._post:
name = f"{name} {self._post}"
elif self._is_multi_channel:
name = f"{name} Channel{self.get_channel_index()}"
# Add a prefix to the name if the homematic ip home has a name.
home_name = getattr(self._home, "name", None)
if name and home_name:
name = f"{home_name} {name}"
return name
def _setup_entity_name(self) -> None:
"""Set up entity naming for has_entity_name mode.
With has_entity_name=True, HA composes the full friendly name as
"{device_name} {entity_name}". This method sets the appropriate
naming attributes.
For multi-channel entities, channel labels provide _attr_name (dynamic).
For entities with _post, _attr_name is derived from the post suffix,
with the first letter capitalized for display consistency.
For primary entities, HA uses device_class as the name.
"""
# Multi-channel entities: use channel label as entity name.
if self._is_multi_channel and self.functional_channel:
label = getattr(self.functional_channel, "label", None)
if label:
label_str = str(label)
device_label = self._device.label or ""
# Strip device name prefix from channel label to avoid
# duplication when HA composes "{device_name} {entity_name}".
# E.g., device "Licht Flur" + channel "Licht Flur 5" -> "5".
if device_label and label_str.startswith(device_label):
stripped = label_str[len(device_label) :].strip()
if stripped:
self._attr_name = stripped
# Otherwise channel label equals device label (modulo
# whitespace); leave _attr_name unset so HA composes just
# the device name without duplicating it.
return
self._attr_name = label_str
return
# Fallback: use post suffix or generic channel name.
if self._post:
self._attr_name = self._post[0].upper() + self._post[1:]
else:
self._attr_name = f"Channel{self.get_channel_index()}"
return
# Entities with a post suffix: use it as the entity name,
# capitalizing the first letter for display consistency.
if self._post:
self._attr_name = self._post[0].upper() + self._post[1:]
return
# Non-multi-channel entities on devices with multiple channels:
# use the first functional channel's label as name context.
# This preserves names like "Treppe CH" for single-function entities
# on multi-channel devices (e.g., HmIP-BSL switch channel).
functional_channels = getattr(self._device, "functionalChannels", None)
if functional_channels and len(functional_channels) > 1:
ch1 = (
functional_channels.get(1)
if isinstance(functional_channels, dict)
else functional_channels[1]
)
label = getattr(ch1, "label", None) if ch1 else None
if label:
label_str = str(label)
device_label = self._device.label or ""
# Strip device name prefix to avoid duplication.
if device_label and label_str.startswith(device_label):
stripped = label_str[len(device_label) :].strip()
if stripped:
self._attr_name = stripped
# Otherwise channel label equals device label (modulo
# whitespace); leave _attr_name unset.
return
self._attr_name = label_str
return
# Primary entity on device: leave unset so HA derives name from
# device_class or translation_key.
@property
def available(self) -> bool:
"""Return if entity is available."""
@@ -82,6 +82,7 @@ class HomematicipDoorBellEvent(HomematicipGenericEntity, EventEntity):
super().__init__(
hap,
device,
post=description.key,
channel=channel,
is_multi_channel=False,
feature_id="doorbell",
@@ -1070,7 +1070,9 @@ class HmipSmokeDetectorSensor(HomematicipGenericEntity, SensorEntity):
description: HmipSmokeDetectorSensorDescription,
) -> None:
"""Initialize the smoke detector sensor."""
super().__init__(hap, device, feature_id="smoke_detector_sensor")
super().__init__(
hap, device, post=description.key, feature_id="smoke_detector_sensor"
)
self.entity_description = description
self._sensor_unique_id = f"{device.id}_{description.key}"
@@ -37,11 +37,6 @@
}
},
"entity": {
"binary_sensor": {
"cloud_connection": {
"name": "Cloud connection"
}
},
"light": {
"optical_signal_light": {
"state_attributes": {
@@ -142,8 +142,6 @@ class HomematicipSwitch(HomematicipMultiSwitch, SwitchEntity):
class HomematicipGroupSwitch(HomematicipGenericEntity, SwitchEntity):
"""Representation of the HomematicIP switching group."""
_attr_has_entity_name = False
def __init__(self, hap: HomematicipHAP, device, post: str = "Group") -> None:
"""Initialize switching group."""
device.modelType = f"HmIP-{post}"
@@ -74,6 +74,11 @@ class HomematicipWeatherSensor(HomematicipGenericEntity, WeatherEntity):
"""Initialize the weather sensor."""
super().__init__(hap, device, feature_id="weather")
@property
def name(self) -> str:
"""Return the name of the sensor."""
return self._device.label
@property
def native_temperature(self) -> float:
"""Return the platform temperature."""
@@ -113,7 +118,6 @@ class HomematicipWeatherSensorPro(HomematicipWeatherSensor):
class HomematicipHomeWeather(HomematicipGenericEntity, WeatherEntity):
"""Representation of the HomematicIP home weather."""
_attr_has_entity_name = False
_attr_native_temperature_unit = UnitOfTemperature.CELSIUS
_attr_native_wind_speed_unit = UnitOfSpeed.KILOMETERS_PER_HOUR
_attr_attribution = "Powered by Homematic IP"
@@ -73,9 +73,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: InComfortConfigEntry) ->
except InvalidHeaterList as exc:
raise NoHeaters from exc
except InvalidGateway as exc:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN, translation_key="incorrect_credentials"
) from exc
# pylint: disable-next=home-assistant-exception-not-translated
raise ConfigEntryAuthFailed("Incorrect credentials") from exc
except ClientResponseError as exc:
if exc.status == 404:
raise NotFound from exc
@@ -15,12 +15,10 @@ from incomfortclient import (
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.exceptions import ConfigEntryError
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
type InComfortConfigEntry = ConfigEntry[InComfortDataCoordinator]
_LOGGER = logging.getLogger(__name__)
@@ -79,20 +77,16 @@ class InComfortDataCoordinator(DataUpdateCoordinator[InComfortData]):
try:
for heater in self.incomfort_data.heaters:
await heater.update()
except TimeoutError as exc:
# pylint: disable-next=home-assistant-exception-not-translated
raise UpdateFailed("Timeout error") from exc
except ClientResponseError as exc:
if exc.status == 401:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN, translation_key="incorrect_credentials"
) from exc
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="update_failed_with_error_message",
translation_placeholders={"error": exc.message},
) from exc
# pylint: disable-next=home-assistant-exception-not-translated
raise ConfigEntryError("Incorrect credentials") from exc
# pylint: disable-next=home-assistant-exception-not-translated
raise UpdateFailed(exc.message) from exc
except InvalidHeaterList as exc:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="update_failed_with_error_message",
translation_placeholders={"error": exc.message},
) from exc
# pylint: disable-next=home-assistant-exception-not-translated
raise UpdateFailed(exc.message) from exc
return self.incomfort_data
@@ -131,7 +131,6 @@
}
},
"exceptions": {
"incorrect_credentials": { "message": "Incorrect credentials." },
"no_heaters": {
"message": "[%key:component::incomfort::config::error::no_heaters%]"
},
@@ -143,9 +142,6 @@
},
"unknown": {
"message": "[%key:component::incomfort::config::error::unknown%]"
},
"update_failed_with_error_message": {
"message": "Update failed, got {error}."
}
},
"options": {
+5 -2
View File
@@ -556,8 +556,9 @@ class KNXConfigFlow(ConfigFlow, domain=DOMAIN):
vol.Required(
CONF_KNX_ROUTE_BACK, default=_route_back
): selector.BooleanSelector(),
vol.Optional(CONF_KNX_LOCAL_IP): _IP_SELECTOR,
}
if self.show_advanced_options:
fields[vol.Optional(CONF_KNX_LOCAL_IP)] = _IP_SELECTOR
if not self._found_tunnels and not errors.get("base"):
errors["base"] = "no_tunnel_discovered"
@@ -889,8 +890,10 @@ class KNXConfigFlow(ConfigFlow, domain=DOMAIN):
): selector.BooleanSelector(),
vol.Required(CONF_KNX_MCAST_GRP, default=_multicast_group): _IP_SELECTOR,
vol.Required(CONF_KNX_MCAST_PORT, default=_multicast_port): _PORT_SELECTOR,
vol.Optional(CONF_KNX_LOCAL_IP): _IP_SELECTOR,
}
if self.show_advanced_options:
# Optional with default doesn't work properly in flow UI
fields[vol.Optional(CONF_KNX_LOCAL_IP)] = _IP_SELECTOR
return self.async_show_form(
step_id="routing", data_schema=vol.Schema(fields), errors=errors
+1 -2
View File
@@ -207,6 +207,7 @@ class LunatoneLight(
await self.coordinator.async_refresh()
# pylint: disable-next=home-assistant-missing-has-entity-name
class LunatoneLineBroadcastLight(
CoordinatorEntity[LunatoneInfoDataUpdateCoordinator], LightEntity
):
@@ -216,8 +217,6 @@ class LunatoneLineBroadcastLight(
_attr_assumed_state = True
_attr_color_mode = ColorMode.BRIGHTNESS
_attr_has_entity_name = True
_attr_name = None
_attr_supported_color_modes = {ColorMode.BRIGHTNESS}
def __init__(
@@ -11,7 +11,7 @@ from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.const import CONF_COUNTRY, CONF_LANGUAGE, CONF_NAME
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
@@ -22,6 +22,10 @@ _LOGGER = logging.getLogger(__name__)
ATTRIBUTION = "Information provided by MeteoAlarm"
# pylint: disable-next=home-assistant-duplicate-const
CONF_COUNTRY = "country"
# pylint: disable-next=home-assistant-duplicate-const
CONF_LANGUAGE = "language"
CONF_PROVINCE = "province"
DEFAULT_NAME = "meteoalarm"
@@ -10,6 +10,8 @@ DEFAULT_DETECTION_TIME: Final = 300
ATTR_MANUFACTURER: Final = "Mikrotik"
ATTR_SERIAL_NUMBER: Final = "serial-number"
ATTR_FIRMWARE: Final = "current-firmware"
# pylint: disable-next=home-assistant-duplicate-const
ATTR_MODEL: Final = "model"
CONF_ARP_PING: Final = "arp_ping"
CONF_FORCE_DHCP: Final = "force_dhcp"
@@ -9,13 +9,7 @@ import librouteros
from librouteros.login import plain as login_plain, token as login_token
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_MODEL,
CONF_HOST,
CONF_PASSWORD,
CONF_USERNAME,
CONF_VERIFY_SSL,
)
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, CONF_VERIFY_SSL
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
@@ -23,6 +17,7 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda
from .const import (
ARP,
ATTR_FIRMWARE,
ATTR_MODEL,
ATTR_SERIAL_NUMBER,
CAPSMAN,
CONF_ARP_PING,
@@ -198,7 +198,7 @@ get_library:
default: 0
selector:
number:
min: 0
min: 1
max: 1000000
step: 1
order_by:
+3 -1
View File
@@ -23,6 +23,7 @@ _ATTRIBUTION = "Data provided by OMIE.es"
SENSOR_DESCRIPTIONS: dict[str, SensorEntityDescription] = {
key: SensorEntityDescription(
key=key,
has_entity_name=True,
translation_key=key,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=f"{CURRENCY_EURO}/{UnitOfEnergy.KILO_WATT_HOUR}",
@@ -32,10 +33,11 @@ SENSOR_DESCRIPTIONS: dict[str, SensorEntityDescription] = {
}
# has_entity_name=True is supplied by every SENSOR_DESCRIPTIONS entry
# pylint: disable-next=home-assistant-missing-has-entity-name
class OMIEPriceSensor(CoordinatorEntity[OMIECoordinator], SensorEntity):
"""OMIE price sensor."""
_attr_has_entity_name = True
_attr_should_poll = False
_attr_attribution = _ATTRIBUTION
@@ -11,7 +11,7 @@ upload_image:
media:
accept:
- image/*
additional_fields:
advanced_options:
collapsed: true
fields:
rotation:
@@ -151,8 +151,8 @@
},
"name": "Upload image",
"sections": {
"additional_fields": {
"name": "Additional options"
"advanced_options": {
"name": "Advanced options"
}
}
}
@@ -8,5 +8,5 @@
"iot_class": "cloud_polling",
"loggers": ["renault_api"],
"quality_scale": "silver",
"requirements": ["renault-api==0.5.10"]
"requirements": ["renault-api==0.5.9"]
}
@@ -401,6 +401,7 @@ class ShellyBlockCoordinator(ShellyCoordinatorBase[BlockDevice]):
"""Fetch data."""
if self.sleep_period:
# Sleeping device, no point polling it, just mark it unavailable
# pylint: disable-next=home-assistant-exception-placeholder-mismatch
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="update_error_sleeping_device",
@@ -670,6 +671,7 @@ class ShellyRpcCoordinator(ShellyCoordinatorBase[RpcDevice]):
if self.sleep_period:
# Sleeping device, no point polling it, just mark it unavailable
# pylint: disable-next=home-assistant-exception-placeholder-mismatch
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="update_error_sleeping_device",
+1 -1
View File
@@ -673,7 +673,7 @@
"message": "An error occurred while reconnecting to {device}"
},
"update_error_sleeping_device": {
"message": "Sleeping device {device} did not update within {period} seconds interval"
"message": "Sleeping device did not update within {period} seconds interval"
}
},
"issues": {
+5 -1
View File
@@ -344,10 +344,14 @@ class RpcUpdateEntity(ShellyRpcAttributeEntity, UpdateEntity):
translation_placeholders={"device": self.coordinator.name},
) from err
except RpcCallError as err:
# pylint: disable-next=home-assistant-exception-placeholder-mismatch
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="ota_update_rpc_error",
translation_placeholders={"device": self.coordinator.name},
translation_placeholders={
"entity": self.entity_id,
"device": self.coordinator.name,
},
) from err
except InvalidAuthError:
await self.coordinator.async_shutdown_device_and_start_reauth()
@@ -22,11 +22,10 @@ async def async_setup_entry(
async_add_entities(SmartThingsScene(scene, client) for scene in scenes.values())
# pylint: disable-next=home-assistant-missing-has-entity-name
class SmartThingsScene(Scene):
"""Define a SmartThings scene."""
_attr_has_entity_name = True
def __init__(self, scene: STScene, client: SmartThings) -> None:
"""Init the scene class."""
self.client = client
+2
View File
@@ -9,6 +9,8 @@ ATTR_HTML: Final = "html"
ATTR_SENDER_NAME: Final = "sender_name"
CONF_ENCRYPTION: Final = "encryption"
# pylint: disable-next=home-assistant-duplicate-const
CONF_DEBUG: Final = "debug"
CONF_SERVER: Final = "server"
CONF_SENDER_NAME: Final = "sender_name"
+1 -1
View File
@@ -24,7 +24,6 @@ from homeassistant.components.notify import (
BaseNotificationService,
)
from homeassistant.const import (
CONF_DEBUG,
CONF_PASSWORD,
CONF_PORT,
CONF_RECIPIENT,
@@ -45,6 +44,7 @@ from homeassistant.util.ssl import create_client_context
from .const import (
ATTR_HTML,
ATTR_IMAGES,
CONF_DEBUG,
CONF_ENCRYPTION,
CONF_SENDER_NAME,
CONF_SERVER,
+1 -1
View File
@@ -191,11 +191,11 @@ class SonosAudioInputFormatSensorEntity(SonosPollingEntity, SensorEntity):
"""Provide a stub for required ABC method."""
# pylint: disable-next=home-assistant-missing-has-entity-name
class SonosFavoritesEntity(SensorEntity):
"""Representation of a Sonos favorites info entity."""
_attr_entity_registry_enabled_default = False
_attr_has_entity_name = True
_attr_name = "Sonos favorites"
_attr_translation_key = "favorites"
_attr_native_unit_of_measurement = "items"
+5 -1
View File
@@ -48,10 +48,14 @@ class LgWebOSNotificationService(BaseNotificationService):
icon_path = data.get(ATTR_ICON) if data else None
if not client.tv_state.is_on:
# pylint: disable-next=home-assistant-exception-placeholder-mismatch
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="notify_device_off",
translation_placeholders={"name": str(self._entry.title)},
translation_placeholders={
"name": str(self._entry.title),
"func": __name__,
},
)
try:
await client.send_message(message, icon_path=icon_path)
@@ -14,9 +14,10 @@ async def async_get_config_entry_diagnostics(
hass: HomeAssistant, config_entry: ZeversolarConfigEntry
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
data: ZeverSolarData = config_entry.runtime_data.data
return {
payload: dict[str, Any] = {
"wifi_enabled": data.wifi_enabled,
"serial_or_registry_id": data.serial_or_registry_id,
"registry_key": data.registry_key,
@@ -32,6 +33,8 @@ async def async_get_config_entry_diagnostics(
"meter_status": data.meter_status.value,
}
return payload
async def async_get_device_diagnostics(
hass: HomeAssistant, entry: ZeversolarConfigEntry, device: DeviceEntry
@@ -39,13 +42,15 @@ async def async_get_device_diagnostics(
"""Return diagnostics for a device entry."""
coordinator = entry.runtime_data
updateInterval = (
None
if coordinator.update_interval is None
else coordinator.update_interval.total_seconds()
)
return {
"name": coordinator.name,
"always_update": coordinator.always_update,
"last_update_success": coordinator.last_update_success,
"update_interval": (
None
if coordinator.update_interval is None
else coordinator.update_interval.total_seconds()
),
"update_interval": updateInterval,
}
-1
View File
@@ -91,7 +91,6 @@ CONF_COMMAND_ON: Final = "command_on"
CONF_COMMAND_OPEN: Final = "command_open"
CONF_COMMAND_STATE: Final = "command_state"
CONF_COMMAND_STOP: Final = "command_stop"
CONF_COMMENT: Final = "comment"
CONF_CONDITION: Final = "condition"
CONF_CONDITIONS: Final = "conditions"
CONF_CONTINUE_ON_ERROR: Final = "continue_on_error"
-1
View File
@@ -192,7 +192,6 @@ FLOWS = {
"ekeybionyx",
"electrasmart",
"electric_kiwi",
"electrolux",
"elevenlabs",
"elgato",
"elkm1",
@@ -1708,12 +1708,6 @@
"config_flow": true,
"iot_class": "cloud_polling"
},
"electrolux": {
"name": "Electrolux",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_push"
},
"elevenlabs": {
"name": "ElevenLabs",
"integration_type": "service",
@@ -38,7 +38,6 @@ from homeassistant.const import (
CONF_ATTRIBUTE,
CONF_BELOW,
CONF_CHOOSE,
CONF_COMMENT,
CONF_CONDITION,
CONF_CONDITIONS,
CONF_CONTINUE_ON_ERROR,
@@ -1459,7 +1458,6 @@ SCRIPT_SCHEMA = vol.All(ensure_list, [script_action])
SCRIPT_ACTION_BASE_SCHEMA: VolDictType = {
vol.Optional(CONF_ALIAS): string,
vol.Remove(CONF_COMMENT): str, # Is only used in frontend
vol.Optional(CONF_CONTINUE_ON_ERROR): boolean,
vol.Optional(CONF_ENABLED): vol.Any(boolean, template),
}
@@ -1527,7 +1525,6 @@ NUMERIC_STATE_THRESHOLD_SCHEMA = vol.Any(
CONDITION_BASE_SCHEMA: VolDictType = {
vol.Optional(CONF_ALIAS): string,
vol.Remove(CONF_COMMENT): str, # Is only used in frontend
vol.Optional(CONF_ENABLED): vol.Any(boolean, template),
}
@@ -1862,7 +1859,6 @@ TRIGGER_BASE_SCHEMA = vol.Schema(
vol.Optional(CONF_ID): str,
vol.Optional(CONF_VARIABLES): SCRIPT_VARIABLES_SCHEMA,
vol.Optional(CONF_ENABLED): vol.Any(boolean, template),
vol.Remove(CONF_COMMENT): str, # Is only used in frontend
}
)
+1 -4
View File
@@ -891,9 +891,6 @@ ekey-bionyxpy==1.0.1
# homeassistant.components.electric_kiwi
electrickiwi-api==0.9.14
# homeassistant.components.electrolux
electrolux-group-developer-sdk==0.5.0
# homeassistant.components.elevenlabs
elevenlabs==2.3.0
@@ -2868,7 +2865,7 @@ refoss-ha==1.2.5
regenmaschine==2024.03.0
# homeassistant.components.renault
renault-api==0.5.10
renault-api==0.5.9
# homeassistant.components.renson
renson-endura-delta==1.7.2
-2
View File
@@ -27,8 +27,6 @@ class CheckKind(StrEnum):
RELEASE_PIPELINE = "release_pipeline"
PR_LINK = "pr_link"
ASYNC_BLOCKING = "async_blocking"
YANKED = "yanked"
VULNERABILITIES = "vulnerabilities"
@dataclass(slots=True)
+2 -50
View File
@@ -1,6 +1,6 @@
"""PyPI metadata + PEP 740 provenance attestation lookups."""
from dataclasses import dataclass, field
from dataclasses import dataclass
import logging
import re
from typing import Any
@@ -69,17 +69,6 @@ _HEADERS = {
_TIMEOUT = 30.0
@dataclass(slots=True, frozen=True)
class Vulnerability:
"""One advisory entry for a specific package version (OSV / PyPA / GHSA)."""
id: str
aliases: tuple[str, ...]
summary: str
fixed_in: tuple[str, ...]
link: str
@dataclass(slots=True)
class PypiPackageInfo:
"""The subset of PyPI metadata we care about for a specific version."""
@@ -88,9 +77,6 @@ class PypiPackageInfo:
repo_url: str | None
file_provenance_urls: list[str] # may be empty
found: bool # False if the version doesn't exist on PyPI
yanked: bool = False
yanked_reason: str | None = None
vulnerabilities: list[Vulnerability] = field(default_factory=list)
@dataclass(slots=True)
@@ -171,43 +157,9 @@ def fetch_package_info(name: str, version: str) -> PypiPackageInfo:
repo_url=_pick_repo_url(project_urls),
file_provenance_urls=provenance_urls,
found=True,
yanked=bool(info.get("yanked")),
yanked_reason=_safe(info.get("yanked_reason")),
vulnerabilities=_parse_vulnerabilities(versioned.get("vulnerabilities")),
)
def _parse_vulnerabilities(raw: Any) -> list[Vulnerability]:
"""Extract non-withdrawn advisories from the PyPI `vulnerabilities` field."""
if not isinstance(raw, list):
return []
out: list[Vulnerability] = []
for entry in raw:
if not isinstance(entry, dict):
continue
if entry.get("withdrawn"):
# Withdrawn means the advisory was removed by the maintainer
# and should not be treated as valid.
continue
vid = _safe(entry.get("id"))
if not vid:
continue
aliases_raw = entry.get("aliases") or []
aliases = tuple(a for a in (_safe(str(x)) for x in aliases_raw if x) if a)
fixed_raw = entry.get("fixed_in") or []
fixed_in = tuple(f for f in (_safe(str(x)) for x in fixed_raw if x) if f)
out.append(
Vulnerability(
id=vid,
aliases=aliases,
summary=_safe(entry.get("summary")) or "",
fixed_in=fixed_in,
link=_safe(entry.get("link")) or "",
)
)
return out
def check_provenance(pkg: PypiPackageInfo) -> ProvenanceResult:
"""Resolve the provenance attestation, if any, to a Step 2b verdict."""
if not pkg.found:
@@ -224,7 +176,7 @@ def check_provenance(pkg: PypiPackageInfo) -> ProvenanceResult:
if not bundle:
continue
any_bundle_fetched = True
for entry in bundle.get("attestation_bundles") or []:
for entry in bundle.get("attestation_bundles", []) or []:
publisher = entry.get("publisher") or {}
kind = publisher.get("kind")
if not kind:
-2
View File
@@ -16,8 +16,6 @@ HEADER = "## Check requirements"
# Column / bullet labels per check kind, in display order.
_CHECK_DISPLAY: tuple[tuple[CheckKind, str], ...] = (
(CheckKind.VULNERABILITIES, "No Advisories"),
(CheckKind.YANKED, "Not Yanked"),
(CheckKind.REPO_PUBLIC, "Repo Public"),
(CheckKind.CI_UPLOAD, "CI Upload"),
(CheckKind.RELEASE_PIPELINE, "Release Pipeline"),
-59
View File
@@ -1,9 +1,6 @@
"""Orchestrate the deterministic requirements checks for one PR.
What the runner resolves itself (deterministic):
- `yanked`: PASS if the new release is live on PyPI, FAIL if it was yanked.
- `vulnerabilities`: FAIL if PyPI reports any non-withdrawn OSV / GHSA / CVE
advisory for the new version; PASS otherwise.
- `ci_upload`: PASS / WARN / FAIL based on PEP 740 attestation on PyPI.
- `release_pipeline`: PASS only when the attestation already identifies a
recognised CI publisher; otherwise NEEDS_AGENT.
@@ -24,60 +21,6 @@ from .pypi import PypiPackageInfo, check_provenance, fetch_package_info
from .render import render_comment
def _resolve_yanked(pkg: PackageChange, pypi_info: PypiPackageInfo) -> None:
"""Mark the release as yanked / not yanked."""
if not pypi_info.found:
pkg.checks[CheckKind.YANKED] = CheckResult(
CheckStatus.FAIL,
f"Version {pkg.new_version} not found on PyPI.",
)
return
if pypi_info.yanked:
reason = pypi_info.yanked_reason or "no reason provided by uploader"
pkg.checks[CheckKind.YANKED] = CheckResult(
CheckStatus.FAIL,
f"Version {pkg.new_version} is yanked on PyPI ({reason}). "
"Home Assistant should not depend on a yanked release.",
)
return
pkg.checks[CheckKind.YANKED] = CheckResult(
CheckStatus.PASS,
f"Version {pkg.new_version} is a live (non-yanked) release.",
)
def _resolve_vulnerabilities(pkg: PackageChange, pypi_info: PypiPackageInfo) -> None:
"""Flag versions with active OSV / GHSA / CVE advisories on PyPI."""
if not pypi_info.found:
pkg.checks[CheckKind.VULNERABILITIES] = CheckResult(
CheckStatus.FAIL,
f"Version {pkg.new_version} not found on PyPI.",
)
return
vulns = pypi_info.vulnerabilities
if not vulns:
pkg.checks[CheckKind.VULNERABILITIES] = CheckResult(
CheckStatus.PASS,
f"No active advisories reported by PyPI for version {pkg.new_version}.",
)
return
entries: list[str] = []
for vuln in vulns:
# Prefer a CVE alias as the primary label when present.
cve = next((a for a in vuln.aliases if a.upper().startswith("CVE-")), None)
label = cve or vuln.id
fixed = ", ".join(vuln.fixed_in) if vuln.fixed_in else "no fix listed"
if vuln.link:
entries.append(f"[{label}]({vuln.link}) (fixed in: {fixed})")
else:
entries.append(f"{label} (fixed in: {fixed})")
pkg.checks[CheckKind.VULNERABILITIES] = CheckResult(
CheckStatus.FAIL,
f"PyPI reports {len(vulns)} active advisories for version "
f"{pkg.new_version}: " + "; ".join(entries) + ".",
)
def _resolve_ci_upload_and_release_pipeline(
pkg: PackageChange, pypi_info: PypiPackageInfo
) -> None:
@@ -126,8 +69,6 @@ def run_checks(
for pkg in packages:
pypi_info = fetch_package_info(pkg.name, pkg.new_version)
pkg.repo_url = pypi_info.repo_url
_resolve_yanked(pkg, pypi_info)
_resolve_vulnerabilities(pkg, pypi_info)
_resolve_ci_upload_and_release_pipeline(pkg, pypi_info)
if not pypi_info.found:
fail = CheckResult(
@@ -3312,7 +3312,7 @@ async def test_handle_intents_filters_results(
)
assert len(mock_recognize.mock_calls) == 2
assert len(mock_process.mock_calls) == 2
assert len(mock_process.mock_calls) == 1
# Check we filtered things
assert len(results) == 2
-52
View File
@@ -1,52 +0,0 @@
"""Tests for the electrolux integration."""
from functools import cache
from electrolux_group_developer_sdk.client.dto.appliance import Appliance
from electrolux_group_developer_sdk.client.dto.appliance_details import ApplianceDetails
from electrolux_group_developer_sdk.client.dto.appliance_state import ApplianceState
from homeassistant.components.electrolux.const import DOMAIN
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry, load_fixture
APPLIANCE_FIXTURES = ["fenix_oven", "pux_oven"]
async def setup_integration(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Set up Electrolux integration for tests."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
@cache
def get_fixture_name(appliance_id: str) -> str:
"""Get the fixture name for the given appliance ID."""
for name in APPLIANCE_FIXTURES:
if load_appliance(name).applianceId == appliance_id:
return name
raise KeyError(f"Fixture name for appliance ID {appliance_id} does not exist")
def load_appliance(appliance_name: str) -> Appliance:
"""Load an Appliance object from a fixture for the given appliance name."""
json_string = load_fixture(f"appliances/{appliance_name}.json", DOMAIN)
return Appliance.model_validate_json(json_string)
def load_appliance_details(appliance_name: str) -> ApplianceDetails:
"""Load an ApplianceDetails object from a fixture for the given appliance name."""
json_string = load_fixture(f"appliance_details/{appliance_name}.json", DOMAIN)
return ApplianceDetails.model_validate_json(json_string)
def load_appliance_state(appliance_name: str) -> ApplianceState:
"""Load an ApplianceState object from a fixture for the given appliance name."""
json_string = load_fixture(f"appliance_states/{appliance_name}.json", DOMAIN)
return ApplianceState.model_validate_json(json_string)
-140
View File
@@ -1,140 +0,0 @@
"""Common fixtures for the electrolux tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
from electrolux_group_developer_sdk.client.appliance_data_factory import (
appliance_data_factory,
)
from electrolux_group_developer_sdk.client.dto.appliance_state import ApplianceState
from electrolux_group_developer_sdk.client.dto.email import Email
import pytest
from homeassistant.components.electrolux.const import CONF_REFRESH_TOKEN, DOMAIN
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_API_KEY
from homeassistant.core import HomeAssistant
from . import (
APPLIANCE_FIXTURES,
get_fixture_name,
load_appliance,
load_appliance_details,
load_appliance_state,
setup_integration,
)
from tests.common import MockConfigEntry
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.electrolux.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
async def init_integration(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Set up Electrolux integration for tests."""
await setup_integration(hass, mock_config_entry)
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Create mocked config entry."""
return MockConfigEntry(
domain=DOMAIN,
title="Electrolux",
unique_id="mock_user_id",
data={
CONF_API_KEY: "mock_api_key",
CONF_ACCESS_TOKEN: "mock_access_token",
CONF_REFRESH_TOKEN: "mock_refresh_token",
},
)
@pytest.fixture
def mock_appliance_client() -> Generator[AsyncMock]:
"""Mock the Electrolux Group Developer SDK client."""
with (
patch(
"homeassistant.components.electrolux.ApplianceClient",
autospec=True,
) as mock_client,
patch(
"homeassistant.components.electrolux.config_flow.ApplianceClient",
new=mock_client,
),
):
client = mock_client.return_value
def get_appliance_state(appliance_id: str) -> ApplianceState | None:
return load_appliance_state(get_fixture_name(appliance_id))
client.get_appliance_state.side_effect = get_appliance_state
client.get_user_email.return_value = Email(email="mock@email.com")
yield client
@pytest.fixture
def mock_token_manager() -> Generator[AsyncMock]:
"""Mock the Electrolux Group Developer SDK token manager."""
with (
patch(
"homeassistant.components.electrolux.TokenManager",
autospec=True,
) as mock_token_manager,
patch(
"homeassistant.components.electrolux.config_flow.TokenManager",
new=mock_token_manager,
),
):
token_manager = mock_token_manager.return_value
token_manager.ensure_credentials.return_value = None
token_manager.get_user_id.return_value = "mock_user_id"
yield token_manager
@pytest.fixture
def appliance_fixture() -> str | None:
"""Return the appliance fixture that should be loaded, or None if all appliances should be loaded."""
return None
@pytest.fixture
def appliances(
mock_appliance_client: AsyncMock, appliance_fixture: str | None
) -> AsyncMock:
"""Mock the list of appliances."""
appliance_names = []
if appliance_fixture is not None:
appliance_names.append(appliance_fixture)
else:
appliance_names.extend(APPLIANCE_FIXTURES)
appliance_data_list = []
for appliance_name in appliance_names:
appliance = load_appliance(appliance_name)
details = load_appliance_details(appliance_name)
state = load_appliance_state(appliance_name)
appliance_data = appliance_data_factory(
appliance=appliance,
details=details,
state=state,
)
appliance_data_list.append(appliance_data)
mock_appliance_client.get_appliance_data.return_value = appliance_data_list
return mock_appliance_client
File diff suppressed because it is too large Load Diff
@@ -1,874 +0,0 @@
{
"applianceInfo": {
"serialNumber": "11112225",
"pnc": "949288049",
"brand": "AEG",
"deviceType": "BUILT-IN OVEN",
"model": "NBX7P631SB",
"variant": "PIZZA-HOTAIRFAN+RING",
"colour": "BLACK"
},
"capabilities": {
"alerts": {
"access": "read",
"type": "alert",
"values": {
"AD_CONVERTER_REFERENCE_ALARM": {},
"BACKLIGHT_ALARM": {},
"BOARD_TEMPERATURE_ALARM": {},
"COMMUNICATION_ALARM_BETWEEN_TWO_CONTROLLERS": {},
"COMMUNICATION_ALARM_OUI_OC": {},
"CONFIGURATION_CHECKSUM_ALARM": {},
"CONFIGURATION_COHERENT_ALARM": {},
"CONFIGURATION_COMPATIBILITY_ALARM": {},
"COOKING_FAN_CONFIG_ALARM": {},
"DATA_FLASH_ALARM_UI": {},
"DOOR_LOCK_ACTUATOR": {},
"DOOR_LOCK_CONFIGURATION_ALARM": {},
"DOOR_LOCK_SENSOR_ALARM": {},
"ELECTRONIC_CLIXON_ALARM": {},
"FIX_SENSOR_DETECTION_ALARM": {},
"FOOD_PROBE_COMMUNICATION_ALARM": {},
"FOOD_PROBE_CONFIGURATION_ALARM": {},
"FUNCTION_SELECTOR_NOT_CONNECTED": {},
"HMI-TOUCH_BOARD_FMEA_ALARM": {},
"HMI-TOUCH_BOARD_SERIAL_COMMUNICATION_ALARM": {},
"HOB_OVEN_COMMUNICATION_ALARM": {},
"HOB_OVEN_POWER_MANAGEMENT_ALARM": {},
"HUMIDITY_SENSOR_OUT_OF_RANGE_ALARM": {},
"INTERNAL_ERROR": {},
"LIB_FMEA_ALARM_ROTARY_GRAB": {},
"MACS_COMMNUNICATION_ERROR": {},
"MEAT_PROBE_OUT_OF_RANGE_ALARM": {},
"NETVM_COMMUNICATION_ALARM": {},
"NIUX_COMMUNICATION_ALARM": {},
"NIUX_ONBOARDING_FAILED_ALARM": {},
"NTC_OUT_OF_RANGE_ALARM": {},
"OTA_FAILURE": {},
"PERIPHERAL_INIT_ALARM_ROTARY_SERIAL_COMMUNICATION_ALARM": {},
"PERIPHERAL_RUNTIME_ALARM_ROTARY_BIT_ENCODER_ALARM": {},
"POWER_ALARM": {},
"PT500_OUT_OF_RANGE_ALARM": {},
"PT500_STEAM_OUT_OF_RANGE_ALARM": {},
"PTO_COMMUNICATION_ALARM_OUI_PTO": {},
"PTO_CONFIGURATION_CHECKSUM_ALARM": {},
"PTO_KEY_ALARM": {},
"PTO_LIB_FMEA_ALARM": {},
"PYR_HOB_ALARM": {},
"ROTARY_TOUCH_ALARM": {},
"RTC_ALARM": {},
"RTC_OUT_OF_RANGE_ALARM": {},
"SMART_AD_CALIBRATION_RUNNING_ERROR": {},
"SMART_ERROR_UNKNOWN": {},
"SMART_INVALID_CONFIGURATION_ERROR": {},
"SMART_NO_START_TEMPERATURE": {},
"SMART_READ_FLASH_ERROR": {},
"SOFTWARE_COMPATIBILITY_CODE_ALARM": {},
"STEAM_MAGNETRON_NTC": {},
"TOO_HIGH_TEMPERATURE_ALARM": {},
"TOUCHSCREEN_DRIVER_ALARM": {},
"TOUCH_KEY_1_ALARM": {},
"TOUCH_KEY_ALARM": {},
"TRIAC_ALARM": {},
"UNKNOWN_STATE_ERROR": {},
"USB_CAMERA_DISCONNECTED_ALARM": {},
"WATER_LEVEL_SENSOR_IN_STEAMER_OUT_OF_RANGE_ALARM": {},
"WATER_LEVEL_SENSOR_IN_STEAM_TANK_DRAWER_OUT_OF_RANGE_ALARM": {},
"WIFI_SIGNAL_MISS_ALARM": {}
}
},
"applianceState": {
"access": "read",
"triggers": [
{
"action": {
"executeCommand": {
"access": "write",
"values": {
"STOPRESET": {}
}
}
},
"condition": {
"operand_1": "value",
"operand_2": "RUNNING",
"operator": "eq"
}
},
{
"action": {
"executeCommand": {
"access": "write",
"values": {
"STOPRESET": {}
}
}
},
"condition": {
"operand_1": "value",
"operand_2": "PAUSED",
"operator": "eq"
}
},
{
"action": {
"executeCommand": {
"access": "write",
"values": {
"START": {}
}
}
},
"condition": {
"operand_1": "value",
"operand_2": "READY_TO_START",
"operator": "eq"
}
},
{
"action": {
"$self": {
"access": "read"
}
},
"condition": {
"operand_1": "value",
"operand_2": "ALARM",
"operator": "eq"
}
},
{
"action": {
"executeCommand": {
"access": "write",
"values": {
"STOPRESET": {}
}
}
},
"condition": {
"operand_1": "value",
"operand_2": "DELAYED_START",
"operator": "eq"
}
},
{
"action": {
"executeCommand": {
"access": "write",
"values": {
"START": {}
}
}
},
"condition": {
"operand_1": "value",
"operand_2": "END_OF_CYCLE",
"operator": "eq"
}
}
],
"type": "string",
"values": {
"ALARM": {},
"DELAYED_START": {},
"END_OF_CYCLE": {},
"IDLE": {},
"OFF": {},
"PAUSED": {},
"READY_TO_START": {},
"RUNNING": {}
}
},
"cavityLight": {
"access": "readwrite",
"type": "boolean",
"values": {
"OFF": {},
"ON": {}
}
},
"displayTemperatureC": {
"access": "read",
"type": "temperature"
},
"doorState": {
"access": "read",
"type": "string",
"values": {
"CLOSED": {},
"OPEN": {}
}
},
"executeCommand": {
"access": "write",
"type": "string",
"values": {
"START": {},
"STOPRESET": {}
}
},
"hideExecuteCommand": {
"access": "constant",
"default": 0,
"triggers": [
{
"action": {
"executeCommand": {
"disabled": false
}
},
"condition": {
"operand_1": "value",
"operand_2": 0,
"operator": "eq"
}
},
{
"action": {
"executeCommand": {
"disabled": true
}
},
"condition": {
"operand_1": "value",
"operand_2": 1,
"operator": "eq"
}
}
],
"type": "int"
},
"keyModel": {
"access": "constant",
"default": "PUX_SP_PYR_PE_A++",
"type": "string"
},
"networkInterface": {
"command": {
"access": "write",
"type": "string",
"values": {
"ABORT": {},
"APPLIANCE_AUTHORIZE": {},
"DOWNLOAD": {},
"START": {},
"USER_AUTHORIZE": {},
"USER_NOT_AUTHORIZE": {}
}
},
"linkQualityIndicator": {
"access": "read",
"type": "string",
"values": {
"EXCELLENT": {},
"GOOD": {},
"POOR": {},
"UNDEFINED": {},
"VERY_GOOD": {},
"VERY_POOR": {}
}
},
"niuSwUpdateCurrentDescription": {
"access": "read",
"type": "string"
},
"otaState": {
"access": "read",
"type": "string",
"values": {
"DESCRIPTION_AVAILABLE": {},
"DESCRIPTION_DOWNLOADING": {},
"DESCRIPTION_READY": {},
"FW_DOWNLOADING": {},
"FW_DOWNLOAD_START": {},
"FW_SIGNATURE_CHECK": {},
"FW_UPDATE_IN_PROGRESS": {},
"IDLE": {},
"READY_TO_UPDATE": {},
"UPDATE_ABORT": {},
"UPDATE_CHECK": {},
"UPDATE_ERROR": {},
"UPDATE_OK": {},
"WAITINGFORAUTHORIZATION": {}
}
},
"startUpCommand": {
"access": "write",
"type": "string",
"values": {
"UNINSTALL": {}
}
},
"swAncAndRevision": {
"access": "read",
"type": "string"
},
"swVersion": {
"access": "read",
"type": "string"
}
},
"pizzaShieldState": {
"access": "read",
"type": "string",
"values": {
"INSERTED_COLD": {},
"INSERTED_HOT": {},
"NOT_INSERTED": {},
"UNKNOWN": {}
}
},
"preheatComplete": {
"access": "read",
"type": "string",
"values": {
"OFF": {},
"PRE_HEAT_COMPLETED": {},
"PRE_HEAT_RUNNING": {},
"RE_HEAT_COMPLETED": {},
"RE_HEAT_RUNNING": {}
}
},
"program": {
"access": "readwrite",
"type": "string",
"values": {
"ASSIST_PIZZA_EXPERT_REHEAT": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 180,
"min": 150,
"step": 1
},
"targetTemperatureC": {
"access": "readwrite",
"default": 340,
"disabled": false,
"max": 340,
"min": 320,
"step": 5,
"type": "temperature"
}
},
"AUGRATIN": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 86340,
"min": 0,
"step": 60
},
"targetTemperatureC": {
"access": "readwrite",
"default": 300,
"disabled": false,
"max": 300,
"min": 80,
"step": 5,
"type": "temperature"
}
},
"BAKE_TRUE_FAN": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 86340,
"min": 0,
"step": 60
},
"targetTemperatureC": {
"access": "readwrite",
"default": 200,
"disabled": false,
"max": 300,
"min": 80,
"step": 5,
"type": "temperature"
}
},
"BOTTOM": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 86340,
"min": 0,
"step": 60
},
"targetTemperatureC": {
"access": "readwrite",
"default": 150,
"disabled": false,
"max": 300,
"min": 80,
"step": 5,
"type": "temperature"
}
},
"BREAD_BAKING": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 86340,
"min": 0,
"step": 60
},
"targetTemperatureC": {
"access": "readwrite",
"default": 220,
"disabled": false,
"max": 300,
"min": 80,
"step": 5,
"type": "temperature"
}
},
"CONVENTIONAL_COOKING": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 86340,
"min": 0,
"step": 60
},
"targetTemperatureC": {
"access": "readwrite",
"default": 200,
"disabled": false,
"max": 300,
"min": 30,
"step": 5,
"type": "temperature"
}
},
"DEFROST": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 86340,
"min": 0,
"step": 60
},
"targetTemperatureC": {
"access": "readwrite",
"default": 30,
"disabled": false,
"max": 30,
"min": 30,
"step": 0,
"type": "temperature"
}
},
"DOUGH_PROVING": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 86340,
"min": 300,
"step": 60
},
"targetTemperatureC": {
"access": "readwrite",
"default": 40,
"disabled": false,
"max": 40,
"min": 40,
"step": 0,
"type": "temperature"
}
},
"DRYING": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 86340,
"min": 0,
"step": 60
},
"targetTemperatureC": {
"access": "readwrite",
"default": 60,
"disabled": false,
"max": 100,
"min": 50,
"step": 5,
"type": "temperature"
}
},
"FROZEN_FOOD": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 86340,
"min": 0,
"step": 60
},
"targetTemperatureC": {
"access": "readwrite",
"default": 220,
"disabled": false,
"max": 300,
"min": 80,
"step": 5,
"type": "temperature"
}
},
"GRILL": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 86340,
"min": 0,
"step": 60
},
"targetTemperatureC": {
"access": "readwrite",
"default": 300,
"disabled": false,
"max": 300,
"min": 80,
"step": 5,
"type": "temperature"
}
},
"GRILL_FAN": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 86340,
"min": 0,
"step": 60
},
"targetTemperatureC": {
"access": "readwrite",
"default": 180,
"disabled": false,
"max": 300,
"min": 80,
"step": 5,
"type": "temperature"
}
},
"KEEP_WARM": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 86340,
"min": 0,
"step": 60
},
"targetTemperatureC": {
"access": "readwrite",
"default": 80,
"disabled": false,
"max": 80,
"min": 80,
"step": 0,
"type": "temperature"
}
},
"MOIST_FAN_BAKING": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 86340,
"min": 0,
"step": 60
},
"targetTemperatureC": {
"access": "readwrite",
"default": 160,
"disabled": false,
"max": 230,
"min": 90,
"step": 5,
"type": "temperature"
}
},
"PIZZA_EXPERT_EXTENSION": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 15,
"min": 15,
"step": 0
},
"targetTemperatureC": {
"access": "readwrite",
"default": 340,
"disabled": false,
"max": 340,
"min": 320,
"step": 5,
"type": "temperature"
}
},
"PIZZA_EXPERT_MAIN": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 180,
"min": 150,
"step": 1
},
"targetTemperatureC": {
"access": "readwrite",
"default": 340,
"disabled": false,
"max": 340,
"min": 320,
"step": 5,
"type": "temperature"
}
},
"PIZZA_EXPERT_PREHEAT": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 86340,
"min": 60,
"step": 60
},
"targetTemperatureC": {
"access": "readwrite",
"default": 340,
"disabled": false,
"max": 340,
"min": 320,
"step": 5,
"type": "temperature"
}
},
"PLATE_WARMING": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 86340,
"min": 0,
"step": 60
},
"targetTemperatureC": {
"access": "readwrite",
"default": 70,
"disabled": false,
"max": 70,
"min": 70,
"step": 0,
"type": "temperature"
}
},
"PRESERVING": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 86340,
"min": 0,
"step": 60
},
"targetTemperatureC": {
"access": "readwrite",
"default": 160,
"disabled": false,
"max": 170,
"min": 100,
"step": 5,
"type": "temperature"
}
},
"PYRO_CLEAN_INTENSE": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 1,
"disabled": true
}
},
"PYRO_CLEAN_LIGHT": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 1,
"disabled": true
}
},
"PYRO_CLEAN_NORMAL": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 1,
"disabled": true
}
},
"SLOW_COOK": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 86340,
"min": 0,
"step": 60
},
"targetTemperatureC": {
"access": "readwrite",
"default": 90,
"disabled": false,
"max": 150,
"min": 80,
"step": 5,
"type": "temperature"
}
},
"TRUE_FAN": {
"hideExecuteCommand": {
"access": "readwrite",
"default": 0,
"disabled": true
},
"targetDuration": {
"access": "readwrite",
"disabled": false,
"max": 86340,
"min": 0,
"step": 60
},
"targetTemperatureC": {
"access": "readwrite",
"default": 150,
"disabled": false,
"max": 300,
"min": 30,
"step": 5,
"type": "temperature"
}
}
}
},
"remoteControl": {
"access": "read",
"type": "string",
"values": {
"DISABLED": {},
"ENABLED": {},
"NOT_SAFETY_RELEVANT_ENABLED": {},
"TEMPORARY_LOCKED": {}
}
},
"runningTime": {
"access": "read",
"default": 0,
"type": "number"
},
"targetDuration": {
"access": "readwrite",
"max": 86340,
"min": 0,
"step": 60,
"type": "number"
},
"targetTemperatureC": {
"access": "readwrite",
"type": "temperature"
},
"timeToEnd": {
"access": "read",
"type": "number"
}
}
}
@@ -1,48 +0,0 @@
{
"applianceId": "900412569_00:43319382-443E0748CCD4",
"connectionState": "connected",
"status": "enabled",
"properties": {
"reported": {
"cleaningReminder": false,
"doorState": "CLOSED",
"remoteControl": "ENABLED",
"targetTemperatureF": 356,
"targetTemperatureC": 180,
"program": "KEY_ERROR",
"targetMicrowavePower": 65535,
"waterTrayInsertionState": "INSERTED",
"waterTankEmpty": "STEAM_TANK_FULL",
"targetDuration": 0,
"startTime": -1,
"applianceInfo": {
"capabilityHash": "ab7b74c6ac8a74614980f77811389dfd47206456617a4d698ffcb97da1cc955b",
"applianceType": "OV"
},
"preheatComplete": "OFF",
"cpv": "00",
"targetFoodProbeTemperatureC": 60,
"targetFoodProbeTemperatureF": 140,
"runningTime": 0,
"applianceState": "READY_TO_START",
"alerts": [],
"networkInterface": {
"swVersion": "v3.0.0S_argo",
"otaState": "IDLE",
"linkQualityIndicator": "VERY_GOOD",
"niuSwUpdateCurrentDescription": "A23642201A-S00007645A",
"swAncAndRevision": "S00007645A"
},
"foodProbeInsertionState": "INSERTED",
"cavityLight": false,
"processPhase": "NONE",
"descalingReminderState": false,
"connectivityState": "connected",
"timeToEnd": -1,
"displayTemperatureC": 30,
"displayFoodProbeTemperatureC": 23,
"displayTemperatureF": 86,
"displayFoodProbeTemperatureF": 73.4
}
}
}
@@ -1,45 +0,0 @@
{
"applianceId": "949288049_00:11112225-443E076A37D6",
"connectionState": "connected",
"status": "enabled",
"properties": {
"reported": {
"applianceInfo": {
"capabilityHash": "212e55b62f19cfec40ff724301b07f80adef9a0ccf072693d45e38049708ef03",
"applianceType": "OV"
},
"doorState": "CLOSED",
"preheatComplete": "OFF",
"remoteControl": "NOT_SAFETY_RELEVANT_ENABLED",
"cpv": "00",
"targetTemperatureF": 302,
"targetTemperatureC": 150,
"runningTime": 0,
"program": "TRUE_FAN",
"applianceState": "READY_TO_START",
"targetMicrowavePower": 65535,
"alerts": [],
"pizzaShieldState": "UNKNOWN",
"networkInterface": {
"autoLocalTimeOffset": -2147483648,
"niuSwUpdateCurrentDescription": "A23642205A-S00008458A",
"swVersion": "v4.0.0S_argo",
"otaState": "IDLE",
"timeZoneDatabaseName": "Etc/UTC",
"swAncAndRevision": "S00008458A",
"linkQualityIndicator": "VERY_GOOD"
},
"foodProbeInsertionState": "NOT_INSERTED",
"cavityLight": false,
"waterTrayInsertionState": "INSERTED",
"waterTankEmpty": "STEAM_TANK_FULL",
"targetDuration": 0,
"startTime": -1,
"processPhase": "NONE",
"connectivityState": "connected",
"timeToEnd": -1,
"displayTemperatureC": 30,
"displayTemperatureF": 86
}
}
}
@@ -1,6 +0,0 @@
{
"applianceId": "900412569_00:43319382-443E0748CCD4",
"applianceName": "Fenix",
"applianceType": "OV",
"created": "2024-10-30T14:00:00.000+00:00"
}
@@ -1,6 +0,0 @@
{
"applianceId": "949288049_00:11112225-443E076A37D6",
"applianceName": "PUX pizza oven",
"applianceType": "OV",
"created": "2026-01-15T14:59:00.000+00:00"
}
@@ -1,63 +0,0 @@
# serializer version: 1
# name: test_all_appliances[fenix_oven]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'config_entries_subentries': <ANY>,
'configuration_url': None,
'connections': set({
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'electrolux',
'900412569_00:43319382-443E0748CCD4',
),
}),
'labels': set({
}),
'manufacturer': 'ELECTROLUX',
'model': 'OE9XS',
'model_id': None,
'name': 'Fenix',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': '43319382',
'sw_version': None,
'via_device_id': None,
})
# ---
# name: test_all_appliances[pux_oven]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'config_entries_subentries': <ANY>,
'configuration_url': None,
'connections': set({
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'electrolux',
'949288049_00:11112225-443E076A37D6',
),
}),
'labels': set({
}),
'manufacturer': 'AEG',
'model': 'NBX7P631SB',
'model_id': None,
'name': 'PUX pizza oven',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': '11112225',
'sw_version': None,
'via_device_id': None,
})
# ---
@@ -1,515 +0,0 @@
# serializer version: 1
# name: test_sensor[sensor.fenix_appliance_state-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'options': list([
'alarm',
'delayed_start',
'end_of_cycle',
'idle',
'off',
'paused',
'ready_to_start',
'running',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.fenix_appliance_state',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Appliance state',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': 'mdi:information-outline',
'original_name': 'Appliance state',
'platform': 'electrolux',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'appliance_state',
'unique_id': '900412569_00:43319382-443E0748CCD4_appliance_state',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[sensor.fenix_appliance_state-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Fenix Appliance state',
'icon': 'mdi:information-outline',
'options': list([
'alarm',
'delayed_start',
'end_of_cycle',
'idle',
'off',
'paused',
'ready_to_start',
'running',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.fenix_appliance_state',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'ready_to_start',
})
# ---
# name: test_sensor[sensor.fenix_current_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': None,
'entity_id': 'sensor.fenix_current_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Current temperature',
'options': dict({
'sensor': dict({
'suggested_display_precision': 1,
}),
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': 'mdi:thermometer',
'original_name': 'Current temperature',
'platform': 'electrolux',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'display_temperature',
'unique_id': '900412569_00:43319382-443E0748CCD4_display_temperature',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_sensor[sensor.fenix_current_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'temperature',
'friendly_name': 'Fenix Current temperature',
'icon': 'mdi:thermometer',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.fenix_current_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '30',
})
# ---
# name: test_sensor[sensor.fenix_food_probe_state-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'options': list([
'inserted',
'not_inserted',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.fenix_food_probe_state',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Food probe state',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': 'mdi:thermometer-probe',
'original_name': 'Food probe state',
'platform': 'electrolux',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'food_probe_state',
'unique_id': '900412569_00:43319382-443E0748CCD4_food_probe_state',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[sensor.fenix_food_probe_state-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Fenix Food probe state',
'icon': 'mdi:thermometer-probe',
'options': list([
'inserted',
'not_inserted',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.fenix_food_probe_state',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'inserted',
})
# ---
# name: test_sensor[sensor.fenix_food_probe_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': None,
'entity_id': 'sensor.fenix_food_probe_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Food probe temperature',
'options': dict({
'sensor': dict({
'suggested_display_precision': 1,
}),
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': 'mdi:thermometer-probe',
'original_name': 'Food probe temperature',
'platform': 'electrolux',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'food_probe_temperature',
'unique_id': '900412569_00:43319382-443E0748CCD4_food_probe_temperature',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_sensor[sensor.fenix_food_probe_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'temperature',
'friendly_name': 'Fenix Food probe temperature',
'icon': 'mdi:thermometer-probe',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.fenix_food_probe_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '23',
})
# ---
# name: test_sensor[sensor.fenix_remote_control-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'options': list([
'disabled',
'enabled',
'not_safety_relevant_enabled',
'temporary_locked',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.fenix_remote_control',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Remote control',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': 'mdi:remote',
'original_name': 'Remote control',
'platform': 'electrolux',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'remote_control',
'unique_id': '900412569_00:43319382-443E0748CCD4_remote_control',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[sensor.fenix_remote_control-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Fenix Remote control',
'icon': 'mdi:remote',
'options': list([
'disabled',
'enabled',
'not_safety_relevant_enabled',
'temporary_locked',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.fenix_remote_control',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'enabled',
})
# ---
# name: test_sensor[sensor.pux_pizza_oven_appliance_state-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'options': list([
'alarm',
'delayed_start',
'end_of_cycle',
'idle',
'off',
'paused',
'ready_to_start',
'running',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.pux_pizza_oven_appliance_state',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Appliance state',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': 'mdi:information-outline',
'original_name': 'Appliance state',
'platform': 'electrolux',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'appliance_state',
'unique_id': '949288049_00:11112225-443E076A37D6_appliance_state',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[sensor.pux_pizza_oven_appliance_state-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'PUX pizza oven Appliance state',
'icon': 'mdi:information-outline',
'options': list([
'alarm',
'delayed_start',
'end_of_cycle',
'idle',
'off',
'paused',
'ready_to_start',
'running',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.pux_pizza_oven_appliance_state',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'ready_to_start',
})
# ---
# name: test_sensor[sensor.pux_pizza_oven_current_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': None,
'entity_id': 'sensor.pux_pizza_oven_current_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Current temperature',
'options': dict({
'sensor': dict({
'suggested_display_precision': 1,
}),
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': 'mdi:thermometer',
'original_name': 'Current temperature',
'platform': 'electrolux',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'display_temperature',
'unique_id': '949288049_00:11112225-443E076A37D6_display_temperature',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_sensor[sensor.pux_pizza_oven_current_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'temperature',
'friendly_name': 'PUX pizza oven Current temperature',
'icon': 'mdi:thermometer',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.pux_pizza_oven_current_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '30',
})
# ---
# name: test_sensor[sensor.pux_pizza_oven_remote_control-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'options': list([
'disabled',
'enabled',
'not_safety_relevant_enabled',
'temporary_locked',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.pux_pizza_oven_remote_control',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Remote control',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': 'mdi:remote',
'original_name': 'Remote control',
'platform': 'electrolux',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'remote_control',
'unique_id': '949288049_00:11112225-443E076A37D6_remote_control',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[sensor.pux_pizza_oven_remote_control-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'PUX pizza oven Remote control',
'icon': 'mdi:remote',
'options': list([
'disabled',
'enabled',
'not_safety_relevant_enabled',
'temporary_locked',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.pux_pizza_oven_remote_control',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'not_safety_relevant_enabled',
})
# ---
@@ -1,176 +0,0 @@
"""Unit test for Electrolux config flow."""
from unittest.mock import AsyncMock
from electrolux_group_developer_sdk.auth.invalid_credentials_exception import (
InvalidCredentialsException,
)
from electrolux_group_developer_sdk.client.bad_credentials_exception import (
BadCredentialsException,
)
from electrolux_group_developer_sdk.client.failed_connection_exception import (
FailedConnectionException,
)
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.electrolux.const import CONF_REFRESH_TOKEN, DOMAIN
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_API_KEY
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
valid_user_input = {
CONF_API_KEY: "test_api_key",
CONF_ACCESS_TOKEN: "test_access_token",
CONF_REFRESH_TOKEN: "test_refresh_token",
}
invalid_user_input = {
CONF_API_KEY: "api_key",
CONF_ACCESS_TOKEN: "invalid_token",
CONF_REFRESH_TOKEN: "invalid_token",
}
async def test_user_flow_success(
hass: HomeAssistant, appliances: AsyncMock, mock_token_manager: AsyncMock
) -> None:
"""Test a successful user config flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is data_entry_flow.FlowResultType.FORM
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=valid_user_input
)
assert result["type"] is data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == "Electrolux for mock@email.com"
assert result["data"] == valid_user_input
assert result["result"].unique_id == "mock_user_id"
async def test_user_flow_invalid_auth(
hass: HomeAssistant, appliances: AsyncMock, mock_token_manager: AsyncMock
) -> None:
"""Test an invalid auth config flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is data_entry_flow.FlowResultType.FORM
mock_token_manager.ensure_credentials.side_effect = InvalidCredentialsException
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=invalid_user_input,
)
assert result["type"] is data_entry_flow.FlowResultType.FORM
assert result["errors"] == {"base": "invalid_auth"}
mock_token_manager.ensure_credentials.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=valid_user_input
)
assert result["type"] is data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == "Electrolux for mock@email.com"
assert result["data"] == valid_user_input
assert result["result"].unique_id == "mock_user_id"
async def test_user_flow_bad_credentials(
hass: HomeAssistant, appliances: AsyncMock, mock_token_manager: AsyncMock
) -> None:
"""Test an invalid auth config flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is data_entry_flow.FlowResultType.FORM
appliances.test_connection.side_effect = BadCredentialsException()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=invalid_user_input,
)
assert result["type"] is data_entry_flow.FlowResultType.FORM
assert result["errors"] == {"base": "invalid_auth"}
appliances.test_connection.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=valid_user_input
)
assert result["type"] is data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == "Electrolux for mock@email.com"
assert result["data"] == valid_user_input
assert result["result"].unique_id == "mock_user_id"
async def test_user_flow_failed_connection(
hass: HomeAssistant, appliances: AsyncMock, mock_token_manager: AsyncMock
) -> None:
"""Test an invalid auth config flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is data_entry_flow.FlowResultType.FORM
appliances.test_connection.side_effect = FailedConnectionException()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=invalid_user_input,
)
assert result["type"] is data_entry_flow.FlowResultType.FORM
assert result["errors"] == {"base": "cannot_connect"}
appliances.test_connection.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=valid_user_input
)
assert result["type"] is data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == "Electrolux for mock@email.com"
assert result["data"] == valid_user_input
assert result["result"].unique_id == "mock_user_id"
async def test_user_flow_duplicate_entry(
hass: HomeAssistant,
appliances: AsyncMock,
mock_token_manager: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test an invalid auth config flow."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is data_entry_flow.FlowResultType.FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=valid_user_input,
)
assert result["type"] is data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "already_configured"
-189
View File
@@ -1,189 +0,0 @@
"""Unit test for Electrolux init flow."""
from unittest.mock import AsyncMock
from electrolux_group_developer_sdk.client.appliance_data_factory import (
appliance_data_factory,
)
from electrolux_group_developer_sdk.client.appliances.appliance_data import (
ApplianceData,
)
from electrolux_group_developer_sdk.client.bad_credentials_exception import (
BadCredentialsException,
)
from electrolux_group_developer_sdk.client.client_exception import (
ApplianceClientException,
)
from electrolux_group_developer_sdk.client.failed_connection_exception import (
FailedConnectionException,
)
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.electrolux import ElectroluxData
from homeassistant.components.electrolux.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from . import (
APPLIANCE_FIXTURES,
get_fixture_name,
load_appliance,
load_appliance_details,
load_appliance_state,
setup_integration,
)
from tests.common import MockConfigEntry
async def test_async_setup_entry_success(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
appliances: AsyncMock,
mock_token_manager: AsyncMock,
) -> None:
"""Test successful setup of the Electrolux integration."""
await setup_integration(hass, mock_config_entry)
# Check integration is loaded
assert mock_config_entry.state is ConfigEntryState.LOADED
assert mock_config_entry.runtime_data is not None
assert isinstance(mock_config_entry.runtime_data, ElectroluxData)
# Unload the config entry
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
def test_appliance_fixture_data() -> None:
"""Test that all appliance fixtures are configured correctly."""
appliance_id_set = set()
for appliance_fixture in APPLIANCE_FIXTURES:
appliance = load_appliance(appliance_fixture)
appliance_id = appliance.applianceId
assert appliance_id not in appliance_id_set, (
f"Duplicate appliance ID {appliance_id} detected in fixture {appliance_fixture}"
)
appliance_state = load_appliance_state(appliance_fixture)
assert appliance_id == appliance_state.applianceId, (
f"Appliance ID in state {appliance_state.applianceId} does not match appliance ID in appliance object {appliance_id}"
)
appliance_id_set.add(appliance_id)
async def test_all_appliances(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
appliances: AsyncMock,
mock_config_entry: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test all entities for all appliance fixtures."""
await setup_integration(hass, mock_config_entry)
appliance_list: list[ApplianceData] = await appliances.get_appliance_data()
for appliance in appliance_list:
appliance_id = appliance.appliance.applianceId
device = device_registry.async_get_device({(DOMAIN, appliance_id)})
assert device is not None
assert device == snapshot(name=get_fixture_name(appliance_id))
async def test_check_for_dynamic_devices(
hass: HomeAssistant,
mock_appliance_client: AsyncMock,
mock_config_entry: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test that the integration adds and removes devices correctly."""
old_appliance_fixture = "fenix_oven"
old_appliance_id = "900412569_00:43319382-443E0748CCD4"
new_appliance_fixture = "pux_oven"
new_appliance_id = "949288049_00:11112225-443E076A37D6"
def set_appliance_fixture_mock(appliance_fixture: str):
appliance = load_appliance(appliance_fixture)
details = load_appliance_details(appliance_fixture)
state = load_appliance_state(appliance_fixture)
appliance_data = appliance_data_factory(
appliance=appliance,
details=details,
state=state,
)
mock_appliance_client.get_appliance_data.return_value = [appliance_data]
set_appliance_fixture_mock(old_appliance_fixture)
await setup_integration(hass, mock_config_entry)
assert device_registry.async_get_device({(DOMAIN, old_appliance_id)}) is not None
assert device_registry.async_get_device({(DOMAIN, new_appliance_id)}) is None
set_appliance_fixture_mock(new_appliance_fixture)
event_stream_call_args = mock_appliance_client.start_event_stream.call_args.args
assert event_stream_call_args is not None and len(event_stream_call_args) > 0, (
"start_event_stream method called without any callbacks specified"
)
callback_list = event_stream_call_args[0]
for callback in callback_list:
await callback()
assert device_registry.async_get_device({(DOMAIN, old_appliance_id)}) is None
assert device_registry.async_get_device({(DOMAIN, new_appliance_id)}) is not None
@pytest.mark.parametrize(
("exception", "entry_state"),
[
(ApplianceClientException(), ConfigEntryState.SETUP_RETRY),
],
)
async def test_appliance_client_exception(
hass: HomeAssistant,
mock_appliance_client: AsyncMock,
mock_config_entry: MockConfigEntry,
exception: Exception,
entry_state: ConfigEntryState,
) -> None:
"""Test for handling errors that occur while getting the data of all appliances."""
mock_appliance_client.get_appliance_data.side_effect = exception
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is entry_state
@pytest.mark.parametrize(
("exception", "entry_state"),
[
(BadCredentialsException(), ConfigEntryState.SETUP_ERROR),
(FailedConnectionException(), ConfigEntryState.SETUP_RETRY),
],
)
async def test_appliance_client_test_connection_bad_credentials(
hass: HomeAssistant,
mock_appliance_client: AsyncMock,
mock_config_entry: MockConfigEntry,
exception: Exception,
entry_state: ConfigEntryState,
) -> None:
"""Test for handling errors that occur while testing the connection."""
mock_appliance_client.test_connection.side_effect = exception
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is entry_state
@@ -1,34 +0,0 @@
"""Sensor tests of Electrolux integration."""
from collections.abc import Generator
from unittest.mock import patch
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
@pytest.fixture(autouse=True)
def override_platforms() -> Generator[None]:
"""Override PLATFORMS."""
with patch("homeassistant.components.electrolux.PLATFORMS", [Platform.SENSOR]):
yield
@pytest.mark.usefixtures("appliances")
async def test_sensor(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test states of the sensor."""
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
+1 -3
View File
@@ -44,9 +44,7 @@ def get_and_check_entity_basics(
assert ha_state is not None
if device_model:
assert ha_state.attributes[ATTR_MODEL_TYPE] == device_model
assert ha_state.name == entity_name, (
f"Expected '{entity_name}', got '{ha_state.name}'"
)
assert ha_state.name == entity_name
hmip_device = mock_hap.hmip_device_by_entity_id.get(entity_id)
@@ -131,8 +131,8 @@ async def test_hmip_home_cloud_connection_sensor(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipCloudConnectionSensor."""
entity_id = "binary_sensor.home_cloud_connection"
entity_name = "Home Cloud connection"
entity_id = "binary_sensor.cloud_connection"
entity_name = "Home Cloud Connection"
device_model = None
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=["Cloud Connection"]
@@ -154,11 +154,11 @@ async def test_hmip_acceleration_sensor(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipAccelerationSensor."""
entity_id = "binary_sensor.garagentor_moving"
entity_name = "Garagentor Moving"
entity_id = "binary_sensor.garagentor"
entity_name = "Garagentor"
device_model = "HmIP-SAM"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=["Garagentor"]
test_devices=[entity_name]
)
ha_state, hmip_device = get_and_check_entity_basics(
@@ -193,11 +193,11 @@ async def test_hmip_tilt_vibration_sensor(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipTiltVibrationSensor."""
entity_id = "binary_sensor.garage_neigungs_und_erschutterungssensor_moving"
entity_name = "Garage Neigungs- und Erschütterungssensor Moving"
entity_id = "binary_sensor.garage_neigungs_und_erschutterungssensor"
entity_name = "Garage Neigungs- und Erschütterungssensor"
device_model = "HmIP-STV"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=["Garage Neigungs- und Erschütterungssensor"]
test_devices=[entity_name]
)
ha_state, hmip_device = get_and_check_entity_basics(
@@ -231,11 +231,11 @@ async def test_hmip_contact_interface(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipContactInterface."""
entity_id = "binary_sensor.kontakt_schnittstelle_unterputz_1_fach_opening"
entity_name = "Kontakt-Schnittstelle Unterputz 1-fach Opening"
entity_id = "binary_sensor.kontakt_schnittstelle_unterputz_1_fach"
entity_name = "Kontakt-Schnittstelle Unterputz 1-fach"
device_model = "HmIP-FCI1"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=["Kontakt-Schnittstelle Unterputz 1-fach"]
test_devices=[entity_name]
)
ha_state, hmip_device = get_and_check_entity_basics(
@@ -256,11 +256,11 @@ async def test_hmip_shutter_contact(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipShutterContact."""
entity_id = "binary_sensor.fenstergriffsensor_door"
entity_name = "Fenstergriffsensor Door"
entity_id = "binary_sensor.fenstergriffsensor"
entity_name = "Fenstergriffsensor"
device_model = "HmIP-SRH"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=["Fenstergriffsensor"]
test_devices=[entity_name]
)
ha_state, hmip_device = get_and_check_entity_basics(
@@ -298,11 +298,11 @@ async def test_hmip_shutter_contact_optical(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipShutterContact."""
entity_id = "binary_sensor.sitzplatzture_door"
entity_name = "Sitzplatzt\u00fcre Door"
entity_id = "binary_sensor.sitzplatzture"
entity_name = "Sitzplatzt\u00fcre"
device_model = "HmIP-SWDO-PL"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=["Sitzplatzt\u00fcre"]
test_devices=[entity_name]
)
ha_state, hmip_device = get_and_check_entity_basics(
@@ -330,11 +330,11 @@ async def test_hmip_motion_detector(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipMotionDetector."""
entity_id = "binary_sensor.bewegungsmelder_fur_55er_rahmen_innen_motion"
entity_name = "Bewegungsmelder für 55er Rahmen innen Motion"
entity_id = "binary_sensor.bewegungsmelder_fur_55er_rahmen_innen"
entity_name = "Bewegungsmelder für 55er Rahmen innen"
device_model = "HmIP-SMI55"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=["Bewegungsmelder für 55er Rahmen innen"]
test_devices=[entity_name]
)
ha_state, hmip_device = get_and_check_entity_basics(
@@ -351,10 +351,12 @@ async def test_hmip_presence_detector(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipPresenceDetector."""
entity_id = "binary_sensor.spi_1_presence"
entity_name = "SPI_1 Presence"
entity_id = "binary_sensor.spi_1"
entity_name = "SPI_1"
device_model = "HmIP-SPI"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(test_devices=["SPI_1"])
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=[entity_name]
)
ha_state, hmip_device = get_and_check_entity_basics(
hass, mock_hap, entity_id, entity_name, device_model
@@ -375,11 +377,11 @@ async def test_hmip_pluggable_mains_failure_surveillance_sensor(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipPresenceDetector."""
entity_id = "binary_sensor.netzausfalluberwachung_power"
entity_name = "Netzausfallüberwachung Power"
entity_id = "binary_sensor.netzausfalluberwachung"
entity_name = "Netzausfallüberwachung"
device_model = "HmIP-PMFS"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=["Netzausfallüberwachung"]
test_devices=[entity_name]
)
ha_state, hmip_device = get_and_check_entity_basics(
@@ -396,11 +398,11 @@ async def test_hmip_smoke_detector(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipSmokeDetector."""
entity_id = "binary_sensor.rauchwarnmelder_smoke"
entity_name = "Rauchwarnmelder Smoke"
entity_id = "binary_sensor.rauchwarnmelder"
entity_name = "Rauchwarnmelder"
device_model = "HmIP-SWSD"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=["Rauchwarnmelder"]
test_devices=[entity_name]
)
ha_state, hmip_device = get_and_check_entity_basics(
@@ -451,11 +453,11 @@ async def test_hmip_water_detector(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipWaterDetector."""
entity_id = "binary_sensor.wassersensor_moisture"
entity_name = "Wassersensor Moisture"
entity_id = "binary_sensor.wassersensor"
entity_name = "Wassersensor"
device_model = "HmIP-SWD"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=["Wassersensor"]
test_devices=[entity_name]
)
ha_state, hmip_device = get_and_check_entity_basics(
@@ -212,8 +212,8 @@ async def test_hap_with_name(
) -> None:
"""Test hap with name."""
home_name = "TestName"
entity_id = "light.testname_treppe_ch"
entity_name = "TestName Treppe CH"
entity_id = "light.treppe_testname_treppe_ch"
entity_name = "Treppe TestName Treppe CH"
device_model = "HmIP-BSL"
hmip_config_entry.add_to_hass(hass)
@@ -1,115 +0,0 @@
"""Unit tests for HomematicipGenericEntity naming helpers."""
from types import SimpleNamespace
import pytest
from homeassistant.components.homematicip_cloud.entity import HomematicipGenericEntity
def _make_entity(
*,
device_label: str,
channels: dict[int, str],
channel: int,
is_multi_channel: bool,
post: str | None = None,
) -> HomematicipGenericEntity:
"""Build a HomematicipGenericEntity bypassing __init__ for unit testing.
Only the attributes read by _setup_entity_name() are populated.
"""
entity = HomematicipGenericEntity.__new__(HomematicipGenericEntity)
entity._device = SimpleNamespace(
label=device_label,
functionalChannels={
idx: SimpleNamespace(label=label, index=idx)
for idx, label in channels.items()
},
)
entity._post = post
entity._channel = channel
entity._channel_real_index = channel
entity._is_multi_channel = is_multi_channel
entity.functional_channel = entity._device.functionalChannels.get(channel)
entity._attr_name = None
return entity
@pytest.mark.parametrize(
("device_label", "channel_label"),
[
("Thermostat EG Wohnzimmer", "Thermostat EG Wohnzimmer"),
("Thermostat EG Wohnzimmer", "Thermostat EG Wohnzimmer "),
("Thermostat EG Wohnzimmer ", "Thermostat EG Wohnzimmer "),
],
ids=["exact-match", "trailing-space-on-channel", "trailing-space-on-both"],
)
def test_multi_channel_label_equals_device_label_leaves_name_unset(
device_label: str, channel_label: str
) -> None:
"""When channel label equals device label, _attr_name must stay None.
Otherwise HA composes "{device_name} {entity_name}" and the user sees the
label duplicated, e.g. "Thermostat EG Wohnzimmer Thermostat EG Wohnzimmer".
"""
entity = _make_entity(
device_label=device_label,
channels={3: channel_label, 1: "primary"},
channel=3,
is_multi_channel=True,
)
entity._setup_entity_name()
assert entity._attr_name is None
def test_multi_channel_label_extends_device_label_keeps_suffix() -> None:
"""When channel label starts with device label + suffix, keep the suffix."""
entity = _make_entity(
device_label="Licht Flur",
channels={5: "Licht Flur 5"},
channel=5,
is_multi_channel=True,
)
entity._setup_entity_name()
assert entity._attr_name == "5"
def test_multi_channel_label_unrelated_to_device_label_uses_full_label() -> None:
"""When channel label does not start with device label, use it verbatim."""
entity = _make_entity(
device_label="DRS8-4",
channels={2: "Flur OG Lampe"},
channel=2,
is_multi_channel=True,
)
entity._setup_entity_name()
assert entity._attr_name == "Flur OG Lampe"
def test_primary_entity_with_channel_1_label_equals_device_label_unset() -> None:
"""Non-multi-channel entity on device whose ch1 label equals device label.
Same duplication risk as the multi-channel case; same fix.
"""
entity = _make_entity(
device_label="Thermostat EG Wohnzimmer",
channels={1: "Thermostat EG Wohnzimmer ", 2: "other"},
channel=2, # non-multi-channel entity sits on some channel
is_multi_channel=False,
)
entity._setup_entity_name()
assert entity._attr_name is None
def test_post_suffix_capitalised() -> None:
"""Post suffix becomes the entity name with first letter uppercased."""
entity = _make_entity(
device_label="Bewegungsmelder Küche",
channels={1: "primary"},
channel=1,
is_multi_channel=False,
post="battery",
)
entity._setup_entity_name()
assert entity._attr_name == "Battery"
@@ -807,7 +807,7 @@ async def test_hmip_water_valve_current_water_flow(
) -> None:
"""Test HomematicipCurrentWaterFlow."""
entity_id = "sensor.bewaesserungsaktor_currentwaterflow"
entity_name = "Bewaesserungsaktor CurrentWaterFlow"
entity_name = "Bewaesserungsaktor currentWaterFlow"
device_model = "ELV-SH-WSM"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=["Bewaesserungsaktor"]
@@ -830,7 +830,7 @@ async def test_hmip_water_valve_water_volume(
) -> None:
"""Test HomematicipWaterVolume."""
entity_id = "sensor.bewaesserungsaktor_watervolume"
entity_name = "Bewaesserungsaktor WaterVolume"
entity_name = "Bewaesserungsaktor waterVolume"
device_model = "ELV-SH-WSM"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=["Bewaesserungsaktor"]
@@ -850,7 +850,7 @@ async def test_hmip_water_valve_water_volume_since_open(
) -> None:
"""Test HomematicipWaterVolumeSinceOpen."""
entity_id = "sensor.bewaesserungsaktor_watervolumesinceopen"
entity_name = "Bewaesserungsaktor WaterVolumeSinceOpen"
entity_name = "Bewaesserungsaktor waterVolumeSinceOpen"
device_model = "ELV-SH-WSM"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=["Bewaesserungsaktor"]
@@ -870,7 +870,7 @@ async def test_hmip_smoke_detector_dirt_level(
) -> None:
"""Test HomematicipSmokeDetectorDirtLevel."""
entity_id = "sensor.rauchwarnmelder_dirt_level"
entity_name = "Rauchwarnmelder Dirt level"
entity_name = "Rauchwarnmelder Dirt_level"
device_model = "HmIP-SWSD"
# Pre-register the entity as enabled before platform loads
@@ -910,7 +910,7 @@ async def test_hmip_smoke_detector_alarm_counter(
) -> None:
"""Test HomematicipSmokeDetectorAlarmCounter."""
entity_id = "sensor.rauchwarnmelder_smoke_alarm_counter"
entity_name = "Rauchwarnmelder Alarm counter"
entity_name = "Rauchwarnmelder Smoke_alarm_counter"
device_model = "HmIP-SWSD"
# Pre-register the entity as enabled before platform loads
@@ -947,7 +947,7 @@ async def test_hmip_smoke_detector_test_counter(
) -> None:
"""Test HomematicipSmokeDetectorTestCounter."""
entity_id = "sensor.rauchwarnmelder_smoke_test_counter"
entity_name = "Rauchwarnmelder Test counter"
entity_name = "Rauchwarnmelder Smoke_test_counter"
device_model = "HmIP-SWSD"
# Pre-register the entity as enabled before platform loads
+10 -4
View File
@@ -220,13 +220,16 @@ async def test_routing_setup(
"homeassistant.components.knx.config_flow.GatewayScanner",
return_value=GatewayScannerMock(),
)
async def test_routing_setup_with_local_ip(
async def test_routing_setup_advanced(
gateway_scanner_mock, hass: HomeAssistant, knx_setup
) -> None:
"""Test routing setup where user specifies a local IP."""
"""Test routing setup with advanced options."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
context={
"source": config_entries.SOURCE_USER,
"show_advanced_options": True,
},
)
assert result["type"] is FlowResultType.FORM
assert not result["errors"]
@@ -725,7 +728,10 @@ async def test_tunneling_setup_for_local_ip(
"""Test tunneling if only one gateway is found."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
context={
"source": config_entries.SOURCE_USER,
"show_advanced_options": True,
},
)
assert result["type"] is FlowResultType.FORM
assert not result["errors"]
@@ -18,7 +18,7 @@
'domain': 'light',
'entity_category': None,
'entity_id': 'light.dali_line_0',
'has_entity_name': True,
'has_entity_name': False,
'hidden_by': None,
'icon': None,
'id': <ANY>,
@@ -79,7 +79,7 @@
'domain': 'light',
'entity_category': None,
'entity_id': 'light.dali_line_1',
'has_entity_name': True,
'has_entity_name': False,
'hidden_by': None,
'icon': None,
'id': <ANY>,
+2 -2
View File
@@ -1027,7 +1027,7 @@ async def test_block_sleeping_device_connection_error(
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert "Sleeping device Test name did not update" in caplog.text
assert "Sleeping device did not update" in caplog.text
assert (state := hass.states.get(entity_id))
assert state.state == STATE_UNAVAILABLE
@@ -1081,7 +1081,7 @@ async def test_rpc_sleeping_device_connection_error(
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert "Sleeping device Test name did not update" in caplog.text
assert "Sleeping device did not update" in caplog.text
assert (state := hass.states.get(entity_id))
assert state.state == STATE_UNAVAILABLE
@@ -14,7 +14,7 @@
'domain': 'scene',
'entity_category': None,
'entity_id': 'scene.away',
'has_entity_name': True,
'has_entity_name': False,
'hidden_by': None,
'icon': None,
'id': <ANY>,
@@ -67,7 +67,7 @@
'domain': 'scene',
'entity_category': None,
'entity_id': 'scene.home',
'has_entity_name': True,
'has_entity_name': False,
'hidden_by': None,
'icon': None,
'id': <ANY>,
-53
View File
@@ -1,7 +1,6 @@
"""Test config validators."""
from collections import OrderedDict
from collections.abc import Callable
from datetime import date, datetime, timedelta
import enum
from functools import partial
@@ -2032,55 +2031,3 @@ def test_stop_action_schema_error_false_with_response() -> None:
# no error with response_variable should work
config = schema({"stop": "Done", "response_variable": "result"})
assert config["response_variable"] == "result"
_COMMENT_SCHEMA_PARAMS = [
pytest.param(
cv.TRIGGER_BASE_SCHEMA,
{"platform": "event"},
id="trigger_base",
),
pytest.param(
cv.CONDITION_SCHEMA,
{"condition": "state", "entity_id": "sun.sun", "state": "above_horizon"},
id="condition",
),
pytest.param(
cv.script_action,
{"action": "test.foo"},
id="script_action",
),
]
@pytest.mark.parametrize(("validator", "base_config"), _COMMENT_SCHEMA_PARAMS)
@pytest.mark.usefixtures("hass")
def test_base_schemas_accept_comment(
validator: Callable[[dict[str, Any]], dict[str, Any]],
base_config: dict[str, Any],
) -> None:
"""Test that the comment field is accepted and stripped from the output."""
validated = validator({**base_config, "comment": "Single line"})
assert "comment" not in validated
@pytest.mark.parametrize(("validator", "base_config"), _COMMENT_SCHEMA_PARAMS)
@pytest.mark.parametrize(
"invalid_comment",
[
pytest.param(None, id="none"),
pytest.param(42, id="int"),
pytest.param(True, id="bool"),
pytest.param([], id="list"),
pytest.param({}, id="dict"),
],
)
@pytest.mark.usefixtures("hass")
def test_base_schemas_reject_invalid_comment(
validator: Callable[[dict[str, Any]], dict[str, Any]],
base_config: dict[str, Any],
invalid_comment: Any,
) -> None:
"""Test that script, condition, trigger base schemas reject non-string comments."""
with pytest.raises(vol.Invalid):
validator({**base_config, "comment": invalid_comment})
@@ -204,85 +204,6 @@ def test_fetch_package_info_picks_repo_url_from_project_urls(
assert info.repo_url == expected_repo_url
def test_fetch_package_info_extracts_yanked_fields(
requests_mock: rm.Mocker,
) -> None:
"""The fetcher lifts `yanked` and `yanked_reason` from PyPI."""
requests_mock.get(
_versioned_url("foo", "1.0"),
json={
"info": {
"project_urls": {},
"yanked": True,
"yanked_reason": "broken on 3.14",
},
"urls": [],
},
)
info = fetch_package_info("foo", "1.0")
assert info.found is True
assert info.yanked is True
assert info.yanked_reason == "broken on 3.14"
def test_fetch_package_info_extracts_vulnerabilities(
requests_mock: rm.Mocker,
) -> None:
"""Active OSV / GHSA / CVE advisories on PyPI are surfaced and parsed."""
requests_mock.get(
_versioned_url("foo", "1.0"),
json={
"info": {"project_urls": {}},
"urls": [],
"vulnerabilities": [
{
"id": "GHSA-aaaa-bbbb-cccc",
"aliases": ["CVE-2099-12345"],
"summary": "remote code execution",
"fixed_in": ["1.1", "1.2"],
"link": "https://osv.dev/vulnerability/GHSA-aaaa-bbbb-cccc",
"withdrawn": None,
},
{
"id": "GHSA-dddd-eeee-ffff",
"aliases": [],
"summary": "withdrawn advisory",
"fixed_in": [],
"link": "https://osv.dev/vulnerability/GHSA-dddd-eeee-ffff",
"withdrawn": "2024-01-01T00:00:00Z",
},
],
},
)
info = fetch_package_info("foo", "1.0")
# The withdrawn advisory is filtered out.
assert len(info.vulnerabilities) == 1
vuln = info.vulnerabilities[0]
assert vuln.id == "GHSA-aaaa-bbbb-cccc"
assert vuln.aliases == ("CVE-2099-12345",)
assert vuln.fixed_in == ("1.1", "1.2")
assert "remote code execution" in vuln.summary
def test_fetch_package_info_defaults_when_yanked_fields_absent(
requests_mock: rm.Mocker,
) -> None:
"""Missing `yanked` keys default to False / None."""
requests_mock.get(
_versioned_url("foo", "1.0"),
json={"info": {"project_urls": {}}, "urls": []},
)
info = fetch_package_info("foo", "1.0")
assert info.yanked is False
assert info.yanked_reason is None
def test_fetch_package_info_strips_dangerous_chars_from_repo_url(
requests_mock: rm.Mocker,
) -> None:
+1 -144
View File
@@ -5,11 +5,7 @@ import json
import pytest
from script.check_requirements.models import CheckKind, CheckStatus
from script.check_requirements.pypi import (
ProvenanceResult,
PypiPackageInfo,
Vulnerability,
)
from script.check_requirements.pypi import ProvenanceResult, PypiPackageInfo
from script.check_requirements.runner import run_checks
@@ -272,145 +268,6 @@ def test_runner_async_blocking_version_bump_diff_only(
assert "diff" in detail
def test_runner_yanked_release_fails(monkeypatch: pytest.MonkeyPatch) -> None:
"""A yanked release on PyPI must FAIL the yanked check."""
_patch_pypi(
monkeypatch,
PypiPackageInfo(
project_urls={"Source": "https://github.com/x/y"},
repo_url="https://github.com/x/y",
file_provenance_urls=["whatever"],
found=True,
yanked=True,
yanked_reason="critical bug",
),
ProvenanceResult(
has_attestation=True,
publisher_kind="GitHub",
recognized_publisher=True,
detail="ok",
),
)
diff = (
"diff --git a/requirements_all.txt b/requirements_all.txt\n"
"--- a/requirements_all.txt\n"
"+++ b/requirements_all.txt\n"
"@@ -1 +1 @@\n"
"-pkg==1.0.0\n"
"+pkg==1.1.0\n"
)
result = run_checks(pr_number=1, diff_text=diff)
pkg = result.packages[0]
assert pkg.checks[CheckKind.YANKED].status == CheckStatus.FAIL
assert "yanked" in pkg.checks[CheckKind.YANKED].details
assert "critical bug" in pkg.checks[CheckKind.YANKED].details
def test_runner_non_yanked_release_passes(monkeypatch: pytest.MonkeyPatch) -> None:
"""A normal (non-yanked) release passes the yanked check."""
_patch_pypi(
monkeypatch,
PypiPackageInfo(
project_urls={"Source": "https://github.com/x/y"},
repo_url="https://github.com/x/y",
file_provenance_urls=["whatever"],
found=True,
),
ProvenanceResult(
has_attestation=True,
publisher_kind="GitHub",
recognized_publisher=True,
detail="ok",
),
)
diff = (
"diff --git a/requirements_all.txt b/requirements_all.txt\n"
"--- a/requirements_all.txt\n"
"+++ b/requirements_all.txt\n"
"@@ -1 +1 @@\n"
"-pkg==1.0.0\n"
"+pkg==1.1.0\n"
)
result = run_checks(pr_number=1, diff_text=diff)
pkg = result.packages[0]
assert pkg.checks[CheckKind.YANKED].status == CheckStatus.PASS
def test_runner_active_vulnerabilities_fail(monkeypatch: pytest.MonkeyPatch) -> None:
"""An active CVE/GHSA on PyPI fails the vulnerabilities check with a link."""
_patch_pypi(
monkeypatch,
PypiPackageInfo(
project_urls={"Source": "https://github.com/x/y"},
repo_url="https://github.com/x/y",
file_provenance_urls=["whatever"],
found=True,
vulnerabilities=[
Vulnerability(
id="GHSA-xxxx-xxxx-xxxx",
aliases=("CVE-2099-12345",),
summary="rce",
fixed_in=("1.2.0",),
link="https://osv.dev/vulnerability/GHSA-xxxx-xxxx-xxxx",
),
],
),
ProvenanceResult(
has_attestation=True,
publisher_kind="GitHub",
recognized_publisher=True,
detail="ok",
),
)
diff = (
"diff --git a/requirements_all.txt b/requirements_all.txt\n"
"--- a/requirements_all.txt\n"
"+++ b/requirements_all.txt\n"
"@@ -1 +1 @@\n"
"-pkg==1.0.0\n"
"+pkg==1.1.0\n"
)
result = run_checks(pr_number=1, diff_text=diff)
pkg = result.packages[0]
assert pkg.checks[CheckKind.VULNERABILITIES].status == CheckStatus.FAIL
details = pkg.checks[CheckKind.VULNERABILITIES].details
# CVE alias is preferred as the link label when present.
assert "[CVE-2099-12345](" in details
assert "[GHSA-xxxx-xxxx-xxxx](" not in details
assert "fixed in: 1.2.0" in details
assert "https://osv.dev/vulnerability/GHSA-xxxx-xxxx-xxxx" in details
def test_runner_no_vulnerabilities_passes(monkeypatch: pytest.MonkeyPatch) -> None:
"""An empty `vulnerabilities` list passes the check."""
_patch_pypi(
monkeypatch,
PypiPackageInfo(
project_urls={"Source": "https://github.com/x/y"},
repo_url="https://github.com/x/y",
file_provenance_urls=["whatever"],
found=True,
),
ProvenanceResult(
has_attestation=True,
publisher_kind="GitHub",
recognized_publisher=True,
detail="ok",
),
)
diff = (
"diff --git a/requirements_all.txt b/requirements_all.txt\n"
"--- a/requirements_all.txt\n"
"+++ b/requirements_all.txt\n"
"@@ -1 +1 @@\n"
"-pkg==1.0.0\n"
"+pkg==1.1.0\n"
)
result = run_checks(pr_number=1, diff_text=diff)
pkg = result.packages[0]
assert pkg.checks[CheckKind.VULNERABILITIES].status == CheckStatus.PASS
def test_runner_serialises_to_json(monkeypatch: pytest.MonkeyPatch) -> None:
"""The artifact contract: `to_dict()` is JSON-serialisable with expected keys."""
_patch_pypi(
-31
View File
@@ -3,7 +3,6 @@
import asyncio
import dataclasses
import logging
from typing import Any
from unittest.mock import Mock, patch
import pytest
@@ -1273,33 +1272,3 @@ def test_nested_section_in_serializer() -> None:
{"collapsed": False},
)
)
@pytest.mark.parametrize(
("context", "expected_show_advanced"),
[
({}, False),
({"show_advanced_options": False}, False),
({"show_advanced_options": True}, True),
],
)
async def test_show_advanced_options(
manager: MockFlowManager, context: dict[str, Any], expected_show_advanced: bool
) -> None:
"""Test FlowHandler show_advanced_options property."""
@manager.mock_reg_handler("test")
class TestFlow(data_entry_flow.FlowHandler):
VERSION = 5
async def async_step_init(self, info):
assert self.show_advanced_options == expected_show_advanced
return self.async_create_entry(title="hello", data={})
await manager.async_init("test", context=context, data={})
assert len(manager.async_progress()) == 0
assert len(manager.mock_created_entries) == 1
entry = manager.mock_created_entries[0]
assert entry["handler"] == "test"
assert entry["title"] == "hello"