Compare commits

..

1 Commits

Author SHA1 Message Date
Bram Kragten
9767aa5e9b Add context support for triggers.yaml 2025-11-13 14:53:36 +01:00
33 changed files with 354 additions and 10475 deletions

2
CODEOWNERS generated
View File

@@ -516,8 +516,6 @@ build.json @home-assistant/supervisor
/tests/components/flo/ @dmulcahey
/homeassistant/components/flume/ @ChrisMandich @bdraco @jeeftor
/tests/components/flume/ @ChrisMandich @bdraco @jeeftor
/homeassistant/components/fluss/ @fluss
/tests/components/fluss/ @fluss
/homeassistant/components/flux_led/ @icemanch
/tests/components/flux_led/ @icemanch
/homeassistant/components/forecast_solar/ @klaasnicolaas @frenck

View File

@@ -1,84 +0,0 @@
rules:
# todo : add get_feed_list to the library
# todo : see if we can drop some extra attributes
# Bronze
action-setup:
status: exempt
comment: |
This integration does not provide additional actions.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage:
status: todo
comment: |
test_reconfigure_api_error should use a mock config entry fixture
test_user_flow_failure should use a mock config entry fixture
move test_user_flow_* to the top of the file
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: |
This integration does not provide additional actions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup:
status: exempt
comment: |
No events are explicitly registered by the integration.
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: done
config-entry-unloading: done
docs-configuration-parameters: done
docs-installation-parameters: done
entity-unavailable: todo
integration-owner: done
log-when-unavailable: done
parallel-updates: todo
reauthentication-flow: todo
test-coverage:
status: todo
comment: |
test the entry state in test_failure
# Gold
devices: todo
diagnostics: todo
discovery-update-info: todo
discovery: todo
docs-data-update: done
docs-examples:
status: exempt
comment: |
This integration does not provide any automation
docs-known-limitations: todo
docs-supported-devices: todo
docs-supported-functions: done
docs-troubleshooting: done
docs-use-cases: todo
dynamic-devices: todo
entity-category: todo
entity-device-class:
status: todo
comment: change device_class=SensorDeviceClass.SIGNAL_STRENGTH to SOUND_PRESSURE
entity-disabled-by-default: todo
entity-translations: done
exception-translations: todo
icon-translations: todo
reconfiguration-flow: done
repair-issues: todo
stale-devices: todo
# Platinum
async-dependency: done
inject-websession: done
strict-typing: todo

View File

@@ -147,8 +147,6 @@ async def async_get_config_entry_diagnostics(
"ctmeter_production_phases": envoy_data.ctmeter_production_phases,
"ctmeter_consumption_phases": envoy_data.ctmeter_consumption_phases,
"ctmeter_storage_phases": envoy_data.ctmeter_storage_phases,
"ctmeters": envoy_data.ctmeters,
"ctmeters_phases": envoy_data.ctmeters_phases,
"dry_contact_status": envoy_data.dry_contact_status,
"dry_contact_settings": envoy_data.dry_contact_settings,
"inverters": envoy_data.inverters,
@@ -181,7 +179,6 @@ async def async_get_config_entry_diagnostics(
"ct_consumption_meter": envoy.consumption_meter_type,
"ct_production_meter": envoy.production_meter_type,
"ct_storage_meter": envoy.storage_meter_type,
"ct_meters": list(envoy_data.ctmeters.keys()),
}
fixture_data: dict[str, Any] = {}

View File

@@ -399,189 +399,117 @@ class EnvoyCTSensorEntityDescription(SensorEntityDescription):
cttype: str | None = None
# All ct types unified in common setup
CT_SENSORS = (
[
EnvoyCTSensorEntityDescription(
key=key,
translation_key=key,
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
state_class=SensorStateClass.TOTAL_INCREASING,
device_class=SensorDeviceClass.ENERGY,
suggested_unit_of_measurement=UnitOfEnergy.MEGA_WATT_HOUR,
suggested_display_precision=3,
value_fn=attrgetter("energy_delivered"),
on_phase=None,
cttype=cttype,
)
for cttype, key in (
(CtType.NET_CONSUMPTION, "lifetime_net_consumption"),
# Production CT energy_delivered is not used
(CtType.STORAGE, "lifetime_battery_discharged"),
)
]
+ [
EnvoyCTSensorEntityDescription(
key=key,
translation_key=key,
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
state_class=SensorStateClass.TOTAL_INCREASING,
device_class=SensorDeviceClass.ENERGY,
suggested_unit_of_measurement=UnitOfEnergy.MEGA_WATT_HOUR,
suggested_display_precision=3,
value_fn=attrgetter("energy_received"),
on_phase=None,
cttype=cttype,
)
for cttype, key in (
(CtType.NET_CONSUMPTION, "lifetime_net_production"),
# Production CT energy_received is not used
(CtType.STORAGE, "lifetime_battery_charged"),
)
]
+ [
EnvoyCTSensorEntityDescription(
key=key,
translation_key=key,
native_unit_of_measurement=UnitOfPower.WATT,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.POWER,
suggested_unit_of_measurement=UnitOfPower.KILO_WATT,
suggested_display_precision=3,
value_fn=attrgetter("active_power"),
on_phase=None,
cttype=cttype,
)
for cttype, key in (
(CtType.NET_CONSUMPTION, "net_consumption"),
# Production CT active_power is not used
(CtType.STORAGE, "battery_discharge"),
)
]
+ [
EnvoyCTSensorEntityDescription(
key=key,
translation_key=(translation_key if translation_key != "" else key),
native_unit_of_measurement=UnitOfFrequency.HERTZ,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.FREQUENCY,
suggested_display_precision=1,
entity_registry_enabled_default=False,
value_fn=attrgetter("frequency"),
on_phase=None,
cttype=cttype,
)
for cttype, key, translation_key in (
(CtType.NET_CONSUMPTION, "frequency", "net_ct_frequency"),
(CtType.PRODUCTION, "production_ct_frequency", ""),
(CtType.STORAGE, "storage_ct_frequency", ""),
)
]
+ [
EnvoyCTSensorEntityDescription(
key=key,
translation_key=(translation_key if translation_key != "" else key),
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.VOLTAGE,
suggested_unit_of_measurement=UnitOfElectricPotential.VOLT,
suggested_display_precision=1,
entity_registry_enabled_default=False,
value_fn=attrgetter("voltage"),
on_phase=None,
cttype=cttype,
)
for cttype, key, translation_key in (
(CtType.NET_CONSUMPTION, "voltage", "net_ct_voltage"),
(CtType.PRODUCTION, "production_ct_voltage", ""),
(CtType.STORAGE, "storage_voltage", "storage_ct_voltage"),
)
]
+ [
EnvoyCTSensorEntityDescription(
key=key,
translation_key=key,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.CURRENT,
suggested_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
suggested_display_precision=3,
entity_registry_enabled_default=False,
value_fn=attrgetter("current"),
on_phase=None,
cttype=cttype,
)
for cttype, key in (
(CtType.NET_CONSUMPTION, "net_ct_current"),
(CtType.PRODUCTION, "production_ct_current"),
(CtType.STORAGE, "storage_ct_current"),
)
]
+ [
EnvoyCTSensorEntityDescription(
key=key,
translation_key=key,
device_class=SensorDeviceClass.POWER_FACTOR,
state_class=SensorStateClass.MEASUREMENT,
suggested_display_precision=2,
entity_registry_enabled_default=False,
value_fn=attrgetter("power_factor"),
on_phase=None,
cttype=cttype,
)
for cttype, key in (
(CtType.NET_CONSUMPTION, "net_ct_powerfactor"),
(CtType.PRODUCTION, "production_ct_powerfactor"),
(CtType.STORAGE, "storage_ct_powerfactor"),
)
]
+ [
EnvoyCTSensorEntityDescription(
key=key,
translation_key=(translation_key if translation_key != "" else key),
device_class=SensorDeviceClass.ENUM,
entity_category=EntityCategory.DIAGNOSTIC,
options=list(CtMeterStatus),
entity_registry_enabled_default=False,
value_fn=attrgetter("metering_status"),
on_phase=None,
cttype=cttype,
)
for cttype, key, translation_key in (
(
CtType.NET_CONSUMPTION,
"net_consumption_ct_metering_status",
"net_ct_metering_status",
),
(CtType.PRODUCTION, "production_ct_metering_status", ""),
(CtType.STORAGE, "storage_ct_metering_status", ""),
)
]
+ [
EnvoyCTSensorEntityDescription(
key=key,
translation_key=(translation_key if translation_key != "" else key),
state_class=None,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda ct: 0 if ct.status_flags is None else len(ct.status_flags),
on_phase=None,
cttype=cttype,
)
for cttype, key, translation_key in (
(
CtType.NET_CONSUMPTION,
"net_consumption_ct_status_flags",
"net_ct_status_flags",
),
(CtType.PRODUCTION, "production_ct_status_flags", ""),
(CtType.STORAGE, "storage_ct_status_flags", ""),
)
]
CT_NET_CONSUMPTION_SENSORS = (
EnvoyCTSensorEntityDescription(
key="lifetime_net_consumption",
translation_key="lifetime_net_consumption",
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
state_class=SensorStateClass.TOTAL_INCREASING,
device_class=SensorDeviceClass.ENERGY,
suggested_unit_of_measurement=UnitOfEnergy.MEGA_WATT_HOUR,
suggested_display_precision=3,
value_fn=attrgetter("energy_delivered"),
on_phase=None,
cttype=CtType.NET_CONSUMPTION,
),
EnvoyCTSensorEntityDescription(
key="lifetime_net_production",
translation_key="lifetime_net_production",
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
state_class=SensorStateClass.TOTAL_INCREASING,
device_class=SensorDeviceClass.ENERGY,
suggested_unit_of_measurement=UnitOfEnergy.MEGA_WATT_HOUR,
suggested_display_precision=3,
value_fn=attrgetter("energy_received"),
on_phase=None,
cttype=CtType.NET_CONSUMPTION,
),
EnvoyCTSensorEntityDescription(
key="net_consumption",
translation_key="net_consumption",
native_unit_of_measurement=UnitOfPower.WATT,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.POWER,
suggested_unit_of_measurement=UnitOfPower.KILO_WATT,
suggested_display_precision=3,
value_fn=attrgetter("active_power"),
on_phase=None,
cttype=CtType.NET_CONSUMPTION,
),
EnvoyCTSensorEntityDescription(
key="frequency",
translation_key="net_ct_frequency",
native_unit_of_measurement=UnitOfFrequency.HERTZ,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.FREQUENCY,
suggested_display_precision=1,
entity_registry_enabled_default=False,
value_fn=attrgetter("frequency"),
on_phase=None,
cttype=CtType.NET_CONSUMPTION,
),
EnvoyCTSensorEntityDescription(
key="voltage",
translation_key="net_ct_voltage",
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.VOLTAGE,
suggested_unit_of_measurement=UnitOfElectricPotential.VOLT,
suggested_display_precision=1,
entity_registry_enabled_default=False,
value_fn=attrgetter("voltage"),
on_phase=None,
cttype=CtType.NET_CONSUMPTION,
),
EnvoyCTSensorEntityDescription(
key="net_ct_current",
translation_key="net_ct_current",
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.CURRENT,
suggested_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
suggested_display_precision=3,
entity_registry_enabled_default=False,
value_fn=attrgetter("current"),
on_phase=None,
cttype=CtType.NET_CONSUMPTION,
),
EnvoyCTSensorEntityDescription(
key="net_ct_powerfactor",
translation_key="net_ct_powerfactor",
device_class=SensorDeviceClass.POWER_FACTOR,
state_class=SensorStateClass.MEASUREMENT,
suggested_display_precision=2,
entity_registry_enabled_default=False,
value_fn=attrgetter("power_factor"),
on_phase=None,
cttype=CtType.NET_CONSUMPTION,
),
EnvoyCTSensorEntityDescription(
key="net_consumption_ct_metering_status",
translation_key="net_ct_metering_status",
device_class=SensorDeviceClass.ENUM,
entity_category=EntityCategory.DIAGNOSTIC,
options=list(CtMeterStatus),
entity_registry_enabled_default=False,
value_fn=attrgetter("metering_status"),
on_phase=None,
cttype=CtType.NET_CONSUMPTION,
),
EnvoyCTSensorEntityDescription(
key="net_consumption_ct_status_flags",
translation_key="net_ct_status_flags",
state_class=None,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda ct: 0 if ct.status_flags is None else len(ct.status_flags),
on_phase=None,
cttype=CtType.NET_CONSUMPTION,
),
)
CT_PHASE_SENSORS = {
CT_NET_CONSUMPTION_PHASE_SENSORS = {
(on_phase := PHASENAMES[phase]): [
replace(
sensor,
@@ -591,7 +519,220 @@ CT_PHASE_SENSORS = {
on_phase=on_phase,
translation_placeholders={"phase_name": f"l{phase + 1}"},
)
for sensor in list(CT_SENSORS)
for sensor in list(CT_NET_CONSUMPTION_SENSORS)
]
for phase in range(3)
}
CT_PRODUCTION_SENSORS = (
EnvoyCTSensorEntityDescription(
key="production_ct_frequency",
translation_key="production_ct_frequency",
native_unit_of_measurement=UnitOfFrequency.HERTZ,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.FREQUENCY,
suggested_display_precision=1,
entity_registry_enabled_default=False,
value_fn=attrgetter("frequency"),
on_phase=None,
cttype=CtType.PRODUCTION,
),
EnvoyCTSensorEntityDescription(
key="production_ct_voltage",
translation_key="production_ct_voltage",
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.VOLTAGE,
suggested_unit_of_measurement=UnitOfElectricPotential.VOLT,
suggested_display_precision=1,
entity_registry_enabled_default=False,
value_fn=attrgetter("voltage"),
on_phase=None,
cttype=CtType.PRODUCTION,
),
EnvoyCTSensorEntityDescription(
key="production_ct_current",
translation_key="production_ct_current",
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.CURRENT,
suggested_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
suggested_display_precision=3,
entity_registry_enabled_default=False,
value_fn=attrgetter("current"),
on_phase=None,
cttype=CtType.PRODUCTION,
),
EnvoyCTSensorEntityDescription(
key="production_ct_powerfactor",
translation_key="production_ct_powerfactor",
device_class=SensorDeviceClass.POWER_FACTOR,
state_class=SensorStateClass.MEASUREMENT,
suggested_display_precision=2,
entity_registry_enabled_default=False,
value_fn=attrgetter("power_factor"),
on_phase=None,
cttype=CtType.PRODUCTION,
),
EnvoyCTSensorEntityDescription(
key="production_ct_metering_status",
translation_key="production_ct_metering_status",
device_class=SensorDeviceClass.ENUM,
options=list(CtMeterStatus),
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=attrgetter("metering_status"),
on_phase=None,
cttype=CtType.PRODUCTION,
),
EnvoyCTSensorEntityDescription(
key="production_ct_status_flags",
translation_key="production_ct_status_flags",
state_class=None,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda ct: 0 if ct.status_flags is None else len(ct.status_flags),
on_phase=None,
cttype=CtType.PRODUCTION,
),
)
CT_PRODUCTION_PHASE_SENSORS = {
(on_phase := PHASENAMES[phase]): [
replace(
sensor,
key=f"{sensor.key}_l{phase + 1}",
translation_key=f"{sensor.translation_key}_phase",
entity_registry_enabled_default=False,
on_phase=on_phase,
translation_placeholders={"phase_name": f"l{phase + 1}"},
)
for sensor in list(CT_PRODUCTION_SENSORS)
]
for phase in range(3)
}
CT_STORAGE_SENSORS = (
EnvoyCTSensorEntityDescription(
key="lifetime_battery_discharged",
translation_key="lifetime_battery_discharged",
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
state_class=SensorStateClass.TOTAL_INCREASING,
device_class=SensorDeviceClass.ENERGY,
suggested_unit_of_measurement=UnitOfEnergy.MEGA_WATT_HOUR,
suggested_display_precision=3,
value_fn=attrgetter("energy_delivered"),
on_phase=None,
cttype=CtType.STORAGE,
),
EnvoyCTSensorEntityDescription(
key="lifetime_battery_charged",
translation_key="lifetime_battery_charged",
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
state_class=SensorStateClass.TOTAL_INCREASING,
device_class=SensorDeviceClass.ENERGY,
suggested_unit_of_measurement=UnitOfEnergy.MEGA_WATT_HOUR,
suggested_display_precision=3,
value_fn=attrgetter("energy_received"),
on_phase=None,
cttype=CtType.STORAGE,
),
EnvoyCTSensorEntityDescription(
key="battery_discharge",
translation_key="battery_discharge",
native_unit_of_measurement=UnitOfPower.WATT,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.POWER,
suggested_unit_of_measurement=UnitOfPower.KILO_WATT,
suggested_display_precision=3,
value_fn=attrgetter("active_power"),
on_phase=None,
cttype=CtType.STORAGE,
),
EnvoyCTSensorEntityDescription(
key="storage_ct_frequency",
translation_key="storage_ct_frequency",
native_unit_of_measurement=UnitOfFrequency.HERTZ,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.FREQUENCY,
suggested_display_precision=1,
entity_registry_enabled_default=False,
value_fn=attrgetter("frequency"),
on_phase=None,
cttype=CtType.STORAGE,
),
EnvoyCTSensorEntityDescription(
key="storage_voltage",
translation_key="storage_ct_voltage",
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.VOLTAGE,
suggested_unit_of_measurement=UnitOfElectricPotential.VOLT,
suggested_display_precision=1,
entity_registry_enabled_default=False,
value_fn=attrgetter("voltage"),
on_phase=None,
cttype=CtType.STORAGE,
),
EnvoyCTSensorEntityDescription(
key="storage_ct_current",
translation_key="storage_ct_current",
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.CURRENT,
suggested_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
suggested_display_precision=3,
entity_registry_enabled_default=False,
value_fn=attrgetter("current"),
on_phase=None,
cttype=CtType.STORAGE,
),
EnvoyCTSensorEntityDescription(
key="storage_ct_powerfactor",
translation_key="storage_ct_powerfactor",
device_class=SensorDeviceClass.POWER_FACTOR,
state_class=SensorStateClass.MEASUREMENT,
suggested_display_precision=2,
entity_registry_enabled_default=False,
value_fn=attrgetter("power_factor"),
on_phase=None,
cttype=CtType.STORAGE,
),
EnvoyCTSensorEntityDescription(
key="storage_ct_metering_status",
translation_key="storage_ct_metering_status",
device_class=SensorDeviceClass.ENUM,
options=list(CtMeterStatus),
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=attrgetter("metering_status"),
on_phase=None,
cttype=CtType.STORAGE,
),
EnvoyCTSensorEntityDescription(
key="storage_ct_status_flags",
translation_key="storage_ct_status_flags",
state_class=None,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda ct: 0 if ct.status_flags is None else len(ct.status_flags),
on_phase=None,
cttype=CtType.STORAGE,
),
)
CT_STORAGE_PHASE_SENSORS = {
(on_phase := PHASENAMES[phase]): [
replace(
sensor,
key=f"{sensor.key}_l{phase + 1}",
translation_key=f"{sensor.translation_key}_phase",
entity_registry_enabled_default=False,
on_phase=on_phase,
translation_placeholders={"phase_name": f"l{phase + 1}"},
)
for sensor in list(CT_STORAGE_SENSORS)
]
for phase in range(3)
}
@@ -919,14 +1060,24 @@ async def async_setup_entry(
if envoy_data.ctmeters:
entities.extend(
EnvoyCTEntity(coordinator, description)
for description in CT_SENSORS
for sensors in (
CT_NET_CONSUMPTION_SENSORS,
CT_PRODUCTION_SENSORS,
CT_STORAGE_SENSORS,
)
for description in sensors
if description.cttype in envoy_data.ctmeters
)
# Add Current Transformer phase entities
if ctmeters_phases := envoy_data.ctmeters_phases:
entities.extend(
EnvoyCTPhaseEntity(coordinator, description)
for phase, descriptions in CT_PHASE_SENSORS.items()
for sensors in (
CT_NET_CONSUMPTION_PHASE_SENSORS,
CT_PRODUCTION_PHASE_SENSORS,
CT_STORAGE_PHASE_SENSORS,
)
for phase, descriptions in sensors.items()
for description in descriptions
if (cttype := description.cttype) in ctmeters_phases
and phase in ctmeters_phases[cttype]

View File

@@ -1,31 +0,0 @@
"""The Fluss+ integration."""
from __future__ import annotations
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY, Platform
from homeassistant.core import HomeAssistant
from .coordinator import FlussDataUpdateCoordinator
PLATFORMS: list[Platform] = [Platform.BUTTON]
type FlussConfigEntry = ConfigEntry[FlussDataUpdateCoordinator]
async def async_setup_entry(
hass: HomeAssistant,
entry: FlussConfigEntry,
) -> bool:
"""Set up Fluss+ from a config entry."""
coordinator = FlussDataUpdateCoordinator(hass, entry, entry.data[CONF_API_KEY])
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: FlussConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

View File

@@ -1,38 +0,0 @@
"""Support for Fluss Devices."""
from homeassistant.components.button import ButtonEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import FlussApiClientError, FlussDataUpdateCoordinator
from .entity import FlussEntity
type FlussConfigEntry = ConfigEntry[FlussDataUpdateCoordinator]
async def async_setup_entry(
hass: HomeAssistant,
entry: FlussConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Fluss Devices, filtering out any invalid payloads."""
coordinator = entry.runtime_data
devices = coordinator.data
async_add_entities(
FlussButton(coordinator, device_id, device)
for device_id, device in devices.items()
)
class FlussButton(FlussEntity, ButtonEntity):
"""Representation of a Fluss button device."""
async def async_press(self) -> None:
"""Handle the button press."""
try:
await self.coordinator.api.async_trigger_device(self.device_id)
except FlussApiClientError as err:
raise HomeAssistantError(f"Failed to trigger device: {err}") from err

View File

@@ -1,54 +0,0 @@
"""Config flow for Fluss+ integration."""
from __future__ import annotations
from typing import Any
from fluss_api import (
FlussApiClient,
FlussApiClientAuthenticationError,
FlussApiClientCommunicationError,
)
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_API_KEY
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN, LOGGER
STEP_USER_DATA_SCHEMA = vol.Schema({vol.Required(CONF_API_KEY): cv.string})
class FlussConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Fluss+."""
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is not None:
api_key = user_input[CONF_API_KEY]
self._async_abort_entries_match({CONF_API_KEY: api_key})
try:
FlussApiClient(
user_input[CONF_API_KEY], session=async_get_clientsession(self.hass)
)
except FlussApiClientCommunicationError:
errors["base"] = "cannot_connect"
except FlussApiClientAuthenticationError:
errors["base"] = "invalid_auth"
except Exception: # noqa: BLE001
LOGGER.exception("Unexpected exception occurred")
errors["base"] = "unknown"
if not errors:
return self.async_create_entry(
title="My Fluss+ Devices", data=user_input
)
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)

View File

@@ -1,9 +0,0 @@
"""Constants for the Fluss+ integration."""
from datetime import timedelta
import logging
DOMAIN = "fluss"
LOGGER = logging.getLogger(__name__)
UPDATE_INTERVAL = 60 # seconds
UPDATE_INTERVAL_TIMEDELTA = timedelta(seconds=UPDATE_INTERVAL)

View File

@@ -1,50 +0,0 @@
"""DataUpdateCoordinator for Fluss+ integration."""
from __future__ import annotations
from typing import Any
from fluss_api import (
FlussApiClient,
FlussApiClientAuthenticationError,
FlussApiClientError,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util import slugify
from .const import LOGGER, UPDATE_INTERVAL_TIMEDELTA
type FlussConfigEntry = ConfigEntry[FlussDataUpdateCoordinator]
class FlussDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""Manages fetching Fluss device data on a schedule."""
def __init__(
self, hass: HomeAssistant, config_entry: FlussConfigEntry, api_key: str
) -> None:
"""Initialize the coordinator."""
self.api = FlussApiClient(api_key, session=async_get_clientsession(hass))
super().__init__(
hass,
LOGGER,
name=f"Fluss+ ({slugify(api_key[:8])})",
config_entry=config_entry,
update_interval=UPDATE_INTERVAL_TIMEDELTA,
)
async def _async_update_data(self) -> dict[str, dict[str, Any]]:
"""Fetch data from the Fluss API and return as a dictionary keyed by deviceId."""
try:
devices = await self.api.async_get_devices()
except FlussApiClientAuthenticationError as err:
raise ConfigEntryAuthFailed(f"Authentication failed: {err}") from err
except FlussApiClientError as err:
raise UpdateFailed(f"Error fetching Fluss devices: {err}") from err
return {device["deviceId"]: device for device in devices.get("devices", [])}

View File

@@ -1,36 +0,0 @@
"""Base entities for the Fluss+ integration."""
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .coordinator import FlussDataUpdateCoordinator
class FlussEntity(CoordinatorEntity[FlussDataUpdateCoordinator]):
"""Base class for Fluss entities."""
_attr_has_entity_name = True
_attr_name = None
def __init__(
self,
coordinator: FlussDataUpdateCoordinator,
device_id: str,
device: dict,
) -> None:
"""Initialize the entity with a device ID and device data."""
super().__init__(coordinator)
self.device_id = device_id
self._device = device
self._attr_unique_id = f"{device_id}"
self._attr_device_info = DeviceInfo(
identifiers={("fluss", device_id)},
name=device.get("deviceName"),
manufacturer="Fluss",
model="Fluss+ Device",
)
@property
def device(self) -> dict:
"""Return the stored device data."""
return self._device

View File

@@ -1,11 +0,0 @@
{
"domain": "fluss",
"name": "Fluss+",
"codeowners": ["@fluss"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/fluss",
"iot_class": "cloud_polling",
"loggers": ["fluss-api"],
"quality_scale": "bronze",
"requirements": ["fluss-api==0.1.9.17"]
}

View File

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

View File

@@ -1,20 +0,0 @@
{
"config": {
"step": {
"user": {
"description": "Your Fluss API key, available in the profile page of the Fluss+ app",
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]"
},
"data_description": {
"api_key": "The API key found in the profile page of the Fluss+ app."
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
}
}
}

View File

@@ -3,17 +3,16 @@
from __future__ import annotations
from datetime import timedelta
from http import HTTPStatus
import logging
from aiosenz import SENZAPI, Thermostat
from httpx import HTTPStatusError, RequestError
from httpx import RequestError
import jwt
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv, httpx_client
from homeassistant.helpers.config_entry_oauth2_flow import (
ImplementationUnavailableError,
@@ -60,21 +59,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: SENZConfigEntry) -> bool
try:
account = await senz_api.get_account()
except HTTPStatusError as err:
if err.response.status_code == HTTPStatus.UNAUTHORIZED:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="config_entry_auth_failed",
) from err
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="config_entry_not_ready",
) from err
except RequestError as err:
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="config_entry_not_ready",
) from err
raise ConfigEntryNotReady from err
coordinator: SENZDataUpdateCoordinator = DataUpdateCoordinator(
hass,

View File

@@ -1,16 +1,10 @@
"""Config flow for nVent RAYCHEM SENZ."""
from collections.abc import Mapping
import logging
from typing import Any
import jwt
from homeassistant.config_entries import (
SOURCE_REAUTH,
SOURCE_RECONFIGURE,
ConfigFlowResult,
)
from homeassistant.config_entries import ConfigFlowResult
from homeassistant.helpers import config_entry_oauth2_flow
from .const import DOMAIN
@@ -35,28 +29,6 @@ class OAuth2FlowHandler(
"""Extra data that needs to be appended to the authorize url."""
return {"scope": "restapi offline_access"}
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Perform reauth upon an API authentication error."""
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Dialog that informs the user that reauth is required."""
if user_input is None:
return self.async_show_form(step_id="reauth_confirm")
return await self.async_step_user()
async def async_step_reconfigure(
self, user_input: Mapping[str, Any] | None = None
) -> ConfigFlowResult:
"""User initiated reconfiguration."""
return await self.async_step_user()
async def async_oauth_create_entry(self, data: dict) -> ConfigFlowResult:
"""Create or update the config entry."""
@@ -66,17 +38,5 @@ class OAuth2FlowHandler(
uid = token["sub"]
await self.async_set_unique_id(uid)
if self.source == SOURCE_REAUTH:
self._abort_if_unique_id_mismatch(reason="account_mismatch")
return self.async_update_reload_and_abort(
self._get_reauth_entry(), data=data
)
if self.source == SOURCE_RECONFIGURE:
self._abort_if_unique_id_mismatch(reason="account_mismatch")
return self.async_update_reload_and_abort(
self._get_reconfigure_entry(), data=data
)
self._abort_if_unique_id_configured()
return await super().async_oauth_create_entry(data)

View File

@@ -1,7 +1,6 @@
{
"config": {
"abort": {
"account_mismatch": "The used account does not match the original account",
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"authorize_url_timeout": "[%key:common::config_flow::abort::oauth2_authorize_url_timeout%]",
@@ -10,9 +9,7 @@
"oauth_error": "[%key:common::config_flow::abort::oauth2_error%]",
"oauth_failed": "[%key:common::config_flow::abort::oauth2_failed%]",
"oauth_timeout": "[%key:common::config_flow::abort::oauth2_timeout%]",
"oauth_unauthorized": "[%key:common::config_flow::abort::oauth2_unauthorized%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]"
"oauth_unauthorized": "[%key:common::config_flow::abort::oauth2_unauthorized%]"
},
"create_entry": {
"default": "[%key:common::config_flow::create_entry::authenticated%]"
@@ -26,20 +23,10 @@
"implementation": "[%key:common::config_flow::description::implementation%]"
},
"title": "[%key:common::config_flow::title::oauth2_pick_implementation%]"
},
"reauth_confirm": {
"description": "The SENZ integration needs to re-authenticate your account",
"title": "[%key:common::config_flow::title::reauth%]"
}
}
},
"exceptions": {
"config_entry_auth_failed": {
"message": "Authentication failed. Please log in again."
},
"config_entry_not_ready": {
"message": "Error while loading the integration."
},
"oauth2_implementation_unavailable": {
"message": "[%key:common::exceptions::oauth2_implementation_unavailable::message%]"
}

View File

@@ -30,5 +30,5 @@
"iot_class": "cloud_push",
"loggers": ["pysmartthings"],
"quality_scale": "bronze",
"requirements": ["pysmartthings==3.3.3"]
"requirements": ["pysmartthings==3.3.2"]
}

View File

@@ -213,7 +213,6 @@ FLOWS = {
"flipr",
"flo",
"flume",
"fluss",
"flux_led",
"folder_watcher",
"forecast_solar",

View File

@@ -2062,12 +2062,6 @@
"config_flow": true,
"iot_class": "cloud_polling"
},
"fluss": {
"name": "Fluss+",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
"flux": {
"name": "Flux",
"integration_type": "hub",

5
requirements_all.txt generated
View File

@@ -968,9 +968,6 @@ flexit_bacnet==2.2.3
# homeassistant.components.flipr
flipr-api==1.6.1
# homeassistant.components.fluss
fluss-api==0.1.9.17
# homeassistant.components.flux_led
flux-led==1.2.0
@@ -2383,7 +2380,7 @@ pysmappee==0.2.29
pysmarlaapi==0.9.2
# homeassistant.components.smartthings
pysmartthings==3.3.3
pysmartthings==3.3.2
# homeassistant.components.smarty
pysmarty2==0.10.3

View File

@@ -844,9 +844,6 @@ flexit_bacnet==2.2.3
# homeassistant.components.flipr
flipr-api==1.6.1
# homeassistant.components.fluss
fluss-api==0.1.9.17
# homeassistant.components.flux_led
flux-led==1.2.0
@@ -1985,7 +1982,7 @@ pysmappee==0.2.29
pysmarlaapi==0.9.2
# homeassistant.components.smartthings
pysmartthings==3.3.3
pysmartthings==3.3.2
# homeassistant.components.smarty
pysmarty2==0.10.3

View File

@@ -331,6 +331,7 @@ INTEGRATIONS_WITHOUT_QUALITY_SCALE_FILE = [
"elv",
"elvia",
"emby",
"emoncms",
"emoncms_history",
"emonitor",
"emulated_hue",

View File

@@ -32,6 +32,7 @@ FIELD_SCHEMA = vol.Schema(
vol.Optional("default"): exists,
vol.Optional("required"): bool,
vol.Optional(CONF_SELECTOR): selector.validate_selector,
vol.Optional("context"): {str: str},
}
)

File diff suppressed because it is too large Load Diff

View File

@@ -94,14 +94,6 @@ async def test_entry_diagnostics_with_fixtures_with_error(
) == snapshot(exclude=limit_diagnostic_attrs)
@pytest.mark.parametrize(
("mock_envoy"),
[
"envoy_metered_batt_relay",
"envoy",
],
indirect=["mock_envoy"],
)
async def test_entry_diagnostics_with_interface_information(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,

View File

@@ -1,92 +0,0 @@
"""Test Script for Fluss+ Initialisation."""
from __future__ import annotations
from unittest.mock import MagicMock, patch
from fluss_api import (
FlussApiClient,
FlussApiClientAuthenticationError,
FlussApiClientCommunicationError,
FlussApiClientError,
)
import pytest
from homeassistant.components.fluss import PLATFORMS
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
@pytest.mark.parametrize(
("side_effect", "expected_exception"),
[
(FlussApiClientAuthenticationError, ConfigEntryAuthFailed),
(FlussApiClientCommunicationError, ConfigEntryNotReady),
(FlussApiClientError, ConfigEntryNotReady),
],
)
async def test_async_setup_entry_errors(
hass: HomeAssistant,
mock_config_entry: MagicMock,
side_effect: Exception,
expected_exception: type[Exception],
) -> None:
"""Test setup errors."""
with (
patch("fluss_api.FlussApiClient", side_effect=side_effect),
pytest.raises(expected_exception),
):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.SETUP_ERROR
@pytest.mark.asyncio
async def test_async_setup_entry_success(
hass: HomeAssistant,
mock_config_entry: MagicMock,
mock_api_client: FlussApiClient,
) -> None:
"""Test successful setup."""
with patch("fluss_api.FlussApiClient", return_value=mock_api_client):
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.LOADED
hass.config_entries.async_forward_entry_setups.assert_called_once_with(
mock_config_entry, PLATFORMS
)
@pytest.mark.asyncio
async def test_async_unload_entry(
hass: HomeAssistant,
mock_config_entry: MagicMock,
mock_api_client: FlussApiClient,
) -> None:
"""Test unloading entry."""
# Set up the config entry first to ensure it's in LOADED state
with patch("fluss_api.FlussApiClient", return_value=mock_api_client):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.LOADED
# Test unloading
with patch(
"homeassistant.components.fluss.async_unload_platforms", return_value=True
):
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
@pytest.mark.asyncio
async def test_platforms_forwarded(
hass: HomeAssistant,
mock_config_entry: MagicMock,
mock_api_client: FlussApiClient,
) -> None:
"""Test platforms are forwarded correctly."""
with patch("fluss_api.FlussApiClient", return_value=mock_api_client):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.LOADED
hass.config_entries.async_forward_entry_setups.assert_called_with(
mock_config_entry, [Platform.BUTTON]
)

View File

@@ -1,74 +0,0 @@
"""Shared test fixtures for Fluss+ integration."""
from __future__ import annotations
from unittest.mock import AsyncMock, patch
import pytest
from homeassistant.components.fluss.const import DOMAIN
from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
domain=DOMAIN,
title="Fluss Integration",
data={CONF_API_KEY: "test_api_key"},
unique_id="test_api_key",
)
@pytest.fixture
def mock_api_client() -> AsyncMock:
"""Mock Fluss API client with single device."""
with (
patch(
"homeassistant.components.fluss.coordinator.FlussApiClient",
autospec=True,
) as mock_client,
patch(
"homeassistant.components.fluss.config_flow.FlussApiClient",
new=mock_client,
),
):
client = mock_client.return_value
client.async_get_devices.return_value = {
"devices": [{"deviceId": "1", "deviceName": "Test Device"}]
}
client.async_trigger_device.return_value = None
yield client
@pytest.fixture
def mock_api_client_multiple_devices() -> AsyncMock:
"""Mock Fluss API client with multiple devices."""
with patch(
"homeassistant.components.fluss.coordinator.FlussApiClient",
autospec=True,
) as mock_client:
client = mock_client.return_value
client.async_get_devices.return_value = {
"devices": [
{"deviceId": "2a303030sdj1", "deviceName": "Device 1"},
{"deviceId": "ape93k9302j2", "deviceName": "Device 2"},
]
}
client.async_trigger_device.return_value = None
yield client
@pytest.fixture
async def init_integration(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_api_client: AsyncMock
) -> MockConfigEntry:
"""Set up the Fluss integration for testing."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
return mock_config_entry

View File

@@ -1,118 +0,0 @@
"""Tests for the Fluss Buttons."""
from __future__ import annotations
from unittest.mock import AsyncMock, patch
from fluss_api import FlussApiClient, FlussApiClientError
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN, SERVICE_PRESS
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr, entity_registry as er
from tests.common import MockConfigEntry, snapshot_platform
async def test_async_setup_entry_multiple_devices(
hass: HomeAssistant,
mock_api_client_multiple_devices: AsyncMock,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test setup with multiple devices."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
mock_api_client_multiple_devices.async_get_devices.assert_called_once()
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_button_press_success(
hass: HomeAssistant,
mock_api_client: FlussApiClient,
entity_registry: er.EntityRegistry,
init_integration: MockConfigEntry,
snapshot: SnapshotAssertion,
) -> None:
"""Test successful button press."""
state = hass.states.get("button.test_device")
assert state
assert state == snapshot(name="button_state")
entry_reg = entity_registry.async_get(state.entity_id)
assert entry_reg
device_registry = dr.async_get(hass)
device = device_registry.async_get(entry_reg.device_id)
assert device == snapshot(name="device_info")
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: "button.test_device"},
blocking=True,
)
mock_api_client.async_trigger_device.assert_called_once_with("1")
async def test_button_press_error(
hass: HomeAssistant,
mock_api_client: FlussApiClient,
init_integration: MockConfigEntry,
) -> None:
"""Test button press with API error."""
state = hass.states.get("button.test_device")
assert state
mock_api_client.async_trigger_device.side_effect = FlussApiClientError("API Boom")
with pytest.raises(HomeAssistantError, match="Failed to trigger device: API Boom"):
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: "button.test_device"},
blocking=True,
)
async def test_no_devices_setup(
hass: HomeAssistant,
mock_api_client: FlussApiClient,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test setup with no devices."""
mock_api_client.async_get_devices.return_value = {"devices": []}
mock_config_entry.add_to_hass(hass)
with patch("fluss_api.FlussApiClient", return_value=mock_api_client):
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("button.test_device") is None
async def test_unload_entry(
hass: HomeAssistant,
init_integration: MockConfigEntry,
) -> None:
"""Test unloading the entry."""
assert init_integration.state is ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(init_integration.entry_id)
await hass.async_block_till_done()
assert init_integration.state is ConfigEntryState.NOT_LOADED
state = hass.states.get("button.test_device")
if state:
assert state.state == "unavailable"
assert state.attributes.get("restored") is True
else:
assert state is None

View File

@@ -1,100 +0,0 @@
"""Tests for the Fluss+ config flow."""
from unittest.mock import AsyncMock, patch
from fluss_api import (
FlussApiClientAuthenticationError,
FlussApiClientCommunicationError,
)
import pytest
from homeassistant.components.fluss.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
async def test_show_form(hass: HomeAssistant) -> None:
"""Test that the form is shown."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
async def test_successful_flow(hass: HomeAssistant, mock_api_client: AsyncMock) -> None:
"""Test successful config flow."""
user_input = {CONF_API_KEY: "valid_api_key"}
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=user_input
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "My Fluss+ Devices"
assert result["data"] == user_input
@pytest.mark.parametrize(
("exception", "expected_error"),
[
(FlussApiClientAuthenticationError, "invalid_auth"),
(FlussApiClientCommunicationError, "cannot_connect"),
(ValueError, "unknown"),
],
)
async def test_step_user_errors(
hass: HomeAssistant,
exception: Exception,
expected_error: str,
mock_api_client: AsyncMock,
) -> None:
"""Test error cases for user step with recovery."""
user_input = {CONF_API_KEY: "some_api_key"}
class_mock = mock_api_client._mock_parent
class_mock.side_effect = exception
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=user_input
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": expected_error}
class_mock.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "My Fluss+ Devices"
assert result["data"] == user_input
async def test_unexpected_exception_logging(
hass: HomeAssistant, mock_api_client: AsyncMock
) -> None:
"""Test logging of unexpected exceptions."""
user_input = {CONF_API_KEY: "some_api_key"}
with patch(
"homeassistant.components.fluss.config_flow.LOGGER.exception"
) as mock_logger:
class_mock = mock_api_client._mock_parent
class_mock.side_effect = Exception("Unexpected error")
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=user_input
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "unknown"}
mock_logger.assert_called_once_with("Unexpected exception occurred")

View File

@@ -1,74 +0,0 @@
"""Test script for Fluss+ integration initialization."""
from unittest.mock import AsyncMock, patch
from fluss_api import (
FlussApiClientAuthenticationError,
FlussApiClientCommunicationError,
FlussApiClientError,
)
import pytest
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def test_async_setup_entry_authentication_error(
hass: HomeAssistant, mock_config_entry
) -> None:
"""Test that an authentication error during setup leads to SETUP_ERROR state."""
mock_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.fluss.coordinator.FlussApiClient.async_get_devices",
side_effect=FlussApiClientAuthenticationError("Invalid credentials"),
):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_ERROR
@pytest.mark.asyncio
@pytest.mark.parametrize(
"error_type",
[
FlussApiClientCommunicationError("Network error"),
FlussApiClientError("General error"),
],
ids=["communication_error", "general_error"],
)
async def test_async_setup_entry_error(
hass: HomeAssistant, mock_config_entry, error_type
) -> None:
"""Test that non-authentication errors during setup lead to SETUP_RETRY state."""
mock_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.fluss.coordinator.FlussApiClient",
side_effect=error_type,
):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_ERROR
@pytest.mark.asyncio
async def test_load_unload_config_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_api_client: AsyncMock,
) -> None:
"""Test the Fluss configuration entry loading/unloading."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
assert len(mock_api_client.async_get_devices.mock_calls) == 1
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

View File

@@ -114,18 +114,12 @@ async def setup_credentials(hass: HomeAssistant) -> None:
@pytest.fixture
def unique_id() -> str:
"""Return a unique ID."""
return ENTRY_UNIQUE_ID
@pytest.fixture
async def access_token(hass: HomeAssistant, unique_id: str) -> str:
async def access_token(hass: HomeAssistant) -> str:
"""Return a valid access token."""
return config_entry_oauth2_flow._encode_jwt(
hass,
{
"sub": unique_id,
"sub": ENTRY_UNIQUE_ID,
"aud": [],
"scp": [
"rest_api",

View File

@@ -16,15 +16,12 @@ from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.setup import async_setup_component
from .const import CLIENT_ID, CLIENT_SECRET, ENTRY_UNIQUE_ID
from .const import CLIENT_ID, CLIENT_SECRET
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import ClientSessionGenerator
REDIRECT_PATH = "/auth/external/callback"
REDIRECT_URL = "https://example.com" + REDIRECT_PATH
@pytest.mark.usefixtures("current_request_with_host")
async def test_full_flow(
@@ -48,18 +45,18 @@ async def test_full_flow(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": REDIRECT_URL,
"redirect_uri": "https://example.com/auth/external/callback",
},
)
assert result["url"] == (
f"{AUTHORIZATION_ENDPOINT}?response_type=code&client_id={CLIENT_ID}"
f"&redirect_uri={REDIRECT_URL}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}&scope=restapi+offline_access"
)
client = await hass_client_no_auth()
resp = await client.get(f"{REDIRECT_PATH}?code=abcd&state={state}")
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
@@ -99,18 +96,18 @@ async def test_duplicate_flow(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": REDIRECT_URL,
"redirect_uri": "https://example.com/auth/external/callback",
},
)
assert result["url"] == (
f"{AUTHORIZATION_ENDPOINT}?response_type=code&client_id={CLIENT_ID}"
f"&redirect_uri={REDIRECT_URL}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}&scope=restapi+offline_access"
)
client = await hass_client_no_auth()
resp = await client.get(f"{REDIRECT_PATH}?code=abcd&state={state}")
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
@@ -129,153 +126,3 @@ async def test_duplicate_flow(
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "already_configured"
@pytest.mark.usefixtures("current_request_with_host")
async def test_reauth_flow(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_config_entry: MockConfigEntry,
access_token: str,
expires_at: float,
) -> None:
"""Test reauth step with correct params."""
CURRENT_TOKEN = {
"auth_implementation": DOMAIN,
"token": {
"access_token": access_token,
"expires_in": 86399,
"refresh_token": "3012bc9f-7a65-4240-b817-9154ffdcc30f",
"token_type": "Bearer",
"expires_at": expires_at,
},
}
assert hass.config_entries.async_update_entry(
mock_config_entry,
data=CURRENT_TOKEN,
)
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
result = await mock_config_entry.start_reauth_flow(hass)
assert result["step_id"] == "reauth_confirm"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["step_id"] == "auth"
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": REDIRECT_URL,
},
)
assert result["url"] == (
f"{AUTHORIZATION_ENDPOINT}?response_type=code&client_id={CLIENT_ID}"
f"&redirect_uri={REDIRECT_URL}"
f"&state={state}&scope=restapi+offline_access"
)
client = await hass_client_no_auth()
resp = await client.get(f"{REDIRECT_PATH}?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
aioclient_mock.post(
TOKEN_ENDPOINT,
json={
"refresh_token": "updated-refresh-token",
"access_token": access_token,
"type": "Bearer",
"expires_in": "60",
},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == "reauth_successful"
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
@pytest.mark.usefixtures("current_request_with_host")
@pytest.mark.parametrize(
("unique_id", "expected_result"),
[
(ENTRY_UNIQUE_ID, "reconfigure_successful"),
("different_unique_id", "account_mismatch"),
],
)
async def test_reconfiguration_flow(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_config_entry: MockConfigEntry,
access_token: str,
unique_id: str,
expected_result: str,
expires_at: float,
) -> None:
"""Test reconfigure step with correct params."""
CURRENT_TOKEN = {
"auth_implementation": DOMAIN,
"token": {
"access_token": access_token,
"expires_in": 86399,
"refresh_token": "3012bc9f-7a65-4240-b817-9154ffdcc30f",
"token_type": "Bearer",
"expires_at": expires_at,
},
}
assert hass.config_entries.async_update_entry(
mock_config_entry,
data=CURRENT_TOKEN,
)
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
result = await mock_config_entry.start_reconfigure_flow(hass)
assert result["step_id"] == "auth"
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": REDIRECT_URL,
},
)
assert result["url"] == (
f"{AUTHORIZATION_ENDPOINT}?response_type=code&client_id={CLIENT_ID}"
f"&redirect_uri={REDIRECT_URL}"
f"&state={state}&scope=restapi+offline_access"
)
client = await hass_client_no_auth()
resp = await client.get(f"{REDIRECT_PATH}?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
aioclient_mock.post(
TOKEN_ENDPOINT,
json={
"refresh_token": "updated-refresh-token",
"access_token": access_token,
"type": "Bearer",
"expires_in": "60",
},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == expected_result
assert len(hass.config_entries.async_entries(DOMAIN)) == 1

View File

@@ -1,13 +1,7 @@
"""Test init of senz integration."""
from http import HTTPStatus
import time
from unittest.mock import MagicMock, patch
from aiosenz import TOKEN_ENDPOINT
from httpx import HTTPStatusError, RequestError
import pytest
from homeassistant.components.senz.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
@@ -19,7 +13,6 @@ from . import setup_integration
from .const import ENTRY_UNIQUE_ID
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
async def test_load_unload_entry(
@@ -85,77 +78,3 @@ async def test_migrate_config_entry(
assert mock_entry_v1_1.version == 1
assert mock_entry_v1_1.minor_version == 2
assert mock_entry_v1_1.unique_id == ENTRY_UNIQUE_ID
@pytest.mark.parametrize(
("expires_at", "status", "expected_state"),
[
(
time.time() - 3600,
HTTPStatus.UNAUTHORIZED,
ConfigEntryState.SETUP_ERROR,
),
(
time.time() - 3600,
HTTPStatus.INTERNAL_SERVER_ERROR,
ConfigEntryState.SETUP_ERROR,
),
],
ids=["unauthorized", "internal_server_error"],
)
async def test_expired_token_refresh_failure(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
aioclient_mock: AiohttpClientMocker,
status: HTTPStatus,
expected_state: ConfigEntryState,
) -> None:
"""Test failure while refreshing token with a transient error."""
aioclient_mock.clear_requests()
aioclient_mock.post(
TOKEN_ENDPOINT,
status=status,
)
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is expected_state
@pytest.mark.parametrize(
("error", "expected_state"),
[
(
HTTPStatusError(
message="Exception",
request=None,
response=MagicMock(status_code=HTTPStatus.UNAUTHORIZED),
),
ConfigEntryState.SETUP_ERROR,
),
(
HTTPStatusError(
message="Exception",
request=None,
response=MagicMock(status_code=HTTPStatus.FORBIDDEN),
),
ConfigEntryState.SETUP_RETRY,
),
(RequestError("Exception"), ConfigEntryState.SETUP_RETRY),
],
ids=["unauthorized", "forbidden", "request_error"],
)
async def test_setup_errors(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_senz_client: MagicMock,
error: Exception,
expected_state: ConfigEntryState,
) -> None:
"""Test setup failure due to unauthorized error."""
mock_senz_client.get_account.side_effect = error
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is expected_state