Compare commits

..

1 Commits

Author SHA1 Message Date
Mike Degatano
bd9d323c46 Add repair for deprecated arch addon issue 2026-03-13 23:06:35 +00:00
86 changed files with 290 additions and 5400 deletions

View File

@@ -570,7 +570,6 @@ homeassistant.components.trafikverket_train.*
homeassistant.components.trafikverket_weatherstation.*
homeassistant.components.transmission.*
homeassistant.components.trend.*
homeassistant.components.trmnl.*
homeassistant.components.tts.*
homeassistant.components.twentemilieu.*
homeassistant.components.unifi.*

4
CODEOWNERS generated
View File

@@ -1770,8 +1770,6 @@ build.json @home-assistant/supervisor
/tests/components/trend/ @jpbede
/homeassistant/components/triggercmd/ @rvmey
/tests/components/triggercmd/ @rvmey
/homeassistant/components/trmnl/ @joostlek
/tests/components/trmnl/ @joostlek
/homeassistant/components/tts/ @home-assistant/core
/tests/components/tts/ @home-assistant/core
/homeassistant/components/tuya/ @Tuya @zlinoliver
@@ -1788,8 +1786,6 @@ build.json @home-assistant/supervisor
/tests/components/ukraine_alarm/ @PaulAnnekov
/homeassistant/components/unifi/ @Kane610
/tests/components/unifi/ @Kane610
/homeassistant/components/unifi_access/ @imhotep @RaHehl
/tests/components/unifi_access/ @imhotep @RaHehl
/homeassistant/components/unifi_direct/ @tofuSCHNITZEL
/homeassistant/components/unifiled/ @florisvdk
/homeassistant/components/unifiprotect/ @RaHehl

View File

@@ -1,12 +1,5 @@
{
"domain": "ubiquiti",
"name": "Ubiquiti",
"integrations": [
"airos",
"unifi",
"unifi_access",
"unifi_direct",
"unifiled",
"unifiprotect"
]
"integrations": ["airos", "unifi", "unifi_direct", "unifiled", "unifiprotect"]
}

View File

@@ -1,5 +1,6 @@
"""Defines a base Alexa Devices entity."""
from aioamazondevices.const.devices import SPEAKER_GROUP_DEVICE_TYPE
from aioamazondevices.structures import AmazonDevice
from homeassistant.helpers.device_registry import DeviceInfo
@@ -24,15 +25,20 @@ class AmazonEntity(CoordinatorEntity[AmazonDevicesCoordinator]):
"""Initialize the entity."""
super().__init__(coordinator)
self._serial_num = serial_num
model = self.device.model
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, serial_num)},
name=self.device.account_name,
model=self.device.model,
model=model,
model_id=self.device.device_type,
manufacturer=self.device.manufacturer or "Amazon",
hw_version=self.device.hardware_version,
sw_version=self.device.software_version,
serial_number=serial_num,
sw_version=(
self.device.software_version
if model != SPEAKER_GROUP_DEVICE_TYPE
else None
),
serial_number=serial_num if model != SPEAKER_GROUP_DEVICE_TYPE else None,
)
self.entity_description = description
self._attr_unique_id = f"{serial_num}-{description.key}"

View File

@@ -8,5 +8,5 @@
"iot_class": "cloud_polling",
"loggers": ["aioamazondevices"],
"quality_scale": "platinum",
"requirements": ["aioamazondevices==13.0.1"]
"requirements": ["aioamazondevices==13.0.0"]
}

View File

@@ -17,7 +17,7 @@ from .coordinator import ArcamFmjConfigEntry, ArcamFmjCoordinator, ArcamFmjRunti
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [Platform.MEDIA_PLAYER, Platform.SENSOR]
PLATFORMS = [Platform.MEDIA_PLAYER]
async def async_setup_entry(hass: HomeAssistant, entry: ArcamFmjConfigEntry) -> bool:

View File

@@ -2,7 +2,6 @@
from __future__ import annotations
from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .coordinator import ArcamFmjCoordinator
@@ -13,16 +12,9 @@ class ArcamFmjEntity(CoordinatorEntity[ArcamFmjCoordinator]):
_attr_has_entity_name = True
def __init__(
self,
coordinator: ArcamFmjCoordinator,
description: EntityDescription | None = None,
) -> None:
def __init__(self, coordinator: ArcamFmjCoordinator) -> None:
"""Initialize the entity."""
super().__init__(coordinator)
self._attr_device_info = coordinator.device_info
self._attr_entity_registry_enabled_default = coordinator.state.zn == 1
self._attr_unique_id = coordinator.zone_unique_id
if description is not None:
self._attr_unique_id = f"{self._attr_unique_id}-{description.key}"
self.entity_description = description

View File

@@ -1,162 +0,0 @@
"""Arcam sensors for incoming stream info."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from arcam.fmj import IncomingVideoAspectRatio, IncomingVideoColorspace
from arcam.fmj.state import IncomingAudioConfig, IncomingAudioFormat, State
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import EntityCategory, UnitOfFrequency
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import ArcamFmjConfigEntry
from .entity import ArcamFmjEntity
@dataclass(frozen=True, kw_only=True)
class ArcamFmjSensorEntityDescription(SensorEntityDescription):
"""Describes an Arcam FMJ sensor entity."""
value_fn: Callable[[State], int | float | str | None]
SENSORS: tuple[ArcamFmjSensorEntityDescription, ...] = (
ArcamFmjSensorEntityDescription(
key="incoming_video_horizontal_resolution",
translation_key="incoming_video_horizontal_resolution",
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement="px",
suggested_display_precision=0,
value_fn=lambda state: (
vp.horizontal_resolution
if (vp := state.get_incoming_video_parameters()) is not None
else None
),
),
ArcamFmjSensorEntityDescription(
key="incoming_video_vertical_resolution",
translation_key="incoming_video_vertical_resolution",
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement="px",
suggested_display_precision=0,
value_fn=lambda state: (
vp.vertical_resolution
if (vp := state.get_incoming_video_parameters()) is not None
else None
),
),
ArcamFmjSensorEntityDescription(
key="incoming_video_refresh_rate",
translation_key="incoming_video_refresh_rate",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.FREQUENCY,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfFrequency.HERTZ,
suggested_display_precision=0,
value_fn=lambda state: (
vp.refresh_rate
if (vp := state.get_incoming_video_parameters()) is not None
else None
),
),
ArcamFmjSensorEntityDescription(
key="incoming_video_aspect_ratio",
translation_key="incoming_video_aspect_ratio",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.ENUM,
options=[member.name.lower() for member in IncomingVideoAspectRatio],
value_fn=lambda state: (
vp.aspect_ratio.name.lower()
if (vp := state.get_incoming_video_parameters()) is not None
else None
),
),
ArcamFmjSensorEntityDescription(
key="incoming_video_colorspace",
translation_key="incoming_video_colorspace",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.ENUM,
options=[member.name.lower() for member in IncomingVideoColorspace],
value_fn=lambda state: (
vp.colorspace.name.lower()
if (vp := state.get_incoming_video_parameters()) is not None
else None
),
),
ArcamFmjSensorEntityDescription(
key="incoming_audio_format",
translation_key="incoming_audio_format",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.ENUM,
options=[member.name.lower() for member in IncomingAudioFormat],
value_fn=lambda state: (
result.name.lower()
if (result := state.get_incoming_audio_format()[0]) is not None
else None
),
),
ArcamFmjSensorEntityDescription(
key="incoming_audio_config",
translation_key="incoming_audio_config",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.ENUM,
options=[member.name.lower() for member in IncomingAudioConfig],
value_fn=lambda state: (
result.name.lower()
if (result := state.get_incoming_audio_format()[1]) is not None
else None
),
),
ArcamFmjSensorEntityDescription(
key="incoming_audio_sample_rate",
translation_key="incoming_audio_sample_rate",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.FREQUENCY,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfFrequency.HERTZ,
suggested_display_precision=0,
value_fn=lambda state: (
None
if (sample_rate := state.get_incoming_audio_sample_rate()) == 0
else sample_rate
),
),
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ArcamFmjConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Arcam FMJ sensors from a config entry."""
coordinators = config_entry.runtime_data.coordinators
entities: list[ArcamFmjSensorEntity] = []
for coordinator in coordinators.values():
entities.extend(
ArcamFmjSensorEntity(coordinator, description) for description in SENSORS
)
async_add_entities(entities)
class ArcamFmjSensorEntity(ArcamFmjEntity, SensorEntity):
"""Representation of an Arcam FMJ sensor."""
entity_description: ArcamFmjSensorEntityDescription
@property
def native_value(self) -> int | float | str | None:
"""Return the sensor value."""
return self.entity_description.value_fn(self.coordinator.state)

View File

@@ -23,116 +23,5 @@
"trigger_type": {
"turn_on": "{entity_name} was requested to turn on"
}
},
"entity": {
"sensor": {
"incoming_audio_config": {
"name": "Incoming audio configuration",
"state": {
"auro_10_1": "Auro 10.1",
"auro_11_1": "Auro 11.1",
"auro_13_1": "Auro 13.1",
"auro_2_2_2": "Auro 2.2.2",
"auro_5_0": "Auro 5.0",
"auro_5_1": "Auro 5.1",
"auro_8_0": "Auro 8.0",
"auro_9_1": "Auro 9.1",
"auro_quad": "Auro quad",
"dual_mono": "Dual mono",
"dual_mono_lfe": "Dual mono + LFE",
"mono": "Mono",
"mono_lfe": "Mono + LFE",
"stereo_center": "Stereo center",
"stereo_center_lfe": "Stereo center + LFE",
"stereo_center_surr_lr": "Stereo center surround L/R",
"stereo_center_surr_lr_back_lr": "Stereo center surround L/R back L/R",
"stereo_center_surr_lr_back_lr_lfe": "Stereo center surround L/R back L/R + LFE",
"stereo_center_surr_lr_back_matrix": "Stereo center surround L/R back matrix",
"stereo_center_surr_lr_back_matrix_lfe": "Stereo center surround L/R back matrix + LFE",
"stereo_center_surr_lr_back_mono": "Stereo center surround L/R back mono",
"stereo_center_surr_lr_back_mono_lfe": "Stereo center surround L/R back mono + LFE",
"stereo_center_surr_lr_lfe": "Stereo center surround L/R + LFE",
"stereo_center_surr_mono": "Stereo center surround mono",
"stereo_center_surr_mono_lfe": "Stereo center surround mono + LFE",
"stereo_downmix": "Stereo downmix",
"stereo_downmix_lfe": "Stereo downmix + LFE",
"stereo_lfe": "Stereo + LFE",
"stereo_only": "Stereo only",
"stereo_only_lo_ro": "Stereo only Lo/Ro",
"stereo_only_lo_ro_lfe": "Stereo only Lo/Ro + LFE",
"stereo_surr_lr": "Stereo surround L/R",
"stereo_surr_lr_back_lr": "Stereo surround L/R back L/R",
"stereo_surr_lr_back_lr_lfe": "Stereo surround L/R back L/R + LFE",
"stereo_surr_lr_back_matrix": "Stereo surround L/R back matrix",
"stereo_surr_lr_back_matrix_lfe": "Stereo surround L/R back matrix + LFE",
"stereo_surr_lr_back_mono": "Stereo surround L/R back mono",
"stereo_surr_lr_back_mono_lfe": "Stereo surround L/R back mono + LFE",
"stereo_surr_lr_lfe": "Stereo surround L/R + LFE",
"stereo_surr_mono": "Stereo surround mono",
"stereo_surr_mono_lfe": "Stereo surround mono + LFE",
"undetected": "Undetected",
"unknown": "Unknown"
}
},
"incoming_audio_format": {
"name": "Incoming audio format",
"state": {
"analogue_direct": "Analogue direct",
"auro_3d": "Auro-3D",
"dolby_atmos": "Dolby Atmos",
"dolby_digital": "Dolby Digital",
"dolby_digital_ex": "Dolby Digital EX",
"dolby_digital_plus": "Dolby Digital Plus",
"dolby_digital_surround": "Dolby Digital Surround",
"dolby_digital_true_hd": "Dolby TrueHD",
"dts": "DTS",
"dts_96_24": "DTS 96/24",
"dts_core": "DTS Core",
"dts_es_discrete": "DTS-ES Discrete",
"dts_es_discrete_96_24": "DTS-ES Discrete 96/24",
"dts_es_matrix": "DTS-ES Matrix",
"dts_es_matrix_96_24": "DTS-ES Matrix 96/24",
"dts_hd_high_res_audio": "DTS-HD High Resolution Audio",
"dts_hd_master_audio": "DTS-HD Master Audio",
"dts_low_bit_rate": "DTS Low Bit Rate",
"dts_x": "DTS:X",
"imax_enhanced": "IMAX Enhanced",
"pcm": "PCM",
"pcm_zero": "PCM zero",
"undetected": "Undetected",
"unsupported": "Unsupported"
}
},
"incoming_audio_sample_rate": {
"name": "Incoming audio sample rate"
},
"incoming_video_aspect_ratio": {
"name": "Incoming video aspect ratio",
"state": {
"aspect_16_9": "16:9",
"aspect_4_3": "4:3",
"undefined": "Undefined"
}
},
"incoming_video_colorspace": {
"name": "Incoming video colorspace",
"state": {
"dolby_vision": "Dolby Vision",
"hdr10": "HDR10",
"hdr10_plus": "HDR10+",
"hlg": "HLG",
"normal": "Normal"
}
},
"incoming_video_horizontal_resolution": {
"name": "Incoming video horizontal resolution"
},
"incoming_video_refresh_rate": {
"name": "Incoming video refresh rate"
},
"incoming_video_vertical_resolution": {
"name": "Incoming video vertical resolution"
}
}
}
}

View File

@@ -15,6 +15,7 @@ async def async_setup_intents(hass: HomeAssistant) -> None:
INTENT_OPEN_COVER,
DOMAIN,
SERVICE_OPEN_COVER,
"Opening {}",
description="Opens a cover",
platforms={DOMAIN},
device_classes={CoverDeviceClass},
@@ -26,6 +27,7 @@ async def async_setup_intents(hass: HomeAssistant) -> None:
INTENT_CLOSE_COVER,
DOMAIN,
SERVICE_CLOSE_COVER,
"Closing {}",
description="Closes a cover",
platforms={DOMAIN},
device_classes={CoverDeviceClass},

View File

@@ -130,6 +130,7 @@ ISSUE_KEY_ADDON_DETACHED_ADDON_REMOVED = "issue_addon_detached_addon_removed"
ISSUE_KEY_ADDON_PWNED = "issue_addon_pwned"
ISSUE_KEY_SYSTEM_FREE_SPACE = "issue_system_free_space"
ISSUE_KEY_ADDON_DEPRECATED = "issue_addon_deprecated_addon"
ISSUE_KEY_ADDON_DEPRECATED_ARCH = "issue_addon_deprecated_arch_addon"
ISSUE_MOUNT_MOUNT_FAILED = "issue_mount_mount_failed"
@@ -170,6 +171,7 @@ EXTRA_PLACEHOLDERS = {
"more_info_pwned": "https://www.home-assistant.io/more-info/pwned-passwords",
},
ISSUE_KEY_ADDON_DEPRECATED: HELP_URLS,
ISSUE_KEY_ADDON_DEPRECATED_ARCH: HELP_URLS,
}

View File

@@ -45,6 +45,7 @@ from .const import (
EVENT_SUPPORTED_CHANGED,
EXTRA_PLACEHOLDERS,
ISSUE_KEY_ADDON_BOOT_FAIL,
ISSUE_KEY_ADDON_DEPRECATED_ARCH,
ISSUE_KEY_ADDON_DETACHED_ADDON_MISSING,
ISSUE_KEY_ADDON_DETACHED_ADDON_REMOVED,
ISSUE_KEY_ADDON_PWNED,
@@ -88,6 +89,7 @@ ISSUE_KEYS_FOR_REPAIRS = {
"issue_system_disk_lifetime",
ISSUE_KEY_SYSTEM_FREE_SPACE,
ISSUE_KEY_ADDON_PWNED,
ISSUE_KEY_ADDON_DEPRECATED_ARCH,
}
_LOGGER = logging.getLogger(__name__)

View File

@@ -19,6 +19,7 @@ from .const import (
EXTRA_PLACEHOLDERS,
ISSUE_KEY_ADDON_BOOT_FAIL,
ISSUE_KEY_ADDON_DEPRECATED,
ISSUE_KEY_ADDON_DEPRECATED_ARCH,
ISSUE_KEY_ADDON_DETACHED_ADDON_REMOVED,
ISSUE_KEY_ADDON_PWNED,
ISSUE_KEY_SYSTEM_DOCKER_CONFIG,
@@ -231,6 +232,7 @@ async def async_create_fix_flow(
ISSUE_KEY_ADDON_DETACHED_ADDON_REMOVED,
ISSUE_KEY_ADDON_BOOT_FAIL,
ISSUE_KEY_ADDON_PWNED,
ISSUE_KEY_ADDON_DEPRECATED_ARCH,
}:
return AddonIssueRepairFlow(hass, issue_id)

View File

@@ -85,6 +85,19 @@
},
"title": "Installed app is deprecated"
},
"issue_addon_deprecated_arch_addon": {
"fix_flow": {
"abort": {
"apply_suggestion_fail": "Could not uninstall the app. Check the Supervisor logs for more details."
},
"step": {
"addon_execute_remove": {
"description": "App {addon} only supports architectures and/or machines which are no longer supported by Home Assistant. It will not be able to receive updates and will stop working in a future release.\n\nSelecting **Submit** will uninstall this deprecated app. Alternatively, you can check [Home Assistant help]({help_url}) and the [community forum]({community_url}) for alternatives to migrate to."
}
}
},
"title": "Installed app does not work on supported architectures and/or machines"
},
"issue_addon_detached_addon_missing": {
"description": "Repository for app {addon} is missing. This means it will not get updates, and backups may not be restored correctly as the Home Assistant Supervisor may not be able to build/download the resources required.\n\nPlease check the [app's documentation]({addon_url}) for installation instructions and add the repository to the store.",
"title": "Missing repository for an installed app"

View File

@@ -10,7 +10,7 @@ from pylitterbot import Account
from pylitterbot.exceptions import LitterRobotException, LitterRobotLoginException
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.helpers.aiohttp_client import async_get_clientsession
@@ -21,7 +21,6 @@ _LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{vol.Required(CONF_USERNAME): str, vol.Required(CONF_PASSWORD): str}
)
STEP_REAUTH_RECONFIGURE_SCHEMA = vol.Schema({vol.Required(CONF_PASSWORD): str})
class LitterRobotConfigFlow(ConfigFlow, domain=DOMAIN):
@@ -44,45 +43,24 @@ class LitterRobotConfigFlow(ConfigFlow, domain=DOMAIN):
self, user_input: dict[str, str] | None = None
) -> ConfigFlowResult:
"""Handle user's reauth credentials."""
errors: dict[str, str] = {}
errors = {}
if user_input:
reauth_entry = self._get_reauth_entry()
result, errors = await self._async_validate_and_update_entry(
reauth_entry, user_input
)
if result is not None:
return result
user_input = user_input | {CONF_USERNAME: self.username}
if not (error := await self._async_validate_input(user_input)):
await self.async_set_unique_id(self._account_user_id)
self._abort_if_unique_id_mismatch()
return self.async_update_reload_and_abort(
self._get_reauth_entry(), data_updates=user_input
)
errors["base"] = error
return self.async_show_form(
step_id="reauth_confirm",
data_schema=STEP_REAUTH_RECONFIGURE_SCHEMA,
data_schema=vol.Schema({vol.Required(CONF_PASSWORD): str}),
description_placeholders={CONF_USERNAME: self.username},
errors=errors,
)
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a reconfiguration flow request."""
reconfigure_entry = self._get_reconfigure_entry()
self.username = reconfigure_entry.data[CONF_USERNAME]
self._async_abort_entries_match({CONF_USERNAME: self.username})
errors: dict[str, str] = {}
if user_input:
result, errors = await self._async_validate_and_update_entry(
reconfigure_entry, user_input
)
if result is not None:
return result
return self.async_show_form(
step_id="reconfigure",
data_schema=STEP_REAUTH_RECONFIGURE_SCHEMA,
errors=errors,
)
async def async_step_user(
self, user_input: Mapping[str, Any] | None = None
) -> ConfigFlowResult:
@@ -103,25 +81,6 @@ class LitterRobotConfigFlow(ConfigFlow, domain=DOMAIN):
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)
async def _async_validate_and_update_entry(
self, entry: ConfigEntry, user_input: dict[str, Any]
) -> tuple[ConfigFlowResult | None, dict[str, str]]:
"""Validate credentials and update an existing entry if valid."""
errors: dict[str, str] = {}
full_input: dict[str, Any] = user_input | {CONF_USERNAME: self.username}
if not (error := await self._async_validate_input(full_input)):
await self.async_set_unique_id(self._account_user_id)
self._abort_if_unique_id_mismatch()
return (
self.async_update_reload_and_abort(
entry,
data_updates=full_input,
),
errors,
)
errors["base"] = error
return None, errors
async def _async_validate_input(self, user_input: Mapping[str, Any]) -> str:
"""Validate login credentials."""
account = Account(websession=async_get_clientsession(self.hass))

View File

@@ -53,14 +53,10 @@ class LitterRobotDataUpdateCoordinator(DataUpdateCoordinator[None]):
# Need to fetch weight history for `get_visits_since`
await pet.fetch_weight_history()
except LitterRobotLoginException as ex:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN, translation_key="invalid_credentials"
) from ex
raise ConfigEntryAuthFailed("Invalid credentials") from ex
except LitterRobotException as ex:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="cannot_connect",
translation_placeholders={"error": str(ex)},
f"Unable to fetch data from the Whisker API: {ex}"
) from ex
async def _async_setup(self) -> None:
@@ -74,15 +70,9 @@ class LitterRobotDataUpdateCoordinator(DataUpdateCoordinator[None]):
load_pets=True,
)
except LitterRobotLoginException as ex:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN, translation_key="invalid_credentials"
) from ex
raise ConfigEntryAuthFailed("Invalid credentials") from ex
except LitterRobotException as ex:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="cannot_connect",
translation_placeholders={"error": str(ex)},
) from ex
raise UpdateFailed("Unable to connect to Whisker API") from ex
def litter_robots(self) -> Generator[LitterRobot]:
"""Get Litter-Robots from the account."""

View File

@@ -49,21 +49,23 @@ rules:
discovery:
status: todo
comment: Need to validate discovery
docs-data-update: done
docs-examples: done
docs-known-limitations: done
docs-supported-devices: done
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
docs-supported-functions: done
docs-troubleshooting: done
docs-use-cases: done
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices: todo
entity-category: done
entity-device-class: done
entity-disabled-by-default: done
entity-translations: done
exception-translations: done
entity-translations:
status: todo
comment: Make sure all translated states are in sentence case
exception-translations: todo
icon-translations: done
reconfiguration-flow: done
reconfiguration-flow: todo
repair-issues:
status: done
comment: |

View File

@@ -3,7 +3,6 @@
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]",
"unique_id_mismatch": "The Whisker account does not match the previously configured account. Please re-authenticate using the same account, or remove this integration and set it up again if you want to use a different account."
},
"error": {
@@ -22,14 +21,6 @@
"description": "Please update your password for {username}",
"title": "[%key:common::config_flow::title::reauth%]"
},
"reconfigure": {
"data": {
"password": "[%key:common::config_flow::data::password%]"
},
"data_description": {
"password": "[%key:component::litterrobot::config::step::user::data_description::password%]"
}
},
"user": {
"data": {
"password": "[%key:common::config_flow::data::password%]",
@@ -205,17 +196,11 @@
}
},
"exceptions": {
"cannot_connect": {
"message": "Unable to fetch data from the Whisker API: {error}"
},
"command_failed": {
"message": "An error occurred while communicating with the device: {error}"
},
"firmware_update_failed": {
"message": "Unable to start firmware update on {name}"
},
"invalid_credentials": {
"message": "Invalid credentials. Please check your username and password, then try again"
}
},
"issues": {

View File

@@ -500,7 +500,6 @@ class DishWasherProgramId(MieleEnum, missing_to_none=True):
pasta_paela = 14
tall_items = 17, 42
glasses_warm = 19
quick_intense = 21
normal = 30
power_wash = 44, 204
comfort_wash = 203

View File

@@ -759,7 +759,6 @@
"pyrolytic": "Pyrolytic",
"quiche_lorraine": "Quiche Lorraine",
"quick_hygiene": "QuickHygiene",
"quick_intense": "QuickIntense",
"quick_mw": "Quick MW",
"quick_power_dry": "QuickPowerDry",
"quick_power_wash": "QuickPowerWash",

View File

@@ -12,5 +12,5 @@
"integration_type": "device",
"iot_class": "local_polling",
"quality_scale": "platinum",
"requirements": ["python-pooldose==0.8.5"]
"requirements": ["python-pooldose==0.8.2"]
}

View File

@@ -200,14 +200,6 @@ CAPABILITY_TO_SENSORS: dict[
supported_states_attributes=Attribute.SUPPORTED_STATUS,
)
},
Capability.CUSTOM_COOKTOP_OPERATING_STATE: {
Attribute.COOKTOP_OPERATING_STATE: SmartThingsBinarySensorEntityDescription(
key=Attribute.COOKTOP_OPERATING_STATE,
translation_key="cooktop_operating_state",
is_on_key="run",
supported_states_attributes=Attribute.SUPPORTED_COOKTOP_OPERATING_STATE,
)
},
}

View File

@@ -25,10 +25,6 @@
"hostname": "hub*",
"macaddress": "286D97*"
},
{
"hostname": "smarthub",
"macaddress": "683A48*"
},
{
"hostname": "samsung-*"
}

View File

@@ -49,9 +49,6 @@
"child_lock": {
"name": "Child lock"
},
"cooktop_operating_state": {
"name": "[%key:component::smartthings::entity::sensor::cooktop_operating_state::name%]"
},
"cool_select_plus_door": {
"name": "CoolSelect+ door"
},

View File

@@ -7,6 +7,7 @@ import io
import logging
import os
from pathlib import Path
from ssl import SSLContext
from types import MappingProxyType
from typing import Any, cast
@@ -47,8 +48,8 @@ from homeassistant.const import (
from homeassistant.core import Context, HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.httpx_client import get_async_client
from homeassistant.util.json import JsonValueType
from homeassistant.util.ssl import get_default_context, get_default_no_verify_context
from .const import (
ATTR_ARGS,
@@ -565,7 +566,11 @@ class TelegramNotificationService:
username=kwargs.get(ATTR_USERNAME, ""),
password=kwargs.get(ATTR_PASSWORD, ""),
authentication=kwargs.get(ATTR_AUTHENTICATION),
verify_ssl=kwargs.get(ATTR_VERIFY_SSL, False),
verify_ssl=(
get_default_context()
if kwargs.get(ATTR_VERIFY_SSL, False)
else get_default_no_verify_context()
),
)
media: InputMedia
@@ -733,7 +738,11 @@ class TelegramNotificationService:
username=kwargs.get(ATTR_USERNAME, ""),
password=kwargs.get(ATTR_PASSWORD, ""),
authentication=kwargs.get(ATTR_AUTHENTICATION),
verify_ssl=kwargs.get(ATTR_VERIFY_SSL, False),
verify_ssl=(
get_default_context()
if kwargs.get(ATTR_VERIFY_SSL, False)
else get_default_no_verify_context()
),
)
if file_type == SERVICE_SEND_PHOTO:
@@ -1046,7 +1055,7 @@ async def load_data(
username: str,
password: str,
authentication: str | None,
verify_ssl: bool,
verify_ssl: SSLContext,
num_retries: int = 5,
) -> io.BytesIO:
"""Load data into ByteIO/File container from a source."""
@@ -1062,13 +1071,16 @@ async def load_data(
elif authentication == HTTP_BASIC_AUTHENTICATION:
params["auth"] = httpx.BasicAuth(username, password)
if verify_ssl is not None:
params["verify"] = verify_ssl
retry_num = 0
async with get_async_client(hass, verify_ssl) as client:
async with httpx.AsyncClient(
timeout=DEFAULT_TIMEOUT_SECONDS, headers=headers, **params
) as client:
while retry_num < num_retries:
try:
response = await client.get(
url, headers=headers, timeout=DEFAULT_TIMEOUT_SECONDS, **params
)
req = await client.get(url)
except (httpx.HTTPError, httpx.InvalidURL) as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
@@ -1076,15 +1088,15 @@ async def load_data(
translation_placeholders={"error": str(err)},
) from err
if response.status_code != 200:
if req.status_code != 200:
_LOGGER.warning(
"Status code %s (retry #%s) loading %s",
response.status_code,
req.status_code,
retry_num + 1,
url,
)
else:
data = io.BytesIO(response.content)
data = io.BytesIO(req.content)
if data.read():
data.seek(0)
data.name = url
@@ -1099,7 +1111,7 @@ async def load_data(
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="failed_to_load_url",
translation_placeholders={"error": str(response.status_code)},
translation_placeholders={"error": str(req.status_code)},
)
elif filepath is not None:
if hass.config.is_allowed_path(filepath):

View File

@@ -1,29 +0,0 @@
"""The TRMNL integration."""
from __future__ import annotations
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from .coordinator import TRMNLConfigEntry, TRMNLCoordinator
PLATFORMS: list[Platform] = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: TRMNLConfigEntry) -> bool:
"""Set up TRMNL from a config entry."""
coordinator = TRMNLCoordinator(hass, entry)
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: TRMNLConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

View File

@@ -1,49 +0,0 @@
"""Config flow for TRMNL."""
from __future__ import annotations
from typing import Any
from trmnl import TRMNLClient
from trmnl.exceptions import TRMNLAuthenticationError, TRMNLError
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_API_KEY
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN, LOGGER
class TRMNLConfigFlow(ConfigFlow, domain=DOMAIN):
"""TRMNL config flow."""
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a flow initialized by the user."""
errors: dict[str, str] = {}
if user_input:
session = async_get_clientsession(self.hass)
client = TRMNLClient(token=user_input[CONF_API_KEY], session=session)
try:
user = await client.get_me()
except TRMNLAuthenticationError:
errors["base"] = "invalid_auth"
except TRMNLError:
errors["base"] = "cannot_connect"
except Exception: # noqa: BLE001
LOGGER.exception("Unexpected error")
errors["base"] = "unknown"
else:
await self.async_set_unique_id(str(user.identifier))
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=user.name,
data={CONF_API_KEY: user_input[CONF_API_KEY]},
)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema({vol.Required(CONF_API_KEY): str}),
errors=errors,
)

View File

@@ -1,7 +0,0 @@
"""Constants for the TRMNL integration."""
import logging
DOMAIN = "trmnl"
LOGGER = logging.getLogger(__package__)

View File

@@ -1,57 +0,0 @@
"""Define an object to manage fetching TRMNL data."""
from __future__ import annotations
from datetime import timedelta
from trmnl import TRMNLClient
from trmnl.exceptions import TRMNLAuthenticationError, TRMNLError
from trmnl.models import Device
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY
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 .const import DOMAIN, LOGGER
type TRMNLConfigEntry = ConfigEntry[TRMNLCoordinator]
class TRMNLCoordinator(DataUpdateCoordinator[dict[int, Device]]):
"""Class to manage fetching TRMNL data."""
config_entry: TRMNLConfigEntry
def __init__(self, hass: HomeAssistant, config_entry: TRMNLConfigEntry) -> None:
"""Initialize coordinator."""
super().__init__(
hass,
logger=LOGGER,
config_entry=config_entry,
name=DOMAIN,
update_interval=timedelta(hours=1),
)
self.client = TRMNLClient(
token=config_entry.data[CONF_API_KEY],
session=async_get_clientsession(hass),
)
async def _async_update_data(self) -> dict[int, Device]:
"""Fetch data from TRMNL."""
try:
devices = await self.client.get_devices()
except TRMNLAuthenticationError as err:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="authentication_error",
) from err
except TRMNLError as err:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="update_error",
translation_placeholders={"error": str(err)},
) from err
return {device.identifier: device for device in devices}

View File

@@ -1,37 +0,0 @@
"""Base class for TRMNL entities."""
from __future__ import annotations
from trmnl.models import Device
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .coordinator import TRMNLCoordinator
class TRMNLEntity(CoordinatorEntity[TRMNLCoordinator]):
"""Defines a base TRMNL entity."""
_attr_has_entity_name = True
def __init__(self, coordinator: TRMNLCoordinator, device_id: int) -> None:
"""Initialize TRMNL entity."""
super().__init__(coordinator)
self._device_id = device_id
device = self._device
self._attr_device_info = DeviceInfo(
connections={(CONNECTION_NETWORK_MAC, device.mac_address)},
name=device.name,
manufacturer="TRMNL",
)
@property
def _device(self) -> Device:
"""Return the device from coordinator data."""
return self.coordinator.data[self._device_id]
@property
def available(self) -> bool:
"""Return if the device is available."""
return super().available and self._device_id in self.coordinator.data

View File

@@ -1,11 +0,0 @@
{
"domain": "trmnl",
"name": "TRMNL",
"codeowners": ["@joostlek"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/trmnl",
"integration_type": "hub",
"iot_class": "cloud_polling",
"quality_scale": "bronze",
"requirements": ["trmnl==0.1.0"]
}

View File

@@ -1,74 +0,0 @@
rules:
# 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: done
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: Entities of this integration do not explicitly subscribe to events.
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:
status: exempt
comment: There are no configuration parameters
docs-installation-parameters: todo
entity-unavailable: done
integration-owner: done
log-when-unavailable: done
parallel-updates: done
reauthentication-flow: todo
test-coverage: todo
# Gold
devices: done
diagnostics: todo
discovery-update-info:
status: exempt
comment: Uses the cloud API
discovery:
status: exempt
comment: Can't be discovered
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
docs-supported-functions: todo
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices: todo
entity-category: todo
entity-device-class: todo
entity-disabled-by-default: todo
entity-translations: done
exception-translations: todo
icon-translations: todo
reconfiguration-flow: todo
repair-issues:
status: exempt
comment: There are no repairable issues
stale-devices: todo
# Platinum
async-dependency: done
inject-websession: done
strict-typing: done

View File

@@ -1,92 +0,0 @@
"""Support for TRMNL sensors."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from trmnl.models import Device
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import (
PERCENTAGE,
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
EntityCategory,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import TRMNLConfigEntry
from .coordinator import TRMNLCoordinator
from .entity import TRMNLEntity
PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True)
class TRMNLSensorEntityDescription(SensorEntityDescription):
"""Describes a TRMNL sensor entity."""
value_fn: Callable[[Device], int | float | None]
SENSOR_DESCRIPTIONS: tuple[TRMNLSensorEntityDescription, ...] = (
TRMNLSensorEntityDescription(
key="battery",
device_class=SensorDeviceClass.BATTERY,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda device: device.percent_charged,
),
TRMNLSensorEntityDescription(
key="rssi",
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda device: device.rssi,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: TRMNLConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up TRMNL sensor entities based on a config entry."""
coordinator = entry.runtime_data
async_add_entities(
TRMNLSensor(coordinator, device_id, description)
for device_id in coordinator.data
for description in SENSOR_DESCRIPTIONS
)
class TRMNLSensor(TRMNLEntity, SensorEntity):
"""Defines a TRMNL sensor."""
entity_description: TRMNLSensorEntityDescription
def __init__(
self,
coordinator: TRMNLCoordinator,
device_id: int,
description: TRMNLSensorEntityDescription,
) -> None:
"""Initialize TRMNL sensor."""
super().__init__(coordinator, device_id)
self.entity_description = description
self._attr_unique_id = f"{device_id}_{description.key}"
@property
def native_value(self) -> int | float | None:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self._device)

View File

@@ -1,30 +0,0 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
},
"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%]"
},
"step": {
"user": {
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]"
},
"data_description": {
"api_key": "The API key for your TRMNL account."
}
}
}
},
"exceptions": {
"authentication_error": {
"message": "Authentication failed. Please check your API key."
},
"update_error": {
"message": "An error occurred while communicating with TRMNL: {error}"
}
}
}

View File

@@ -1,54 +0,0 @@
"""The UniFi Access integration."""
from __future__ import annotations
from unifi_access_api import ApiAuthError, ApiConnectionError, UnifiAccessApiClient
from homeassistant.const import CONF_API_TOKEN, CONF_HOST, CONF_VERIFY_SSL, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .coordinator import UnifiAccessConfigEntry, UnifiAccessCoordinator
PLATFORMS: list[Platform] = [Platform.BUTTON, Platform.EVENT, Platform.SWITCH]
async def async_setup_entry(hass: HomeAssistant, entry: UnifiAccessConfigEntry) -> bool:
"""Set up UniFi Access from a config entry."""
session = async_get_clientsession(hass, verify_ssl=entry.data[CONF_VERIFY_SSL])
client = UnifiAccessApiClient(
host=entry.data[CONF_HOST],
api_token=entry.data[CONF_API_TOKEN],
session=session,
verify_ssl=entry.data[CONF_VERIFY_SSL],
)
try:
await client.authenticate()
except ApiAuthError as err:
raise ConfigEntryNotReady(
f"Authentication failed for UniFi Access at {entry.data[CONF_HOST]}"
) from err
except ApiConnectionError as err:
raise ConfigEntryNotReady(
f"Unable to connect to UniFi Access at {entry.data[CONF_HOST]}"
) from err
coordinator = UnifiAccessCoordinator(hass, entry, client)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
entry.async_on_unload(client.close)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(
hass: HomeAssistant, entry: UnifiAccessConfigEntry
) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

View File

@@ -1,53 +0,0 @@
"""Button platform for the UniFi Access integration."""
from __future__ import annotations
from unifi_access_api import Door, UnifiAccessError
from homeassistant.components.button import ButtonEntity
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN
from .coordinator import UnifiAccessConfigEntry, UnifiAccessCoordinator
from .entity import UnifiAccessEntity
PARALLEL_UPDATES = 1
async def async_setup_entry(
hass: HomeAssistant,
entry: UnifiAccessConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up UniFi Access button entities."""
coordinator = entry.runtime_data
async_add_entities(
UnifiAccessUnlockButton(coordinator, door)
for door in coordinator.data.doors.values()
)
class UnifiAccessUnlockButton(UnifiAccessEntity, ButtonEntity):
"""Representation of a UniFi Access door unlock button."""
_attr_translation_key = "unlock"
def __init__(
self,
coordinator: UnifiAccessCoordinator,
door: Door,
) -> None:
"""Initialize the button entity."""
super().__init__(coordinator, door, "unlock")
async def async_press(self) -> None:
"""Unlock the door."""
try:
await self.coordinator.client.unlock_door(self._door_id)
except UnifiAccessError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="unlock_failed",
) from err

View File

@@ -1,68 +0,0 @@
"""Config flow for UniFi Access integration."""
from __future__ import annotations
import logging
from typing import Any
from unifi_access_api import ApiAuthError, ApiConnectionError, UnifiAccessApiClient
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_API_TOKEN, CONF_HOST, CONF_VERIFY_SSL
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
class UnifiAccessConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for UniFi Access."""
VERSION = 1
MINOR_VERSION = 1
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:
session = async_get_clientsession(
self.hass, verify_ssl=user_input[CONF_VERIFY_SSL]
)
client = UnifiAccessApiClient(
host=user_input[CONF_HOST],
api_token=user_input[CONF_API_TOKEN],
session=session,
verify_ssl=user_input[CONF_VERIFY_SSL],
)
try:
await client.authenticate()
except ApiAuthError:
errors["base"] = "invalid_auth"
except ApiConnectionError:
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
self._async_abort_entries_match({CONF_HOST: user_input[CONF_HOST]})
return self.async_create_entry(
title="UniFi Access",
data=user_input,
)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Required(CONF_API_TOKEN): str,
vol.Required(CONF_VERIFY_SSL, default=False): bool,
}
),
errors=errors,
)

View File

@@ -1,3 +0,0 @@
"""Constants for the UniFi Access integration."""
DOMAIN = "unifi_access"

View File

@@ -1,240 +0,0 @@
"""Data update coordinator for the UniFi Access integration."""
from __future__ import annotations
import asyncio
from collections.abc import Callable
from dataclasses import dataclass
import logging
from typing import Any, cast
from unifi_access_api import (
ApiAuthError,
ApiConnectionError,
ApiError,
Door,
EmergencyStatus,
UnifiAccessApiClient,
WsMessageHandler,
)
from unifi_access_api.models.websocket import (
HwDoorbell,
InsightsAdd,
LocationUpdateState,
LocationUpdateV2,
SettingUpdate,
V2LocationState,
V2LocationUpdate,
WebsocketMessage,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
type UnifiAccessConfigEntry = ConfigEntry[UnifiAccessCoordinator]
@dataclass(frozen=True)
class DoorEvent:
"""Represent a door event from WebSocket."""
door_id: str
category: str
event_type: str
event_data: dict[str, Any]
@dataclass(frozen=True)
class UnifiAccessData:
"""Data provided by the UniFi Access coordinator."""
doors: dict[str, Door]
emergency: EmergencyStatus
class UnifiAccessCoordinator(DataUpdateCoordinator[UnifiAccessData]):
"""Coordinator for fetching UniFi Access door data."""
config_entry: UnifiAccessConfigEntry
def __init__(
self,
hass: HomeAssistant,
entry: UnifiAccessConfigEntry,
client: UnifiAccessApiClient,
) -> None:
"""Initialize the coordinator."""
super().__init__(
hass,
_LOGGER,
config_entry=entry,
name=DOMAIN,
update_interval=None,
)
self.client = client
self._event_listeners: list[Callable[[DoorEvent], None]] = []
@callback
def async_subscribe_door_events(
self,
event_callback: Callable[[DoorEvent], None],
) -> CALLBACK_TYPE:
"""Subscribe to door events (doorbell, access)."""
def _unsubscribe() -> None:
self._event_listeners.remove(event_callback)
self._event_listeners.append(event_callback)
return _unsubscribe
async def _async_setup(self) -> None:
"""Set up the WebSocket connection for push updates."""
handlers: dict[str, WsMessageHandler] = {
"access.data.device.location_update_v2": self._handle_location_update,
"access.data.v2.location.update": self._handle_v2_location_update,
"access.hw.door_bell": self._handle_doorbell,
"access.logs.insights.add": self._handle_insights_add,
"access.data.setting.update": self._handle_setting_update,
}
self.client.start_websocket(
handlers,
on_connect=self._on_ws_connect,
on_disconnect=self._on_ws_disconnect,
)
async def _async_update_data(self) -> UnifiAccessData:
"""Fetch all doors and emergency status from the API."""
try:
async with asyncio.timeout(10):
doors, emergency = await asyncio.gather(
self.client.get_doors(),
self.client.get_emergency_status(),
)
except ApiAuthError as err:
raise UpdateFailed(f"Authentication failed: {err}") from err
except ApiConnectionError as err:
raise UpdateFailed(f"Error connecting to API: {err}") from err
except ApiError as err:
raise UpdateFailed(f"Error communicating with API: {err}") from err
except TimeoutError as err:
raise UpdateFailed("Timeout communicating with UniFi Access API") from err
return UnifiAccessData(
doors={door.id: door for door in doors},
emergency=emergency,
)
def _on_ws_connect(self) -> None:
"""Handle WebSocket connection established."""
_LOGGER.debug("WebSocket connected to UniFi Access")
if not self.last_update_success:
self.config_entry.async_create_background_task(
self.hass,
self.async_request_refresh(),
"unifi_access_reconnect_refresh",
)
def _on_ws_disconnect(self) -> None:
"""Handle WebSocket disconnection."""
_LOGGER.warning("WebSocket disconnected from UniFi Access")
self.async_set_update_error(
UpdateFailed("WebSocket disconnected from UniFi Access")
)
async def _handle_location_update(self, msg: WebsocketMessage) -> None:
"""Handle location_update_v2 messages."""
update = cast(LocationUpdateV2, msg)
self._process_door_update(update.data.id, update.data.state)
async def _handle_v2_location_update(self, msg: WebsocketMessage) -> None:
"""Handle V2 location update messages."""
update = cast(V2LocationUpdate, msg)
self._process_door_update(update.data.id, update.data.state)
def _process_door_update(
self, door_id: str, ws_state: LocationUpdateState | V2LocationState | None
) -> None:
"""Process a door state update from WebSocket."""
if self.data is None or door_id not in self.data.doors:
return
if ws_state is None:
return
current_door = self.data.doors[door_id]
updates: dict[str, object] = {}
if ws_state.dps is not None:
updates["door_position_status"] = ws_state.dps
if ws_state.lock == "locked":
updates["door_lock_relay_status"] = "lock"
elif ws_state.lock == "unlocked":
updates["door_lock_relay_status"] = "unlock"
if not updates:
return
updated_door = current_door.with_updates(**updates)
self.async_set_updated_data(
UnifiAccessData(
doors={**self.data.doors, door_id: updated_door},
emergency=self.data.emergency,
)
)
async def _handle_setting_update(self, msg: WebsocketMessage) -> None:
"""Handle settings update messages (evacuation/lockdown)."""
if self.data is None:
return
update = cast(SettingUpdate, msg)
self.async_set_updated_data(
UnifiAccessData(
doors=self.data.doors,
emergency=EmergencyStatus(
evacuation=update.data.evacuation,
lockdown=update.data.lockdown,
),
)
)
async def _handle_doorbell(self, msg: WebsocketMessage) -> None:
"""Handle doorbell press events."""
doorbell = cast(HwDoorbell, msg)
self._dispatch_door_event(
doorbell.data.door_id,
"doorbell",
"ring",
{},
)
async def _handle_insights_add(self, msg: WebsocketMessage) -> None:
"""Handle access insights events (entry/exit)."""
insights = cast(InsightsAdd, msg)
door = insights.data.metadata.door
if not door.id:
return
event_type = (
"access_granted" if insights.data.result == "ACCESS" else "access_denied"
)
attrs: dict[str, Any] = {}
if insights.data.metadata.actor.display_name:
attrs["actor"] = insights.data.metadata.actor.display_name
if insights.data.metadata.authentication.display_name:
attrs["authentication"] = insights.data.metadata.authentication.display_name
if insights.data.result:
attrs["result"] = insights.data.result
self._dispatch_door_event(door.id, "access", event_type, attrs)
@callback
def _dispatch_door_event(
self,
door_id: str,
category: str,
event_type: str,
event_data: dict[str, Any],
) -> None:
"""Dispatch a door event to all subscribed listeners."""
event = DoorEvent(door_id, category, event_type, event_data)
for listener in self._event_listeners:
listener(event)

View File

@@ -1,58 +0,0 @@
"""Base entity for the UniFi Access integration."""
from __future__ import annotations
from unifi_access_api import Door
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import UnifiAccessCoordinator
class UnifiAccessEntity(CoordinatorEntity[UnifiAccessCoordinator]):
"""Base entity for UniFi Access doors."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: UnifiAccessCoordinator,
door: Door,
key: str,
) -> None:
"""Initialize the entity."""
super().__init__(coordinator)
self._door_id = door.id
self._attr_unique_id = f"{door.id}-{key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, door.id)},
name=door.name,
manufacturer="Ubiquiti",
)
@property
def available(self) -> bool:
"""Return True if entity is available."""
return super().available and self._door_id in self.coordinator.data.doors
@property
def _door(self) -> Door:
"""Return the current door state from coordinator data."""
return self.coordinator.data.doors[self._door_id]
class UnifiAccessHubEntity(CoordinatorEntity[UnifiAccessCoordinator]):
"""Base entity for hub-level (controller-wide) UniFi Access entities."""
_attr_has_entity_name = True
def __init__(self, coordinator: UnifiAccessCoordinator) -> None:
"""Initialize the hub entity."""
super().__init__(coordinator)
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, coordinator.config_entry.entry_id)},
name="UniFi Access",
manufacturer="Ubiquiti",
)

View File

@@ -1,96 +0,0 @@
"""Event platform for the UniFi Access integration."""
from __future__ import annotations
from dataclasses import dataclass
from homeassistant.components.event import (
EventDeviceClass,
EventEntity,
EventEntityDescription,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import DoorEvent, UnifiAccessConfigEntry, UnifiAccessCoordinator
from .entity import UnifiAccessEntity
PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True)
class UnifiAccessEventEntityDescription(EventEntityDescription):
"""Describes a UniFi Access event entity."""
category: str
DOORBELL_EVENT_DESCRIPTION = UnifiAccessEventEntityDescription(
key="doorbell",
translation_key="doorbell",
device_class=EventDeviceClass.DOORBELL,
event_types=["ring"],
category="doorbell",
)
ACCESS_EVENT_DESCRIPTION = UnifiAccessEventEntityDescription(
key="access",
translation_key="access",
event_types=["access_granted", "access_denied"],
category="access",
)
EVENT_DESCRIPTIONS: list[UnifiAccessEventEntityDescription] = [
DOORBELL_EVENT_DESCRIPTION,
ACCESS_EVENT_DESCRIPTION,
]
async def async_setup_entry(
hass: HomeAssistant,
entry: UnifiAccessConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up UniFi Access event entities."""
coordinator = entry.runtime_data
async_add_entities(
UnifiAccessEventEntity(coordinator, door_id, description)
for door_id in coordinator.data.doors
for description in EVENT_DESCRIPTIONS
)
class UnifiAccessEventEntity(UnifiAccessEntity, EventEntity):
"""Representation of a UniFi Access event entity."""
entity_description: UnifiAccessEventEntityDescription
def __init__(
self,
coordinator: UnifiAccessCoordinator,
door_id: str,
description: UnifiAccessEventEntityDescription,
) -> None:
"""Initialize the event entity."""
door = coordinator.data.doors[door_id]
super().__init__(coordinator, door, description.key)
self.entity_description = description
async def async_added_to_hass(self) -> None:
"""Subscribe to door events when added to hass."""
await super().async_added_to_hass()
self.async_on_remove(
self.coordinator.async_subscribe_door_events(self._async_handle_event)
)
@callback
def _async_handle_event(self, event: DoorEvent) -> None:
"""Handle incoming event from coordinator."""
if (
event.door_id != self._door_id
or event.category != self.entity_description.category
or event.event_type not in self.event_types
):
return
self._trigger_event(event.event_type, event.event_data)
self.async_write_ha_state()

View File

@@ -1,22 +0,0 @@
{
"entity": {
"button": {
"unlock": {
"default": "mdi:lock-open"
}
},
"event": {
"access": {
"default": "mdi:door"
}
},
"switch": {
"evacuation": {
"default": "mdi:exit-run"
},
"lockdown": {
"default": "mdi:lock-alert"
}
}
}
}

View File

@@ -1,12 +0,0 @@
{
"domain": "unifi_access",
"name": "UniFi Access",
"codeowners": ["@imhotep", "@RaHehl"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/unifi_access",
"integration_type": "hub",
"iot_class": "local_push",
"loggers": ["unifi_access_api"],
"quality_scale": "bronze",
"requirements": ["py-unifi-access==1.0.0"]
}

View File

@@ -1,66 +0,0 @@
rules:
# Bronze
action-setup:
status: exempt
comment: Integration does not register custom actions.
appropriate-polling:
status: exempt
comment: Integration uses WebSocket push updates, no polling.
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: Integration does not register custom actions.
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: done
config-entry-unloading: done
docs-configuration-parameters: todo
docs-installation-parameters: todo
entity-unavailable: done
integration-owner: done
log-when-unavailable: done
parallel-updates: done
reauthentication-flow: todo
test-coverage: done
# Gold
devices: done
diagnostics: todo
discovery-update-info: todo
discovery: todo
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
docs-supported-functions: todo
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices: todo
entity-category: todo
entity-device-class: todo
entity-disabled-by-default: todo
entity-translations: todo
exception-translations: done
icon-translations: done
reconfiguration-flow: todo
repair-issues: todo
stale-devices: todo
# Platinum
async-dependency: done
inject-websession: done
strict-typing: todo

View File

@@ -1,72 +0,0 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
},
"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%]"
},
"step": {
"user": {
"data": {
"api_token": "[%key:common::config_flow::data::api_token%]",
"host": "[%key:common::config_flow::data::host%]",
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
},
"data_description": {
"api_token": "API token generated in the UniFi Access settings.",
"host": "Hostname or IP address of the UniFi Access controller.",
"verify_ssl": "Verify the SSL certificate of the controller."
}
}
}
},
"entity": {
"button": {
"unlock": {
"name": "Unlock"
}
},
"event": {
"access": {
"name": "Access",
"state_attributes": {
"event_type": {
"state": {
"access_denied": "Access denied",
"access_granted": "Access granted"
}
}
}
},
"doorbell": {
"name": "Doorbell",
"state_attributes": {
"event_type": {
"state": {
"ring": "Ring"
}
}
}
}
},
"switch": {
"evacuation": {
"name": "Evacuation"
},
"lockdown": {
"name": "Lockdown"
}
}
},
"exceptions": {
"emergency_failed": {
"message": "Failed to set emergency status."
},
"unlock_failed": {
"message": "Failed to unlock the door."
}
}
}

View File

@@ -1,110 +0,0 @@
"""Switch platform for the UniFi Access integration."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any
from unifi_access_api import EmergencyStatus, UnifiAccessError
from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN
from .coordinator import UnifiAccessConfigEntry, UnifiAccessCoordinator, UnifiAccessData
from .entity import UnifiAccessHubEntity
PARALLEL_UPDATES = 1
@dataclass(frozen=True, kw_only=True)
class UnifiAccessSwitchEntityDescription(SwitchEntityDescription):
"""Describes a UniFi Access switch entity."""
value_fn: Callable[[EmergencyStatus], bool]
set_fn: Callable[[EmergencyStatus, bool], EmergencyStatus]
SWITCH_DESCRIPTIONS: tuple[UnifiAccessSwitchEntityDescription, ...] = (
UnifiAccessSwitchEntityDescription(
key="evacuation",
translation_key="evacuation",
value_fn=lambda s: s.evacuation,
set_fn=lambda s, v: EmergencyStatus(evacuation=v, lockdown=s.lockdown),
),
UnifiAccessSwitchEntityDescription(
key="lockdown",
translation_key="lockdown",
value_fn=lambda s: s.lockdown,
set_fn=lambda s, v: EmergencyStatus(evacuation=s.evacuation, lockdown=v),
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: UnifiAccessConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up UniFi Access switch entities."""
coordinator = entry.runtime_data
async_add_entities(
UnifiAccessEmergencySwitch(coordinator, description)
for description in SWITCH_DESCRIPTIONS
)
class UnifiAccessEmergencySwitch(UnifiAccessHubEntity, SwitchEntity):
"""Representation of a UniFi Access emergency switch."""
entity_description: UnifiAccessSwitchEntityDescription
def __init__(
self,
coordinator: UnifiAccessCoordinator,
description: UnifiAccessSwitchEntityDescription,
) -> None:
"""Initialize the switch entity."""
super().__init__(coordinator)
self._attr_unique_id = f"{coordinator.config_entry.entry_id}-{description.key}"
self.entity_description = description
@property
def is_on(self) -> bool:
"""Return True if the switch is on."""
return self.entity_description.value_fn(self.coordinator.data.emergency)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on."""
await self._async_set_emergency(True)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off."""
await self._async_set_emergency(False)
async def _async_set_emergency(self, value: bool) -> None:
"""Set emergency status."""
new_status = self.entity_description.set_fn(
self.coordinator.data.emergency, value
)
try:
await self.coordinator.client.set_emergency_status(new_status)
except UnifiAccessError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="emergency_failed",
) from err
# Optimistically update state; the WebSocket confirmation via
# access.data.setting.update typically arrives ~200ms later.
# Guard against flipping coordinator.last_update_success back to True
# while the WebSocket is disconnected and all entities are unavailable.
if self.coordinator.last_update_success:
self.coordinator.async_set_updated_data(
UnifiAccessData(
doors=self.coordinator.data.doors,
emergency=new_status,
)
)

View File

@@ -116,6 +116,13 @@ SENSOR_DESCRIPTIONS: tuple[UptimeKumaSensorEntityDescription, ...] = (
value_fn=lambda m: m.monitor_port,
create_entity=lambda t: t in HAS_PORT,
),
UptimeKumaSensorEntityDescription(
key=UptimeKumaSensor.PORT,
translation_key=UptimeKumaSensor.PORT,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda m: m.monitor_port,
create_entity=lambda t: t in HAS_PORT,
),
UptimeKumaSensorEntityDescription(
key=UptimeKumaSensor.UPTIME_RATIO_1D,
translation_key=UptimeKumaSensor.UPTIME_RATIO_1D,

View File

@@ -7,5 +7,5 @@
"integration_type": "service",
"iot_class": "cloud_polling",
"loggers": ["pywaze", "homeassistant.helpers.location"],
"requirements": ["pywaze==1.2.0"]
"requirements": ["pywaze==1.1.1"]
}

View File

@@ -156,7 +156,6 @@ class WebDavBackupAgent(BackupAgent):
f"{self._backup_path}/{filename_tar}",
timeout=BACKUP_TIMEOUT,
content_length=backup.size,
progress=lambda current, total: on_progress(bytes_uploaded=current),
)
_LOGGER.debug(

View File

@@ -942,7 +942,7 @@
"name": "Start-up color temperature"
},
"start_up_current_level": {
"name": "Power-on level"
"name": "Start-up current level"
},
"startup_time": {
"name": "Startup time"
@@ -1298,7 +1298,7 @@
"name": "Speed"
},
"start_up_on_off": {
"name": "Power-on behavior"
"name": "Start-up behavior"
},
"status_indication": {
"name": "Status indication"

View File

@@ -743,7 +743,6 @@ FLOWS = {
"trane",
"transmission",
"triggercmd",
"trmnl",
"tuya",
"twentemilieu",
"twilio",
@@ -752,7 +751,6 @@ FLOWS = {
"uhoo",
"ukraine_alarm",
"unifi",
"unifi_access",
"unifiprotect",
"upb",
"upcloud",

View File

@@ -817,11 +817,6 @@ DHCP: Final[list[dict[str, str | bool]]] = [
"hostname": "hub*",
"macaddress": "286D97*",
},
{
"domain": "smartthings",
"hostname": "smarthub",
"macaddress": "683A48*",
},
{
"domain": "smartthings",
"hostname": "samsung-*",

View File

@@ -7244,12 +7244,6 @@
"config_flow": true,
"iot_class": "cloud_polling"
},
"trmnl": {
"name": "TRMNL",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
"tuya": {
"name": "Tuya",
"integration_type": "hub",
@@ -7331,12 +7325,6 @@
"iot_class": "local_push",
"name": "UniFi Network"
},
"unifi_access": {
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_push",
"name": "UniFi Access"
},
"unifi_direct": {
"integration_type": "hub",
"config_flow": false,

View File

@@ -184,52 +184,6 @@ class IntentUnexpectedError(IntentError):
"""Unexpected error while handling intent."""
class MatchFailedError(IntentError):
"""Error when target matching fails."""
def __init__(
self,
result: MatchTargetsResult,
constraints: MatchTargetsConstraints,
preferences: MatchTargetsPreferences | None = None,
) -> None:
"""Initialize error."""
super().__init__()
self.result = result
self.constraints = constraints
self.preferences = preferences
def __str__(self) -> str:
"""Return string representation."""
return f"<MatchFailedError result={self.result}, constraints={self.constraints}, preferences={self.preferences}>"
class NoStatesMatchedError(MatchFailedError):
"""Error when no states match the intent's constraints."""
def __init__(
self,
reason: MatchFailedReason,
name: str | None = None,
area: str | None = None,
floor: str | None = None,
domains: set[str] | None = None,
device_classes: set[str] | None = None,
) -> None:
"""Initialize error."""
super().__init__(
result=MatchTargetsResult(False, reason),
constraints=MatchTargetsConstraints(
name=name,
area_name=area,
floor_name=floor,
domains=domains,
device_classes=device_classes,
),
)
class MatchFailedReason(Enum):
"""Possible reasons for match failure in async_match_targets."""
@@ -278,29 +232,6 @@ class MatchFailedReason(Enum):
)
@dataclass
class MatchTargetsResult:
"""Result from async_match_targets."""
is_match: bool
"""True if one or more entities matched."""
no_match_reason: MatchFailedReason | None = None
"""Reason for failed match when is_match = False."""
states: list[State] = field(default_factory=list)
"""List of matched entity states."""
no_match_name: str | None = None
"""Name of invalid area/floor or duplicate name when match fails for those reasons."""
areas: list[ar.AreaEntry] = field(default_factory=list)
"""Areas that were targeted."""
floors: list[fr.FloorEntry] = field(default_factory=list)
"""Floors that were targeted."""
@dataclass
class MatchTargetsConstraints:
"""Constraints for async_match_targets."""
@@ -361,6 +292,75 @@ class MatchTargetsPreferences:
"""Id of floor to use when deduplicating names."""
@dataclass
class MatchTargetsResult:
"""Result from async_match_targets."""
is_match: bool
"""True if one or more entities matched."""
no_match_reason: MatchFailedReason | None = None
"""Reason for failed match when is_match = False."""
states: list[State] = field(default_factory=list)
"""List of matched entity states."""
no_match_name: str | None = None
"""Name of invalid area/floor or duplicate name when match fails for those reasons."""
areas: list[ar.AreaEntry] = field(default_factory=list)
"""Areas that were targeted."""
floors: list[fr.FloorEntry] = field(default_factory=list)
"""Floors that were targeted."""
class MatchFailedError(IntentError):
"""Error when target matching fails."""
def __init__(
self,
result: MatchTargetsResult,
constraints: MatchTargetsConstraints,
preferences: MatchTargetsPreferences | None = None,
) -> None:
"""Initialize error."""
super().__init__()
self.result = result
self.constraints = constraints
self.preferences = preferences
def __str__(self) -> str:
"""Return string representation."""
return f"<MatchFailedError result={self.result}, constraints={self.constraints}, preferences={self.preferences}>"
class NoStatesMatchedError(MatchFailedError):
"""Error when no states match the intent's constraints."""
def __init__(
self,
reason: MatchFailedReason,
name: str | None = None,
area: str | None = None,
floor: str | None = None,
domains: set[str] | None = None,
device_classes: set[str] | None = None,
) -> None:
"""Initialize error."""
super().__init__(
result=MatchTargetsResult(False, reason),
constraints=MatchTargetsConstraints(
name=name,
area_name=area,
floor_name=floor,
domains=domains,
device_classes=device_classes,
),
)
@dataclass
class MatchTargetsCandidate:
"""Candidate for async_match_targets."""
@@ -915,7 +915,7 @@ class DynamicServiceIntentHandler(IntentHandler):
def __init__(
self,
intent_type: str,
*,
speech: str | None = None,
required_slots: _IntentSlotsType | None = None,
optional_slots: _IntentSlotsType | None = None,
required_domains: set[str] | None = None,
@@ -927,6 +927,7 @@ class DynamicServiceIntentHandler(IntentHandler):
) -> None:
"""Create Service Intent Handler."""
self.intent_type = intent_type
self.speech = speech
self.required_domains = required_domains
self.required_features = required_features
self.required_states = required_states
@@ -1113,6 +1114,7 @@ class DynamicServiceIntentHandler(IntentHandler):
)
for floor in match_result.floors
)
speech_name = match_result.floors[0].name
elif match_result.areas:
success_results.extend(
IntentResponseTarget(
@@ -1120,6 +1122,9 @@ class DynamicServiceIntentHandler(IntentHandler):
)
for area in match_result.areas
)
speech_name = match_result.areas[0].name
else:
speech_name = states[0].name
service_coros: list[Coroutine[Any, Any, None]] = []
for state in states:
@@ -1161,6 +1166,9 @@ class DynamicServiceIntentHandler(IntentHandler):
states = [hass.states.get(state.entity_id) or state for state in states]
response.async_set_states(states)
if self.speech is not None:
response.async_set_speech(self.speech.format(speech_name))
return response
async def async_call_service(
@@ -1223,7 +1231,7 @@ class ServiceIntentHandler(DynamicServiceIntentHandler):
intent_type: str,
domain: str,
service: str,
*,
speech: str | None = None,
required_slots: _IntentSlotsType | None = None,
optional_slots: _IntentSlotsType | None = None,
required_domains: set[str] | None = None,
@@ -1236,6 +1244,7 @@ class ServiceIntentHandler(DynamicServiceIntentHandler):
"""Create service handler."""
super().__init__(
intent_type,
speech=speech,
required_slots=required_slots,
optional_slots=optional_slots,
required_domains=required_domains,

10
mypy.ini generated
View File

@@ -5458,16 +5458,6 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.trmnl.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.tts.*]
check_untyped_defs = true
disallow_incomplete_defs = true

12
requirements_all.txt generated
View File

@@ -190,7 +190,7 @@ aioairzone-cloud==0.7.2
aioairzone==1.0.5
# homeassistant.components.alexa_devices
aioamazondevices==13.0.1
aioamazondevices==13.0.0
# homeassistant.components.ambient_network
# homeassistant.components.ambient_station
@@ -1882,9 +1882,6 @@ py-sucks==0.9.11
# homeassistant.components.synology_dsm
py-synologydsm-api==2.7.3
# homeassistant.components.unifi_access
py-unifi-access==1.0.0
# homeassistant.components.atome
pyAtome==0.1.1
@@ -2636,7 +2633,7 @@ python-overseerr==0.9.0
python-picnic-api2==1.3.1
# homeassistant.components.pooldose
python-pooldose==0.8.5
python-pooldose==0.8.2
# homeassistant.components.rabbitair
python-rabbitair==0.0.8
@@ -2730,7 +2727,7 @@ pyvlx==0.2.30
pyvolumio==0.1.5
# homeassistant.components.waze_travel_time
pywaze==1.2.0
pywaze==1.1.1
# homeassistant.components.weatherflow
pyweatherflowudp==1.5.0
@@ -3129,9 +3126,6 @@ transmission-rpc==7.0.3
# homeassistant.components.triggercmd
triggercmd==0.0.36
# homeassistant.components.trmnl
trmnl==0.1.0
# homeassistant.components.twinkly
ttls==1.8.3

View File

@@ -181,7 +181,7 @@ aioairzone-cloud==0.7.2
aioairzone==1.0.5
# homeassistant.components.alexa_devices
aioamazondevices==13.0.1
aioamazondevices==13.0.0
# homeassistant.components.ambient_network
# homeassistant.components.ambient_station
@@ -1631,9 +1631,6 @@ py-sucks==0.9.11
# homeassistant.components.synology_dsm
py-synologydsm-api==2.7.3
# homeassistant.components.unifi_access
py-unifi-access==1.0.0
# homeassistant.components.hdmi_cec
pyCEC==0.5.2
@@ -2235,7 +2232,7 @@ python-overseerr==0.9.0
python-picnic-api2==1.3.1
# homeassistant.components.pooldose
python-pooldose==0.8.5
python-pooldose==0.8.2
# homeassistant.components.rabbitair
python-rabbitair==0.0.8
@@ -2314,7 +2311,7 @@ pyvlx==0.2.30
pyvolumio==0.1.5
# homeassistant.components.waze_travel_time
pywaze==1.2.0
pywaze==1.1.1
# homeassistant.components.weatherflow
pyweatherflowudp==1.5.0
@@ -2632,9 +2629,6 @@ transmission-rpc==7.0.3
# homeassistant.components.triggercmd
triggercmd==0.0.36
# homeassistant.components.trmnl
trmnl==0.1.0
# homeassistant.components.twinkly
ttls==1.8.3

View File

@@ -2,7 +2,10 @@
from unittest.mock import AsyncMock
from aioamazondevices.const.devices import SPEAKER_GROUP_FAMILY
from aioamazondevices.const.devices import (
SPEAKER_GROUP_DEVICE_TYPE,
SPEAKER_GROUP_FAMILY,
)
from aioamazondevices.exceptions import CannotConnect, CannotRetrieveData
import pytest
@@ -114,7 +117,7 @@ async def test_alexa_dnd_group_removal(
identifiers={(DOMAIN, mock_config_entry.entry_id)},
name=mock_config_entry.title,
manufacturer="Amazon",
model="Speaker Group",
model=SPEAKER_GROUP_DEVICE_TYPE,
entry_type=dr.DeviceEntryType.SERVICE,
)
@@ -153,7 +156,7 @@ async def test_alexa_unsupported_notification_sensor_removal(
identifiers={(DOMAIN, mock_config_entry.entry_id)},
name=mock_config_entry.title,
manufacturer="Amazon",
model="Speaker Group",
model=SPEAKER_GROUP_DEVICE_TYPE,
entry_type=dr.DeviceEntryType.SERVICE,
)

File diff suppressed because it is too large Load Diff

View File

@@ -1,94 +0,0 @@
"""Tests for Arcam FMJ sensor entities."""
from collections.abc import Generator
from unittest.mock import Mock, patch
from arcam.fmj import IncomingVideoAspectRatio, IncomingVideoColorspace
from arcam.fmj.state import IncomingAudioConfig, IncomingAudioFormat, State
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry, snapshot_platform
@pytest.fixture(autouse=True)
def sensor_only() -> Generator[None]:
"""Limit platform setup to sensor only."""
with patch("homeassistant.components.arcam_fmj.PLATFORMS", [Platform.SENSOR]):
yield
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "player_setup")
async def test_setup(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test snapshot of the sensor platform."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.usefixtures("player_setup")
async def test_sensor_video_parameters(
hass: HomeAssistant,
state_1: State,
client: Mock,
) -> None:
"""Test video parameter sensors with actual data."""
video_params = Mock()
video_params.horizontal_resolution = 1920
video_params.vertical_resolution = 1080
video_params.refresh_rate = 60.0
video_params.aspect_ratio = IncomingVideoAspectRatio.ASPECT_16_9
video_params.colorspace = IncomingVideoColorspace.HDR10
state_1.get_incoming_video_parameters.return_value = video_params
client.notify_data_updated()
await hass.async_block_till_done()
expected = {
"incoming_video_horizontal_resolution": "1920",
"incoming_video_vertical_resolution": "1080",
"incoming_video_refresh_rate": "60.0",
"incoming_video_aspect_ratio": "aspect_16_9",
"incoming_video_colorspace": "hdr10",
}
for key, value in expected.items():
state = hass.states.get(f"sensor.arcam_fmj_127_0_0_1_{key}")
assert state is not None, f"State missing for {key}"
assert state.state == value, f"Expected {value} for {key}, got {state.state}"
@pytest.mark.usefixtures("player_setup")
async def test_sensor_audio_parameters(
hass: HomeAssistant,
state_1: State,
client: Mock,
) -> None:
"""Test audio parameter sensors with actual data."""
state_1.get_incoming_audio_format.return_value = (
IncomingAudioFormat.PCM,
IncomingAudioConfig.STEREO_ONLY,
)
state_1.get_incoming_audio_sample_rate.return_value = 48000
client.notify_data_updated()
await hass.async_block_till_done()
assert (
hass.states.get("sensor.arcam_fmj_127_0_0_1_incoming_audio_format").state
== "pcm"
)
assert (
hass.states.get("sensor.arcam_fmj_127_0_0_1_incoming_audio_configuration").state
== "stereo_only"
)
assert (
hass.states.get("sensor.arcam_fmj_127_0_0_1_incoming_audio_sample_rate").state
== "48000"
)

View File

@@ -43,7 +43,7 @@ async def test_open_cover_intent(hass: HomeAssistant, slots: dict[str, Any]) ->
)
await hass.async_block_till_done()
assert response.response_type == intent.IntentResponseType.ACTION_DONE
assert response.speech["plain"]["speech"] == "Opening garage door"
assert len(calls) == 1
call = calls[0]
assert call.domain == DOMAIN
@@ -75,7 +75,7 @@ async def test_close_cover_intent(hass: HomeAssistant, slots: dict[str, Any]) ->
)
await hass.async_block_till_done()
assert response.response_type == intent.IntentResponseType.ACTION_DONE
assert response.speech["plain"]["speech"] == "Closing garage door"
assert len(calls) == 1
call = calls[0]
assert call.domain == DOMAIN

View File

@@ -1087,3 +1087,90 @@ async def test_supervisor_issue_deprecated_addon(
assert not issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex)
supervisor_client.resolution.apply_suggestion.assert_called_once_with(sugg_uuid)
@pytest.mark.parametrize(
"all_setup_requests", [{"include_addons": True}], indirect=True
)
@pytest.mark.usefixtures("all_setup_requests")
async def test_supervisor_issue_deprecated_arch_addon(
hass: HomeAssistant,
supervisor_client: AsyncMock,
hass_client: ClientSessionGenerator,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test fix flow for supervisor issue for add-on using deprecated architecture or machine."""
mock_resolution_info(
supervisor_client,
issues=[
Issue(
type=IssueType.DEPRECATED_ARCH_ADDON,
context=ContextType.ADDON,
reference="test",
uuid=(issue_uuid := uuid4()),
),
],
suggestions_by_issue={
issue_uuid: [
Suggestion(
type=SuggestionType.EXECUTE_REMOVE,
context=ContextType.ADDON,
reference="test",
uuid=(sugg_uuid := uuid4()),
auto=False,
),
]
},
)
assert await async_setup_component(hass, "hassio", {})
repair_issue = issue_registry.async_get_issue(
domain="hassio", issue_id=issue_uuid.hex
)
assert repair_issue
client = await hass_client()
resp = await client.post(
"/api/repairs/issues/fix",
json={"handler": "hassio", "issue_id": repair_issue.issue_id},
)
assert resp.status == HTTPStatus.OK
data = await resp.json()
flow_id = data["flow_id"]
assert data == {
"type": "form",
"flow_id": flow_id,
"handler": "hassio",
"step_id": "addon_execute_remove",
"data_schema": [],
"errors": None,
"description_placeholders": {
"reference": "test",
"addon": "test",
"help_url": "https://www.home-assistant.io/help/",
"community_url": "https://community.home-assistant.io/",
},
"last_step": True,
"preview": None,
}
resp = await client.post(f"/api/repairs/issues/fix/{flow_id}")
assert resp.status == HTTPStatus.OK
data = await resp.json()
flow_id = data["flow_id"]
assert data == {
"type": "create_entry",
"flow_id": flow_id,
"handler": "hassio",
"description": None,
"description_placeholders": None,
}
assert not issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex)
supervisor_client.resolution.apply_suggestion.assert_called_once_with(sugg_uuid)

View File

@@ -248,11 +248,12 @@ async def test_cover_intents_loading(hass: HomeAssistant) -> None:
hass.states.async_set("cover.garage_door", "closed")
calls = async_mock_service(hass, "cover", SERVICE_OPEN_COVER)
await intent.async_handle(
response = await intent.async_handle(
hass, "test", "HassOpenCover", {"name": {"value": "garage door"}}
)
await hass.async_block_till_done()
assert response.speech["plain"]["speech"] == "Opening garage door"
assert len(calls) == 1
call = calls[0]
assert call.domain == "cover"

View File

@@ -201,59 +201,3 @@ async def test_reauth_wrong_account(hass: HomeAssistant) -> None:
assert result["reason"] == "unique_id_mismatch"
assert entry.unique_id == ACCOUNT_USER_ID
assert entry.data == CONFIG[DOMAIN]
async def test_reconfigure(hass: HomeAssistant, mock_account: Account) -> None:
"""Test reconfiguration flow (with fail and recover)."""
entry = MockConfigEntry(
domain=DOMAIN,
data=CONFIG[DOMAIN],
unique_id=ACCOUNT_USER_ID,
)
entry.add_to_hass(hass)
original_password = entry.data[CONF_PASSWORD]
new_password = f"{original_password}_new"
result = await entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
with patch(
"homeassistant.components.litterrobot.config_flow.Account.connect",
side_effect=LitterRobotLoginException,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_PASSWORD: new_password},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "invalid_auth"}
assert entry.data[CONF_PASSWORD] == original_password
with (
patch(
"homeassistant.components.litterrobot.config_flow.Account.connect",
return_value=mock_account,
),
patch(
"homeassistant.components.litterrobot.config_flow.Account.user_id",
new_callable=PropertyMock,
return_value=ACCOUNT_USER_ID,
),
patch(
"homeassistant.components.litterrobot.async_setup_entry",
return_value=True,
) as mock_setup_entry,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_PASSWORD: new_password},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert entry.unique_id == ACCOUNT_USER_ID
assert entry.data[CONF_PASSWORD] == new_password
assert len(mock_setup_entry.mock_calls) == 1

View File

@@ -298,55 +298,6 @@
'state': 'off',
})
# ---
# name: test_all_entities[da_ks_cooktop_000001][binary_sensor.table_de_cuisson_operating_state-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': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.table_de_cuisson_operating_state',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Operating state',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Operating state',
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'cooktop_operating_state',
'unique_id': '5c202ad1-d112-d746-50b8-bd76a554b362_main_custom.cooktopOperatingState_cooktopOperatingState_cooktopOperatingState',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[da_ks_cooktop_000001][binary_sensor.table_de_cuisson_operating_state-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Table de cuisson Operating state',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.table_de_cuisson_operating_state',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_entities[da_ks_cooktop_000001][binary_sensor.table_de_cuisson_power-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
@@ -1089,55 +1040,6 @@
'state': 'off',
})
# ---
# name: test_all_entities[da_ks_range_0101x][binary_sensor.vulcan_operating_state-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': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.vulcan_operating_state',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Operating state',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Operating state',
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'cooktop_operating_state',
'unique_id': '2c3cbaa0-1899-5ddc-7b58-9d657bd48f18_main_custom.cooktopOperatingState_cooktopOperatingState_cooktopOperatingState',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[da_ks_range_0101x][binary_sensor.vulcan_operating_state-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Vulcan Operating state',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.vulcan_operating_state',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_entities[da_ks_range_0101x][binary_sensor.vulcan_remote_control-entry]
EntityRegistryEntrySnapshot({
'aliases': set({

View File

@@ -1,13 +0,0 @@
"""Tests for the TRMNL integration."""
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Set up the TRMNL integration for testing."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()

View File

@@ -1,55 +0,0 @@
"""Common fixtures for the TRMNL tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
import pytest
from trmnl.models import DevicesResponse, UserResponse
from homeassistant.components.trmnl.const import DOMAIN
from homeassistant.const import CONF_API_KEY
from tests.common import MockConfigEntry, load_fixture
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.trmnl.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Mock config entry."""
return MockConfigEntry(
domain=DOMAIN,
title="Test",
unique_id="30561",
data={CONF_API_KEY: "user_aaaaaaaaaa"},
)
@pytest.fixture
def mock_trmnl_client() -> Generator[AsyncMock]:
"""Mock TRMNL client."""
with (
patch(
"homeassistant.components.trmnl.coordinator.TRMNLClient",
autospec=True,
) as mock_client,
patch(
"homeassistant.components.trmnl.config_flow.TRMNLClient",
new=mock_client,
),
):
client = mock_client.return_value
client.get_me.return_value = UserResponse.from_json(
load_fixture("me.json", DOMAIN)
).data
client.get_devices.return_value = DevicesResponse.from_json(
load_fixture("devices.json", DOMAIN)
).data
yield client

View File

@@ -1,17 +0,0 @@
{
"data": [
{
"id": 42793,
"name": "Test TRMNL",
"friendly_id": "1RJXS4",
"mac_address": "B0:A6:04:AA:BB:CC",
"battery_voltage": 3.87,
"rssi": -64,
"sleep_mode_enabled": false,
"sleep_start_time": 1320,
"sleep_end_time": 480,
"percent_charged": 72.5,
"wifi_strength": 50
}
]
}

View File

@@ -1,14 +0,0 @@
{
"data": {
"id": 30561,
"name": "Test",
"email": "test@outlook.com",
"first_name": "test",
"last_name": "test",
"locale": "en",
"time_zone": "Amsterdam",
"time_zone_iana": "Europe/Amsterdam",
"utc_offset": 3600,
"api_key": "user_aaaaaaaaaa"
}
}

View File

@@ -1,32 +0,0 @@
# serializer version: 1
# name: test_device
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'config_entries_subentries': <ANY>,
'configuration_url': None,
'connections': set({
tuple(
'mac',
'b0:a6:04:aa:bb:cc',
),
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
}),
'labels': set({
}),
'manufacturer': 'TRMNL',
'model': None,
'model_id': None,
'name': 'Test TRMNL',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': None,
'sw_version': None,
'via_device_id': None,
})
# ---

View File

@@ -1,109 +0,0 @@
# serializer version: 1
# name: test_all_entities[sensor.test_trmnl_battery-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.test_trmnl_battery',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Battery',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.BATTERY: 'battery'>,
'original_icon': None,
'original_name': 'Battery',
'platform': 'trmnl',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '42793_battery',
'unit_of_measurement': '%',
})
# ---
# name: test_all_entities[sensor.test_trmnl_battery-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'battery',
'friendly_name': 'Test TRMNL Battery',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'sensor.test_trmnl_battery',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '72.5',
})
# ---
# name: test_all_entities[sensor.test_trmnl_signal_strength-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.test_trmnl_signal_strength',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Signal strength',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.SIGNAL_STRENGTH: 'signal_strength'>,
'original_icon': None,
'original_name': 'Signal strength',
'platform': 'trmnl',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '42793_rssi',
'unit_of_measurement': 'dBm',
})
# ---
# name: test_all_entities[sensor.test_trmnl_signal_strength-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'signal_strength',
'friendly_name': 'Test TRMNL Signal strength',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': 'dBm',
}),
'context': <ANY>,
'entity_id': 'sensor.test_trmnl_signal_strength',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '-64',
})
# ---

View File

@@ -1,92 +0,0 @@
"""Test the TRMNL config flow."""
from unittest.mock import AsyncMock
import pytest
from trmnl.exceptions import TRMNLAuthenticationError, TRMNLError
from homeassistant.components.trmnl.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
from tests.common import MockConfigEntry
@pytest.mark.usefixtures("mock_trmnl_client")
async def test_full_flow(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> None:
"""Test the full config flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_API_KEY: "user_aaaaaaaaaa"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Test"
assert result["data"] == {CONF_API_KEY: "user_aaaaaaaaaa"}
assert result["result"].unique_id == "30561"
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
("exception", "error"),
[
(TRMNLAuthenticationError, "invalid_auth"),
(TRMNLError, "cannot_connect"),
(Exception, "unknown"),
],
)
async def test_form_errors(
hass: HomeAssistant,
mock_trmnl_client: AsyncMock,
mock_setup_entry: AsyncMock,
exception: type[Exception],
error: str,
) -> None:
"""Test we handle form errors."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
mock_trmnl_client.get_me.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_API_KEY: "user_aaaaaaaaaa"}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": error}
mock_trmnl_client.get_me.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_API_KEY: "user_aaaaaaaaaa"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.usefixtures("mock_trmnl_client")
async def test_duplicate_entry(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test we handle duplicate entries."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_API_KEY: "user_aaaaaaaaaa"}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"

View File

@@ -1,47 +0,0 @@
"""Test the TRMNL initialization."""
from unittest.mock import AsyncMock
from syrupy.assertion import SnapshotAssertion
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
from . import setup_integration
from tests.common import MockConfigEntry
async def test_load_unload_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_trmnl_client: AsyncMock,
) -> None:
"""Test loading and unloading a config entry."""
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
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
async def test_device(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
mock_config_entry: MockConfigEntry,
mock_trmnl_client: AsyncMock,
snapshot: SnapshotAssertion,
) -> None:
"""Test the TRMNL device."""
await setup_integration(hass, mock_config_entry)
device = device_registry.async_get_device(
connections={(CONNECTION_NETWORK_MAC, "B0:A6:04:AA:BB:CC")}
)
assert device
assert device == snapshot

View File

@@ -1,29 +0,0 @@
"""Tests for the TRMNL sensor."""
from unittest.mock import AsyncMock, patch
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_all_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
mock_trmnl_client: AsyncMock,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test all sensor entities."""
with patch("homeassistant.components.trmnl.PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)

View File

@@ -1,14 +0,0 @@
"""Tests for the UniFi Access integration."""
from __future__ import annotations
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Set up the UniFi Access integration for testing."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()

View File

@@ -1,117 +0,0 @@
"""Fixtures for UniFi Access integration tests."""
from __future__ import annotations
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from unifi_access_api import (
Door,
DoorLockRelayStatus,
DoorPositionStatus,
EmergencyStatus,
)
from homeassistant.components.unifi_access.const import DOMAIN
from homeassistant.const import CONF_API_TOKEN, CONF_HOST, CONF_VERIFY_SSL
from homeassistant.core import HomeAssistant
from . import setup_integration
from tests.common import MockConfigEntry
MOCK_HOST = "192.168.1.1"
MOCK_API_TOKEN = "test-api-token-12345"
MOCK_ENTRY_ID = "mock-unifi-access-entry-id"
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return a mock config entry."""
return MockConfigEntry(
domain=DOMAIN,
entry_id=MOCK_ENTRY_ID,
title="UniFi Access",
data={
CONF_HOST: MOCK_HOST,
CONF_API_TOKEN: MOCK_API_TOKEN,
CONF_VERIFY_SSL: False,
},
version=1,
minor_version=1,
)
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Mock setting up a config entry."""
with patch(
"homeassistant.components.unifi_access.async_setup_entry", return_value=True
) as mock_setup:
yield mock_setup
def _make_door(
door_id: str = "door-001",
name: str = "Front Door",
lock_status: DoorLockRelayStatus = DoorLockRelayStatus.LOCK,
position_status: DoorPositionStatus = DoorPositionStatus.CLOSE,
) -> Door:
"""Create a mock Door object."""
return Door(
id=door_id,
name=name,
door_lock_relay_status=lock_status,
door_position_status=position_status,
)
MOCK_DOORS = [
_make_door("door-001", "Front Door"),
_make_door(
"door-002",
"Back Door",
lock_status=DoorLockRelayStatus.UNLOCK,
position_status=DoorPositionStatus.OPEN,
),
]
@pytest.fixture
def mock_client() -> Generator[MagicMock]:
"""Return a mocked UniFi Access API client."""
with (
patch(
"homeassistant.components.unifi_access.UnifiAccessApiClient",
autospec=True,
) as client_mock,
patch(
"homeassistant.components.unifi_access.config_flow.UnifiAccessApiClient",
new=client_mock,
),
):
client = client_mock.return_value
client.authenticate = AsyncMock()
client.get_doors = AsyncMock(return_value=MOCK_DOORS)
client.get_emergency_status = AsyncMock(
return_value=EmergencyStatus(evacuation=False, lockdown=False)
)
client.set_emergency_status = AsyncMock()
client.unlock_door = AsyncMock()
client.close = AsyncMock()
client.start_websocket = MagicMock()
yield client
@pytest.fixture
async def init_integration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
) -> MockConfigEntry:
"""Set up the UniFi Access integration for testing."""
await setup_integration(hass, mock_config_entry)
return mock_config_entry

View File

@@ -1,99 +0,0 @@
# serializer version: 1
# name: test_button_entities[button.back_door_unlock-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': 'button',
'entity_category': None,
'entity_id': 'button.back_door_unlock',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Unlock',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Unlock',
'platform': 'unifi_access',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'unlock',
'unique_id': 'door-002-unlock',
'unit_of_measurement': None,
})
# ---
# name: test_button_entities[button.back_door_unlock-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Back Door Unlock',
}),
'context': <ANY>,
'entity_id': 'button.back_door_unlock',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_button_entities[button.front_door_unlock-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': 'button',
'entity_category': None,
'entity_id': 'button.front_door_unlock',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Unlock',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Unlock',
'platform': 'unifi_access',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'unlock',
'unique_id': 'door-001-unlock',
'unit_of_measurement': None,
})
# ---
# name: test_button_entities[button.front_door_unlock-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Front Door Unlock',
}),
'context': <ANY>,
'entity_id': 'button.front_door_unlock',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---

View File

@@ -1,235 +0,0 @@
# serializer version: 1
# name: test_event_entities[event.back_door_access-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'event_types': list([
'access_granted',
'access_denied',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.back_door_access',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Access',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Access',
'platform': 'unifi_access',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'access',
'unique_id': 'door-002-access',
'unit_of_measurement': None,
})
# ---
# name: test_event_entities[event.back_door_access-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'event_type': None,
'event_types': list([
'access_granted',
'access_denied',
]),
'friendly_name': 'Back Door Access',
}),
'context': <ANY>,
'entity_id': 'event.back_door_access',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_event_entities[event.back_door_doorbell-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'event_types': list([
'ring',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.back_door_doorbell',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Doorbell',
'options': dict({
}),
'original_device_class': <EventDeviceClass.DOORBELL: 'doorbell'>,
'original_icon': None,
'original_name': 'Doorbell',
'platform': 'unifi_access',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'doorbell',
'unique_id': 'door-002-doorbell',
'unit_of_measurement': None,
})
# ---
# name: test_event_entities[event.back_door_doorbell-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'doorbell',
'event_type': None,
'event_types': list([
'ring',
]),
'friendly_name': 'Back Door Doorbell',
}),
'context': <ANY>,
'entity_id': 'event.back_door_doorbell',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_event_entities[event.front_door_access-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'event_types': list([
'access_granted',
'access_denied',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.front_door_access',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Access',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Access',
'platform': 'unifi_access',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'access',
'unique_id': 'door-001-access',
'unit_of_measurement': None,
})
# ---
# name: test_event_entities[event.front_door_access-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'event_type': None,
'event_types': list([
'access_granted',
'access_denied',
]),
'friendly_name': 'Front Door Access',
}),
'context': <ANY>,
'entity_id': 'event.front_door_access',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_event_entities[event.front_door_doorbell-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'event_types': list([
'ring',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.front_door_doorbell',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Doorbell',
'options': dict({
}),
'original_device_class': <EventDeviceClass.DOORBELL: 'doorbell'>,
'original_icon': None,
'original_name': 'Doorbell',
'platform': 'unifi_access',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'doorbell',
'unique_id': 'door-001-doorbell',
'unit_of_measurement': None,
})
# ---
# name: test_event_entities[event.front_door_doorbell-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'doorbell',
'event_type': None,
'event_types': list([
'ring',
]),
'friendly_name': 'Front Door Doorbell',
}),
'context': <ANY>,
'entity_id': 'event.front_door_doorbell',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---

View File

@@ -1,99 +0,0 @@
# serializer version: 1
# name: test_switch_entities[switch.unifi_access_evacuation-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': 'switch',
'entity_category': None,
'entity_id': 'switch.unifi_access_evacuation',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Evacuation',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Evacuation',
'platform': 'unifi_access',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'evacuation',
'unique_id': 'mock-unifi-access-entry-id-evacuation',
'unit_of_measurement': None,
})
# ---
# name: test_switch_entities[switch.unifi_access_evacuation-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'UniFi Access Evacuation',
}),
'context': <ANY>,
'entity_id': 'switch.unifi_access_evacuation',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_switch_entities[switch.unifi_access_lockdown-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': 'switch',
'entity_category': None,
'entity_id': 'switch.unifi_access_lockdown',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Lockdown',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Lockdown',
'platform': 'unifi_access',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'lockdown',
'unique_id': 'mock-unifi-access-entry-id-lockdown',
'unit_of_measurement': None,
})
# ---
# name: test_switch_entities[switch.unifi_access_lockdown-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'UniFi Access Lockdown',
}),
'context': <ANY>,
'entity_id': 'switch.unifi_access_lockdown',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---

View File

@@ -1,130 +0,0 @@
"""Tests for the UniFi Access button platform."""
from __future__ import annotations
from collections.abc import Callable
from unittest.mock import MagicMock, patch
import pytest
from syrupy.assertion import SnapshotAssertion
from unifi_access_api import ApiError
from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN, SERVICE_PRESS
from homeassistant.const import ATTR_ENTITY_ID, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
FRONT_DOOR_ENTITY = "button.front_door_unlock"
BACK_DOOR_ENTITY = "button.back_door_unlock"
def _get_on_connect(mock_client: MagicMock) -> Callable[[], None]:
"""Extract on_connect callback from mock client."""
return mock_client.start_websocket.call_args[1]["on_connect"]
def _get_on_disconnect(mock_client: MagicMock) -> Callable[[], None]:
"""Extract on_disconnect callback from mock client."""
return mock_client.start_websocket.call_args[1]["on_disconnect"]
async def test_button_entities(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test button entities are created with expected state."""
with patch("homeassistant.components.unifi_access.PLATFORMS", [Platform.BUTTON]):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_unlock_door(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test pressing the unlock button."""
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: "button.front_door_unlock"},
blocking=True,
)
mock_client.unlock_door.assert_awaited_once_with("door-001")
async def test_unlock_door_api_error(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test pressing the unlock button raises on API error."""
mock_client.unlock_door.side_effect = ApiError("unlock failed")
with pytest.raises(HomeAssistantError, match="Failed to unlock the door"):
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: "button.front_door_unlock"},
blocking=True,
)
async def test_ws_disconnect_marks_entities_unavailable(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test WebSocket disconnect marks entities as unavailable."""
assert hass.states.get(FRONT_DOOR_ENTITY).state == "unknown"
on_disconnect = _get_on_disconnect(mock_client)
on_disconnect()
await hass.async_block_till_done()
assert hass.states.get(FRONT_DOOR_ENTITY).state == "unavailable"
assert hass.states.get(BACK_DOOR_ENTITY).state == "unavailable"
async def test_ws_reconnect_restores_entities(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test WebSocket reconnect restores entity availability."""
on_disconnect = _get_on_disconnect(mock_client)
on_connect = _get_on_connect(mock_client)
on_disconnect()
await hass.async_block_till_done()
assert hass.states.get(FRONT_DOOR_ENTITY).state == "unavailable"
on_connect()
await hass.async_block_till_done()
assert hass.states.get(FRONT_DOOR_ENTITY).state == "unknown"
assert hass.states.get(BACK_DOOR_ENTITY).state == "unknown"
async def test_ws_connect_no_refresh_when_healthy(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test WebSocket connect does not trigger redundant refresh when healthy."""
on_connect = _get_on_connect(mock_client)
on_connect()
await hass.async_block_till_done()
assert mock_client.get_doors.call_count == 1

View File

@@ -1,149 +0,0 @@
"""Tests for the UniFi Access config flow."""
from __future__ import annotations
from unittest.mock import AsyncMock, MagicMock
import pytest
from unifi_access_api import ApiAuthError, ApiConnectionError
from homeassistant.components.unifi_access.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_API_TOKEN, CONF_HOST, CONF_VERIFY_SSL
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .conftest import MOCK_API_TOKEN, MOCK_HOST
from tests.common import MockConfigEntry
async def test_user_flow(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_client: MagicMock,
) -> None:
"""Test successful user config flow."""
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"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: MOCK_HOST,
CONF_API_TOKEN: MOCK_API_TOKEN,
CONF_VERIFY_SSL: False,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "UniFi Access"
assert result["data"] == {
CONF_HOST: MOCK_HOST,
CONF_API_TOKEN: MOCK_API_TOKEN,
CONF_VERIFY_SSL: False,
}
mock_client.authenticate.assert_awaited_once()
@pytest.mark.parametrize(
("exception", "error"),
[
(ApiConnectionError("Connection failed"), "cannot_connect"),
(ApiAuthError(), "invalid_auth"),
(RuntimeError("boom"), "unknown"),
],
)
async def test_user_flow_errors(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_client: MagicMock,
exception: Exception,
error: str,
) -> None:
"""Test user config flow errors and recovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
mock_client.authenticate.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: MOCK_HOST,
CONF_API_TOKEN: MOCK_API_TOKEN,
CONF_VERIFY_SSL: False,
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": error}
mock_client.authenticate.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: MOCK_HOST,
CONF_API_TOKEN: MOCK_API_TOKEN,
CONF_VERIFY_SSL: False,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_user_flow_already_configured(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test user config flow aborts when already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: MOCK_HOST,
CONF_API_TOKEN: MOCK_API_TOKEN,
CONF_VERIFY_SSL: False,
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_user_flow_different_host(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test user config flow allows different host."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: "10.0.0.1",
CONF_API_TOKEN: MOCK_API_TOKEN,
CONF_VERIFY_SSL: False,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY

View File

@@ -1,282 +0,0 @@
"""Tests for the UniFi Access event platform."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from unittest.mock import MagicMock, patch
import pytest
from syrupy.assertion import SnapshotAssertion
from unifi_access_api.models.websocket import (
HwDoorbell,
HwDoorbellData,
InsightsAdd,
InsightsAddData,
InsightsMetadata,
InsightsMetadataEntry,
WebsocketMessage,
)
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
FRONT_DOOR_DOORBELL_ENTITY = "event.front_door_doorbell"
FRONT_DOOR_ACCESS_ENTITY = "event.front_door_access"
BACK_DOOR_DOORBELL_ENTITY = "event.back_door_doorbell"
BACK_DOOR_ACCESS_ENTITY = "event.back_door_access"
def _get_ws_handlers(
mock_client: MagicMock,
) -> dict[str, Callable[[WebsocketMessage], Awaitable[None]]]:
"""Extract WebSocket handlers from mock client."""
return mock_client.start_websocket.call_args[0][0]
async def test_event_entities(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test event entities are created with expected state."""
with patch("homeassistant.components.unifi_access.PLATFORMS", [Platform.EVENT]):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.freeze_time("2025-01-01 00:00:00+00:00")
async def test_doorbell_ring_event(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test doorbell ring event is fired when WebSocket message arrives."""
handlers = _get_ws_handlers(mock_client)
doorbell_msg = HwDoorbell(
event="access.hw.door_bell",
data=HwDoorbellData(
door_id="door-001",
door_name="Front Door",
request_id="req-123",
),
)
await handlers["access.hw.door_bell"](doorbell_msg)
await hass.async_block_till_done()
state = hass.states.get(FRONT_DOOR_DOORBELL_ENTITY)
assert state is not None
assert state.attributes["event_type"] == "ring"
assert state.state == "2025-01-01T00:00:00.000+00:00"
async def test_doorbell_ring_event_wrong_door(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test doorbell ring event for unknown door is ignored."""
handlers = _get_ws_handlers(mock_client)
doorbell_msg = HwDoorbell(
event="access.hw.door_bell",
data=HwDoorbellData(
door_id="door-unknown",
door_name="Unknown Door",
request_id="req-999",
),
)
await handlers["access.hw.door_bell"](doorbell_msg)
await hass.async_block_till_done()
# Front door entity should still have no event
state = hass.states.get(FRONT_DOOR_DOORBELL_ENTITY)
assert state is not None
assert state.state == "unknown"
@pytest.mark.parametrize(
(
"result",
"expected_event_type",
"door_id",
"entity_id",
"actor",
"authentication",
),
[
(
"ACCESS",
"access_granted",
"door-001",
FRONT_DOOR_ACCESS_ENTITY,
"John Doe",
"NFC",
),
(
"BLOCKED",
"access_denied",
"door-002",
BACK_DOOR_ACCESS_ENTITY,
"Unknown",
"PIN_CODE",
),
],
)
@pytest.mark.freeze_time("2025-01-01 00:00:00+00:00")
async def test_access_event(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
result: str,
expected_event_type: str,
door_id: str,
entity_id: str,
actor: str,
authentication: str,
) -> None:
"""Test access event is fired with correct mapping from API result."""
handlers = _get_ws_handlers(mock_client)
insights_msg = InsightsAdd(
event="access.logs.insights.add",
data=InsightsAddData(
event_type="access.door.unlock",
result=result,
metadata=InsightsMetadata(
door=InsightsMetadataEntry(
id=door_id,
display_name="Door",
),
actor=InsightsMetadataEntry(
display_name=actor,
),
authentication=InsightsMetadataEntry(
display_name=authentication,
),
),
),
)
await handlers["access.logs.insights.add"](insights_msg)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state is not None
assert state.attributes["event_type"] == expected_event_type
assert state.attributes["actor"] == actor
assert state.attributes["authentication"] == authentication
assert state.attributes["result"] == result
assert state.state == "2025-01-01T00:00:00.000+00:00"
async def test_insights_no_door_id_ignored(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test insights event without door_id is ignored."""
handlers = _get_ws_handlers(mock_client)
insights_msg = InsightsAdd(
event="access.logs.insights.add",
data=InsightsAddData(
event_type="access.door.unlock",
result="ACCESS",
metadata=InsightsMetadata(
door=InsightsMetadataEntry(id="", display_name=""),
),
),
)
await handlers["access.logs.insights.add"](insights_msg)
await hass.async_block_till_done()
state = hass.states.get(FRONT_DOOR_ACCESS_ENTITY)
assert state is not None
assert state.state == "unknown"
@pytest.mark.parametrize(
("result", "expected_event_type", "expected_result_attr"),
[
("ACCESS", "access_granted", "ACCESS"),
("BLOCKED", "access_denied", "BLOCKED"),
("TIMEOUT", "access_denied", "TIMEOUT"),
("", "access_denied", None),
],
)
@pytest.mark.freeze_time("2025-01-01 00:00:00+00:00")
async def test_access_event_result_mapping(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
result: str,
expected_event_type: str,
expected_result_attr: str | None,
) -> None:
"""Test result-to-event-type mapping with minimal attributes."""
handlers = _get_ws_handlers(mock_client)
insights_msg = InsightsAdd(
event="access.logs.insights.add",
data=InsightsAddData(
event_type="access.door.unlock",
result=result,
metadata=InsightsMetadata(
door=InsightsMetadataEntry(
id="door-001",
display_name="Front Door",
),
),
),
)
await handlers["access.logs.insights.add"](insights_msg)
await hass.async_block_till_done()
state = hass.states.get(FRONT_DOOR_ACCESS_ENTITY)
assert state is not None
assert state.attributes["event_type"] == expected_event_type
assert "actor" not in state.attributes
assert "authentication" not in state.attributes
assert state.attributes.get("result") == expected_result_attr
assert state.state == "2025-01-01T00:00:00.000+00:00"
async def test_unload_entry_removes_listeners(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test that events are not processed after config entry is unloaded."""
handlers = _get_ws_handlers(mock_client)
await hass.config_entries.async_unload(init_integration.entry_id)
await hass.async_block_till_done()
doorbell_msg = HwDoorbell(
event="access.hw.door_bell",
data=HwDoorbellData(
door_id="door-001",
door_name="Front Door",
request_id="req-after-unload",
),
)
await handlers["access.hw.door_bell"](doorbell_msg)
await hass.async_block_till_done()
state = hass.states.get(FRONT_DOOR_DOORBELL_ENTITY)
assert state is not None
assert state.state == "unavailable"

View File

@@ -1,243 +0,0 @@
"""Tests for the UniFi Access integration setup."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from unittest.mock import MagicMock
import pytest
from unifi_access_api import (
ApiAuthError,
ApiConnectionError,
ApiError,
DoorPositionStatus,
)
from unifi_access_api.models.websocket import (
LocationUpdateData,
LocationUpdateState,
LocationUpdateV2,
V2LocationState,
V2LocationUpdate,
V2LocationUpdateData,
WebsocketMessage,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from . import setup_integration
from tests.common import MockConfigEntry
def _get_ws_handlers(
mock_client: MagicMock,
) -> dict[str, Callable[[WebsocketMessage], Awaitable[None]]]:
"""Extract WebSocket handlers from mock client."""
return mock_client.start_websocket.call_args[0][0]
async def test_setup_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test successful setup of a config entry."""
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
mock_client.authenticate.assert_awaited_once()
mock_client.get_doors.assert_awaited_once()
@pytest.mark.parametrize(
("exception", "expected_state"),
[
(ApiAuthError(), ConfigEntryState.SETUP_RETRY),
(ApiConnectionError("Connection failed"), ConfigEntryState.SETUP_RETRY),
],
)
async def test_setup_entry_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
exception: Exception,
expected_state: ConfigEntryState,
) -> None:
"""Test setup handles errors correctly."""
mock_client.authenticate.side_effect = exception
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 expected_state
@pytest.mark.parametrize(
("failing_method", "exception"),
[
("get_doors", ApiAuthError()),
("get_doors", ApiConnectionError("Connection failed")),
("get_doors", ApiError("API error")),
("get_emergency_status", ApiAuthError()),
("get_emergency_status", ApiConnectionError("Connection failed")),
("get_emergency_status", ApiError("API error")),
],
)
async def test_coordinator_update_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
failing_method: str,
exception: Exception,
) -> None:
"""Test coordinator handles update errors from get_doors or get_emergency_status."""
getattr(mock_client, failing_method).side_effect = exception
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_unload_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test unloading a config entry."""
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
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
async def test_ws_location_update_v2(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test location_update_v2 WebSocket message updates door state."""
await setup_integration(hass, mock_config_entry)
coordinator = mock_config_entry.runtime_data
assert coordinator.data.doors["door-001"].door_lock_relay_status == "lock"
handlers = _get_ws_handlers(mock_client)
msg = LocationUpdateV2(
event="access.data.device.location_update_v2",
data=LocationUpdateData(
id="door-001",
location_type="DOOR",
state=LocationUpdateState(
dps=DoorPositionStatus.OPEN,
lock="unlocked",
),
),
)
await handlers["access.data.device.location_update_v2"](msg)
await hass.async_block_till_done()
door = coordinator.data.doors["door-001"]
assert door.door_position_status == "open"
assert door.door_lock_relay_status == "unlock"
async def test_ws_v2_location_update(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test V2 location update WebSocket message updates door state."""
await setup_integration(hass, mock_config_entry)
coordinator = mock_config_entry.runtime_data
handlers = _get_ws_handlers(mock_client)
msg = V2LocationUpdate(
event="access.data.v2.location.update",
data=V2LocationUpdateData(
id="door-002",
location_type="DOOR",
name="Back Door",
up_id="up-1",
device_ids=[],
state=V2LocationState(
lock="locked",
dps=DoorPositionStatus.CLOSE,
dps_connected=True,
is_unavailable=False,
),
),
)
await handlers["access.data.v2.location.update"](msg)
await hass.async_block_till_done()
door = coordinator.data.doors["door-002"]
assert door.door_lock_relay_status == "lock"
assert door.door_position_status == "close"
async def test_ws_location_update_unknown_door_ignored(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test location update for unknown door is silently ignored."""
await setup_integration(hass, mock_config_entry)
coordinator = mock_config_entry.runtime_data
original_data = coordinator.data
handlers = _get_ws_handlers(mock_client)
msg = LocationUpdateV2(
event="access.data.device.location_update_v2",
data=LocationUpdateData(
id="door-unknown",
location_type="DOOR",
state=LocationUpdateState(
dps=DoorPositionStatus.OPEN,
lock="unlocked",
),
),
)
await handlers["access.data.device.location_update_v2"](msg)
await hass.async_block_till_done()
# Data should be unchanged
assert coordinator.data is original_data
async def test_ws_location_update_no_state_ignored(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test location update with no state is silently ignored."""
await setup_integration(hass, mock_config_entry)
coordinator = mock_config_entry.runtime_data
original_data = coordinator.data
handlers = _get_ws_handlers(mock_client)
msg = LocationUpdateV2(
event="access.data.device.location_update_v2",
data=LocationUpdateData(
id="door-001",
location_type="DOOR",
state=None,
),
)
await handlers["access.data.device.location_update_v2"](msg)
await hass.async_block_till_done()
assert coordinator.data is original_data

View File

@@ -1,318 +0,0 @@
"""Tests for the UniFi Access switch platform."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from typing import Any
from unittest.mock import MagicMock, patch
import pytest
from syrupy.assertion import SnapshotAssertion
from unifi_access_api import (
ApiAuthError,
ApiConnectionError,
ApiError,
ApiSSLError,
EmergencyStatus,
)
from unifi_access_api.models.websocket import SettingUpdate, SettingUpdateData
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
EVACUATION_ENTITY = "switch.unifi_access_evacuation"
LOCKDOWN_ENTITY = "switch.unifi_access_lockdown"
def _get_ws_handlers(
mock_client: MagicMock,
) -> dict[str, Callable[[object], Awaitable[None]]]:
"""Extract WebSocket handlers from mock client."""
return mock_client.start_websocket.call_args[0][0]
def _get_on_disconnect(mock_client: MagicMock) -> Callable[[], Any]:
"""Extract on_disconnect callback from mock client."""
return mock_client.start_websocket.call_args[1]["on_disconnect"]
def _get_on_connect(mock_client: MagicMock) -> Callable[[], Any]:
"""Extract on_connect callback from mock client."""
return mock_client.start_websocket.call_args[1]["on_connect"]
async def test_switch_entities(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test switch entities are created with expected state."""
with patch("homeassistant.components.unifi_access.PLATFORMS", [Platform.SWITCH]):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.parametrize(
("entity_id", "expected_status"),
[
(EVACUATION_ENTITY, EmergencyStatus(evacuation=True, lockdown=False)),
(LOCKDOWN_ENTITY, EmergencyStatus(evacuation=False, lockdown=True)),
],
)
async def test_turn_on_switch(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
entity_id: str,
expected_status: EmergencyStatus,
) -> None:
"""Test turning on emergency switch."""
assert hass.states.get(entity_id).state == "off"
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_client.set_emergency_status.assert_awaited_once_with(expected_status)
assert hass.states.get(entity_id).state == "on"
@pytest.mark.parametrize(
("entity_id", "expected_status"),
[
(EVACUATION_ENTITY, EmergencyStatus(evacuation=False, lockdown=False)),
(LOCKDOWN_ENTITY, EmergencyStatus(evacuation=False, lockdown=False)),
],
)
async def test_turn_off_switch(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
entity_id: str,
expected_status: EmergencyStatus,
) -> None:
"""Test turning off emergency switch."""
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
assert hass.states.get(entity_id).state == "on"
mock_client.set_emergency_status.reset_mock()
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_client.set_emergency_status.assert_awaited_once_with(expected_status)
assert hass.states.get(entity_id).state == "off"
@pytest.mark.parametrize(
"exception",
[
ApiError("api error"),
ApiAuthError(),
ApiConnectionError("connection failed"),
ApiSSLError("ssl error"),
],
)
async def test_switch_api_error(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
exception: Exception,
) -> None:
"""Test switch raises HomeAssistantError on API failure."""
mock_client.set_emergency_status.side_effect = exception
with pytest.raises(HomeAssistantError, match="Failed to set emergency status"):
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: EVACUATION_ENTITY},
blocking=True,
)
async def test_switches_are_independent(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test that toggling one switch does not affect the other."""
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: LOCKDOWN_ENTITY},
blocking=True,
)
assert hass.states.get(LOCKDOWN_ENTITY).state == "on"
assert hass.states.get(EVACUATION_ENTITY).state == "off"
mock_client.set_emergency_status.assert_awaited_once_with(
EmergencyStatus(evacuation=False, lockdown=True)
)
mock_client.set_emergency_status.reset_mock()
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: EVACUATION_ENTITY},
blocking=True,
)
assert hass.states.get(EVACUATION_ENTITY).state == "on"
assert hass.states.get(LOCKDOWN_ENTITY).state == "on"
mock_client.set_emergency_status.assert_awaited_once_with(
EmergencyStatus(evacuation=True, lockdown=True)
)
async def test_ws_disconnect_marks_switches_unavailable(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test WebSocket disconnect marks switch entities as unavailable."""
assert hass.states.get(EVACUATION_ENTITY).state == "off"
assert hass.states.get(LOCKDOWN_ENTITY).state == "off"
on_disconnect = _get_on_disconnect(mock_client)
on_disconnect()
await hass.async_block_till_done()
assert hass.states.get(EVACUATION_ENTITY).state == "unavailable"
assert hass.states.get(LOCKDOWN_ENTITY).state == "unavailable"
async def test_ws_reconnect_restores_switches(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test WebSocket reconnect restores switch availability."""
on_disconnect = _get_on_disconnect(mock_client)
on_connect = _get_on_connect(mock_client)
on_disconnect()
await hass.async_block_till_done()
assert hass.states.get(EVACUATION_ENTITY).state == "unavailable"
on_connect()
await hass.async_block_till_done()
assert hass.states.get(EVACUATION_ENTITY).state == "off"
assert hass.states.get(LOCKDOWN_ENTITY).state == "off"
async def test_ws_setting_update(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test WebSocket setting update refreshes emergency switch state."""
assert hass.states.get(EVACUATION_ENTITY).state == "off"
assert hass.states.get(LOCKDOWN_ENTITY).state == "off"
handlers = _get_ws_handlers(mock_client)
setting_handler = handlers["access.data.setting.update"]
await setting_handler(
SettingUpdate(
event="access.data.setting.update",
data=SettingUpdateData(evacuation=True, lockdown=True),
)
)
await hass.async_block_till_done()
assert hass.states.get(EVACUATION_ENTITY).state == "on"
assert hass.states.get(LOCKDOWN_ENTITY).state == "on"
async def test_optimistic_update_before_ws_confirmation(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test state is optimistically set immediately, then corrected by WS confirmation.
Verifies that the optimistic update happens synchronously after the API
call, without waiting for the WebSocket confirmation message.
If the WS returns a different value (e.g. hardware rejected the command),
the state is corrected accordingly.
"""
assert hass.states.get(EVACUATION_ENTITY).state == "off"
# Turn on evacuation — state should be optimistically "on" immediately
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: EVACUATION_ENTITY},
blocking=True,
)
assert hass.states.get(EVACUATION_ENTITY).state == "on"
# Simulate WS confirmation arriving — hardware reported evacuation stayed off
# (e.g. rejected by the controller), so state should be corrected
handlers = _get_ws_handlers(mock_client)
await handlers["access.data.setting.update"](
SettingUpdate(
event="access.data.setting.update",
data=SettingUpdateData(evacuation=False, lockdown=False),
)
)
await hass.async_block_till_done()
assert hass.states.get(EVACUATION_ENTITY).state == "off"
async def test_no_optimistic_update_when_ws_disconnected(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test that optimistic update is skipped when WebSocket is disconnected.
Prevents async_set_updated_data from flipping last_update_success back
to True while the coordinator is unavailable due to WS disconnection.
"""
on_disconnect = _get_on_disconnect(mock_client)
on_disconnect()
await hass.async_block_till_done()
assert hass.states.get(EVACUATION_ENTITY).state == "unavailable"
# API call succeeds but optimistic update must NOT restore availability
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: EVACUATION_ENTITY},
blocking=True,
)
await hass.async_block_till_done()
assert hass.states.get(EVACUATION_ENTITY).state == "unavailable"
assert hass.states.get(LOCKDOWN_ENTITY).state == "unavailable"

View File

@@ -188,80 +188,6 @@ async def test_agents_upload(
assert webdav_client.upload_iter.call_count == 2
async def test_agents_upload_emits_progress_events(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
hass_client: ClientSessionGenerator,
webdav_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test upload emits progress events with bytes from upload_iter callbacks."""
test_backup = AgentBackup.from_dict(BACKUP_METADATA)
client = await hass_client()
ws_client = await hass_ws_client(hass)
observed_progress_bytes: list[int] = []
await ws_client.send_json_auto_id({"type": "backup/subscribe_events"})
response = await ws_client.receive_json()
assert response["event"] == {"manager_state": "idle"}
response = await ws_client.receive_json()
assert response["success"] is True
async def _mock_upload_iter(*args: object, **kwargs: object) -> None:
"""Mock upload and trigger progress callback for backup upload."""
path = args[1]
if path.endswith(".tar"):
progress = kwargs.get("progress")
assert callable(progress)
progress(1024, test_backup.size)
progress(test_backup.size, test_backup.size)
with (
patch(
"homeassistant.components.backup.manager.BackupManager.async_get_backup",
) as fetch_backup,
patch(
"homeassistant.components.backup.manager.read_backup",
return_value=test_backup,
),
patch("pathlib.Path.open") as mocked_open,
):
mocked_open.return_value.read = Mock(side_effect=[b"test", b""])
webdav_client.upload_iter.side_effect = _mock_upload_iter
fetch_backup.return_value = test_backup
resp = await client.post(
f"/api/backup/upload?agent_id={DOMAIN}.{mock_config_entry.entry_id}",
data={"file": StringIO("test")},
)
await hass.async_block_till_done()
assert resp.status == 201
# Gather progress events from the upload flow.
reached_idle = False
for _ in range(20):
response = await ws_client.receive_json()
event = response.get("event")
if event is None:
continue
if (
event.get("manager_state") == "receive_backup"
and event.get("agent_id") == f"{DOMAIN}.{mock_config_entry.entry_id}"
and "uploaded_bytes" in event
):
observed_progress_bytes.append(event["uploaded_bytes"])
if event == {"manager_state": "idle"}:
reached_idle = True
break
assert reached_idle
assert 1024 in observed_progress_bytes
assert test_backup.size in observed_progress_bytes
async def test_agents_download(
hass_client: ClientSessionGenerator,
webdav_client: AsyncMock,

View File

@@ -691,7 +691,9 @@ async def test_validate_then_run_in_background(hass: HomeAssistant) -> None:
hass.services.async_register("light", "turn_on", mock_service)
# Create intent handler with a service timeout of 0.05 seconds
handler = intent.ServiceIntentHandler("TestType", "light", "turn_on")
handler = intent.ServiceIntentHandler(
"TestType", "light", "turn_on", "Turned {} on"
)
handler.service_timeout = 0.05
intent.async_register(hass, handler)
@@ -713,7 +715,9 @@ async def test_validate_then_run_in_background(hass: HomeAssistant) -> None:
async def test_invalid_area_floor_names(hass: HomeAssistant) -> None:
"""Test that we throw an appropriate errors with invalid area/floor names."""
handler = intent.ServiceIntentHandler("TestType", "light", "turn_on")
handler = intent.ServiceIntentHandler(
"TestType", "light", "turn_on", "Turned {} on"
)
intent.async_register(hass, handler)
# Need a light to avoid domain error
@@ -748,6 +752,7 @@ async def test_service_intent_handler_required_domains(hass: HomeAssistant) -> N
"TestType",
"homeassistant",
"turn_on",
"Turned {} on",
required_domains={"light"},
)
intent.async_register(hass, handler)
@@ -787,6 +792,7 @@ async def test_service_handler_empty_strings(hass: HomeAssistant) -> None:
"TestType",
"light",
"turn_on",
"Turned {} on",
)
intent.async_register(hass, handler)
@@ -812,7 +818,9 @@ async def test_service_handler_empty_strings(hass: HomeAssistant) -> None:
async def test_service_handler_no_filter(hass: HomeAssistant) -> None:
"""Test that targeting all devices in the house fails."""
handler = intent.ServiceIntentHandler("TestType", "light", "turn_on")
handler = intent.ServiceIntentHandler(
"TestType", "light", "turn_on", "Turned {} on"
)
intent.async_register(hass, handler)
with pytest.raises(intent.IntentHandleError):
@@ -844,6 +852,7 @@ async def test_service_handler_device_classes(
"TestType",
"switch",
"turn_on",
"Turned {} on",
device_classes={switch.SwitchDeviceClass},
)
intent.async_register(hass, handler)