Compare commits

..

28 Commits

Author SHA1 Message Date
jbouwh
bb97822db9 Cleanup 2025-11-13 06:48:59 +00:00
jbouwh
33ffccabd1 Refactor 2025-11-13 06:48:59 +00:00
jbouwh
56de03ce33 Rework private _included_entities attribute 2025-11-13 06:48:59 +00:00
jbouwh
0cbf7002a8 Add docstring 2025-11-13 06:48:59 +00:00
jbouwh
cffceffe04 Move setup code to add_to_platform_finish 2025-11-13 06:48:59 +00:00
jbouwh
253189805e Remove final 2025-11-13 06:48:59 +00:00
jbouwh
2e91725ac0 Use cached_properties 2025-11-13 06:48:58 +00:00
jbouwh
3b54dddc08 Fix attrbute check - make property final 2025-11-13 06:48:58 +00:00
jbouwh
9bc3d83a55 Update docstring 2025-11-13 06:48:58 +00:00
jbouwh
d62a554cbf Remove the need to manually call async_set_included_entities 2025-11-13 06:48:58 +00:00
jbouwh
f071b7cd46 Improve docstring 2025-11-13 06:48:58 +00:00
jbouwh
37f34f6189 Remove _included_entities property 2025-11-13 06:48:58 +00:00
jbouwh
27dc5b6d18 Do not set included entities if no unique IDs are set 2025-11-13 06:48:58 +00:00
jbouwh
0bbc2f49a6 Upfdate docstr 2025-11-13 06:48:58 +00:00
jbouwh
c121fa25e8 Call async_set_included_entities from add_to_platform_finish 2025-11-13 06:48:58 +00:00
jbouwh
660cea8b65 Handle the entity_id attribute in the Entity base class 2025-11-13 06:48:58 +00:00
jbouwh
c7749ebae1 Fix device tracker 2025-11-13 06:48:58 +00:00
jbouwh
a2acb744b3 Use platform name 2025-11-13 06:48:58 +00:00
jbouwh
0d9158689d Fix device tracker state attrs 2025-11-13 06:48:58 +00:00
jbouwh
f85e8d6c1f Also implement as default in base entity 2025-11-13 06:48:58 +00:00
jbouwh
9be4cc5af1 Integrate with base entity component state attributes 2025-11-13 06:48:58 +00:00
jbouwh
a141eedf2c Update docstr 2025-11-13 06:48:58 +00:00
jbouwh
03040c131c Move logic into Entity class 2025-11-13 06:48:58 +00:00
jbouwh
3eef50632c Use platform domain attribute 2025-11-13 06:48:58 +00:00
jbouwh
eff150cd54 Fix typo 2025-11-13 06:48:58 +00:00
jbouwh
6dcc94b0a1 Follow up on code review 2025-11-13 06:48:58 +00:00
jbouwh
7201903877 Implement mixin class and add feature to maintain included entities from unique IDs 2025-11-13 06:48:58 +00:00
jbouwh
5b776307ea Add included_entities attribute to base Entity class 2025-11-13 06:48:57 +00:00
73 changed files with 930 additions and 11640 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

@@ -386,260 +386,11 @@ def _async_validate_auto_generated_cost_entity(
issues.add_issue(hass, "recorder_untracked", cost_entity_id)
def _validate_grid_source(
hass: HomeAssistant,
source: data.GridSourceType,
statistics_metadata: dict[str, tuple[int, recorder.models.StatisticMetaData]],
wanted_statistics_metadata: set[str],
source_result: ValidationIssues,
validate_calls: list[functools.partial[None]],
) -> None:
"""Validate grid energy source."""
flow_from: data.FlowFromGridSourceType
for flow_from in source["flow_from"]:
wanted_statistics_metadata.add(flow_from["stat_energy_from"])
validate_calls.append(
functools.partial(
_async_validate_usage_stat,
hass,
statistics_metadata,
flow_from["stat_energy_from"],
ENERGY_USAGE_DEVICE_CLASSES,
ENERGY_USAGE_UNITS,
ENERGY_UNIT_ERROR,
source_result,
)
)
if (stat_cost := flow_from.get("stat_cost")) is not None:
wanted_statistics_metadata.add(stat_cost)
validate_calls.append(
functools.partial(
_async_validate_cost_stat,
hass,
statistics_metadata,
stat_cost,
source_result,
)
)
elif (entity_energy_price := flow_from.get("entity_energy_price")) is not None:
validate_calls.append(
functools.partial(
_async_validate_price_entity,
hass,
entity_energy_price,
source_result,
ENERGY_PRICE_UNITS,
ENERGY_PRICE_UNIT_ERROR,
)
)
if (
flow_from.get("entity_energy_price") is not None
or flow_from.get("number_energy_price") is not None
):
validate_calls.append(
functools.partial(
_async_validate_auto_generated_cost_entity,
hass,
flow_from["stat_energy_from"],
source_result,
)
)
flow_to: data.FlowToGridSourceType
for flow_to in source["flow_to"]:
wanted_statistics_metadata.add(flow_to["stat_energy_to"])
validate_calls.append(
functools.partial(
_async_validate_usage_stat,
hass,
statistics_metadata,
flow_to["stat_energy_to"],
ENERGY_USAGE_DEVICE_CLASSES,
ENERGY_USAGE_UNITS,
ENERGY_UNIT_ERROR,
source_result,
)
)
if (stat_compensation := flow_to.get("stat_compensation")) is not None:
wanted_statistics_metadata.add(stat_compensation)
validate_calls.append(
functools.partial(
_async_validate_cost_stat,
hass,
statistics_metadata,
stat_compensation,
source_result,
)
)
elif (entity_energy_price := flow_to.get("entity_energy_price")) is not None:
validate_calls.append(
functools.partial(
_async_validate_price_entity,
hass,
entity_energy_price,
source_result,
ENERGY_PRICE_UNITS,
ENERGY_PRICE_UNIT_ERROR,
)
)
if (
flow_to.get("entity_energy_price") is not None
or flow_to.get("number_energy_price") is not None
):
validate_calls.append(
functools.partial(
_async_validate_auto_generated_cost_entity,
hass,
flow_to["stat_energy_to"],
source_result,
)
)
for power_stat in source.get("power", []):
wanted_statistics_metadata.add(power_stat["stat_rate"])
validate_calls.append(
functools.partial(
_async_validate_power_stat,
hass,
statistics_metadata,
power_stat["stat_rate"],
POWER_USAGE_DEVICE_CLASSES,
POWER_USAGE_UNITS,
POWER_UNIT_ERROR,
source_result,
)
)
def _validate_gas_source(
hass: HomeAssistant,
source: data.GasSourceType,
statistics_metadata: dict[str, tuple[int, recorder.models.StatisticMetaData]],
wanted_statistics_metadata: set[str],
source_result: ValidationIssues,
validate_calls: list[functools.partial[None]],
) -> None:
"""Validate gas energy source."""
wanted_statistics_metadata.add(source["stat_energy_from"])
validate_calls.append(
functools.partial(
_async_validate_usage_stat,
hass,
statistics_metadata,
source["stat_energy_from"],
GAS_USAGE_DEVICE_CLASSES,
GAS_USAGE_UNITS,
GAS_UNIT_ERROR,
source_result,
)
)
if (stat_cost := source.get("stat_cost")) is not None:
wanted_statistics_metadata.add(stat_cost)
validate_calls.append(
functools.partial(
_async_validate_cost_stat,
hass,
statistics_metadata,
stat_cost,
source_result,
)
)
elif (entity_energy_price := source.get("entity_energy_price")) is not None:
validate_calls.append(
functools.partial(
_async_validate_price_entity,
hass,
entity_energy_price,
source_result,
GAS_PRICE_UNITS,
GAS_PRICE_UNIT_ERROR,
)
)
if (
source.get("entity_energy_price") is not None
or source.get("number_energy_price") is not None
):
validate_calls.append(
functools.partial(
_async_validate_auto_generated_cost_entity,
hass,
source["stat_energy_from"],
source_result,
)
)
def _validate_water_source(
hass: HomeAssistant,
source: data.WaterSourceType,
statistics_metadata: dict[str, tuple[int, recorder.models.StatisticMetaData]],
wanted_statistics_metadata: set[str],
source_result: ValidationIssues,
validate_calls: list[functools.partial[None]],
) -> None:
"""Validate water energy source."""
wanted_statistics_metadata.add(source["stat_energy_from"])
validate_calls.append(
functools.partial(
_async_validate_usage_stat,
hass,
statistics_metadata,
source["stat_energy_from"],
WATER_USAGE_DEVICE_CLASSES,
WATER_USAGE_UNITS,
WATER_UNIT_ERROR,
source_result,
)
)
if (stat_cost := source.get("stat_cost")) is not None:
wanted_statistics_metadata.add(stat_cost)
validate_calls.append(
functools.partial(
_async_validate_cost_stat,
hass,
statistics_metadata,
stat_cost,
source_result,
)
)
elif (entity_energy_price := source.get("entity_energy_price")) is not None:
validate_calls.append(
functools.partial(
_async_validate_price_entity,
hass,
entity_energy_price,
source_result,
WATER_PRICE_UNITS,
WATER_PRICE_UNIT_ERROR,
)
)
if (
source.get("entity_energy_price") is not None
or source.get("number_energy_price") is not None
):
validate_calls.append(
functools.partial(
_async_validate_auto_generated_cost_entity,
hass,
source["stat_energy_from"],
source_result,
)
)
async def async_validate(hass: HomeAssistant) -> EnergyPreferencesValidation:
"""Validate the energy configuration."""
manager: data.EnergyManager = await data.async_get_manager(hass)
statistics_metadata: dict[str, tuple[int, recorder.models.StatisticMetaData]] = {}
validate_calls: list[functools.partial[None]] = []
validate_calls = []
wanted_statistics_metadata: set[str] = set()
result = EnergyPreferencesValidation()
@@ -653,35 +404,230 @@ async def async_validate(hass: HomeAssistant) -> EnergyPreferencesValidation:
result.energy_sources.append(source_result)
if source["type"] == "grid":
_validate_grid_source(
hass,
source,
statistics_metadata,
wanted_statistics_metadata,
source_result,
validate_calls,
)
flow: data.FlowFromGridSourceType | data.FlowToGridSourceType
for flow in source["flow_from"]:
wanted_statistics_metadata.add(flow["stat_energy_from"])
validate_calls.append(
functools.partial(
_async_validate_usage_stat,
hass,
statistics_metadata,
flow["stat_energy_from"],
ENERGY_USAGE_DEVICE_CLASSES,
ENERGY_USAGE_UNITS,
ENERGY_UNIT_ERROR,
source_result,
)
)
if (stat_cost := flow.get("stat_cost")) is not None:
wanted_statistics_metadata.add(stat_cost)
validate_calls.append(
functools.partial(
_async_validate_cost_stat,
hass,
statistics_metadata,
stat_cost,
source_result,
)
)
elif (
entity_energy_price := flow.get("entity_energy_price")
) is not None:
validate_calls.append(
functools.partial(
_async_validate_price_entity,
hass,
entity_energy_price,
source_result,
ENERGY_PRICE_UNITS,
ENERGY_PRICE_UNIT_ERROR,
)
)
if (
flow.get("entity_energy_price") is not None
or flow.get("number_energy_price") is not None
):
validate_calls.append(
functools.partial(
_async_validate_auto_generated_cost_entity,
hass,
flow["stat_energy_from"],
source_result,
)
)
for flow in source["flow_to"]:
wanted_statistics_metadata.add(flow["stat_energy_to"])
validate_calls.append(
functools.partial(
_async_validate_usage_stat,
hass,
statistics_metadata,
flow["stat_energy_to"],
ENERGY_USAGE_DEVICE_CLASSES,
ENERGY_USAGE_UNITS,
ENERGY_UNIT_ERROR,
source_result,
)
)
if (stat_compensation := flow.get("stat_compensation")) is not None:
wanted_statistics_metadata.add(stat_compensation)
validate_calls.append(
functools.partial(
_async_validate_cost_stat,
hass,
statistics_metadata,
stat_compensation,
source_result,
)
)
elif (
entity_energy_price := flow.get("entity_energy_price")
) is not None:
validate_calls.append(
functools.partial(
_async_validate_price_entity,
hass,
entity_energy_price,
source_result,
ENERGY_PRICE_UNITS,
ENERGY_PRICE_UNIT_ERROR,
)
)
if (
flow.get("entity_energy_price") is not None
or flow.get("number_energy_price") is not None
):
validate_calls.append(
functools.partial(
_async_validate_auto_generated_cost_entity,
hass,
flow["stat_energy_to"],
source_result,
)
)
for power_stat in source.get("power", []):
wanted_statistics_metadata.add(power_stat["stat_rate"])
validate_calls.append(
functools.partial(
_async_validate_power_stat,
hass,
statistics_metadata,
power_stat["stat_rate"],
POWER_USAGE_DEVICE_CLASSES,
POWER_USAGE_UNITS,
POWER_UNIT_ERROR,
source_result,
)
)
elif source["type"] == "gas":
_validate_gas_source(
hass,
source,
statistics_metadata,
wanted_statistics_metadata,
source_result,
validate_calls,
wanted_statistics_metadata.add(source["stat_energy_from"])
validate_calls.append(
functools.partial(
_async_validate_usage_stat,
hass,
statistics_metadata,
source["stat_energy_from"],
GAS_USAGE_DEVICE_CLASSES,
GAS_USAGE_UNITS,
GAS_UNIT_ERROR,
source_result,
)
)
if (stat_cost := source.get("stat_cost")) is not None:
wanted_statistics_metadata.add(stat_cost)
validate_calls.append(
functools.partial(
_async_validate_cost_stat,
hass,
statistics_metadata,
stat_cost,
source_result,
)
)
elif (entity_energy_price := source.get("entity_energy_price")) is not None:
validate_calls.append(
functools.partial(
_async_validate_price_entity,
hass,
entity_energy_price,
source_result,
GAS_PRICE_UNITS,
GAS_PRICE_UNIT_ERROR,
)
)
if (
source.get("entity_energy_price") is not None
or source.get("number_energy_price") is not None
):
validate_calls.append(
functools.partial(
_async_validate_auto_generated_cost_entity,
hass,
source["stat_energy_from"],
source_result,
)
)
elif source["type"] == "water":
_validate_water_source(
hass,
source,
statistics_metadata,
wanted_statistics_metadata,
source_result,
validate_calls,
wanted_statistics_metadata.add(source["stat_energy_from"])
validate_calls.append(
functools.partial(
_async_validate_usage_stat,
hass,
statistics_metadata,
source["stat_energy_from"],
WATER_USAGE_DEVICE_CLASSES,
WATER_USAGE_UNITS,
WATER_UNIT_ERROR,
source_result,
)
)
if (stat_cost := source.get("stat_cost")) is not None:
wanted_statistics_metadata.add(stat_cost)
validate_calls.append(
functools.partial(
_async_validate_cost_stat,
hass,
statistics_metadata,
stat_cost,
source_result,
)
)
elif (entity_energy_price := source.get("entity_energy_price")) is not None:
validate_calls.append(
functools.partial(
_async_validate_price_entity,
hass,
entity_energy_price,
source_result,
WATER_PRICE_UNITS,
WATER_PRICE_UNIT_ERROR,
)
)
if (
source.get("entity_energy_price") is not None
or source.get("number_energy_price") is not None
):
validate_calls.append(
functools.partial(
_async_validate_auto_generated_cost_entity,
hass,
source["stat_energy_from"],
source_result,
)
)
elif source["type"] == "solar":
wanted_statistics_metadata.add(source["stat_energy_from"])
validate_calls.append(

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,7 +3,6 @@
from __future__ import annotations
import asyncio
from urllib.parse import quote
import voluptuous as vol
@@ -153,9 +152,7 @@ class HassFoscamCamera(FoscamEntity, Camera):
async def stream_source(self) -> str | None:
"""Return the stream source."""
if self._rtsp_port:
_username = quote(self._username)
_password = quote(self._password)
return f"rtsp://{_username}:{_password}@{self._foscam_session.host}:{self._rtsp_port}/video{self._stream}"
return f"rtsp://{self._username}:{self._password}@{self._foscam_session.host}:{self._rtsp_port}/video{self._stream}"
return None

View File

@@ -6,12 +6,6 @@
"dependencies": ["hardware", "usb", "homeassistant_hardware"],
"documentation": "https://www.home-assistant.io/integrations/homeassistant_connect_zbt2",
"integration_type": "hardware",
"loggers": [
"bellows",
"universal_silabs_flasher",
"zigpy.serial",
"serial_asyncio_fast"
],
"quality_scale": "bronze",
"usb": [
{

View File

@@ -6,12 +6,6 @@
"dependencies": ["hardware", "usb", "homeassistant_hardware"],
"documentation": "https://www.home-assistant.io/integrations/homeassistant_sky_connect",
"integration_type": "hardware",
"loggers": [
"bellows",
"universal_silabs_flasher",
"zigpy.serial",
"serial_asyncio_fast"
],
"usb": [
{
"description": "*skyconnect v1.0*",

View File

@@ -7,11 +7,5 @@
"dependencies": ["hardware", "homeassistant_hardware"],
"documentation": "https://www.home-assistant.io/integrations/homeassistant_yellow",
"integration_type": "hardware",
"loggers": [
"bellows",
"universal_silabs_flasher",
"zigpy.serial",
"serial_asyncio_fast"
],
"single_config_entry": true
}

View File

@@ -37,5 +37,5 @@
"iot_class": "cloud_push",
"loggers": ["pylamarzocco"],
"quality_scale": "platinum",
"requirements": ["pylamarzocco==2.1.3"]
"requirements": ["pylamarzocco==2.1.2"]
}

View File

@@ -22,7 +22,7 @@ _LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Ruuvi BLE device from a config entry."""
"""Set up Ruuvitag BLE device from a config entry."""
address = entry.unique_id
assert address is not None
data = RuuvitagBluetoothDeviceData()

View File

@@ -1,6 +1,6 @@
{
"domain": "ruuvitag_ble",
"name": "Ruuvi BLE",
"name": "RuuviTag BLE",
"bluetooth": [
{
"connectable": false,

View File

@@ -191,7 +191,7 @@ async def async_setup_entry(
entry: config_entries.ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Ruuvi BLE sensors."""
"""Set up the Ruuvitag BLE sensors."""
coordinator: PassiveBluetoothProcessorCoordinator = hass.data[DOMAIN][
entry.entry_id
]
@@ -210,7 +210,7 @@ class RuuvitagBluetoothSensorEntity(
],
SensorEntity,
):
"""Representation of a Ruuvi BLE sensor."""
"""Representation of a Ruuvitag BLE sensor."""
@property
def native_value(self) -> int | float | None:

View File

@@ -13,19 +13,20 @@ from homeassistant.components.alarm_control_panel import (
AlarmControlPanelState,
CodeFormat,
)
from homeassistant.config_entries import ConfigSubentry
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import (
CONF_ARM_HOME_MODE,
CONF_PARTITION_NUMBER,
DOMAIN,
SIGNAL_PANEL_MESSAGE,
SUBENTRY_TYPE_PARTITION,
SatelConfigEntry,
)
from .entity import SatelIntegraEntity
ALARM_STATE_MAP = {
AlarmState.TRIGGERED: AlarmControlPanelState.TRIGGERED,
@@ -58,48 +59,53 @@ async def async_setup_entry(
for subentry in partition_subentries:
partition_num: int = subentry.data[CONF_PARTITION_NUMBER]
zone_name: str = subentry.data[CONF_NAME]
arm_home_mode: int = subentry.data[CONF_ARM_HOME_MODE]
async_add_entities(
[
SatelIntegraAlarmPanel(
controller,
config_entry.entry_id,
subentry,
partition_num,
zone_name,
arm_home_mode,
partition_num,
config_entry.entry_id,
)
],
config_subentry_id=subentry.subentry_id,
)
class SatelIntegraAlarmPanel(SatelIntegraEntity, AlarmControlPanelEntity):
class SatelIntegraAlarmPanel(AlarmControlPanelEntity):
"""Representation of an AlarmDecoder-based alarm panel."""
_attr_code_format = CodeFormat.NUMBER
_attr_should_poll = False
_attr_supported_features = (
AlarmControlPanelEntityFeature.ARM_HOME
| AlarmControlPanelEntityFeature.ARM_AWAY
)
_attr_has_entity_name = True
_attr_name = None
def __init__(
self,
controller: AsyncSatel,
config_entry_id: str,
subentry: ConfigSubentry,
device_number: int,
device_name: str,
arm_home_mode: int,
partition_id: int,
config_entry_id: str,
) -> None:
"""Initialize the alarm panel."""
super().__init__(
controller,
config_entry_id,
subentry,
device_number,
)
self._attr_unique_id = f"{config_entry_id}_alarm_panel_{partition_id}"
self._arm_home_mode = arm_home_mode
self._partition_id = partition_id
self._satel = controller
self._attr_device_info = DeviceInfo(
name=device_name, identifiers={(DOMAIN, self._attr_unique_id)}
)
async def async_added_to_hass(self) -> None:
"""Update alarm status and register callbacks for future updates."""
@@ -130,7 +136,7 @@ class SatelIntegraAlarmPanel(SatelIntegraEntity, AlarmControlPanelEntity):
for satel_state, ha_state in ALARM_STATE_MAP.items():
if (
satel_state in self._satel.partition_states
and self._device_number in self._satel.partition_states[satel_state]
and self._partition_id in self._satel.partition_states[satel_state]
):
return ha_state
@@ -146,21 +152,21 @@ class SatelIntegraAlarmPanel(SatelIntegraEntity, AlarmControlPanelEntity):
self._attr_alarm_state == AlarmControlPanelState.TRIGGERED
)
await self._satel.disarm(code, [self._device_number])
await self._satel.disarm(code, [self._partition_id])
if clear_alarm_necessary:
# Wait 1s before clearing the alarm
await asyncio.sleep(1)
await self._satel.clear_alarm(code, [self._device_number])
await self._satel.clear_alarm(code, [self._partition_id])
async def async_alarm_arm_away(self, code: str | None = None) -> None:
"""Send arm away command."""
if code:
await self._satel.arm(code, [self._device_number])
await self._satel.arm(code, [self._partition_id])
async def async_alarm_arm_home(self, code: str | None = None) -> None:
"""Send arm home command."""
if code:
await self._satel.arm(code, [self._device_number], self._arm_home_mode)
await self._satel.arm(code, [self._partition_id], self._arm_home_mode)

View File

@@ -8,22 +8,25 @@ from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.config_entries import ConfigSubentry
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import (
CONF_OUTPUT_NUMBER,
CONF_OUTPUTS,
CONF_ZONE_NUMBER,
CONF_ZONE_TYPE,
CONF_ZONES,
DOMAIN,
SIGNAL_OUTPUTS_UPDATED,
SIGNAL_ZONES_UPDATED,
SUBENTRY_TYPE_OUTPUT,
SUBENTRY_TYPE_ZONE,
SatelConfigEntry,
)
from .entity import SatelIntegraEntity
async def async_setup_entry(
@@ -43,16 +46,18 @@ async def async_setup_entry(
for subentry in zone_subentries:
zone_num: int = subentry.data[CONF_ZONE_NUMBER]
zone_type: BinarySensorDeviceClass = subentry.data[CONF_ZONE_TYPE]
zone_name: str = subentry.data[CONF_NAME]
async_add_entities(
[
SatelIntegraBinarySensor(
controller,
config_entry.entry_id,
subentry,
zone_num,
zone_name,
zone_type,
CONF_ZONES,
SIGNAL_ZONES_UPDATED,
config_entry.entry_id,
)
],
config_subentry_id=subentry.subentry_id,
@@ -66,44 +71,51 @@ async def async_setup_entry(
for subentry in output_subentries:
output_num: int = subentry.data[CONF_OUTPUT_NUMBER]
ouput_type: BinarySensorDeviceClass = subentry.data[CONF_ZONE_TYPE]
output_name: str = subentry.data[CONF_NAME]
async_add_entities(
[
SatelIntegraBinarySensor(
controller,
config_entry.entry_id,
subentry,
output_num,
output_name,
ouput_type,
CONF_OUTPUTS,
SIGNAL_OUTPUTS_UPDATED,
config_entry.entry_id,
)
],
config_subentry_id=subentry.subentry_id,
)
class SatelIntegraBinarySensor(SatelIntegraEntity, BinarySensorEntity):
class SatelIntegraBinarySensor(BinarySensorEntity):
"""Representation of an Satel Integra binary sensor."""
_attr_should_poll = False
_attr_has_entity_name = True
_attr_name = None
def __init__(
self,
controller: AsyncSatel,
config_entry_id: str,
subentry: ConfigSubentry,
device_number: int,
device_name: str,
device_class: BinarySensorDeviceClass,
sensor_type: str,
react_to_signal: str,
config_entry_id: str,
) -> None:
"""Initialize the binary_sensor."""
super().__init__(
controller,
config_entry_id,
subentry,
device_number,
)
self._device_number = device_number
self._attr_unique_id = f"{config_entry_id}_{sensor_type}_{device_number}"
self._react_to_signal = react_to_signal
self._satel = controller
self._attr_device_class = device_class
self._react_to_signal = react_to_signal
self._attr_device_info = DeviceInfo(
name=device_name, identifiers={(DOMAIN, self._attr_unique_id)}
)
async def async_added_to_hass(self) -> None:
"""Register callbacks."""

View File

@@ -1,58 +0,0 @@
"""Satel Integra base entity."""
from __future__ import annotations
from typing import TYPE_CHECKING
from satel_integra.satel_integra import AsyncSatel
from homeassistant.config_entries import ConfigSubentry
from homeassistant.const import CONF_NAME
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import Entity
from .const import (
DOMAIN,
SUBENTRY_TYPE_OUTPUT,
SUBENTRY_TYPE_PARTITION,
SUBENTRY_TYPE_SWITCHABLE_OUTPUT,
SUBENTRY_TYPE_ZONE,
)
SubentryTypeToEntityType: dict[str, str] = {
SUBENTRY_TYPE_PARTITION: "alarm_panel",
SUBENTRY_TYPE_SWITCHABLE_OUTPUT: "switch",
SUBENTRY_TYPE_ZONE: "zones",
SUBENTRY_TYPE_OUTPUT: "outputs",
}
class SatelIntegraEntity(Entity):
"""Defines a base Satel Integra entity."""
_attr_should_poll = False
_attr_has_entity_name = True
_attr_name = None
def __init__(
self,
controller: AsyncSatel,
config_entry_id: str,
subentry: ConfigSubentry,
device_number: int,
) -> None:
"""Initialize the Satel Integra entity."""
self._satel = controller
self._device_number = device_number
entity_type = SubentryTypeToEntityType[subentry.subentry_type]
if TYPE_CHECKING:
assert entity_type is not None
self._attr_unique_id = f"{config_entry_id}_{entity_type}_{device_number}"
self._attr_device_info = DeviceInfo(
name=subentry.data[CONF_NAME], identifiers={(DOMAIN, self._attr_unique_id)}
)

View File

@@ -7,19 +7,19 @@ from typing import Any
from satel_integra.satel_integra import AsyncSatel
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigSubentry
from homeassistant.const import CONF_CODE
from homeassistant.const import CONF_CODE, CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import (
CONF_SWITCHABLE_OUTPUT_NUMBER,
DOMAIN,
SIGNAL_OUTPUTS_UPDATED,
SUBENTRY_TYPE_SWITCHABLE_OUTPUT,
SatelConfigEntry,
)
from .entity import SatelIntegraEntity
async def async_setup_entry(
@@ -38,41 +38,46 @@ async def async_setup_entry(
for subentry in switchable_output_subentries:
switchable_output_num: int = subentry.data[CONF_SWITCHABLE_OUTPUT_NUMBER]
switchable_output_name: str = subentry.data[CONF_NAME]
async_add_entities(
[
SatelIntegraSwitch(
controller,
config_entry.entry_id,
subentry,
switchable_output_num,
switchable_output_name,
config_entry.options.get(CONF_CODE),
config_entry.entry_id,
),
],
config_subentry_id=subentry.subentry_id,
)
class SatelIntegraSwitch(SatelIntegraEntity, SwitchEntity):
"""Representation of an Satel Integra switch."""
class SatelIntegraSwitch(SwitchEntity):
"""Representation of an Satel switch."""
_attr_should_poll = False
_attr_has_entity_name = True
_attr_name = None
def __init__(
self,
controller: AsyncSatel,
config_entry_id: str,
subentry: ConfigSubentry,
device_number: int,
device_name: str,
code: str | None,
config_entry_id: str,
) -> None:
"""Initialize the switch."""
super().__init__(
controller,
config_entry_id,
subentry,
device_number,
)
self._device_number = device_number
self._attr_unique_id = f"{config_entry_id}_switch_{device_number}"
self._code = code
self._satel = controller
self._attr_device_info = DeviceInfo(
name=device_name, identifiers={(DOMAIN, self._attr_unique_id)}
)
async def async_added_to_hass(self) -> None:
"""Register callbacks."""

View File

@@ -3,17 +3,15 @@
from __future__ import annotations
from datetime import timedelta
from http import HTTPStatus
import logging
from aiosenz import SENZAPI, Thermostat
from httpx import HTTPStatusError, RequestError
import jwt
from httpx import RequestError
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 +58,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,
@@ -97,27 +82,3 @@ async def async_setup_entry(hass: HomeAssistant, entry: SENZConfigEntry) -> bool
async def async_unload_entry(hass: HomeAssistant, entry: SENZConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def async_migrate_entry(
hass: HomeAssistant, config_entry: SENZConfigEntry
) -> bool:
"""Migrate old entry."""
# Use sub(ject) from access_token as unique_id
if config_entry.version == 1 and config_entry.minor_version == 1:
token = jwt.decode(
config_entry.data["token"]["access_token"],
options={"verify_signature": False},
)
uid = token["sub"]
hass.config_entries.async_update_entry(
config_entry, unique_id=uid, minor_version=2
)
_LOGGER.info(
"Migration to version %s.%s successful",
config_entry.version,
config_entry.minor_version,
)
return True

View File

@@ -1,16 +1,7 @@
"""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.helpers import config_entry_oauth2_flow
from .const import DOMAIN
@@ -21,8 +12,6 @@ class OAuth2FlowHandler(
):
"""Config flow to handle SENZ OAuth2 authentication."""
VERSION = 1
MINOR_VERSION = 2
DOMAIN = DOMAIN
@property
@@ -34,49 +23,3 @@ class OAuth2FlowHandler(
def extra_authorize_data(self) -> dict:
"""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."""
token = jwt.decode(
data["token"]["access_token"], options={"verify_signature": False}
)
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

@@ -219,6 +219,7 @@ class AbstractTemplateAlarmControlPanel(
self._attr_code_arm_required: bool = config[CONF_CODE_ARM_REQUIRED]
self._attr_code_format = config[CONF_CODE_FORMAT].value
self._state: AlarmControlPanelState | None = None
self._attr_supported_features: AlarmControlPanelEntityFeature = (
AlarmControlPanelEntityFeature(0)
)
@@ -243,6 +244,11 @@ class AbstractTemplateAlarmControlPanel(
if (action_config := config.get(action_id)) is not None:
yield (action_id, action_config, supported_feature)
@property
def alarm_state(self) -> AlarmControlPanelState | None:
"""Return the state of the device."""
return self._state
async def _async_handle_restored_state(self) -> None:
if (
(last_state := await self.async_get_last_state()) is not None
@@ -250,14 +256,14 @@ class AbstractTemplateAlarmControlPanel(
and last_state.state in _VALID_STATES
# The trigger might have fired already while we waited for stored data,
# then we should not restore state
and self._attr_alarm_state is None
and self._state is None
):
self._attr_alarm_state = AlarmControlPanelState(last_state.state)
self._state = AlarmControlPanelState(last_state.state)
def _handle_state(self, result: Any) -> None:
# Validate state
if result in _VALID_STATES:
self._attr_alarm_state = result
self._state = result
_LOGGER.debug("Valid state - %s", result)
return
@@ -267,7 +273,7 @@ class AbstractTemplateAlarmControlPanel(
self.entity_id,
", ".join(_VALID_STATES),
)
self._attr_alarm_state = None
self._state = None
async def _async_alarm_arm(self, state: Any, script: Script | None, code: Any):
"""Arm the panel to specified state with supplied script."""
@@ -278,7 +284,7 @@ class AbstractTemplateAlarmControlPanel(
)
if self._attr_assumed_state:
self._attr_alarm_state = state
self._state = state
self.async_write_ha_state()
async def async_alarm_arm_away(self, code: str | None = None) -> None:
@@ -370,7 +376,7 @@ class StateAlarmControlPanelEntity(TemplateEntity, AbstractTemplateAlarmControlP
@callback
def _update_state(self, result):
if isinstance(result, TemplateError):
self._attr_alarm_state = None
self._state = None
return
self._handle_state(result)
@@ -380,7 +386,7 @@ class StateAlarmControlPanelEntity(TemplateEntity, AbstractTemplateAlarmControlP
"""Set up templates."""
if self._template:
self.add_template_attribute(
"_attr_alarm_state", self._template, None, self._update_state
"_state", self._template, None, self._update_state
)
super()._async_setup_templates()

View File

@@ -709,7 +709,6 @@ class DPCode(StrEnum):
DEW_POINT_TEMP = "dew_point_temp"
DISINFECTION = "disinfection"
DO_NOT_DISTURB = "do_not_disturb"
DOORBELL_PIC = "doorbell_pic"
DOORCONTACT_STATE = "doorcontact_state" # Status of door window sensor
DOORCONTACT_STATE_2 = "doorcontact_state_2"
DOORCONTACT_STATE_3 = "doorcontact_state_3"

View File

@@ -15,13 +15,6 @@ from homeassistant.util import dt as dt_util
from . import TuyaConfigEntry
from .const import DOMAIN, DPCode
_REDACTED_DPCODES = {
DPCode.ALARM_MESSAGE,
DPCode.ALARM_MSG,
DPCode.DOORBELL_PIC,
DPCode.MOVEMENT_DETECT_PIC,
}
async def async_get_config_entry_diagnostics(
hass: HomeAssistant, entry: TuyaConfigEntry
@@ -102,7 +95,7 @@ def _async_device_as_dict(
# Gather Tuya states
for dpcode, value in device.status.items():
# These statuses may contain sensitive information, redact these..
if dpcode in _REDACTED_DPCODES:
if dpcode in {DPCode.ALARM_MESSAGE, DPCode.MOVEMENT_DETECT_PIC}:
data["status"][dpcode] = REDACTED
continue

View File

@@ -56,32 +56,37 @@ class VeluxCover(VeluxEntity, CoverEntity):
def __init__(self, node: OpeningDevice, config_entry_id: str) -> None:
"""Initialize VeluxCover."""
super().__init__(node, config_entry_id)
# Features common to all covers
self._attr_supported_features = (
CoverEntityFeature.OPEN
| CoverEntityFeature.CLOSE
| CoverEntityFeature.SET_POSITION
| CoverEntityFeature.STOP
)
# Window is the default device class for covers
self._attr_device_class = CoverDeviceClass.WINDOW
if isinstance(node, Awning):
self._attr_device_class = CoverDeviceClass.AWNING
if isinstance(node, Blind):
self._attr_device_class = CoverDeviceClass.BLIND
self._is_blind = True
if isinstance(node, GarageDoor):
self._attr_device_class = CoverDeviceClass.GARAGE
if isinstance(node, Gate):
self._attr_device_class = CoverDeviceClass.GATE
if isinstance(node, RollerShutter):
self._attr_device_class = CoverDeviceClass.SHUTTER
if isinstance(node, Blind):
self._attr_device_class = CoverDeviceClass.BLIND
self._is_blind = True
self._attr_supported_features |= (
@property
def supported_features(self) -> CoverEntityFeature:
"""Flag supported features."""
supported_features = (
CoverEntityFeature.OPEN
| CoverEntityFeature.CLOSE
| CoverEntityFeature.SET_POSITION
| CoverEntityFeature.STOP
)
if self.current_cover_tilt_position is not None:
supported_features |= (
CoverEntityFeature.OPEN_TILT
| CoverEntityFeature.CLOSE_TILT
| CoverEntityFeature.SET_TILT_POSITION
| CoverEntityFeature.STOP_TILT
)
return supported_features
@property
def current_cover_position(self) -> int:

View File

@@ -59,7 +59,7 @@ from .utils import (
get_burners,
get_circuits,
get_compressors,
get_condensers,
get_condensors,
get_device_serial,
get_evaporators,
is_supported,
@@ -1237,10 +1237,10 @@ COMPRESSOR_SENSORS: tuple[ViCareSensorEntityDescription, ...] = (
),
)
CONDENSER_SENSORS: tuple[ViCareSensorEntityDescription, ...] = (
CONDENSOR_SENSORS: tuple[ViCareSensorEntityDescription, ...] = (
ViCareSensorEntityDescription(
key="condenser_liquid_temperature",
translation_key="condenser_liquid_temperature",
key="condensor_liquid_temperature",
translation_key="condensor_liquid_temperature",
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
value_getter=lambda api: api.getCondensorLiquidTemperature(),
@@ -1248,8 +1248,8 @@ CONDENSER_SENSORS: tuple[ViCareSensorEntityDescription, ...] = (
entity_registry_enabled_default=False,
),
ViCareSensorEntityDescription(
key="condenser_subcooling_temperature",
translation_key="condenser_subcooling_temperature",
key="condensor_subcooling_temperature",
translation_key="condensor_subcooling_temperature",
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
value_getter=lambda api: api.getCondensorSubcoolingTemperature(),
@@ -1303,7 +1303,7 @@ def _build_entities(
(get_circuits(device.api), CIRCUIT_SENSORS),
(get_burners(device.api), BURNER_SENSORS),
(get_compressors(device.api), COMPRESSOR_SENSORS),
(get_condensers(device.api), CONDENSER_SENSORS),
(get_condensors(device.api), CONDENSOR_SENSORS),
(get_evaporators(device.api), EVAPORATOR_SENSORS),
):
entities.extend(

View File

@@ -244,11 +244,11 @@
"compressor_starts": {
"name": "Compressor starts"
},
"condenser_liquid_temperature": {
"name": "Condenser liquid temperature"
"condensor_liquid_temperature": {
"name": "Condensor liquid temperature"
},
"condenser_subcooling_temperature": {
"name": "Condenser subcooling temperature"
"condensor_subcooling_temperature": {
"name": "Condensor subcooling temperature"
},
"dhw_storage_bottom_temperature": {
"name": "DHW storage bottom temperature"

View File

@@ -130,14 +130,14 @@ def get_compressors(device: PyViCareDevice) -> list[PyViCareHeatingDeviceCompone
return []
def get_condensers(device: PyViCareDevice) -> list[PyViCareHeatingDeviceComponent]:
"""Return the list of condensers."""
def get_condensors(device: PyViCareDevice) -> list[PyViCareHeatingDeviceComponent]:
"""Return the list of condensors."""
try:
return device.condensors
except PyViCareNotSupportedFeatureError:
_LOGGER.debug("No condensers found")
_LOGGER.debug("No condensors found")
except AttributeError as error:
_LOGGER.debug("No condensers found: %s", error)
_LOGGER.debug("No condensors found: %s", error)
return []

View File

@@ -71,6 +71,8 @@ class ZWaveBaseEntity(Entity):
)
# Entity class attributes
self._attr_name = self.generate_name()
self._attr_unique_id = get_unique_id(driver, self.info.primary_value.value_id)
if isinstance(info, NewZwaveDiscoveryInfo):
self.entity_description = info.entity_description
else:
@@ -78,8 +80,6 @@ class ZWaveBaseEntity(Entity):
self._attr_entity_registry_enabled_default = enabled_default
if (entity_category := info.entity_category) is not None:
self._attr_entity_category = entity_category
self._attr_name = self.generate_name()
self._attr_unique_id = get_unique_id(driver, self.info.primary_value.value_id)
self._attr_assumed_state = self.info.assumed_state
# device is precreated in main handler
self._attr_device_info = DeviceInfo(

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",
@@ -5695,7 +5689,7 @@
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_push",
"name": "Ruuvi BLE"
"name": "RuuviTag BLE"
}
}
},

View File

@@ -25,6 +25,7 @@ from homeassistant.const import (
ATTR_ASSUMED_STATE,
ATTR_ATTRIBUTION,
ATTR_DEVICE_CLASS,
ATTR_ENTITY_ID,
ATTR_ENTITY_PICTURE,
ATTR_FRIENDLY_NAME,
ATTR_ICON,
@@ -417,6 +418,7 @@ CACHED_PROPERTIES_WITH_ATTR_ = {
"extra_state_attributes",
"force_update",
"icon",
"included_unique_ids",
"name",
"should_poll",
"state",
@@ -524,6 +526,9 @@ class Entity(
__capabilities_updated_at_reported: bool = False
__remove_future: asyncio.Future[None] | None = None
# A list of included entity IDs in case the entity represents a group
_included_entities: list[str] | None = None
# Entity Properties
_attr_assumed_state: bool = False
_attr_attribution: str | None = None
@@ -539,6 +544,7 @@ class Entity(
_attr_extra_state_attributes: dict[str, Any]
_attr_force_update: bool
_attr_icon: str | None
_attr_included_unique_ids: list[str]
_attr_name: str | None
_attr_should_poll: bool = True
_attr_state: StateType = STATE_UNKNOWN
@@ -1085,6 +1091,21 @@ class Entity(
available = self.available # only call self.available once per update cycle
state = self._stringify_state(available)
if available:
if self.included_unique_ids is not None:
entity_registry = er.async_get(self.hass)
self._included_entities = [
entity_id
for included_id in self.included_unique_ids
if (
entity_id := entity_registry.async_get_entity_id(
self.platform.domain,
self.platform.platform_name,
included_id,
)
)
is not None
]
attr[ATTR_ENTITY_ID] = self._included_entities.copy()
if state_attributes := self.state_attributes:
attr |= state_attributes
if extra_state_attributes := self.extra_state_attributes:
@@ -1374,6 +1395,30 @@ class Entity(
async def add_to_platform_finish(self) -> None:
"""Finish adding an entity to a platform."""
entity_registry = er.async_get(self.hass)
async def _handle_entity_registry_updated(event: Event[Any]) -> None:
"""Handle registry create or update event."""
if (
event.data["action"] in {"create", "update"}
and (entry := entity_registry.async_get(event.data["entity_id"]))
and self.included_unique_ids is not None
and entry.unique_id in self.included_unique_ids
) or (
event.data["action"] == "remove"
and self._included_entities is not None
and event.data["entity_id"] in self._included_entities
):
self.async_write_ha_state()
if self.included_unique_ids is not None:
self.async_on_remove(
self.hass.bus.async_listen(
er.EVENT_ENTITY_REGISTRY_UPDATED,
_handle_entity_registry_updated,
)
)
await self.async_internal_added_to_hass()
await self.async_added_to_hass()
self._platform_state = EntityPlatformState.ADDED
@@ -1633,6 +1678,16 @@ class Entity(
self.hass, integration_domain=platform_name, module=type(self).__module__
)
@cached_property
def included_unique_ids(self) -> list[str] | None:
"""Return the list of unique IDs if the entity represents a group.
The corresponding entities will be shown as members in the UI.
"""
if hasattr(self, "_attr_included_unique_ids"):
return self._attr_included_unique_ids
return None
class ToggleEntityDescription(EntityDescription, frozen_or_thawed=True):
"""A class that describes toggle entities."""

7
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
@@ -2131,7 +2128,7 @@ pykwb==0.0.8
pylacrosse==0.4
# homeassistant.components.lamarzocco
pylamarzocco==2.1.3
pylamarzocco==2.1.2
# homeassistant.components.lastfm
pylast==5.1.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
@@ -1775,7 +1772,7 @@ pykrakenapi==0.1.8
pykulersky==0.5.8
# homeassistant.components.lamarzocco
pylamarzocco==2.1.3
pylamarzocco==2.1.2
# homeassistant.components.lastfm
pylast==5.1.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",

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

@@ -1,7 +1,4 @@
"""The tests for LG Netcast device triggers."""
from collections.abc import Generator
from unittest.mock import patch
"""The tests for LG NEtcast device triggers."""
import pytest
@@ -22,13 +19,6 @@ from . import ENTITY_ID, UNIQUE_ID, setup_lgnetcast
from tests.common import MockConfigEntry, async_get_device_automations
@pytest.fixture(autouse=True)
def mock_lg_netcast() -> Generator[None]:
"""Mock LG Netcast library."""
with patch("homeassistant.components.lg_netcast.LgNetCastClient"):
yield
async def test_get_triggers(
hass: HomeAssistant, device_registry: dr.DeviceRegistry
) -> None:

View File

@@ -1,6 +1,5 @@
"""The tests for LG Netcast device triggers."""
from collections.abc import Generator
from unittest.mock import patch
import pytest
@@ -18,13 +17,6 @@ from . import ENTITY_ID, UNIQUE_ID, setup_lgnetcast
from tests.common import MockEntity, MockEntityPlatform
@pytest.fixture(autouse=True)
def mock_lg_netcast() -> Generator[None]:
"""Mock LG Netcast library."""
with patch("homeassistant.components.lg_netcast.LgNetCastClient"):
yield
async def test_lg_netcast_turn_on_trigger_device_id(
hass: HomeAssistant,
service_calls: list[ServiceCall],

View File

@@ -1 +1 @@
"""Test package for Ruuvi BLE sensor integration."""
"""Test package for RuuviTag BLE sensor integration."""

View File

@@ -1,4 +1,4 @@
"""Fixtures for testing Ruuvi BLE."""
"""Fixtures for testing RuuviTag BLE."""
from homeassistant.helpers.service_info.bluetooth import BluetoothServiceInfo

View File

@@ -1,4 +1,4 @@
"""Test the Ruuvi BLE sensors."""
"""Test the Ruuvitag BLE sensors."""
from __future__ import annotations
@@ -35,7 +35,7 @@ async def test_sensors(
snapshot: SnapshotAssertion,
service_info: BluetoothServiceInfo,
) -> None:
"""Test the Ruuvi BLE sensors."""
"""Test the RuuviTag BLE sensors."""
entry = MockConfigEntry(domain=DOMAIN, unique_id=service_info.address)
entry.add_to_hass(hass)

View File

@@ -14,10 +14,9 @@ from homeassistant.components.application_credentials import (
)
from homeassistant.components.senz.const import DOMAIN
from homeassistant.core import HomeAssistant
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,
@@ -64,7 +63,7 @@ def mock_expires_at() -> float:
def mock_config_entry(hass: HomeAssistant, expires_at: float) -> MockConfigEntry:
"""Return the default mocked config entry."""
config_entry = MockConfigEntry(
minor_version=2,
minor_version=1,
domain=DOMAIN,
title="Senz test",
data={
@@ -78,7 +77,6 @@ def mock_config_entry(hass: HomeAssistant, expires_at: float) -> MockConfigEntry
},
},
entry_id="senz_test",
unique_id=ENTRY_UNIQUE_ID,
)
config_entry.add_to_hass(hass)
return config_entry
@@ -111,26 +109,3 @@ async def setup_credentials(hass: HomeAssistant) -> None:
),
DOMAIN,
)
@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:
"""Return a valid access token."""
return config_entry_oauth2_flow._encode_jwt(
hass,
{
"sub": unique_id,
"aud": [],
"scp": [
"rest_api",
"offline_access",
],
"ou_code": "NA",
},
)

View File

@@ -2,5 +2,3 @@
CLIENT_ID = "test_client_id"
CLIENT_SECRET = "test_client_secret"
ENTRY_UNIQUE_ID = "test_unique_id"

View File

@@ -12,26 +12,20 @@ from homeassistant.components.application_credentials import (
)
from homeassistant.components.senz.const import DOMAIN
from homeassistant.core import HomeAssistant
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(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
access_token: str,
) -> None:
"""Check full flow."""
await async_setup_component(hass, DOMAIN, {})
@@ -48,18 +42,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"
@@ -67,7 +61,7 @@ async def test_full_flow(
TOKEN_ENDPOINT,
json={
"refresh_token": "mock-refresh-token",
"access_token": access_token,
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
},
@@ -80,202 +74,3 @@ async def test_full_flow(
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert len(mock_setup.mock_calls) == 1
@pytest.mark.usefixtures("current_request_with_host")
async def test_duplicate_flow(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_config_entry: MockConfigEntry,
access_token: str,
) -> None:
"""Check full flow with duplicate entry."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
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": "mock-refresh-token",
"access_token": access_token,
"type": "Bearer",
"expires_in": 60,
},
)
with patch("homeassistant.components.senz.async_setup_entry", return_value=True):
result2 = await hass.config_entries.flow.async_configure(result["flow_id"])
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,14 +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
from homeassistant.helpers.config_entry_oauth2_flow import (
@@ -16,10 +9,8 @@ from homeassistant.helpers.config_entry_oauth2_flow import (
)
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(
@@ -52,110 +43,3 @@ async def test_oauth_implementation_not_available(
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_migrate_config_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_senz_client: MagicMock,
expires_at: float,
access_token: str,
) -> None:
"""Test migration of config entry."""
mock_entry_v1_1 = MockConfigEntry(
version=1,
minor_version=1,
domain=DOMAIN,
title="SENZ test",
data={
"auth_implementation": DOMAIN,
"token": {
"access_token": access_token,
"scope": "rest_api offline_access",
"expires_in": 86399,
"refresh_token": "3012bc9f-7a65-4240-b817-9154ffdcc30f",
"token_type": "Bearer",
"expires_at": expires_at,
},
},
entry_id="senz_test",
)
await setup_integration(hass, mock_entry_v1_1)
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

View File

@@ -192,7 +192,7 @@ async def _create_device(hass: HomeAssistant, mock_device_code: str) -> Customer
device.function = {
key: DeviceFunction(
code=key,
code=value.get("code"),
type=value["type"],
values=json_dumps(value["value"]),
)
@@ -200,7 +200,7 @@ async def _create_device(hass: HomeAssistant, mock_device_code: str) -> Customer
}
device.status_range = {
key: DeviceStatusRange(
code=key,
code=value.get("code"),
type=value["type"],
values=json_dumps(value["value"]),
)

View File

@@ -211,7 +211,7 @@
"switch_kb_light": false,
"telnet_state": "sim_card_no",
"muffling": false,
"alarm_msg": "**REDACTED**",
"alarm_msg": "AFMAZQBuAHMAbwByACAATABvAHcAIABCAGEAdAB0AGUAcgB5AAoAWgBvAG4AZQA6ADAAMAA1AEUAbgB0AHIAYQBuAGMAZQ==",
"switch_alarm_propel": true,
"alarm_delay_time": 20,
"master_state": "normal",

View File

@@ -217,7 +217,7 @@
"wireless_lowpower": 10,
"wireless_awake": false,
"pir_switch": 3,
"doorbell_pic": "**REDACTED**",
"doorbell_pic": "",
"basic_device_volume": 51,
"humanoid_filter": false,
"alarm_message": "**REDACTED**",

View File

@@ -120,8 +120,8 @@
"sd_format_state": 0,
"motion_switch": false,
"doorbell_active": "",
"doorbell_pic": "**REDACTED**",
"doorbell_pic": "aHR0cHM6Ly90eS1ldS1zdG9yYWdlMzAtcGljLnMzLmV1LWNlbnRyYWwtMS5hbWF6b25hd3MuY29tL2U0ODYwMy0yMjU2NjYxOC1zempzYjU0ZDE2ZGI0ZTQ3OTAxYS9kZXRlY3QvMTc2MjE5OTIyMS5qcGVnP1gtQW16LVNlY3VyaXR5LVRva2VuPUZ3b0daWEl2WVhkekVLMyUyRiUyRiUyRiUyRiUyRiUyRiUyRiUyRiUyRiUyRndFYURDUmJiZDNWWldORmtsWUliQ0tDQXZCZCUyQnEwY2EzRURkZzdONTJqUDhmWUI3WVNSS0huSDNnRXZDRjh6OHpMSU92bkZrdG1UQWFLVldSNkxsMDlMMTJ6b09wR2ptekwwRGIyR1NRSG1uSmJNZXRhSm9nWlRQeGI4eGdMbTRwVkhidTkyZndib29UVVllMUwycmhNJTJCdiUyQkFtVG9DTVdwWE9sNThXUDVwZDAwSmdIWGlBUzVGWnhndVR5UWNJcmxFeG5JeW4wYzgwa0VRMjlVa3d2VThMRVpDeUtwTFlIRjJlYTElMkYlMkZPaUk2b1hrdVF3TU0lMkZCWHMlMkJYMWVYYWdnJTJGaW1oRUVhJTJCQ1REODUlMkYlMkZlSHVqZm1KRSUyQnIyeERkdmgwSUJPTFMwYWc1Zm9EbyUyRjZpRHpXMHNKZE1tTjdPNVhiMnMwRnM4MUxwWG5wTXdKRFRxbUklMkJFSDVyYzlxT0NHemY1SUZqbnZZMGF3TjY1blVsMWlpeWphVElCaklwZWVva2htU1F6WlBVJTJGdERzRHlGYUJRRXFWNjkyemlGdVluWHozdnlqdHlzOU5JWG1aJTJGd1hRaTglM0QmWC1BbXotQWxnb3JpdGhtPUFXUzQtSE1BQy1TSEEyNTYmWC1BbXotQ3JlZGVudGlhbD1BU0lBVVRQTVVKSkpRTlVFUEozSCUyRjIwMjUxMTAzJTJGZXUtY2VudHJhbC0xJTJGczMlMkZhd3M0X3JlcXVlc3QmWC1BbXotRGF0ZT0yMDI1MTEwM1QxOTQ3MDRaJlgtQW16LUV4cGlyZXM9NjAmWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0JmJ1Y2tldD10eS1ldS1zdG9yYWdlMzAtcGljJnY9MS4wJlgtQW16LVNpZ25hdHVyZT05YTFlZTYyNWVlMGM5NmQ5NzViMjg2OGQxOGNlOTA3YzU0YTExNjgyNGMzZjkwYzI3YTlmNTNjYjNhN2E0MjA0",
"device_restart": false,
"alarm_message": "**REDACTED**"
"alarm_message": "eyJ2IjoiNS4wIiwiZmlsZXMiOlt7ImRhdGEiOiJhMThiNDM0YmJmZDY1NGM3N2UzNTc2MWRlMDgyZTc2OGZjM2JmYmQ2NThlZDAyMGIwZGJhZjQ2OTE1YTEwY2NjZDI5YjUxZTY1YjBkNjJiMzAxNmVlZDU0YjU1MTU1ZjE1NzkwNTk2ZDc2YzgwYWFlOWU3ODQ0N2QwYzFlOWNmNmIzMWRlN2ZiOWQyOWU4ZWEwODhlYzAxOGJhYTRhNWMzZjBlMDFmYThiOTRiNGQzYWVkNDk4ZGIwOTUyOTc1ZWQ5ODY2OTNlNmM1NDMyYWY3YTE5N2FiYTA3ZWE3YjJkZGNmZDRjMzQ2N2Q5ZDAwMmJkMDc4OWQ0OTYzNWI1NzkyIiwia2V5SWQiOiJkZWZhdWx0IiwiaXYiOiJjN2JiMTk2Mjc1MWRmOThhZWRiM2VjMGU3Mjk4MWVmMCJ9XSwiY21kIjoiaXBjX2Rvb3JiZWxsIiwidHlwZSI6ImltYWdlIn0="
}
}

View File

@@ -364,7 +364,7 @@
"record_switch": true,
"record_mode": 1,
"pir_switch": 2,
"doorbell_pic": "**REDACTED**",
"doorbell_pic": "",
"siren_switch": false,
"basic_device_volume": 1,
"motion_tracking": true,

View File

@@ -1,305 +1,4 @@
# serializer version: 1
# name: test_device_diagnostics[mal_gyitctrjj1kefxp2]
dict({
'active_time': '2024-12-02T20:08:56+00:00',
'category': 'mal',
'create_time': '2024-12-02T20:08:56+00:00',
'disabled_by': None,
'disabled_polling': False,
'endpoint': 'https://apigw.tuyaeu.com',
'function': dict({
'alarm_delay_time': dict({
'type': 'Integer',
'value': '{"unit":"s","min":0,"max":999,"scale":0,"step":1}',
}),
'alarm_time': dict({
'type': 'Integer',
'value': '{"unit":"min","min":0,"max":999,"scale":0,"step":1}',
}),
'delay_set': dict({
'type': 'Integer',
'value': '{"unit":"s","min":0,"max":999,"scale":0,"step":1}',
}),
'master_mode': dict({
'type': 'Enum',
'value': '{"range":["disarmed","arm","home","sos"]}',
}),
'master_state': dict({
'type': 'Enum',
'value': '{"range":["normal","alarm"]}',
}),
'muffling': dict({
'type': 'Boolean',
'value': '{}',
}),
'sub_admin': dict({
'type': 'Raw',
'value': '{}',
}),
'sub_class': dict({
'type': 'Enum',
'value': '{"range":["remote_controller","detector"]}',
}),
'switch_alarm_light': dict({
'type': 'Boolean',
'value': '{}',
}),
'switch_alarm_propel': dict({
'type': 'Boolean',
'value': '{}',
}),
'switch_alarm_sound': dict({
'type': 'Boolean',
'value': '{}',
}),
'switch_kb_light': dict({
'type': 'Boolean',
'value': '{}',
}),
'switch_kb_sound': dict({
'type': 'Boolean',
'value': '{}',
}),
'switch_mode_sound': dict({
'type': 'Boolean',
'value': '{}',
}),
}),
'home_assistant': dict({
'disabled': False,
'disabled_by': None,
'entities': list([
dict({
'device_class': None,
'disabled': False,
'disabled_by': None,
'entity_category': None,
'icon': None,
'original_device_class': None,
'original_icon': None,
'state': dict({
'attributes': dict({
'changed_by': None,
'code_arm_required': False,
'code_format': None,
'friendly_name': 'Multifunction alarm',
'supported_features': 11,
}),
'entity_id': 'alarm_control_panel.multifunction_alarm',
'state': 'disarmed',
}),
'unit_of_measurement': None,
}),
dict({
'device_class': None,
'disabled': False,
'disabled_by': None,
'entity_category': 'config',
'icon': None,
'original_device_class': 'duration',
'original_icon': None,
'state': dict({
'attributes': dict({
'device_class': 'duration',
'friendly_name': 'Multifunction alarm Arm delay',
'max': 999.0,
'min': 0.0,
'mode': 'auto',
'step': 1.0,
'unit_of_measurement': 's',
}),
'entity_id': 'number.multifunction_alarm_arm_delay',
'state': '15.0',
}),
'unit_of_measurement': 's',
}),
dict({
'device_class': None,
'disabled': False,
'disabled_by': None,
'entity_category': 'config',
'icon': None,
'original_device_class': 'duration',
'original_icon': None,
'state': dict({
'attributes': dict({
'device_class': 'duration',
'friendly_name': 'Multifunction alarm Alarm delay',
'max': 999.0,
'min': 0.0,
'mode': 'auto',
'step': 1.0,
'unit_of_measurement': 's',
}),
'entity_id': 'number.multifunction_alarm_alarm_delay',
'state': '20.0',
}),
'unit_of_measurement': 's',
}),
dict({
'device_class': None,
'disabled': False,
'disabled_by': None,
'entity_category': 'config',
'icon': None,
'original_device_class': 'duration',
'original_icon': None,
'state': dict({
'attributes': dict({
'device_class': 'duration',
'friendly_name': 'Multifunction alarm Siren duration',
'max': 999.0,
'min': 0.0,
'mode': 'auto',
'step': 1.0,
'unit_of_measurement': 'min',
}),
'entity_id': 'number.multifunction_alarm_siren_duration',
'state': '3.0',
}),
'unit_of_measurement': 'min',
}),
dict({
'device_class': None,
'disabled': False,
'disabled_by': None,
'entity_category': 'config',
'icon': None,
'original_device_class': None,
'original_icon': None,
'state': dict({
'attributes': dict({
'friendly_name': 'Multifunction alarm Arm beep',
}),
'entity_id': 'switch.multifunction_alarm_arm_beep',
'state': 'on',
}),
'unit_of_measurement': None,
}),
dict({
'device_class': None,
'disabled': False,
'disabled_by': None,
'entity_category': 'config',
'icon': None,
'original_device_class': None,
'original_icon': None,
'state': dict({
'attributes': dict({
'friendly_name': 'Multifunction alarm Siren',
}),
'entity_id': 'switch.multifunction_alarm_siren',
'state': 'on',
}),
'unit_of_measurement': None,
}),
]),
'name': 'Multifunction alarm',
'name_by_user': None,
}),
'id': '2pxfek1jjrtctiyglam',
'mqtt_connected': True,
'name': 'Multifunction alarm',
'online': True,
'product_id': 'gyitctrjj1kefxp2',
'product_name': 'Multifunction alarm',
'set_up': True,
'status': dict({
'alarm_delay_time': 20,
'alarm_msg': '**REDACTED**',
'alarm_time': 3,
'delay_set': 15,
'master_mode': 'disarmed',
'master_state': 'normal',
'muffling': False,
'sub_admin': 'AgEFCggC////HABLAGkAdABjAGgAZQBuACAAUwBtAG8AawBlACBjAAL///8gAHUAbgBkAGUAbABlAHQAYQBiAGwAZQA6AEUATwBMADFkAAL///8gAHUAbgBkAGUAbABlAHQAYQBiAGwAZQA6AEUATwBMADJlAAL///8gAHUAbgBkAGUAbABlAHQAYQBiAGwAZQA6AEUATwBMADNmAAL///8gAHUAbgBkAGUAbABlAHQAYQBiAGwAZQA6AEUATwBMADQ=',
'sub_class': 'remote_controller',
'sub_state': 'normal',
'switch_alarm_light': True,
'switch_alarm_propel': True,
'switch_alarm_sound': True,
'switch_kb_light': False,
'switch_kb_sound': False,
'switch_mode_sound': True,
'telnet_state': 'sim_card_no',
}),
'status_range': dict({
'alarm_delay_time': dict({
'type': 'Integer',
'value': '{"unit":"s","min":0,"max":999,"scale":0,"step":1}',
}),
'alarm_msg': dict({
'type': 'Raw',
'value': '{}',
}),
'alarm_time': dict({
'type': 'Integer',
'value': '{"unit":"min","min":0,"max":999,"scale":0,"step":1}',
}),
'delay_set': dict({
'type': 'Integer',
'value': '{"unit":"s","min":0,"max":999,"scale":0,"step":1}',
}),
'master_mode': dict({
'type': 'Enum',
'value': '{"range":["disarmed","arm","home","sos"]}',
}),
'master_state': dict({
'type': 'Enum',
'value': '{"range":["normal","alarm"]}',
}),
'muffling': dict({
'type': 'Boolean',
'value': '{}',
}),
'sub_admin': dict({
'type': 'Raw',
'value': '{}',
}),
'sub_class': dict({
'type': 'Enum',
'value': '{"range":["remote_controller","detector"]}',
}),
'sub_state': dict({
'type': 'Enum',
'value': '{"range":["normal","alarm","fault","others"]}',
}),
'switch_alarm_light': dict({
'type': 'Boolean',
'value': '{}',
}),
'switch_alarm_propel': dict({
'type': 'Boolean',
'value': '{}',
}),
'switch_alarm_sound': dict({
'type': 'Boolean',
'value': '{}',
}),
'switch_kb_light': dict({
'type': 'Boolean',
'value': '{}',
}),
'switch_kb_sound': dict({
'type': 'Boolean',
'value': '{}',
}),
'switch_mode_sound': dict({
'type': 'Boolean',
'value': '{}',
}),
'telnet_state': dict({
'type': 'Enum',
'value': '{"range":["normal","network_no","phone_no","sim_card_no","network_search","signal_level_1","signal_level_2","signal_level_3","signal_level_4","signal_level_5"]}',
}),
}),
'sub': False,
'support_local': True,
'terminal_id': '7cd96aff-6ec8-4006-b093-3dbff7947591',
'time_zone': '+02:00',
'update_time': '2024-12-02T20:08:56+00:00',
})
# ---
# name: test_device_diagnostics[rqbj_4iqe2hsfyd86kwwc]
dict({
'active_time': '2025-06-24T20:33:10+00:00',
@@ -309,15 +8,7 @@
'disabled_polling': False,
'endpoint': 'https://apigw.tuyaeu.com',
'function': dict({
'alarm_time': dict({
'type': 'Integer',
'value': '{"unit":"s","min":0,"max":3600,"scale":0,"step":1}',
}),
'muffling': dict({
'type': 'Boolean',
'value': '{}',
}),
'self_checking': dict({
'null': dict({
'type': 'Boolean',
'value': '{}',
}),
@@ -383,27 +74,7 @@
'self_checking': False,
}),
'status_range': dict({
'alarm_time': dict({
'type': 'Integer',
'value': '{"unit":"s","min":0,"max":3600,"scale":0,"step":1}',
}),
'checking_result': dict({
'type': 'Enum',
'value': '{"range":["checking","check_success","check_failure","others"]}',
}),
'gas_sensor_status': dict({
'type': 'Enum',
'value': '{"range":["alarm","normal"]}',
}),
'gas_sensor_value': dict({
'type': 'Integer',
'value': '{"unit":"ppm","min":0,"max":999,"scale":0,"step":1}',
}),
'muffling': dict({
'type': 'Boolean',
'value': '{}',
}),
'self_checking': dict({
'null': dict({
'type': 'Boolean',
'value': '{}',
}),
@@ -423,15 +94,7 @@
'category': 'rqbj',
'create_time': '2025-06-24T20:33:10+00:00',
'function': dict({
'alarm_time': dict({
'type': 'Integer',
'value': '{"unit":"s","min":0,"max":3600,"scale":0,"step":1}',
}),
'muffling': dict({
'type': 'Boolean',
'value': '{}',
}),
'self_checking': dict({
'null': dict({
'type': 'Boolean',
'value': '{}',
}),
@@ -496,27 +159,7 @@
'self_checking': False,
}),
'status_range': dict({
'alarm_time': dict({
'type': 'Integer',
'value': '{"unit":"s","min":0,"max":3600,"scale":0,"step":1}',
}),
'checking_result': dict({
'type': 'Enum',
'value': '{"range":["checking","check_success","check_failure","others"]}',
}),
'gas_sensor_status': dict({
'type': 'Enum',
'value': '{"range":["alarm","normal"]}',
}),
'gas_sensor_value': dict({
'type': 'Integer',
'value': '{"unit":"ppm","min":0,"max":999,"scale":0,"step":1}',
}),
'muffling': dict({
'type': 'Boolean',
'value': '{}',
}),
'self_checking': dict({
'null': dict({
'type': 'Boolean',
'value': '{}',
}),

View File

@@ -80,63 +80,54 @@ async def test_service(
mock_manager.send_commands.assert_called_once_with(mock_device.id, [command])
@patch("homeassistant.components.tuya.PLATFORMS", [Platform.ALARM_CONTROL_PANEL])
@pytest.mark.parametrize(
"mock_device_code",
["mal_gyitctrjj1kefxp2"],
)
@pytest.mark.parametrize(
("status_updates", "expected_state"),
[
(
{"master_mode": "disarmed"},
AlarmControlPanelState.DISARMED,
),
(
{"master_mode": "arm"},
AlarmControlPanelState.ARMED_AWAY,
),
(
{"master_mode": "home"},
AlarmControlPanelState.ARMED_HOME,
),
(
{"master_mode": "sos"},
AlarmControlPanelState.TRIGGERED,
),
(
{
"master_mode": "home",
"master_state": "alarm",
# "Test Sensor" in UTF-16BE
"alarm_msg": "AFQAZQBzAHQAIABTAGUAbgBzAG8Acg==",
},
AlarmControlPanelState.TRIGGERED,
),
(
{
"master_mode": "home",
"master_state": "alarm",
# "Sensor Low Battery Test Sensor" in UTF-16BE
"alarm_msg": "AFMAZQBuAHMAbwByACAATABvAHcAIABCAGEAdAB0AGUAcgB5ACAAVABlAHMAdAAgAFMAZQBuAHMAbwBy",
},
AlarmControlPanelState.ARMED_HOME,
),
],
)
async def test_state(
async def test_alarm_state_triggered(
hass: HomeAssistant,
mock_manager: Manager,
mock_config_entry: MockConfigEntry,
mock_device: CustomerDevice,
status_updates: dict[str, Any],
expected_state: str,
) -> None:
"""Test state."""
"""Test alarm state returns TRIGGERED for non-battery alarms."""
entity_id = "alarm_control_panel.multifunction_alarm"
mock_device.status.update(status_updates)
# Set up alarm state without battery warning
mock_device.status["master_state"] = "alarm"
mock_device.status["alarm_msg"] = (
"AFQAZQBzAHQAIABTAGUAbgBzAG8Acg==" # "Test Sensor" in UTF-16BE
)
await initialize_entry(hass, mock_manager, mock_config_entry, mock_device)
state = hass.states.get(entity_id)
assert state is not None, f"{entity_id} does not exist"
assert state.state == expected_state
assert state.state == AlarmControlPanelState.TRIGGERED
@pytest.mark.parametrize(
"mock_device_code",
["mal_gyitctrjj1kefxp2"],
)
async def test_alarm_state_battery_warning(
hass: HomeAssistant,
mock_manager: Manager,
mock_config_entry: MockConfigEntry,
mock_device: CustomerDevice,
) -> None:
"""Test alarm state ignores battery warnings."""
entity_id = "alarm_control_panel.multifunction_alarm"
# Set up alarm state with battery warning
mock_device.status["master_state"] = "alarm"
mock_device.status["alarm_msg"] = (
"AFMAZQBuAHMAbwByACAATABvAHcAIABCAGEAdAB0AGUAcgB5ACAAVABlAHMAdAAgAFMAZQBuAHMAbwBy" # "Sensor Low Battery Test Sensor" in UTF-16BE
)
await initialize_entry(hass, mock_manager, mock_config_entry, mock_device)
state = hass.states.get(entity_id)
assert state is not None, f"{entity_id} does not exist"
# Should not be triggered for battery warnings
assert state.state != AlarmControlPanelState.TRIGGERED

View File

@@ -42,13 +42,7 @@ async def test_entry_diagnostics(
)
@pytest.mark.parametrize(
"mock_device_code",
[
"mal_gyitctrjj1kefxp2",
"rqbj_4iqe2hsfyd86kwwc",
],
)
@pytest.mark.parametrize("mock_device_code", ["rqbj_4iqe2hsfyd86kwwc"])
async def test_device_diagnostics(
hass: HomeAssistant,
mock_manager: Manager,

View File

@@ -6,7 +6,6 @@ from syrupy.assertion import SnapshotAssertion
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.tuya.const import DOMAIN
from homeassistant.components.tuya.diagnostics import _REDACTED_DPCODES
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
@@ -66,11 +65,3 @@ async def test_fixtures_valid(hass: HomeAssistant) -> None:
assert key not in details, (
f"Please remove data[`'{key}']` from {device_code}.json"
)
if "status" in details:
statuses = details["status"]
for key in statuses:
if key in _REDACTED_DPCODES:
assert statuses[key] == "**REDACTED**", (
f"Please mark `data['status']['{key}']` as `**REDACTED**`"
f" in {device_code}.json"
)

View File

@@ -1,52 +0,0 @@
# serializer version: 1
# name: test_cover_setup[cover.test_window-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'cover',
'entity_category': None,
'entity_id': 'cover.test_window',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <CoverDeviceClass.WINDOW: 'window'>,
'original_icon': None,
'original_name': None,
'platform': 'velux',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <CoverEntityFeature: 15>,
'translation_key': None,
'unique_id': '123456789',
'unit_of_measurement': None,
})
# ---
# name: test_cover_setup[cover.test_window-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'current_position': 70,
'device_class': 'window',
'friendly_name': 'Test Window',
'supported_features': <CoverEntityFeature: 15>,
}),
'context': <ANY>,
'entity_id': 'cover.test_window',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'open',
})
# ---

View File

@@ -1,69 +1,32 @@
"""Tests for the Velux cover platform."""
from unittest.mock import AsyncMock
from unittest.mock import AsyncMock, patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from homeassistant.components.velux import DOMAIN
from homeassistant.const import STATE_CLOSED, STATE_OPEN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from . import update_callback_entity
from tests.common import MockConfigEntry, SnapshotAssertion, snapshot_platform
from tests.common import MockConfigEntry
@pytest.fixture
def platform() -> Platform:
"""Fixture to specify platform to test."""
return Platform.COVER
@pytest.mark.usefixtures("setup_integration")
async def test_cover_setup(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Snapshot the cover entity (registry + state)."""
await snapshot_platform(
hass,
entity_registry,
snapshot,
mock_config_entry.entry_id,
)
# Get the cover entity setup and test device association
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
assert len(entity_entries) == 1
entry = entity_entries[0]
assert entry.device_id is not None
device_entry = device_registry.async_get(entry.device_id)
assert device_entry is not None
assert (DOMAIN, f"{123456789}") in device_entry.identifiers
assert device_entry.via_device_id is not None
via_device_entry = device_registry.async_get(device_entry.via_device_id)
assert via_device_entry is not None
assert (
DOMAIN,
f"gateway_{mock_config_entry.entry_id}",
) in via_device_entry.identifiers
@pytest.mark.usefixtures("setup_integration")
@pytest.mark.usefixtures("mock_pyvlx")
async def test_cover_closed(
hass: HomeAssistant,
mock_window: AsyncMock,
mock_config_entry: MockConfigEntry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test the cover closed state."""
mock_config_entry.add_to_hass(hass)
with patch("homeassistant.components.velux.PLATFORMS", [Platform.COVER]):
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
test_entity_id = "cover.test_window"
# Initial state should be open

View File

@@ -3283,7 +3283,7 @@
'state': '5067',
})
# ---
# name: test_all_entities[sensor.model2_condenser_subcooling_temperature-entry]
# name: test_all_entities[sensor.model2_condensor_subcooling_temperature-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
@@ -3296,7 +3296,7 @@
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.model2_condenser_subcooling_temperature',
'entity_id': 'sensor.model2_condensor_subcooling_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
@@ -3311,25 +3311,25 @@
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': None,
'original_name': 'Condenser subcooling temperature',
'original_name': 'Condensor subcooling temperature',
'platform': 'vicare',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'condenser_subcooling_temperature',
'unique_id': 'gateway2_################-condenser_subcooling_temperature-0',
'translation_key': 'condensor_subcooling_temperature',
'unique_id': 'gateway2_################-condensor_subcooling_temperature-0',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_all_entities[sensor.model2_condenser_subcooling_temperature-state]
# name: test_all_entities[sensor.model2_condensor_subcooling_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'temperature',
'friendly_name': 'model2 Condenser subcooling temperature',
'friendly_name': 'model2 Condensor subcooling temperature',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.model2_condenser_subcooling_temperature',
'entity_id': 'sensor.model2_condensor_subcooling_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,

View File

@@ -6,7 +6,7 @@ import dataclasses
from datetime import timedelta
import logging
import threading
from typing import Any
from typing import Any, final
from unittest.mock import MagicMock, PropertyMock, patch
from freezegun.api import FrozenDateTimeFactory
@@ -20,6 +20,7 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_ATTRIBUTION,
ATTR_DEVICE_CLASS,
ATTR_ENTITY_ID,
ATTR_FRIENDLY_NAME,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
@@ -1878,6 +1879,7 @@ async def test_change_entity_id(
self.remove_calls = []
async def async_added_to_hass(self):
await super().async_added_to_hass()
self.added_calls.append(None)
self.async_on_remove(lambda: result.append(1))
@@ -2896,3 +2898,103 @@ async def test_platform_state_write_from_init_unique_id(
# The early attempt to write is interpreted as a unique ID collision
assert "Platform test_platform does not generate unique IDs." in caplog.text
assert "Entity id already exists - ignoring: test.test" not in caplog.text
async def test_included_entities(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
) -> None:
"""Test included entities are exposed via the entity_id attribute."""
entity_registry.async_get_or_create(
domain="hello",
platform="test",
unique_id="very_unique_oceans",
suggested_object_id="oceans",
)
entity_registry.async_get_or_create(
domain="hello",
platform="test",
unique_id="very_unique_continents",
suggested_object_id="continents",
)
entity_registry.async_get_or_create(
domain="hello",
platform="test",
unique_id="very_unique_moon",
suggested_object_id="moon",
)
class MockHelloBaseClass(entity.Entity):
"""Domain base entity platform domain Hello."""
@final
@property
def state_attributes(self) -> dict[str, Any]:
"""Return the state attributes."""
return {"extra": "beer"}
class MockHelloIncludedEntitiesClass(MockHelloBaseClass, entity.Entity):
"""Mock hello grouped entity class for a test integration."""
platform = MockEntityPlatform(hass, domain="hello", platform_name="test")
mock_entity = MockHelloIncludedEntitiesClass()
mock_entity.hass = hass
mock_entity.entity_id = "hello.universe"
mock_entity.unique_id = "very_unique_universe"
mock_entity._attr_included_unique_ids = [
"very_unique_continents",
"very_unique_oceans",
]
await platform.async_add_entities([mock_entity])
# Initiate mock grouped entity for hello domain
mock_entity.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
state = hass.states.get(mock_entity.entity_id)
assert state.attributes.get(ATTR_ENTITY_ID) == ["hello.continents", "hello.oceans"]
# Add an entity to the group of included entities
mock_entity._attr_included_unique_ids = [
"very_unique_continents",
"very_unique_moon",
"very_unique_oceans",
]
mock_entity.async_write_ha_state()
await hass.async_block_till_done()
state = hass.states.get(mock_entity.entity_id)
assert state.attributes.get("extra") == "beer"
assert state.attributes.get(ATTR_ENTITY_ID) == [
"hello.continents",
"hello.moon",
"hello.oceans",
]
# Remove an entity from the group of included entities
mock_entity._attr_included_unique_ids = ["very_unique_moon", "very_unique_oceans"]
mock_entity.async_write_ha_state()
await hass.async_block_till_done()
state = hass.states.get(mock_entity.entity_id)
assert state.attributes.get(ATTR_ENTITY_ID) == ["hello.moon", "hello.oceans"]
# Rename an included entity via the registry entity
entity_registry.async_update_entity(
entity_id="hello.moon", new_entity_id="hello.moon_light"
)
await hass.async_block_till_done()
state = hass.states.get(mock_entity.entity_id)
assert state.attributes.get(ATTR_ENTITY_ID) == ["hello.moon_light", "hello.oceans"]
# Remove an included entity from the registry entity
entity_registry.async_remove(entity_id="hello.oceans")
await hass.async_block_till_done()
state = hass.states.get(mock_entity.entity_id)
assert state.attributes.get(ATTR_ENTITY_ID) == ["hello.moon_light"]