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

Merge branch 'dev' into epenet-20250602-1736

Browse Source
This commit is contained in:
epenet 2025-06-03 11:03:21 +02:00 committed by GitHub
commit 4996d5c3cf
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
31 changed files with 1003 additions and 497 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/adax/__init__.py
View File

@ -8,7 +8,7 @@ from homeassistant.core import HomeAssistant
from .const import CONNECTION_TYPE, LOCAL
from .coordinator import AdaxCloudCoordinator, AdaxConfigEntry, AdaxLocalCoordinator
PLATFORMS = [Platform.CLIMATE]
PLATFORMS = [Platform.CLIMATE, Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: AdaxConfigEntry) -> bool:

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

@ -41,7 +41,30 @@ class AdaxCloudCoordinator(DataUpdateCoordinator[dict[str, dict[str, Any]]]):
async def _async_update_data(self) -> dict[str, dict[str, Any]]:
"""Fetch data from the Adax."""
rooms = await self.adax_data_handler.get_rooms() or []
try:
if hasattr(self.adax_data_handler, "fetch_rooms_info"):
rooms = await self.adax_data_handler.fetch_rooms_info() or []
_LOGGER.debug("fetch_rooms_info returned: %s", rooms)
else:
_LOGGER.debug("fetch_rooms_info method not available, using get_rooms")
rooms = []
if not rooms:
_LOGGER.debug(
"No rooms from fetch_rooms_info, trying get_rooms as fallback"
)
rooms = await self.adax_data_handler.get_rooms() or []
_LOGGER.debug("get_rooms fallback returned: %s", rooms)
if not rooms:
raise UpdateFailed("No rooms available from Adax API")
except OSError as e:
raise UpdateFailed(f"Error communicating with API: {e}") from e
for room in rooms:
room["energyWh"] = int(room.get("energyWh", 0))
return {r["id"]: r for r in rooms}

77
homeassistant/components/adax/sensor.py Normal file
View File

@ -0,0 +1,77 @@
"""Support for Adax energy sensors."""
from __future__ import annotations
from typing import cast
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorStateClass,
)
from homeassistant.const import UnitOfEnergy
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import AdaxConfigEntry
from .const import CONNECTION_TYPE, DOMAIN, LOCAL
from .coordinator import AdaxCloudCoordinator
async def async_setup_entry(
hass: HomeAssistant,
entry: AdaxConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Adax energy sensors with config flow."""
if entry.data.get(CONNECTION_TYPE) != LOCAL:
cloud_coordinator = cast(AdaxCloudCoordinator, entry.runtime_data)
# Create individual energy sensors for each device
async_add_entities(
AdaxEnergySensor(cloud_coordinator, device_id)
for device_id in cloud_coordinator.data
)
class AdaxEnergySensor(CoordinatorEntity[AdaxCloudCoordinator], SensorEntity):
"""Representation of an Adax energy sensor."""
_attr_has_entity_name = True
_attr_translation_key = "energy"
_attr_device_class = SensorDeviceClass.ENERGY
_attr_native_unit_of_measurement = UnitOfEnergy.WATT_HOUR
_attr_suggested_unit_of_measurement = UnitOfEnergy.KILO_WATT_HOUR
_attr_state_class = SensorStateClass.TOTAL_INCREASING
_attr_suggested_display_precision = 3
def __init__(
self,
coordinator: AdaxCloudCoordinator,
device_id: str,
) -> None:
"""Initialize the energy sensor."""
super().__init__(coordinator)
self._device_id = device_id
room = coordinator.data[device_id]
self._attr_unique_id = f"{room['homeId']}_{device_id}_energy"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, device_id)},
name=room["name"],
manufacturer="Adax",
)
@property
def available(self) -> bool:
"""Return True if entity is available."""
return (
super().available and "energyWh" in self.coordinator.data[self._device_id]
)
@property
def native_value(self) -> int:
"""Return the native value of the sensor."""
return int(self.coordinator.data[self._device_id]["energyWh"])

4
homeassistant/components/homematicip_cloud/__init__.py
View File

@ -21,7 +21,7 @@ from .const import (
HMIPC_NAME,
)
from .hap import HomematicIPConfigEntry, HomematicipHAP
from .services import async_setup_services, async_unload_services
from .services import async_setup_services
CONFIG_SCHEMA = vol.Schema(
{
@ -116,8 +116,6 @@ async def async_unload_entry(
assert hap.reset_connection_listener is not None
hap.reset_connection_listener()
await async_unload_services(hass)
return await hap.async_reset()

88
homeassistant/components/homematicip_cloud/services.py
View File

@ -123,32 +123,29 @@ SCHEMA_SET_HOME_COOLING_MODE = vol.Schema(
async def async_setup_services(hass: HomeAssistant) -> None:
"""Set up the HomematicIP Cloud services."""
if hass.services.async_services_for_domain(DOMAIN):
return
@verify_domain_entity_control(DOMAIN)
@verify_domain_entity_control(hass, DOMAIN)
async def async_call_hmipc_service(service: ServiceCall) -> None:
"""Call correct HomematicIP Cloud service."""
service_name = service.service
if service_name == SERVICE_ACTIVATE_ECO_MODE_WITH_DURATION:
await _async_activate_eco_mode_with_duration(hass, service)
await _async_activate_eco_mode_with_duration(service)
elif service_name == SERVICE_ACTIVATE_ECO_MODE_WITH_PERIOD:
await _async_activate_eco_mode_with_period(hass, service)
await _async_activate_eco_mode_with_period(service)
elif service_name == SERVICE_ACTIVATE_VACATION:
await _async_activate_vacation(hass, service)
await _async_activate_vacation(service)
elif service_name == SERVICE_DEACTIVATE_ECO_MODE:
await _async_deactivate_eco_mode(hass, service)
await _async_deactivate_eco_mode(service)
elif service_name == SERVICE_DEACTIVATE_VACATION:
await _async_deactivate_vacation(hass, service)
await _async_deactivate_vacation(service)
elif service_name == SERVICE_DUMP_HAP_CONFIG:
await _async_dump_hap_config(hass, service)
await _async_dump_hap_config(service)
elif service_name == SERVICE_RESET_ENERGY_COUNTER:
await _async_reset_energy_counter(hass, service)
await _async_reset_energy_counter(service)
elif service_name == SERVICE_SET_ACTIVE_CLIMATE_PROFILE:
await _set_active_climate_profile(hass, service)
await _set_active_climate_profile(service)
elif service_name == SERVICE_SET_HOME_COOLING_MODE:
await _async_set_home_cooling_mode(hass, service)
await _async_set_home_cooling_mode(service)
hass.services.async_register(
domain=DOMAIN,
@ -217,90 +214,75 @@ async def async_setup_services(hass: HomeAssistant) -> None:
)
async def async_unload_services(hass: HomeAssistant):
"""Unload HomematicIP Cloud services."""
if hass.config_entries.async_loaded_entries(DOMAIN):
return
for hmipc_service in HMIPC_SERVICES:
hass.services.async_remove(domain=DOMAIN, service=hmipc_service)
async def _async_activate_eco_mode_with_duration(
hass: HomeAssistant, service: ServiceCall
) -> None:
async def _async_activate_eco_mode_with_duration(service: ServiceCall) -> None:
"""Service to activate eco mode with duration."""
duration = service.data[ATTR_DURATION]
if hapid := service.data.get(ATTR_ACCESSPOINT_ID):
if home := _get_home(hass, hapid):
if home := _get_home(service.hass, hapid):
await home.activate_absence_with_duration_async(duration)
else:
entry: HomematicIPConfigEntry
for entry in hass.config_entries.async_loaded_entries(DOMAIN):
for entry in service.hass.config_entries.async_loaded_entries(DOMAIN):
await entry.runtime_data.home.activate_absence_with_duration_async(duration)
async def _async_activate_eco_mode_with_period(
hass: HomeAssistant, service: ServiceCall
) -> None:
async def _async_activate_eco_mode_with_period(service: ServiceCall) -> None:
"""Service to activate eco mode with period."""
endtime = service.data[ATTR_ENDTIME]
if hapid := service.data.get(ATTR_ACCESSPOINT_ID):
if home := _get_home(hass, hapid):
if home := _get_home(service.hass, hapid):
await home.activate_absence_with_period_async(endtime)
else:
entry: HomematicIPConfigEntry
for entry in hass.config_entries.async_loaded_entries(DOMAIN):
for entry in service.hass.config_entries.async_loaded_entries(DOMAIN):
await entry.runtime_data.home.activate_absence_with_period_async(endtime)
async def _async_activate_vacation(hass: HomeAssistant, service: ServiceCall) -> None:
async def _async_activate_vacation(service: ServiceCall) -> None:
"""Service to activate vacation."""
endtime = service.data[ATTR_ENDTIME]
temperature = service.data[ATTR_TEMPERATURE]
if hapid := service.data.get(ATTR_ACCESSPOINT_ID):
if home := _get_home(hass, hapid):
if home := _get_home(service.hass, hapid):
await home.activate_vacation_async(endtime, temperature)
else:
entry: HomematicIPConfigEntry
for entry in hass.config_entries.async_loaded_entries(DOMAIN):
for entry in service.hass.config_entries.async_loaded_entries(DOMAIN):
await entry.runtime_data.home.activate_vacation_async(endtime, temperature)
async def _async_deactivate_eco_mode(hass: HomeAssistant, service: ServiceCall) -> None:
async def _async_deactivate_eco_mode(service: ServiceCall) -> None:
"""Service to deactivate eco mode."""
if hapid := service.data.get(ATTR_ACCESSPOINT_ID):
if home := _get_home(hass, hapid):
if home := _get_home(service.hass, hapid):
await home.deactivate_absence_async()
else:
entry: HomematicIPConfigEntry
for entry in hass.config_entries.async_loaded_entries(DOMAIN):
for entry in service.hass.config_entries.async_loaded_entries(DOMAIN):
await entry.runtime_data.home.deactivate_absence_async()
async def _async_deactivate_vacation(hass: HomeAssistant, service: ServiceCall) -> None:
async def _async_deactivate_vacation(service: ServiceCall) -> None:
"""Service to deactivate vacation."""
if hapid := service.data.get(ATTR_ACCESSPOINT_ID):
if home := _get_home(hass, hapid):
if home := _get_home(service.hass, hapid):
await home.deactivate_vacation_async()
else:
entry: HomematicIPConfigEntry
for entry in hass.config_entries.async_loaded_entries(DOMAIN):
for entry in service.hass.config_entries.async_loaded_entries(DOMAIN):
await entry.runtime_data.home.deactivate_vacation_async()
async def _set_active_climate_profile(
hass: HomeAssistant, service: ServiceCall
) -> None:
async def _set_active_climate_profile(service: ServiceCall) -> None:
"""Service to set the active climate profile."""
entity_id_list = service.data[ATTR_ENTITY_ID]
climate_profile_index = service.data[ATTR_CLIMATE_PROFILE_INDEX] - 1
entry: HomematicIPConfigEntry
for entry in hass.config_entries.async_loaded_entries(DOMAIN):
for entry in service.hass.config_entries.async_loaded_entries(DOMAIN):
if entity_id_list != "all":
for entity_id in entity_id_list:
group = entry.runtime_data.hmip_device_by_entity_id.get(entity_id)
@ -312,16 +294,16 @@ async def _set_active_climate_profile(
await group.set_active_profile_async(climate_profile_index)
async def _async_dump_hap_config(hass: HomeAssistant, service: ServiceCall) -> None:
async def _async_dump_hap_config(service: ServiceCall) -> None:
"""Service to dump the configuration of a Homematic IP Access Point."""
config_path: str = (
service.data.get(ATTR_CONFIG_OUTPUT_PATH) or hass.config.config_dir
service.data.get(ATTR_CONFIG_OUTPUT_PATH) or service.hass.config.config_dir
)
config_file_prefix = service.data[ATTR_CONFIG_OUTPUT_FILE_PREFIX]
anonymize = service.data[ATTR_ANONYMIZE]
entry: HomematicIPConfigEntry
for entry in hass.config_entries.async_loaded_entries(DOMAIN):
for entry in service.hass.config_entries.async_loaded_entries(DOMAIN):
hap_sgtin = entry.unique_id
assert hap_sgtin is not None
@ -338,12 +320,12 @@ async def _async_dump_hap_config(hass: HomeAssistant, service: ServiceCall) -> N
config_file.write_text(json_state, encoding="utf8")
async def _async_reset_energy_counter(hass: HomeAssistant, service: ServiceCall):
async def _async_reset_energy_counter(service: ServiceCall):
"""Service to reset the energy counter."""
entity_id_list = service.data[ATTR_ENTITY_ID]
entry: HomematicIPConfigEntry
for entry in hass.config_entries.async_loaded_entries(DOMAIN):
for entry in service.hass.config_entries.async_loaded_entries(DOMAIN):
if entity_id_list != "all":
for entity_id in entity_id_list:
device = entry.runtime_data.hmip_device_by_entity_id.get(entity_id)
@ -355,16 +337,16 @@ async def _async_reset_energy_counter(hass: HomeAssistant, service: ServiceCall)
await device.reset_energy_counter_async()
async def _async_set_home_cooling_mode(hass: HomeAssistant, service: ServiceCall):
async def _async_set_home_cooling_mode(service: ServiceCall):
"""Service to set the cooling mode."""
cooling = service.data[ATTR_COOLING]
if hapid := service.data.get(ATTR_ACCESSPOINT_ID):
if home := _get_home(hass, hapid):
if home := _get_home(service.hass, hapid):
await home.set_cooling_async(cooling)
else:
entry: HomematicIPConfigEntry
for entry in hass.config_entries.async_loaded_entries(DOMAIN):
for entry in service.hass.config_entries.async_loaded_entries(DOMAIN):
await entry.runtime_data.home.set_cooling_async(cooling)

15
homeassistant/components/icloud/__init__.py
View File

@ -6,19 +6,32 @@ from typing import Any
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.storage import Store
from homeassistant.helpers.typing import ConfigType
from .account import IcloudAccount, IcloudConfigEntry
from .const import (
CONF_GPS_ACCURACY_THRESHOLD,
CONF_MAX_INTERVAL,
CONF_WITH_FAMILY,
DOMAIN,
PLATFORMS,
STORAGE_KEY,
STORAGE_VERSION,
)
from .services import register_services
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up iCloud integration."""
register_services(hass)
return True
async def async_setup_entry(hass: HomeAssistant, entry: IcloudConfigEntry) -> bool:
"""Set up an iCloud account from a config entry."""
@ -51,8 +64,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: IcloudConfigEntry) -> bo
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
register_services(hass)
return True

2
homeassistant/components/insteon/__init__.py
View File

@ -25,9 +25,9 @@ from .const import (
DOMAIN,
INSTEON_PLATFORMS,
)
from .services import async_register_services
from .utils import (
add_insteon_events,
async_register_services,
get_device_platforms,
register_new_device_callback,
)

291
homeassistant/components/insteon/services.py Normal file
View File

@ -0,0 +1,291 @@
"""Utilities used by insteon component."""
from __future__ import annotations
import asyncio
import logging
from pyinsteon import devices
from pyinsteon.address import Address
from pyinsteon.managers.link_manager import (
async_enter_linking_mode,
async_enter_unlinking_mode,
)
from pyinsteon.managers.scene_manager import (
async_trigger_scene_off,
async_trigger_scene_on,
)
from pyinsteon.managers.x10_manager import (
async_x10_all_lights_off,
async_x10_all_lights_on,
async_x10_all_units_off,
)
from pyinsteon.x10_address import create as create_x10_address
from homeassistant.const import (
CONF_ADDRESS,
CONF_ENTITY_ID,
CONF_PLATFORM,
ENTITY_MATCH_ALL,
)
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
async_dispatcher_send,
dispatcher_send,
)
from .const import (
CONF_CAT,
CONF_DIM_STEPS,
CONF_HOUSECODE,
CONF_SUBCAT,
CONF_UNITCODE,
DOMAIN,
SIGNAL_ADD_DEFAULT_LINKS,
SIGNAL_ADD_DEVICE_OVERRIDE,
SIGNAL_ADD_X10_DEVICE,
SIGNAL_LOAD_ALDB,
SIGNAL_PRINT_ALDB,
SIGNAL_REMOVE_DEVICE_OVERRIDE,
SIGNAL_REMOVE_ENTITY,
SIGNAL_REMOVE_HA_DEVICE,
SIGNAL_REMOVE_INSTEON_DEVICE,
SIGNAL_REMOVE_X10_DEVICE,
SIGNAL_SAVE_DEVICES,
SRV_ADD_ALL_LINK,
SRV_ADD_DEFAULT_LINKS,
SRV_ALL_LINK_GROUP,
SRV_ALL_LINK_MODE,
SRV_CONTROLLER,
SRV_DEL_ALL_LINK,
SRV_HOUSECODE,
SRV_LOAD_ALDB,
SRV_LOAD_DB_RELOAD,
SRV_PRINT_ALDB,
SRV_PRINT_IM_ALDB,
SRV_SCENE_OFF,
SRV_SCENE_ON,
SRV_X10_ALL_LIGHTS_OFF,
SRV_X10_ALL_LIGHTS_ON,
SRV_X10_ALL_UNITS_OFF,
)
from .schemas import (
ADD_ALL_LINK_SCHEMA,
ADD_DEFAULT_LINKS_SCHEMA,
DEL_ALL_LINK_SCHEMA,
LOAD_ALDB_SCHEMA,
PRINT_ALDB_SCHEMA,
TRIGGER_SCENE_SCHEMA,
X10_HOUSECODE_SCHEMA,
)
from .utils import print_aldb_to_log
_LOGGER = logging.getLogger(__name__)
@callback
def async_register_services(hass: HomeAssistant) -> None: # noqa: C901
"""Register services used by insteon component."""
save_lock = asyncio.Lock()
async def async_srv_add_all_link(service: ServiceCall) -> None:
"""Add an INSTEON All-Link between two devices."""
group = service.data[SRV_ALL_LINK_GROUP]
mode = service.data[SRV_ALL_LINK_MODE]
link_mode = mode.lower() == SRV_CONTROLLER
await async_enter_linking_mode(link_mode, group)
async def async_srv_del_all_link(service: ServiceCall) -> None:
"""Delete an INSTEON All-Link between two devices."""
group = service.data.get(SRV_ALL_LINK_GROUP)
await async_enter_unlinking_mode(group)
async def async_srv_load_aldb(service: ServiceCall) -> None:
"""Load the device All-Link database."""
entity_id = service.data[CONF_ENTITY_ID]
reload = service.data[SRV_LOAD_DB_RELOAD]
if entity_id.lower() == ENTITY_MATCH_ALL:
await async_srv_load_aldb_all(reload)
else:
signal = f"{entity_id}_{SIGNAL_LOAD_ALDB}"
async_dispatcher_send(hass, signal, reload)
async def async_srv_load_aldb_all(reload):
"""Load the All-Link database for all devices."""
# Cannot be done concurrently due to issues with the underlying protocol.
for address in devices:
device = devices[address]
if device != devices.modem and device.cat != 0x03:
await device.aldb.async_load(refresh=reload)
await async_srv_save_devices()
async def async_srv_save_devices():
"""Write the Insteon device configuration to file."""
async with save_lock:
_LOGGER.debug("Saving Insteon devices")
await devices.async_save(hass.config.config_dir)
def print_aldb(service: ServiceCall) -> None:
"""Print the All-Link Database for a device."""
# For now this sends logs to the log file.
# Future direction is to create an INSTEON control panel.
entity_id = service.data[CONF_ENTITY_ID]
signal = f"{entity_id}_{SIGNAL_PRINT_ALDB}"
dispatcher_send(hass, signal)
def print_im_aldb(service: ServiceCall) -> None:
"""Print the All-Link Database for a device."""
# For now this sends logs to the log file.
# Future direction is to create an INSTEON control panel.
print_aldb_to_log(devices.modem.aldb)
async def async_srv_x10_all_units_off(service: ServiceCall) -> None:
"""Send the X10 All Units Off command."""
housecode = service.data.get(SRV_HOUSECODE)
await async_x10_all_units_off(housecode)
async def async_srv_x10_all_lights_off(service: ServiceCall) -> None:
"""Send the X10 All Lights Off command."""
housecode = service.data.get(SRV_HOUSECODE)
await async_x10_all_lights_off(housecode)
async def async_srv_x10_all_lights_on(service: ServiceCall) -> None:
"""Send the X10 All Lights On command."""
housecode = service.data.get(SRV_HOUSECODE)
await async_x10_all_lights_on(housecode)
async def async_srv_scene_on(service: ServiceCall) -> None:
"""Trigger an INSTEON scene ON."""
group = service.data.get(SRV_ALL_LINK_GROUP)
await async_trigger_scene_on(group)
async def async_srv_scene_off(service: ServiceCall) -> None:
"""Trigger an INSTEON scene ON."""
group = service.data.get(SRV_ALL_LINK_GROUP)
await async_trigger_scene_off(group)
@callback
def async_add_default_links(service: ServiceCall) -> None:
"""Add the default All-Link entries to a device."""
entity_id = service.data[CONF_ENTITY_ID]
signal = f"{entity_id}_{SIGNAL_ADD_DEFAULT_LINKS}"
async_dispatcher_send(hass, signal)
async def async_add_device_override(override):
"""Remove an Insten device and associated entities."""
address = Address(override[CONF_ADDRESS])
await async_remove_ha_device(address)
devices.set_id(address, override[CONF_CAT], override[CONF_SUBCAT], 0)
await async_srv_save_devices()
async def async_remove_device_override(address):
"""Remove an Insten device and associated entities."""
address = Address(address)
await async_remove_ha_device(address)
devices.set_id(address, None, None, None)
await devices.async_identify_device(address)
await async_srv_save_devices()
@callback
def async_add_x10_device(x10_config):
"""Add X10 device."""
housecode = x10_config[CONF_HOUSECODE]
unitcode = x10_config[CONF_UNITCODE]
platform = x10_config[CONF_PLATFORM]
steps = x10_config.get(CONF_DIM_STEPS, 22)
x10_type = "on_off"
if platform == "light":
x10_type = "dimmable"
elif platform == "binary_sensor":
x10_type = "sensor"
_LOGGER.debug(
"Adding X10 device to Insteon: %s %d %s", housecode, unitcode, x10_type
)
# This must be run in the event loop
devices.add_x10_device(housecode, unitcode, x10_type, steps)
async def async_remove_x10_device(housecode, unitcode):
"""Remove an X10 device and associated entities."""
address = create_x10_address(housecode, unitcode)
devices.pop(address)
await async_remove_ha_device(address)
async def async_remove_ha_device(address: Address, remove_all_refs: bool = False):
"""Remove the device and all entities from hass."""
signal = f"{address.id}_{SIGNAL_REMOVE_ENTITY}"
async_dispatcher_send(hass, signal)
dev_registry = dr.async_get(hass)
device = dev_registry.async_get_device(identifiers={(DOMAIN, str(address))})
if device:
dev_registry.async_remove_device(device.id)
async def async_remove_insteon_device(
address: Address, remove_all_refs: bool = False
):
"""Remove the underlying Insteon device from the network."""
await devices.async_remove_device(
address=address, force=False, remove_all_refs=remove_all_refs
)
await async_srv_save_devices()
hass.services.async_register(
DOMAIN, SRV_ADD_ALL_LINK, async_srv_add_all_link, schema=ADD_ALL_LINK_SCHEMA
)
hass.services.async_register(
DOMAIN, SRV_DEL_ALL_LINK, async_srv_del_all_link, schema=DEL_ALL_LINK_SCHEMA
)
hass.services.async_register(
DOMAIN, SRV_LOAD_ALDB, async_srv_load_aldb, schema=LOAD_ALDB_SCHEMA
)
hass.services.async_register(
DOMAIN, SRV_PRINT_ALDB, print_aldb, schema=PRINT_ALDB_SCHEMA
)
hass.services.async_register(DOMAIN, SRV_PRINT_IM_ALDB, print_im_aldb, schema=None)
hass.services.async_register(
DOMAIN,
SRV_X10_ALL_UNITS_OFF,
async_srv_x10_all_units_off,
schema=X10_HOUSECODE_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SRV_X10_ALL_LIGHTS_OFF,
async_srv_x10_all_lights_off,
schema=X10_HOUSECODE_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SRV_X10_ALL_LIGHTS_ON,
async_srv_x10_all_lights_on,
schema=X10_HOUSECODE_SCHEMA,
)
hass.services.async_register(
DOMAIN, SRV_SCENE_ON, async_srv_scene_on, schema=TRIGGER_SCENE_SCHEMA
)
hass.services.async_register(
DOMAIN, SRV_SCENE_OFF, async_srv_scene_off, schema=TRIGGER_SCENE_SCHEMA
)
hass.services.async_register(
DOMAIN,
SRV_ADD_DEFAULT_LINKS,
async_add_default_links,
schema=ADD_DEFAULT_LINKS_SCHEMA,
)
async_dispatcher_connect(hass, SIGNAL_SAVE_DEVICES, async_srv_save_devices)
async_dispatcher_connect(
hass, SIGNAL_ADD_DEVICE_OVERRIDE, async_add_device_override
)
async_dispatcher_connect(
hass, SIGNAL_REMOVE_DEVICE_OVERRIDE, async_remove_device_override
)
async_dispatcher_connect(hass, SIGNAL_ADD_X10_DEVICE, async_add_x10_device)
async_dispatcher_connect(hass, SIGNAL_REMOVE_X10_DEVICE, async_remove_x10_device)
async_dispatcher_connect(hass, SIGNAL_REMOVE_HA_DEVICE, async_remove_ha_device)
async_dispatcher_connect(
hass, SIGNAL_REMOVE_INSTEON_DEVICE, async_remove_insteon_device
)
_LOGGER.debug("Insteon Services registered")

280
homeassistant/components/insteon/utils.py
View File

@ -2,7 +2,6 @@
from __future__ import annotations
import asyncio
from collections.abc import Callable
import logging
from typing import TYPE_CHECKING, Any
@ -12,90 +11,25 @@ from pyinsteon.address import Address
from pyinsteon.constants import ALDBStatus, DeviceAction
from pyinsteon.device_types.device_base import Device
from pyinsteon.events import OFF_EVENT, OFF_FAST_EVENT, ON_EVENT, ON_FAST_EVENT, Event
from pyinsteon.managers.link_manager import (
async_enter_linking_mode,
async_enter_unlinking_mode,
)
from pyinsteon.managers.scene_manager import (
async_trigger_scene_off,
async_trigger_scene_on,
)
from pyinsteon.managers.x10_manager import (
async_x10_all_lights_off,
async_x10_all_lights_on,
async_x10_all_units_off,
)
from pyinsteon.x10_address import create as create_x10_address
from serial.tools import list_ports
from homeassistant.components import usb
from homeassistant.const import (
CONF_ADDRESS,
CONF_ENTITY_ID,
CONF_PLATFORM,
ENTITY_MATCH_ALL,
Platform,
)
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.const import CONF_ADDRESS, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
async_dispatcher_send,
dispatcher_send,
)
from homeassistant.helpers.dispatcher import dispatcher_send
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import (
CONF_CAT,
CONF_DIM_STEPS,
CONF_HOUSECODE,
CONF_SUBCAT,
CONF_UNITCODE,
DOMAIN,
EVENT_CONF_BUTTON,
EVENT_GROUP_OFF,
EVENT_GROUP_OFF_FAST,
EVENT_GROUP_ON,
EVENT_GROUP_ON_FAST,
SIGNAL_ADD_DEFAULT_LINKS,
SIGNAL_ADD_DEVICE_OVERRIDE,
SIGNAL_ADD_ENTITIES,
SIGNAL_ADD_X10_DEVICE,
SIGNAL_LOAD_ALDB,
SIGNAL_PRINT_ALDB,
SIGNAL_REMOVE_DEVICE_OVERRIDE,
SIGNAL_REMOVE_ENTITY,
SIGNAL_REMOVE_HA_DEVICE,
SIGNAL_REMOVE_INSTEON_DEVICE,
SIGNAL_REMOVE_X10_DEVICE,
SIGNAL_SAVE_DEVICES,
SRV_ADD_ALL_LINK,
SRV_ADD_DEFAULT_LINKS,
SRV_ALL_LINK_GROUP,
SRV_ALL_LINK_MODE,
SRV_CONTROLLER,
SRV_DEL_ALL_LINK,
SRV_HOUSECODE,
SRV_LOAD_ALDB,
SRV_LOAD_DB_RELOAD,
SRV_PRINT_ALDB,
SRV_PRINT_IM_ALDB,
SRV_SCENE_OFF,
SRV_SCENE_ON,
SRV_X10_ALL_LIGHTS_OFF,
SRV_X10_ALL_LIGHTS_ON,
SRV_X10_ALL_UNITS_OFF,
)
from .ipdb import get_device_platform_groups, get_device_platforms
from .schemas import (
ADD_ALL_LINK_SCHEMA,
ADD_DEFAULT_LINKS_SCHEMA,
DEL_ALL_LINK_SCHEMA,
LOAD_ALDB_SCHEMA,
PRINT_ALDB_SCHEMA,
TRIGGER_SCENE_SCHEMA,
X10_HOUSECODE_SCHEMA,
)
if TYPE_CHECKING:
from .entity import InsteonEntity
@ -154,7 +88,7 @@ def add_insteon_events(hass: HomeAssistant, device: Device) -> None:
_register_event(event, async_fire_insteon_event)
def register_new_device_callback(hass):
def register_new_device_callback(hass: HomeAssistant) -> None:
"""Register callback for new Insteon device."""
@callback
@ -180,212 +114,6 @@ def register_new_device_callback(hass):
devices.subscribe(async_new_insteon_device, force_strong_ref=True)
@callback
def async_register_services(hass): # noqa: C901
"""Register services used by insteon component."""
save_lock = asyncio.Lock()
async def async_srv_add_all_link(service: ServiceCall) -> None:
"""Add an INSTEON All-Link between two devices."""
group = service.data[SRV_ALL_LINK_GROUP]
mode = service.data[SRV_ALL_LINK_MODE]
link_mode = mode.lower() == SRV_CONTROLLER
await async_enter_linking_mode(link_mode, group)
async def async_srv_del_all_link(service: ServiceCall) -> None:
"""Delete an INSTEON All-Link between two devices."""
group = service.data.get(SRV_ALL_LINK_GROUP)
await async_enter_unlinking_mode(group)
async def async_srv_load_aldb(service: ServiceCall) -> None:
"""Load the device All-Link database."""
entity_id = service.data[CONF_ENTITY_ID]
reload = service.data[SRV_LOAD_DB_RELOAD]
if entity_id.lower() == ENTITY_MATCH_ALL:
await async_srv_load_aldb_all(reload)
else:
signal = f"{entity_id}_{SIGNAL_LOAD_ALDB}"
async_dispatcher_send(hass, signal, reload)
async def async_srv_load_aldb_all(reload):
"""Load the All-Link database for all devices."""
# Cannot be done concurrently due to issues with the underlying protocol.
for address in devices:
device = devices[address]
if device != devices.modem and device.cat != 0x03:
await device.aldb.async_load(refresh=reload)
await async_srv_save_devices()
async def async_srv_save_devices():
"""Write the Insteon device configuration to file."""
async with save_lock:
_LOGGER.debug("Saving Insteon devices")
await devices.async_save(hass.config.config_dir)
def print_aldb(service: ServiceCall) -> None:
"""Print the All-Link Database for a device."""
# For now this sends logs to the log file.
# Future direction is to create an INSTEON control panel.
entity_id = service.data[CONF_ENTITY_ID]
signal = f"{entity_id}_{SIGNAL_PRINT_ALDB}"
dispatcher_send(hass, signal)
def print_im_aldb(service: ServiceCall) -> None:
"""Print the All-Link Database for a device."""
# For now this sends logs to the log file.
# Future direction is to create an INSTEON control panel.
print_aldb_to_log(devices.modem.aldb)
async def async_srv_x10_all_units_off(service: ServiceCall) -> None:
"""Send the X10 All Units Off command."""
housecode = service.data.get(SRV_HOUSECODE)
await async_x10_all_units_off(housecode)
async def async_srv_x10_all_lights_off(service: ServiceCall) -> None:
"""Send the X10 All Lights Off command."""
housecode = service.data.get(SRV_HOUSECODE)
await async_x10_all_lights_off(housecode)
async def async_srv_x10_all_lights_on(service: ServiceCall) -> None:
"""Send the X10 All Lights On command."""
housecode = service.data.get(SRV_HOUSECODE)
await async_x10_all_lights_on(housecode)
async def async_srv_scene_on(service: ServiceCall) -> None:
"""Trigger an INSTEON scene ON."""
group = service.data.get(SRV_ALL_LINK_GROUP)
await async_trigger_scene_on(group)
async def async_srv_scene_off(service: ServiceCall) -> None:
"""Trigger an INSTEON scene ON."""
group = service.data.get(SRV_ALL_LINK_GROUP)
await async_trigger_scene_off(group)
@callback
def async_add_default_links(service: ServiceCall) -> None:
"""Add the default All-Link entries to a device."""
entity_id = service.data[CONF_ENTITY_ID]
signal = f"{entity_id}_{SIGNAL_ADD_DEFAULT_LINKS}"
async_dispatcher_send(hass, signal)
async def async_add_device_override(override):
"""Remove an Insten device and associated entities."""
address = Address(override[CONF_ADDRESS])
await async_remove_ha_device(address)
devices.set_id(address, override[CONF_CAT], override[CONF_SUBCAT], 0)
await async_srv_save_devices()
async def async_remove_device_override(address):
"""Remove an Insten device and associated entities."""
address = Address(address)
await async_remove_ha_device(address)
devices.set_id(address, None, None, None)
await devices.async_identify_device(address)
await async_srv_save_devices()
@callback
def async_add_x10_device(x10_config):
"""Add X10 device."""
housecode = x10_config[CONF_HOUSECODE]
unitcode = x10_config[CONF_UNITCODE]
platform = x10_config[CONF_PLATFORM]
steps = x10_config.get(CONF_DIM_STEPS, 22)
x10_type = "on_off"
if platform == "light":
x10_type = "dimmable"
elif platform == "binary_sensor":
x10_type = "sensor"
_LOGGER.debug(
"Adding X10 device to Insteon: %s %d %s", housecode, unitcode, x10_type
)
# This must be run in the event loop
devices.add_x10_device(housecode, unitcode, x10_type, steps)
async def async_remove_x10_device(housecode, unitcode):
"""Remove an X10 device and associated entities."""
address = create_x10_address(housecode, unitcode)
devices.pop(address)
await async_remove_ha_device(address)
async def async_remove_ha_device(address: Address, remove_all_refs: bool = False):
"""Remove the device and all entities from hass."""
signal = f"{address.id}_{SIGNAL_REMOVE_ENTITY}"
async_dispatcher_send(hass, signal)
dev_registry = dr.async_get(hass)
device = dev_registry.async_get_device(identifiers={(DOMAIN, str(address))})
if device:
dev_registry.async_remove_device(device.id)
async def async_remove_insteon_device(
address: Address, remove_all_refs: bool = False
):
"""Remove the underlying Insteon device from the network."""
await devices.async_remove_device(
address=address, force=False, remove_all_refs=remove_all_refs
)
await async_srv_save_devices()
hass.services.async_register(
DOMAIN, SRV_ADD_ALL_LINK, async_srv_add_all_link, schema=ADD_ALL_LINK_SCHEMA
)
hass.services.async_register(
DOMAIN, SRV_DEL_ALL_LINK, async_srv_del_all_link, schema=DEL_ALL_LINK_SCHEMA
)
hass.services.async_register(
DOMAIN, SRV_LOAD_ALDB, async_srv_load_aldb, schema=LOAD_ALDB_SCHEMA
)
hass.services.async_register(
DOMAIN, SRV_PRINT_ALDB, print_aldb, schema=PRINT_ALDB_SCHEMA
)
hass.services.async_register(DOMAIN, SRV_PRINT_IM_ALDB, print_im_aldb, schema=None)
hass.services.async_register(
DOMAIN,
SRV_X10_ALL_UNITS_OFF,
async_srv_x10_all_units_off,
schema=X10_HOUSECODE_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SRV_X10_ALL_LIGHTS_OFF,
async_srv_x10_all_lights_off,
schema=X10_HOUSECODE_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SRV_X10_ALL_LIGHTS_ON,
async_srv_x10_all_lights_on,
schema=X10_HOUSECODE_SCHEMA,
)
hass.services.async_register(
DOMAIN, SRV_SCENE_ON, async_srv_scene_on, schema=TRIGGER_SCENE_SCHEMA
)
hass.services.async_register(
DOMAIN, SRV_SCENE_OFF, async_srv_scene_off, schema=TRIGGER_SCENE_SCHEMA
)
hass.services.async_register(
DOMAIN,
SRV_ADD_DEFAULT_LINKS,
async_add_default_links,
schema=ADD_DEFAULT_LINKS_SCHEMA,
)
async_dispatcher_connect(hass, SIGNAL_SAVE_DEVICES, async_srv_save_devices)
async_dispatcher_connect(
hass, SIGNAL_ADD_DEVICE_OVERRIDE, async_add_device_override
)
async_dispatcher_connect(
hass, SIGNAL_REMOVE_DEVICE_OVERRIDE, async_remove_device_override
)
async_dispatcher_connect(hass, SIGNAL_ADD_X10_DEVICE, async_add_x10_device)
async_dispatcher_connect(hass, SIGNAL_REMOVE_X10_DEVICE, async_remove_x10_device)
async_dispatcher_connect(hass, SIGNAL_REMOVE_HA_DEVICE, async_remove_ha_device)
async_dispatcher_connect(
hass, SIGNAL_REMOVE_INSTEON_DEVICE, async_remove_insteon_device
)
_LOGGER.debug("Insteon Services registered")
def print_aldb_to_log(aldb):
"""Print the All-Link Database to the log file."""
logger = logging.getLogger(f"{__name__}.links")

54
homeassistant/components/nzbget/__init__.py
View File

@ -1,30 +1,25 @@
"""The NZBGet integration."""
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType
from .const import (
ATTR_SPEED,
DATA_COORDINATOR,
DATA_UNDO_UPDATE_LISTENER,
DEFAULT_SPEED_LIMIT,
DOMAIN,
SERVICE_PAUSE,
SERVICE_RESUME,
SERVICE_SET_SPEED,
)
from .const import DATA_COORDINATOR, DATA_UNDO_UPDATE_LISTENER, DOMAIN
from .coordinator import NZBGetDataUpdateCoordinator
from .services import async_register_services
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
PLATFORMS = [Platform.SENSOR, Platform.SWITCH]
SPEED_LIMIT_SCHEMA = vol.Schema(
{vol.Optional(ATTR_SPEED, default=DEFAULT_SPEED_LIMIT): cv.positive_int}
)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up NZBGet integration."""
async_register_services(hass)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
@ -44,8 +39,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
_async_register_services(hass, coordinator)
return True
@ -60,31 +53,6 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok
def _async_register_services(
hass: HomeAssistant,
coordinator: NZBGetDataUpdateCoordinator,
) -> None:
"""Register integration-level services."""
def pause(call: ServiceCall) -> None:
"""Service call to pause downloads in NZBGet."""
coordinator.nzbget.pausedownload()
def resume(call: ServiceCall) -> None:
"""Service call to resume downloads in NZBGet."""
coordinator.nzbget.resumedownload()
def set_speed(call: ServiceCall) -> None:
"""Service call to rate limit speeds in NZBGet."""
coordinator.nzbget.rate(call.data[ATTR_SPEED])
hass.services.async_register(DOMAIN, SERVICE_PAUSE, pause, schema=vol.Schema({}))
hass.services.async_register(DOMAIN, SERVICE_RESUME, resume, schema=vol.Schema({}))
hass.services.async_register(
DOMAIN, SERVICE_SET_SPEED, set_speed, schema=SPEED_LIMIT_SCHEMA
)
async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle options update."""
await hass.config_entries.async_reload(entry.entry_id)

58
homeassistant/components/nzbget/services.py Normal file
View File

@ -0,0 +1,58 @@
"""The NZBGet integration."""
import voluptuous as vol
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import config_validation as cv
from .const import (
ATTR_SPEED,
DATA_COORDINATOR,
DEFAULT_SPEED_LIMIT,
DOMAIN,
SERVICE_PAUSE,
SERVICE_RESUME,
SERVICE_SET_SPEED,
)
from .coordinator import NZBGetDataUpdateCoordinator
SPEED_LIMIT_SCHEMA = vol.Schema(
{vol.Optional(ATTR_SPEED, default=DEFAULT_SPEED_LIMIT): cv.positive_int}
)
def _get_coordinator(call: ServiceCall) -> NZBGetDataUpdateCoordinator:
"""Service call to pause downloads in NZBGet."""
entries = call.hass.config_entries.async_loaded_entries(DOMAIN)
if not entries:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="invalid_config_entry",
)
return call.hass.data[DOMAIN][entries[0].entry_id][DATA_COORDINATOR]
def pause(call: ServiceCall) -> None:
"""Service call to pause downloads in NZBGet."""
_get_coordinator(call).nzbget.pausedownload()
def resume(call: ServiceCall) -> None:
"""Service call to resume downloads in NZBGet."""
_get_coordinator(call).nzbget.resumedownload()
def set_speed(call: ServiceCall) -> None:
"""Service call to rate limit speeds in NZBGet."""
_get_coordinator(call).nzbget.rate(call.data[ATTR_SPEED])
def async_register_services(hass: HomeAssistant) -> None:
"""Register integration-level services."""
hass.services.async_register(DOMAIN, SERVICE_PAUSE, pause, schema=vol.Schema({}))
hass.services.async_register(DOMAIN, SERVICE_RESUME, resume, schema=vol.Schema({}))
hass.services.async_register(
DOMAIN, SERVICE_SET_SPEED, set_speed, schema=SPEED_LIMIT_SCHEMA
)

5
homeassistant/components/nzbget/strings.json
View File

@ -64,6 +64,11 @@
}
}
},
"exceptions": {
"invalid_config_entry": {
"message": "Config entry not found or not loaded!"
}
},
"services": {
"pause": {
"name": "[%key:common::action::pause%]",

2
homeassistant/components/picnic/manifest.json
View File

@ -6,5 +6,5 @@
"documentation": "https://www.home-assistant.io/integrations/picnic",
"iot_class": "cloud_polling",
"loggers": ["python_picnic_api2"],
"requirements": ["python-picnic-api2==1.2.4"]
"requirements": ["python-picnic-api2==1.3.1"]
}

4
homeassistant/components/teslemetry/sensor.py
View File

@ -205,7 +205,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
key="charge_state_charging_state",
polling=True,
streaming_listener=lambda vehicle, callback: vehicle.listen_DetailedChargeState(
lambda value: None if value is None else callback(value.lower())
lambda value: callback(None if value is None else CHARGE_STATES.get(value))
),
polling_value_fn=lambda value: CHARGE_STATES.get(str(value)),
options=list(CHARGE_STATES.values()),
@ -533,7 +533,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="bms_state",
streaming_listener=lambda vehicle, callback: vehicle.listen_BMSState(
lambda value: None if value is None else callback(BMS_STATES.get(value))
lambda value: callback(None if value is None else BMS_STATES.get(value))
),
device_class=SensorDeviceClass.ENUM,
options=list(BMS_STATES.values()),

2
requirements_all.txt generated
View File

@ -2486,7 +2486,7 @@ python-otbr-api==2.7.0
python-overseerr==0.7.1
# homeassistant.components.picnic
python-picnic-api2==1.2.4
python-picnic-api2==1.3.1
# homeassistant.components.rabbitair
python-rabbitair==0.0.8

2
requirements_test_all.txt generated
View File

@ -2053,7 +2053,7 @@ python-otbr-api==2.7.0
python-overseerr==0.7.1
# homeassistant.components.picnic
python-picnic-api2==1.2.4
python-picnic-api2==1.3.1
# homeassistant.components.rabbitair
python-rabbitair==0.0.8

9
tests/components/adax/conftest.py
View File

@ -43,6 +43,7 @@ CLOUD_DEVICE_DATA: dict[str, Any] = [
"temperature": 15,
"targetTemperature": 20,
"heatingEnabled": True,
"energyWh": 1500,
}
]
@ -70,9 +71,17 @@ def mock_adax_cloud():
with patch("homeassistant.components.adax.coordinator.Adax") as mock_adax:
mock_adax_class = mock_adax.return_value
mock_adax_class.fetch_rooms_info = AsyncMock()
mock_adax_class.fetch_rooms_info.return_value = CLOUD_DEVICE_DATA
mock_adax_class.get_rooms = AsyncMock()
mock_adax_class.get_rooms.return_value = CLOUD_DEVICE_DATA
mock_adax_class.fetch_energy_info = AsyncMock()
mock_adax_class.fetch_energy_info.return_value = [
{"deviceId": "1", "energyWh": 1500}
]
mock_adax_class.update = AsyncMock()
mock_adax_class.update.return_value = None
yield mock_adax_class

237
tests/components/adax/snapshots/test_sensor.ambr Normal file
View File

@ -0,0 +1,237 @@
# serializer version: 1
# name: test_fallback_to_get_rooms[sensor.room_1_energy-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.room_1_energy',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 3,
}),
'sensor.private': dict({
'suggested_unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
}),
}),
'original_device_class': <SensorDeviceClass.ENERGY: 'energy'>,
'original_icon': None,
'original_name': 'Energy',
'platform': 'adax',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'energy',
'unique_id': '1_1_energy',
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
})
# ---
# name: test_fallback_to_get_rooms[sensor.room_1_energy-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'energy',
'friendly_name': 'Room 1 Energy',
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
}),
'context': <ANY>,
'entity_id': 'sensor.room_1_energy',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.0',
})
# ---
# name: test_multiple_devices_create_individual_sensors[sensor.room_1_energy-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.room_1_energy',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 3,
}),
'sensor.private': dict({
'suggested_unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
}),
}),
'original_device_class': <SensorDeviceClass.ENERGY: 'energy'>,
'original_icon': None,
'original_name': 'Energy',
'platform': 'adax',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'energy',
'unique_id': '1_1_energy',
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
})
# ---
# name: test_multiple_devices_create_individual_sensors[sensor.room_1_energy-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'energy',
'friendly_name': 'Room 1 Energy',
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
}),
'context': <ANY>,
'entity_id': 'sensor.room_1_energy',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '1.5',
})
# ---
# name: test_multiple_devices_create_individual_sensors[sensor.room_2_energy-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.room_2_energy',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 3,
}),
'sensor.private': dict({
'suggested_unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
}),
}),
'original_device_class': <SensorDeviceClass.ENERGY: 'energy'>,
'original_icon': None,
'original_name': 'Energy',
'platform': 'adax',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'energy',
'unique_id': '1_2_energy',
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
})
# ---
# name: test_multiple_devices_create_individual_sensors[sensor.room_2_energy-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'energy',
'friendly_name': 'Room 2 Energy',
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
}),
'context': <ANY>,
'entity_id': 'sensor.room_2_energy',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2.5',
})
# ---
# name: test_sensor_cloud[sensor.room_1_energy-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.room_1_energy',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 3,
}),
'sensor.private': dict({
'suggested_unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
}),
}),
'original_device_class': <SensorDeviceClass.ENERGY: 'energy'>,
'original_icon': None,
'original_name': 'Energy',
'platform': 'adax',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'energy',
'unique_id': '1_1_energy',
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
})
# ---
# name: test_sensor_cloud[sensor.room_1_energy-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'energy',
'friendly_name': 'Room 1 Energy',
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
}),
'context': <ANY>,
'entity_id': 'sensor.room_1_energy',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '1.5',
})
# ---

4
tests/components/adax/test_climate.py
View File

@ -20,7 +20,7 @@ async def test_climate_cloud(
) -> None:
"""Test states of the (cloud) Climate entity."""
await setup_integration(hass, mock_cloud_config_entry)
mock_adax_cloud.get_rooms.assert_called_once()
mock_adax_cloud.fetch_rooms_info.assert_called_once()
assert len(hass.states.async_entity_ids(Platform.CLIMATE)) == 1
entity_id = hass.states.async_entity_ids(Platform.CLIMATE)[0]
@ -37,7 +37,7 @@ async def test_climate_cloud(
== CLOUD_DEVICE_DATA[0]["temperature"]
)
mock_adax_cloud.get_rooms.side_effect = Exception()
mock_adax_cloud.fetch_rooms_info.side_effect = Exception()
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()

121
tests/components/adax/test_sensor.py Normal file
View File

@ -0,0 +1,121 @@
"""Test Adax sensor entity."""
from unittest.mock import AsyncMock, patch
from syrupy.assertion import SnapshotAssertion
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
async def test_sensor_cloud(
hass: HomeAssistant,
mock_adax_cloud: AsyncMock,
mock_cloud_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
) -> None:
"""Test sensor setup for cloud connection."""
with patch("homeassistant.components.adax.PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_cloud_config_entry)
# Now we use fetch_rooms_info as primary method
mock_adax_cloud.fetch_rooms_info.assert_called_once()
await snapshot_platform(
hass, entity_registry, snapshot, mock_cloud_config_entry.entry_id
)
async def test_sensor_local_not_created(
hass: HomeAssistant,
mock_adax_local: AsyncMock,
mock_local_config_entry: MockConfigEntry,
) -> None:
"""Test that sensors are not created for local connection."""
with patch("homeassistant.components.adax.PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_local_config_entry)
# No sensor entities should be created for local connection
sensor_entities = hass.states.async_entity_ids("sensor")
adax_sensors = [e for e in sensor_entities if "adax" in e or "room" in e]
assert len(adax_sensors) == 0
async def test_multiple_devices_create_individual_sensors(
hass: HomeAssistant,
mock_adax_cloud: AsyncMock,
mock_cloud_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that multiple devices create individual sensors."""
# Mock multiple devices for both fetch_rooms_info and get_rooms (fallback)
multiple_devices_data = [
{
"id": "1",
"homeId": "1",
"name": "Room 1",
"temperature": 15,
"targetTemperature": 20,
"heatingEnabled": True,
"energyWh": 1500,
},
{
"id": "2",
"homeId": "1",
"name": "Room 2",
"temperature": 18,
"targetTemperature": 22,
"heatingEnabled": True,
"energyWh": 2500,
},
]
mock_adax_cloud.fetch_rooms_info.return_value = multiple_devices_data
mock_adax_cloud.get_rooms.return_value = multiple_devices_data
with patch("homeassistant.components.adax.PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_cloud_config_entry)
await snapshot_platform(
hass, entity_registry, snapshot, mock_cloud_config_entry.entry_id
)
async def test_fallback_to_get_rooms(
hass: HomeAssistant,
mock_adax_cloud: AsyncMock,
mock_cloud_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
) -> None:
"""Test fallback to get_rooms when fetch_rooms_info returns empty list."""
# Mock fetch_rooms_info to return empty list, get_rooms to return data
mock_adax_cloud.fetch_rooms_info.return_value = []
mock_adax_cloud.get_rooms.return_value = [
{
"id": "1",
"homeId": "1",
"name": "Room 1",
"temperature": 15,
"targetTemperature": 20,
"heatingEnabled": True,
"energyWh": 0, # No energy data from get_rooms
}
]
with patch("homeassistant.components.adax.PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_cloud_config_entry)
# Should call both methods
mock_adax_cloud.fetch_rooms_info.assert_called_once()
mock_adax_cloud.get_rooms.assert_called_once()
await snapshot_platform(
hass, entity_registry, snapshot, mock_cloud_config_entry.entry_id
)

8
tests/components/hassio/test_backup.py
View File

@ -54,7 +54,7 @@ from homeassistant.setup import async_setup_component
from .test_init import MOCK_ENVIRON
from tests.common import load_json_object_fixture, mock_platform
from tests.common import async_load_json_object_fixture, mock_platform
from tests.typing import ClientSessionGenerator, WebSocketGenerator
TEST_BACKUP = supervisor_backups.Backup(
@ -1018,8 +1018,10 @@ async def test_reader_writer_create_addon_folder_error(
supervisor_client.jobs.get_job.side_effect = [
TEST_JOB_NOT_DONE,
supervisor_jobs.Job.from_dict(
load_json_object_fixture(
"backup_done_with_addon_folder_errors.json", DOMAIN
(
await async_load_json_object_fixture(
hass, "backup_done_with_addon_folder_errors.json", DOMAIN
)
)["data"]
),
]

47
tests/components/homematicip_cloud/test_init.py
View File

@ -176,8 +176,8 @@ async def test_hmip_dump_hap_config_services(
assert write_mock.mock_calls
async def test_setup_services_and_unload_services(hass: HomeAssistant) -> None:
"""Test setup services and unload services."""
async def test_setup_services(hass: HomeAssistant) -> None:
"""Test setup services."""
mock_config = {HMIPC_AUTHTOKEN: "123", HMIPC_HAPID: "ABC123", HMIPC_NAME: "name"}
MockConfigEntry(domain=DOMAIN, data=mock_config).add_to_hass(hass)
@ -201,46 +201,3 @@ async def test_setup_services_and_unload_services(hass: HomeAssistant) -> None:
assert len(config_entries) == 1
await hass.config_entries.async_unload(config_entries[0].entry_id)
# Check services are removed
assert not hass.services.async_services().get(DOMAIN)
async def test_setup_two_haps_unload_one_by_one(hass: HomeAssistant) -> None:
"""Test setup two access points and unload one by one and check services."""
# Setup AP1
mock_config = {HMIPC_AUTHTOKEN: "123", HMIPC_HAPID: "ABC123", HMIPC_NAME: "name"}
MockConfigEntry(domain=DOMAIN, data=mock_config).add_to_hass(hass)
# Setup AP2
mock_config2 = {HMIPC_AUTHTOKEN: "123", HMIPC_HAPID: "ABC1234", HMIPC_NAME: "name2"}
MockConfigEntry(domain=DOMAIN, data=mock_config2).add_to_hass(hass)
with patch("homeassistant.components.homematicip_cloud.HomematicipHAP") as mock_hap:
instance = mock_hap.return_value
instance.async_setup = AsyncMock(return_value=True)
instance.home.id = "1"
instance.home.modelType = "mock-type"
instance.home.name = "mock-name"
instance.home.label = "mock-label"
instance.home.currentAPVersion = "mock-ap-version"
instance.async_reset = AsyncMock(return_value=True)
assert await async_setup_component(hass, DOMAIN, {})
hmipc_services = hass.services.async_services()[DOMAIN]
assert len(hmipc_services) == 9
config_entries = hass.config_entries.async_entries(DOMAIN)
assert len(config_entries) == 2
# unload the first AP
await hass.config_entries.async_unload(config_entries[0].entry_id)
# services still exists
hmipc_services = hass.services.async_services()[DOMAIN]
assert len(hmipc_services) == 9
# unload the second AP
await hass.config_entries.async_unload(config_entries[1].entry_id)
# Check services are removed
assert not hass.services.async_services().get(DOMAIN)

6
tests/components/hunterdouglas_powerview/test_config_flow.py
View File

@ -15,7 +15,7 @@ from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
from .const import DHCP_DATA, DISCOVERY_DATA, HOMEKIT_DATA, MOCK_SERIAL
from tests.common import MockConfigEntry, load_json_object_fixture
from tests.common import MockConfigEntry, async_load_json_object_fixture
@pytest.mark.usefixtures("mock_hunterdouglas_hub")
@ -330,7 +330,9 @@ async def test_form_unsupported_device(
# Simulate a gen 3 secondary hub
with patch(
"homeassistant.components.hunterdouglas_powerview.util.Hub.request_raw_data",
return_value=load_json_object_fixture("gen3/gateway/secondary.json", DOMAIN),
return_value=await async_load_json_object_fixture(
hass, "gen3/gateway/secondary.json", DOMAIN
),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],

2
tests/components/nexia/test_switch.py
View File

@ -29,7 +29,7 @@ async def test_nexia_sensor_switch(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test NexiaRoomIQSensorSwitch."""
await async_init_integration(hass, house_fixture="nexia/sensors_xl1050_house.json")
await async_init_integration(hass, house_fixture="sensors_xl1050_house.json")
sw1_id = f"{Platform.SWITCH}.center_nativezone_include_center"
sw1 = {ATTR_ENTITY_ID: sw1_id}
sw2_id = f"{Platform.SWITCH}.center_nativezone_include_upstairs"

19
tests/components/nexia/util.py
View File

@ -9,7 +9,7 @@ from homeassistant.components.nexia.const import DOMAIN
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry, load_fixture
from tests.common import MockConfigEntry, async_load_fixture
from tests.test_util.aiohttp import mock_aiohttp_client
@ -18,13 +18,13 @@ async def async_init_integration(
skip_setup: bool = False,
exception: Exception | None = None,
*,
house_fixture="nexia/mobile_houses_123456.json",
house_fixture="mobile_houses_123456.json",
) -> MockConfigEntry:
"""Set up the nexia integration in Home Assistant."""
session_fixture = "nexia/session_123456.json"
sign_in_fixture = "nexia/sign_in.json"
set_fan_speed_fixture = "nexia/set_fan_speed_2293892.json"
session_fixture = "session_123456.json"
sign_in_fixture = "sign_in.json"
set_fan_speed_fixture = "set_fan_speed_2293892.json"
with (
mock_aiohttp_client() as mock_session,
patch("nexia.home.load_or_create_uuid", return_value=uuid.uuid4()),
@ -40,19 +40,20 @@ async def async_init_integration(
)
else:
mock_session.post(
nexia.API_MOBILE_SESSION_URL, text=load_fixture(session_fixture)
nexia.API_MOBILE_SESSION_URL,
text=await async_load_fixture(hass, session_fixture, DOMAIN),
)
mock_session.get(
nexia.API_MOBILE_HOUSES_URL.format(house_id=123456),
text=load_fixture(house_fixture),
text=await async_load_fixture(hass, house_fixture, DOMAIN),
)
mock_session.post(
nexia.API_MOBILE_ACCOUNTS_SIGN_IN_URL,
text=load_fixture(sign_in_fixture),
text=await async_load_fixture(hass, sign_in_fixture, DOMAIN),
)
mock_session.post(
"https://www.mynexia.com/mobile/xxl_thermostats/2293892/fan_speed",
text=load_fixture(set_fan_speed_fixture),
text=await async_load_fixture(hass, set_fan_speed_fixture, DOMAIN),
)
entry = MockConfigEntry(
domain=DOMAIN,

14
tests/components/nuki/__init__.py
View File

@ -9,8 +9,8 @@ from .mock import MOCK_INFO, setup_nuki_integration
from tests.common import (
MockConfigEntry,
load_json_array_fixture,
load_json_object_fixture,
async_load_json_array_fixture,
async_load_json_object_fixture,
)
@ -21,15 +21,19 @@ async def init_integration(hass: HomeAssistant) -> MockConfigEntry:
mock.get("http://1.1.1.1:8080/info", json=MOCK_INFO)
mock.get(
"http://1.1.1.1:8080/list",
json=load_json_array_fixture("list.json", DOMAIN),
json=await async_load_json_array_fixture(hass, "list.json", DOMAIN),
)
mock.get(
"http://1.1.1.1:8080/callback/list",
json=load_json_object_fixture("callback_list.json", DOMAIN),
json=await async_load_json_object_fixture(
hass, "callback_list.json", DOMAIN
),
)
mock.get(
"http://1.1.1.1:8080/callback/add",
json=load_json_object_fixture("callback_add.json", DOMAIN),
json=await async_load_json_object_fixture(
hass, "callback_add.json", DOMAIN
),
)
entry = await setup_nuki_integration(hass)
await hass.config_entries.async_setup(entry.entry_id)

10
tests/components/overkiz/test_diagnostics.py
View File

@ -8,7 +8,7 @@ from homeassistant.components.overkiz.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from tests.common import MockConfigEntry, load_json_object_fixture
from tests.common import MockConfigEntry, async_load_json_object_fixture
from tests.components.diagnostics import (
get_diagnostics_for_config_entry,
get_diagnostics_for_device,
@ -23,7 +23,9 @@ async def test_diagnostics(
snapshot: SnapshotAssertion,
) -> None:
"""Test diagnostics."""
diagnostic_data = load_json_object_fixture("overkiz/setup_tahoma_switch.json")
diagnostic_data = await async_load_json_object_fixture(
hass, "setup_tahoma_switch.json", DOMAIN
)
with patch.multiple(
"pyoverkiz.client.OverkizClient",
@ -44,7 +46,9 @@ async def test_device_diagnostics(
snapshot: SnapshotAssertion,
) -> None:
"""Test device diagnostics."""
diagnostic_data = load_json_object_fixture("overkiz/setup_tahoma_switch.json")
diagnostic_data = await async_load_json_object_fixture(
hass, "setup_tahoma_switch.json", DOMAIN
)
device = device_registry.async_get_device(
identifiers={(DOMAIN, "rts://****-****-6867/16756006")}

37
tests/components/skybell/conftest.py
View File

@ -12,7 +12,7 @@ from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from tests.common import MockConfigEntry, load_fixture
from tests.common import MockConfigEntry, async_load_fixture
from tests.test_util.aiohttp import AiohttpClientMocker
USERNAME = "user"
@ -53,39 +53,41 @@ def create_entry(hass: HomeAssistant) -> MockConfigEntry:
return entry
async def set_aioclient_responses(aioclient_mock: AiohttpClientMocker) -> None:
async def set_aioclient_responses(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Set AioClient responses."""
aioclient_mock.get(
f"{BASE_URL}devices/{DEVICE_ID}/info/",
text=load_fixture("skybell/device_info.json"),
text=await async_load_fixture(hass, "device_info.json", DOMAIN),
)
aioclient_mock.get(
f"{BASE_URL}devices/{DEVICE_ID}/settings/",
text=load_fixture("skybell/device_settings.json"),
text=await async_load_fixture(hass, "device_settings.json", DOMAIN),
)
aioclient_mock.get(
f"{BASE_URL}devices/{DEVICE_ID}/activities/",
text=load_fixture("skybell/activities.json"),
text=await async_load_fixture(hass, "activities.json", DOMAIN),
)
aioclient_mock.get(
f"{BASE_URL}devices/",
text=load_fixture("skybell/device.json"),
text=await async_load_fixture(hass, "device.json", DOMAIN),
)
aioclient_mock.get(
USERS_ME_URL,
text=load_fixture("skybell/me.json"),
text=await async_load_fixture(hass, "me.json", DOMAIN),
)
aioclient_mock.post(
f"{BASE_URL}login/",
text=load_fixture("skybell/login.json"),
text=await async_load_fixture(hass, "login.json", DOMAIN),
)
aioclient_mock.get(
f"{BASE_URL}devices/{DEVICE_ID}/activities/1234567890ab1234567890ac/video/",
text=load_fixture("skybell/video.json"),
text=await async_load_fixture(hass, "video.json", DOMAIN),
)
aioclient_mock.get(
f"{BASE_URL}devices/{DEVICE_ID}/avatar/",
text=load_fixture("skybell/avatar.json"),
text=await async_load_fixture(hass, "avatar.json", DOMAIN),
)
aioclient_mock.get(
f"https://v3-production-devices-avatar.s3.us-west-2.amazonaws.com/{DEVICE_ID}.jpg",
@ -96,12 +98,12 @@ async def set_aioclient_responses(aioclient_mock: AiohttpClientMocker) -> None:
@pytest.fixture
async def connection(aioclient_mock: AiohttpClientMocker) -> None:
async def connection(hass: HomeAssistant, aioclient_mock: AiohttpClientMocker) -> None:
"""Fixture for good connection responses."""
await set_aioclient_responses(aioclient_mock)
await set_aioclient_responses(hass, aioclient_mock)
def create_skybell(hass: HomeAssistant) -> Skybell:
async def create_skybell(hass: HomeAssistant) -> Skybell:
"""Create Skybell object."""
skybell = Skybell(
username=USERNAME,
@ -109,14 +111,15 @@ def create_skybell(hass: HomeAssistant) -> Skybell:
get_devices=True,
session=async_get_clientsession(hass),
)
skybell._cache = orjson.loads(load_fixture("skybell/cache.json"))
skybell._cache = orjson.loads(await async_load_fixture(hass, "cache.json", DOMAIN))
return skybell
def mock_skybell(hass: HomeAssistant):
async def mock_skybell(hass: HomeAssistant):
"""Mock Skybell object."""
return patch(
"homeassistant.components.skybell.Skybell", return_value=create_skybell(hass)
"homeassistant.components.skybell.Skybell",
return_value=await create_skybell(hass),
)
@ -124,7 +127,7 @@ async def async_init_integration(hass: HomeAssistant) -> MockConfigEntry:
"""Set up the Skybell integration in Home Assistant."""
config_entry = create_entry(hass)
with mock_skybell(hass), patch("aioskybell.utils.async_save_cache"):
with await mock_skybell(hass), patch("aioskybell.utils.async_save_cache"):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()

6
tests/components/switchbot_cloud/test_sensor.py
View File

@ -12,7 +12,7 @@ from homeassistant.helpers import entity_registry as er
from . import configure_integration
from tests.common import load_json_object_fixture, snapshot_platform
from tests.common import async_load_json_object_fixture, snapshot_platform
async def test_meter(
@ -33,7 +33,9 @@ async def test_meter(
hubDeviceId="test-hub-id",
),
]
mock_get_status.return_value = load_json_object_fixture("meter_status.json", DOMAIN)
mock_get_status.return_value = await async_load_json_object_fixture(
hass, "meter_status.json", DOMAIN
)
with patch("homeassistant.components.switchbot_cloud.PLATFORMS", [Platform.SENSOR]):
entry = await configure_integration(hass)

53
tests/components/tplink_omada/conftest.py
View File

@ -1,6 +1,7 @@
"""Test fixtures for TP-Link Omada integration."""
from collections.abc import AsyncIterable, Generator
from collections.abc import AsyncGenerator, Generator
from functools import partial
import json
from unittest.mock import AsyncMock, MagicMock, patch
@ -23,7 +24,7 @@ from homeassistant.components.tplink_omada.const import DOMAIN
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, CONF_VERIFY_SSL
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry, load_fixture
from tests.common import MockConfigEntry, async_load_fixture
@pytest.fixture
@ -53,29 +54,33 @@ def mock_setup_entry() -> Generator[AsyncMock]:
@pytest.fixture
def mock_omada_site_client() -> Generator[AsyncMock]:
async def mock_omada_site_client(hass: HomeAssistant) -> AsyncGenerator[AsyncMock]:
"""Mock Omada site client."""
site_client = MagicMock()
gateway_data = json.loads(load_fixture("gateway-TL-ER7212PC.json", DOMAIN))
gateway_data = json.loads(
await async_load_fixture(hass, "gateway-TL-ER7212PC.json", DOMAIN)
)
gateway = OmadaGateway(gateway_data)
site_client.get_gateway = AsyncMock(return_value=gateway)
switch1_data = json.loads(load_fixture("switch-TL-SG3210XHP-M2.json", DOMAIN))
switch1_data = json.loads(
await async_load_fixture(hass, "switch-TL-SG3210XHP-M2.json", DOMAIN)
)
switch1 = OmadaSwitch(switch1_data)
site_client.get_switches = AsyncMock(return_value=[switch1])
devices_data = json.loads(load_fixture("devices.json", DOMAIN))
devices_data = json.loads(await async_load_fixture(hass, "devices.json", DOMAIN))
devices = [OmadaListDevice(d) for d in devices_data]
site_client.get_devices = AsyncMock(return_value=devices)
switch1_ports_data = json.loads(
load_fixture("switch-ports-TL-SG3210XHP-M2.json", DOMAIN)
await async_load_fixture(hass, "switch-ports-TL-SG3210XHP-M2.json", DOMAIN)
)
switch1_ports = [OmadaSwitchPortDetails(p) for p in switch1_ports_data]
site_client.get_switch_ports = AsyncMock(return_value=switch1_ports)
async def async_empty() -> AsyncIterable:
async def async_empty() -> AsyncGenerator:
for c in ():
yield c
@ -85,24 +90,30 @@ def mock_omada_site_client() -> Generator[AsyncMock]:
@pytest.fixture
def mock_omada_clients_only_site_client() -> Generator[AsyncMock]:
def mock_omada_clients_only_site_client(hass: HomeAssistant) -> Generator[AsyncMock]:
"""Mock Omada site client containing only client connection data."""
site_client = MagicMock()
site_client.get_switches = AsyncMock(return_value=[])
site_client.get_devices = AsyncMock(return_value=[])
site_client.get_switch_ports = AsyncMock(return_value=[])
site_client.get_client = AsyncMock(side_effect=_get_mock_client)
site_client.get_client = AsyncMock(side_effect=partial(_get_mock_client, hass))
site_client.get_known_clients.side_effect = _get_mock_known_clients
site_client.get_connected_clients.side_effect = _get_mock_connected_clients
site_client.get_known_clients.side_effect = partial(_get_mock_known_clients, hass)
site_client.get_connected_clients.side_effect = partial(
_get_mock_connected_clients, hass
)
return site_client
async def _get_mock_known_clients() -> AsyncIterable[OmadaNetworkClient]:
async def _get_mock_known_clients(
hass: HomeAssistant,
) -> AsyncGenerator[OmadaNetworkClient]:
"""Mock known clients of the Omada network."""
known_clients_data = json.loads(load_fixture("known-clients.json", DOMAIN))
known_clients_data = json.loads(
await async_load_fixture(hass, "known-clients.json", DOMAIN)
)
for c in known_clients_data:
if c["wireless"]:
yield OmadaWirelessClient(c)
@ -110,9 +121,13 @@ async def _get_mock_known_clients() -> AsyncIterable[OmadaNetworkClient]:
yield OmadaWiredClient(c)
async def _get_mock_connected_clients() -> AsyncIterable[OmadaConnectedClient]:
async def _get_mock_connected_clients(
hass: HomeAssistant,
) -> AsyncGenerator[OmadaConnectedClient]:
"""Mock connected clients of the Omada network."""
connected_clients_data = json.loads(load_fixture("connected-clients.json", DOMAIN))
connected_clients_data = json.loads(
await async_load_fixture(hass, "connected-clients.json", DOMAIN)
)
for c in connected_clients_data:
if c["wireless"]:
yield OmadaWirelessClient(c)
@ -120,9 +135,11 @@ async def _get_mock_connected_clients() -> AsyncIterable[OmadaConnectedClient]:
yield OmadaWiredClient(c)
def _get_mock_client(mac: str) -> OmadaNetworkClient:
async def _get_mock_client(hass: HomeAssistant, mac: str) -> OmadaNetworkClient:
"""Mock an Omada client."""
connected_clients_data = json.loads(load_fixture("connected-clients.json", DOMAIN))
connected_clients_data = json.loads(
await async_load_fixture(hass, "connected-clients.json", DOMAIN)
)
for c in connected_clients_data:
if c["mac"] == mac:

16
tests/components/youless/__init__.py
View File

@ -8,8 +8,8 @@ from homeassistant.core import HomeAssistant
from tests.common import (
MockConfigEntry,
load_json_array_fixture,
load_json_object_fixture,
async_load_json_array_fixture,
async_load_json_object_fixture,
)
@ -18,16 +18,22 @@ async def init_component(hass: HomeAssistant) -> MockConfigEntry:
with requests_mock.Mocker() as mock:
mock.get(
"http://1.1.1.1/d",
json=load_json_object_fixture("device.json", youless.DOMAIN),
json=await async_load_json_object_fixture(
hass, "device.json", youless.DOMAIN
),
)
mock.get(
"http://1.1.1.1/e",
json=load_json_array_fixture("enologic.json", youless.DOMAIN),
json=await async_load_json_array_fixture(
hass, "enologic.json", youless.DOMAIN
),
headers={"Content-Type": "application/json"},
)
mock.get(
"http://1.1.1.1/f",
json=load_json_object_fixture("phase.json", youless.DOMAIN),
json=await async_load_json_object_fixture(
hass, "phase.json", youless.DOMAIN
),
headers={"Content-Type": "application/json"},
)