Compare commits

..

31 Commits

Author SHA1 Message Date
mib1185
ee0230f3b1 use renamed helpers 2025-12-17 20:06:03 +00:00
mib1185
851fd467fe Merge branch 'dev' into input_boolean/add-domain-driven-triggers 2025-12-17 20:05:20 +00:00
hanwg
c418d9750b Remove ALLOW_EXTRA from Telegram bot action schema (#158886) 2025-12-17 19:49:34 +01:00
Joost Lekkerkerker
e96d614076 Add integration_type service to meteo_france (#159315) 2025-12-17 19:19:14 +01:00
Abílio Costa
f0a5e0a023 Enable duplicated log file on supervised when env var is set (#158679) 2025-12-17 17:44:54 +00:00
Klaas Schoute
6ac6b86060 Set quality scale in Autarco manifest (#159263) 2025-12-17 16:17:19 +01:00
PaulCavill
3909171b1a Login exception reason (#159259) 2025-12-17 16:13:54 +01:00
Luke Lashley
769029505f Bump python-roborock to 3.18.0 (#159271) 2025-12-17 06:39:06 -08:00
Paul Tarjan
080ec3524b Fix flaky camera stream teardown (#158507)
Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>
2025-12-17 13:47:22 +01:00
Matthew Vance
48d671ad5f Update py-improv-ble-client to 2.0.1 (#159233) 2025-12-17 08:27:06 +01:00
alorente
7115db5d22 Change device class from PRESSURE to ATMOSPHERIC_PRESSURE (#159149) 2025-12-17 07:16:46 +01:00
Jordan Harvey
d0c8792e4b Improve Nintendo Switch parental controls exception handling (#159199) 2025-12-17 07:15:26 +01:00
Richard
84d7c37502 Bump mill-local to 0.5.0 (#159220) 2025-12-16 20:41:28 +01:00
Jordan Harvey
8a10638470 Add select platform to Nintendo Switch parental controls (#159217) 2025-12-16 19:06:43 +01:00
Abílio Costa
10dd53ffc2 Rename base trigger class and methods (#159213) 2025-12-16 18:01:37 +00:00
ryanjones-gentex
36aefce9e1 Store unique user configurations for HomeLink integration (#159111) 2025-12-16 17:14:49 +01:00
Raphael Hehl
fe34da19e2 Use typed HassKey for hass.data in unifiprotect (#158798)
Co-authored-by: RaHehl <rahehl@users.noreply.github.com>
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
2025-12-16 17:12:57 +01:00
Jordan Harvey
fe94dea1db Add missing tests for Nintendo parental controls code coverage (#159210)
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
2025-12-16 17:12:36 +01:00
Anthony Garera
3f57b46756 Add issue sensors to Overseerr integration (#158888) 2025-12-16 17:11:28 +01:00
Raphael Hehl
7e141533bb Improve config flow tests to verify error recovery (#158484)
Co-authored-by: RaHehl <rahehl@users.noreply.github.com>
2025-12-16 17:04:38 +01:00
Joost Lekkerkerker
391ccbafae Add integration_type service to ipma (#159179) 2025-12-16 17:04:29 +01:00
epenet
6af674e64e Use is over == comparison for ConfigEntryState in tests (#159212) 2025-12-16 16:51:39 +01:00
Manu
7b1653c77b Migrate friends to subentries in Xbox integration (#156101) 2025-12-16 16:22:28 +01:00
peteS-UK
c87dafa2e6 Create Squeezebox initial Quality Scale entry (#153993) 2025-12-16 15:56:03 +01:00
Abílio Costa
8375acf315 Add device_tracker home enter/leave triggers (#158083) 2025-12-16 14:50:56 +00:00
Paul Tarjan
4df5a41b57 Migrate Hikvision integration to config flow (#158279)
Co-authored-by: Kamil Breguła <mik-laj@users.noreply.github.com>
2025-12-16 15:44:23 +01:00
Niracler
5796b4c0d9 Enhance Sunricher DALI with update gateway IP from DHCP discovery (#157809)
Co-authored-by: Joostlek <joostlek@outlook.com>
2025-12-16 15:19:37 +01:00
Andrew Jackson
5f4f07803b Add a delay to switch statuses on Transmission (#157493) 2025-12-16 15:11:10 +01:00
Richard Polzer
a0a444e3c8 Bump ekey-bionyxpy to version 1.0.1 (#159196) 2025-12-16 14:30:58 +01:00
epenet
30cfe987ed Bump pyinsteon to 1.6.4 (#159067) 2025-12-16 14:29:06 +01:00
mib1185
d10148a175 add turned_off and turned_on triggers 2025-12-12 20:53:03 +00:00
143 changed files with 4517 additions and 732 deletions

View File

@@ -1328,8 +1328,6 @@ jobs:
- pytest-postgres
- pytest-mariadb
timeout-minutes: 10
permissions:
id-token: write
# codecov/test-results-action currently doesn't support tokenless uploads
# therefore we can't run it on forks
if: |
@@ -1341,9 +1339,8 @@ jobs:
with:
pattern: test-results-*
- name: Upload test results to Codecov
uses: codecov/codecov-action@671740ac38dd9b0130fbe1cec585b89eea48d3de # v5.5.2
uses: codecov/test-results-action@47f89e9acb64b76debcd5ea40642d25a4adced9f # v1.1.1
with:
report_type: test_results
fail_ci_if_error: true
verbose: true
use_oidc: true
token: ${{ secrets.CODECOV_TOKEN }}

1
CODEOWNERS generated
View File

@@ -665,6 +665,7 @@ build.json @home-assistant/supervisor
/homeassistant/components/here_travel_time/ @eifinger
/tests/components/here_travel_time/ @eifinger
/homeassistant/components/hikvision/ @mezz64
/tests/components/hikvision/ @mezz64
/homeassistant/components/hikvisioncam/ @fbradyirl
/homeassistant/components/hisense_aehw4a1/ @bannhead
/tests/components/hisense_aehw4a1/ @bannhead

View File

@@ -624,13 +624,16 @@ async def async_enable_logging(
if log_file is None:
default_log_path = hass.config.path(ERROR_LOG_FILENAME)
if "SUPERVISOR" in os.environ:
_LOGGER.info("Running in Supervisor, not logging to file")
if "SUPERVISOR" in os.environ and "HA_DUPLICATE_LOG_FILE" not in os.environ:
# Rename the default log file if it exists, since previous versions created
# it even on Supervisor
if os.path.isfile(default_log_path):
with contextlib.suppress(OSError):
os.rename(default_log_path, f"{default_log_path}.old")
def rename_old_file() -> None:
"""Rename old log file in executor."""
if os.path.isfile(default_log_path):
with contextlib.suppress(OSError):
os.rename(default_log_path, f"{default_log_path}.old")
await hass.async_add_executor_job(rename_old_file)
err_log_path = None
else:
err_log_path = default_log_path

View File

@@ -4,10 +4,10 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity import get_supported_features
from homeassistant.helpers.trigger import (
EntityStateTriggerBase,
EntityTargetStateTriggerBase,
Trigger,
make_conditional_entity_state_trigger,
make_entity_state_trigger,
make_entity_target_state_trigger,
make_entity_transition_trigger,
)
from .const import DOMAIN, AlarmControlPanelEntityFeature, AlarmControlPanelState
@@ -21,7 +21,7 @@ def supports_feature(hass: HomeAssistant, entity_id: str, features: int) -> bool
return False
class EntityStateTriggerRequiredFeatures(EntityStateTriggerBase):
class EntityStateTriggerRequiredFeatures(EntityTargetStateTriggerBase):
"""Trigger for entity state changes."""
_required_features: int
@@ -38,7 +38,7 @@ class EntityStateTriggerRequiredFeatures(EntityStateTriggerBase):
def make_entity_state_trigger_required_features(
domain: str, to_state: str, required_features: int
) -> type[EntityStateTriggerBase]:
) -> type[EntityTargetStateTriggerBase]:
"""Create an entity state trigger class."""
class CustomTrigger(EntityStateTriggerRequiredFeatures):
@@ -52,7 +52,7 @@ def make_entity_state_trigger_required_features(
TRIGGERS: dict[str, type[Trigger]] = {
"armed": make_conditional_entity_state_trigger(
"armed": make_entity_transition_trigger(
DOMAIN,
from_states={
AlarmControlPanelState.ARMING,
@@ -89,8 +89,12 @@ TRIGGERS: dict[str, type[Trigger]] = {
AlarmControlPanelState.ARMED_VACATION,
AlarmControlPanelEntityFeature.ARM_VACATION,
),
"disarmed": make_entity_state_trigger(DOMAIN, AlarmControlPanelState.DISARMED),
"triggered": make_entity_state_trigger(DOMAIN, AlarmControlPanelState.TRIGGERED),
"disarmed": make_entity_target_state_trigger(
DOMAIN, AlarmControlPanelState.DISARMED
),
"triggered": make_entity_target_state_trigger(
DOMAIN, AlarmControlPanelState.TRIGGERED
),
}

View File

@@ -1,16 +1,22 @@
"""Provides triggers for assist satellites."""
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import Trigger, make_entity_state_trigger
from homeassistant.helpers.trigger import Trigger, make_entity_target_state_trigger
from .const import DOMAIN
from .entity import AssistSatelliteState
TRIGGERS: dict[str, type[Trigger]] = {
"idle": make_entity_state_trigger(DOMAIN, AssistSatelliteState.IDLE),
"listening": make_entity_state_trigger(DOMAIN, AssistSatelliteState.LISTENING),
"processing": make_entity_state_trigger(DOMAIN, AssistSatelliteState.PROCESSING),
"responding": make_entity_state_trigger(DOMAIN, AssistSatelliteState.RESPONDING),
"idle": make_entity_target_state_trigger(DOMAIN, AssistSatelliteState.IDLE),
"listening": make_entity_target_state_trigger(
DOMAIN, AssistSatelliteState.LISTENING
),
"processing": make_entity_target_state_trigger(
DOMAIN, AssistSatelliteState.PROCESSING
),
"responding": make_entity_target_state_trigger(
DOMAIN, AssistSatelliteState.RESPONDING
),
}

View File

@@ -6,5 +6,6 @@
"documentation": "https://www.home-assistant.io/integrations/autarco",
"integration_type": "hub",
"iot_class": "cloud_polling",
"quality_scale": "silver",
"requirements": ["autarco==3.2.0"]
}

View File

@@ -128,7 +128,9 @@ _EXPERIMENTAL_TRIGGER_PLATFORMS = {
"button",
"climate",
"cover",
"device_tracker",
"fan",
"input_boolean",
"lawn_mower",
"light",
"media_player",

View File

@@ -4,7 +4,7 @@ from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity import get_device_class
from homeassistant.helpers.trigger import EntityStateTriggerBase, Trigger
from homeassistant.helpers.trigger import EntityTargetStateTriggerBase, Trigger
from homeassistant.helpers.typing import UNDEFINED, UndefinedType
from . import DOMAIN, BinarySensorDeviceClass
@@ -20,7 +20,7 @@ def get_device_class_or_undefined(
return UNDEFINED
class BinarySensorOnOffTrigger(EntityStateTriggerBase):
class BinarySensorOnOffTrigger(EntityTargetStateTriggerBase):
"""Class for binary sensor on/off triggers."""
_device_class: BinarySensorDeviceClass | None

View File

@@ -3,22 +3,22 @@
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import (
Trigger,
make_conditional_entity_state_trigger,
make_entity_state_attribute_trigger,
make_entity_state_trigger,
make_entity_target_state_attribute_trigger,
make_entity_target_state_trigger,
make_entity_transition_trigger,
)
from .const import ATTR_HVAC_ACTION, DOMAIN, HVACAction, HVACMode
TRIGGERS: dict[str, type[Trigger]] = {
"started_cooling": make_entity_state_attribute_trigger(
"started_cooling": make_entity_target_state_attribute_trigger(
DOMAIN, ATTR_HVAC_ACTION, HVACAction.COOLING
),
"started_drying": make_entity_state_attribute_trigger(
"started_drying": make_entity_target_state_attribute_trigger(
DOMAIN, ATTR_HVAC_ACTION, HVACAction.DRYING
),
"turned_off": make_entity_state_trigger(DOMAIN, HVACMode.OFF),
"turned_on": make_conditional_entity_state_trigger(
"turned_off": make_entity_target_state_trigger(DOMAIN, HVACMode.OFF),
"turned_on": make_entity_transition_trigger(
DOMAIN,
from_states={
HVACMode.OFF,
@@ -32,7 +32,7 @@ TRIGGERS: dict[str, type[Trigger]] = {
HVACMode.HEAT_COOL,
},
),
"started_heating": make_entity_state_attribute_trigger(
"started_heating": make_entity_target_state_attribute_trigger(
DOMAIN, ATTR_HVAC_ACTION, HVACAction.HEATING
),
}

View File

@@ -11,5 +11,13 @@
"see": {
"service": "mdi:account-eye"
}
},
"triggers": {
"entered_home": {
"trigger": "mdi:account-arrow-left"
},
"left_home": {
"trigger": "mdi:account-arrow-right"
}
}
}

View File

@@ -1,4 +1,8 @@
{
"common": {
"trigger_behavior_description": "The behavior of the targeted device trackers to trigger on.",
"trigger_behavior_name": "Behavior"
},
"device_automation": {
"condition_type": {
"is_home": "{entity_name} is home",
@@ -44,6 +48,15 @@
}
}
},
"selector": {
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
}
}
},
"services": {
"see": {
"description": "Manually update the records of a seen legacy device tracker in the known_devices.yaml file.",
@@ -80,5 +93,27 @@
"name": "See"
}
},
"title": "Device tracker"
"title": "Device tracker",
"triggers": {
"entered_home": {
"description": "Triggers when one or more device trackers enter home.",
"fields": {
"behavior": {
"description": "[%key:component::device_tracker::common::trigger_behavior_description%]",
"name": "[%key:component::device_tracker::common::trigger_behavior_name%]"
}
},
"name": "Entered home"
},
"left_home": {
"description": "Triggers when one or more device trackers leave home.",
"fields": {
"behavior": {
"description": "[%key:component::device_tracker::common::trigger_behavior_description%]",
"name": "[%key:component::device_tracker::common::trigger_behavior_name%]"
}
},
"name": "Left home"
}
}
}

View File

@@ -0,0 +1,21 @@
"""Provides triggers for device_trackers."""
from homeassistant.const import STATE_HOME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import (
Trigger,
make_entity_origin_state_trigger,
make_entity_target_state_trigger,
)
from .const import DOMAIN
TRIGGERS: dict[str, type[Trigger]] = {
"entered_home": make_entity_target_state_trigger(DOMAIN, STATE_HOME),
"left_home": make_entity_origin_state_trigger(DOMAIN, from_state=STATE_HOME),
}
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
"""Return the triggers for device trackers."""
return TRIGGERS

View File

@@ -0,0 +1,18 @@
.trigger_common: &trigger_common
target:
entity:
domain: device_tracker
fields:
behavior:
required: true
default: any
selector:
select:
options:
- first
- last
- any
translation_key: trigger_behavior
entered_home: *trigger_common
left_home: *trigger_common

View File

@@ -8,5 +8,5 @@
"integration_type": "hub",
"iot_class": "local_push",
"quality_scale": "bronze",
"requirements": ["ekey-bionyxpy==1.0.0"]
"requirements": ["ekey-bionyxpy==1.0.1"]
}

View File

@@ -2,13 +2,13 @@
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import Trigger, make_entity_state_trigger
from homeassistant.helpers.trigger import Trigger, make_entity_target_state_trigger
from . import DOMAIN
TRIGGERS: dict[str, type[Trigger]] = {
"turned_off": make_entity_state_trigger(DOMAIN, STATE_OFF),
"turned_on": make_entity_state_trigger(DOMAIN, STATE_ON),
"turned_off": make_entity_target_state_trigger(DOMAIN, STATE_OFF),
"turned_on": make_entity_target_state_trigger(DOMAIN, STATE_ON),
}

View File

@@ -5,6 +5,7 @@ from typing import Any
import botocore.exceptions
from homelink.auth.srp_auth import SRPAuth
import jwt
import voluptuous as vol
from homeassistant.config_entries import ConfigFlowResult
@@ -38,8 +39,6 @@ class SRPFlowHandler(AbstractOAuth2FlowHandler, domain=DOMAIN):
"""Ask for username and password."""
errors: dict[str, str] = {}
if user_input is not None:
self._async_abort_entries_match({CONF_EMAIL: user_input[CONF_EMAIL]})
srp_auth = SRPAuth()
try:
tokens = await self.hass.async_add_executor_job(
@@ -48,12 +47,17 @@ class SRPFlowHandler(AbstractOAuth2FlowHandler, domain=DOMAIN):
user_input[CONF_PASSWORD],
)
except botocore.exceptions.ClientError:
_LOGGER.exception("Error authenticating homelink account")
errors["base"] = "srp_auth_failed"
except Exception:
_LOGGER.exception("An unexpected error occurred")
errors["base"] = "unknown"
else:
access_token = jwt.decode(
tokens["AuthenticationResult"]["AccessToken"],
options={"verify_signature": False},
)
await self.async_set_unique_id(access_token["sub"])
self._abort_if_unique_id_configured()
self.external_data = {"tokens": tokens}
return await self.async_step_creation()

View File

@@ -1,10 +1,9 @@
"""Makes requests to the state server and stores the resulting data so that the buttons can access it."""
"""Establish MQTT connection and listen for event data."""
from __future__ import annotations
from collections.abc import Callable
from functools import partial
import logging
from typing import TypedDict
from homelink.model.device import Device
@@ -14,8 +13,6 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.util.ssl import get_default_context
_LOGGER = logging.getLogger(__name__)
type HomeLinkConfigEntry = ConfigEntry[HomeLinkCoordinator]
type EventCallback = Callable[[HomeLinkEventData], None]

View File

@@ -1 +1,87 @@
"""The hikvision component."""
"""The Hikvision integration."""
from __future__ import annotations
from dataclasses import dataclass
import logging
from pyhik.hikvision import HikCamera
import requests
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
CONF_SSL,
CONF_USERNAME,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [Platform.BINARY_SENSOR]
@dataclass
class HikvisionData:
"""Data class for Hikvision runtime data."""
camera: HikCamera
device_id: str
device_name: str
device_type: str
type HikvisionConfigEntry = ConfigEntry[HikvisionData]
async def async_setup_entry(hass: HomeAssistant, entry: HikvisionConfigEntry) -> bool:
"""Set up Hikvision from a config entry."""
host = entry.data[CONF_HOST]
port = entry.data[CONF_PORT]
username = entry.data[CONF_USERNAME]
password = entry.data[CONF_PASSWORD]
ssl = entry.data[CONF_SSL]
protocol = "https" if ssl else "http"
url = f"{protocol}://{host}"
try:
camera = await hass.async_add_executor_job(
HikCamera, url, port, username, password
)
except requests.exceptions.RequestException as err:
raise ConfigEntryNotReady(f"Unable to connect to {host}") from err
device_id = camera.get_id()
if device_id is None:
raise ConfigEntryNotReady(f"Unable to get device ID from {host}")
device_name = camera.get_name or host
device_type = camera.get_type or "Camera"
entry.runtime_data = HikvisionData(
camera=camera,
device_id=device_id,
device_name=device_name,
device_type=device_type,
)
# Start the event stream
await hass.async_add_executor_job(camera.start_stream)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: HikvisionConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
# Stop the event stream
await hass.async_add_executor_job(entry.runtime_data.camera.disconnect)
return unload_ok

View File

@@ -2,10 +2,9 @@
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any
from pyhik.hikvision import HikCamera
import voluptuous as vol
from homeassistant.components.binary_sensor import (
@@ -13,6 +12,7 @@ from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import (
ATTR_LAST_TRIP_TIME,
CONF_CUSTOMIZE,
@@ -23,27 +23,27 @@ from homeassistant.const import (
CONF_PORT,
CONF_SSL,
CONF_USERNAME,
EVENT_HOMEASSISTANT_START,
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import track_point_in_utc_time
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import config_validation as cv, issue_registry as ir
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import (
AddConfigEntryEntitiesCallback,
AddEntitiesCallback,
)
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util.dt import utcnow
_LOGGER = logging.getLogger(__name__)
from . import HikvisionConfigEntry
from .const import DEFAULT_PORT, DOMAIN
CONF_IGNORED = "ignored"
DEFAULT_PORT = 80
DEFAULT_IGNORED = False
DEFAULT_DELAY = 0
DEFAULT_IGNORED = False
ATTR_DELAY = "delay"
DEVICE_CLASS_MAP = {
# Device class mapping for Hikvision event types
DEVICE_CLASS_MAP: dict[str, BinarySensorDeviceClass | None] = {
"Motion": BinarySensorDeviceClass.MOTION,
"Line Crossing": BinarySensorDeviceClass.MOTION,
"Field Detection": BinarySensorDeviceClass.MOTION,
@@ -67,6 +67,8 @@ DEVICE_CLASS_MAP = {
"Entering Region": BinarySensorDeviceClass.MOTION,
}
_LOGGER = logging.getLogger(__name__)
CUSTOMIZE_SCHEMA = vol.Schema(
{
vol.Optional(CONF_IGNORED, default=DEFAULT_IGNORED): cv.boolean,
@@ -88,214 +90,144 @@ PLATFORM_SCHEMA = BINARY_SENSOR_PLATFORM_SCHEMA.extend(
}
)
PARALLEL_UPDATES = 0
def setup_platform(
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the Hikvision binary sensor devices."""
name = config.get(CONF_NAME)
host = config[CONF_HOST]
port = config[CONF_PORT]
username = config[CONF_USERNAME]
password = config[CONF_PASSWORD]
"""Set up the Hikvision binary sensor platform from YAML."""
# Trigger the import flow to migrate YAML config to config entry
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=config
)
customize = config[CONF_CUSTOMIZE]
protocol = "https" if config[CONF_SSL] else "http"
url = f"{protocol}://{host}"
data = HikvisionData(hass, url, port, name, username, password)
if data.sensors is None:
_LOGGER.error("Hikvision event stream has no data, unable to set up")
if (
result.get("type") is FlowResultType.ABORT
and result.get("reason") != "already_configured"
):
ir.async_create_issue(
hass,
DOMAIN,
f"deprecated_yaml_import_issue_{result.get('reason')}",
is_fixable=False,
issue_domain=DOMAIN,
severity=ir.IssueSeverity.WARNING,
translation_key="deprecated_yaml_import_issue",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "Hikvision",
},
)
return
entities = []
for sensor, channel_list in data.sensors.items():
for channel in channel_list:
# Build sensor name, then parse customize config.
if data.type == "NVR":
sensor_name = f"{sensor.replace(' ', '_')}_{channel[1]}"
else:
sensor_name = sensor.replace(" ", "_")
custom = customize.get(sensor_name.lower(), {})
ignore = custom.get(CONF_IGNORED)
delay = custom.get(CONF_DELAY)
_LOGGER.debug(
"Entity: %s - %s, Options - Ignore: %s, Delay: %s",
data.name,
sensor_name,
ignore,
delay,
)
if not ignore:
entities.append(
HikvisionBinarySensor(hass, sensor, channel[1], data, delay)
)
add_entities(entities)
ir.async_create_issue(
hass,
HOMEASSISTANT_DOMAIN,
f"deprecated_yaml_{DOMAIN}",
is_fixable=False,
issue_domain=DOMAIN,
severity=ir.IssueSeverity.WARNING,
translation_key="deprecated_yaml",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "Hikvision",
},
)
class HikvisionData:
"""Hikvision device event stream object."""
async def async_setup_entry(
hass: HomeAssistant,
entry: HikvisionConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Hikvision binary sensors from a config entry."""
data = entry.runtime_data
camera = data.camera
def __init__(self, hass, url, port, name, username, password):
"""Initialize the data object."""
self._url = url
self._port = port
self._name = name
self._username = username
self._password = password
sensors = camera.current_event_states
if sensors is None or not sensors:
_LOGGER.warning("Hikvision device has no sensors available")
return
# Establish camera
self.camdata = HikCamera(self._url, self._port, self._username, self._password)
if self._name is None:
self._name = self.camdata.get_name
hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, self.stop_hik)
hass.bus.listen_once(EVENT_HOMEASSISTANT_START, self.start_hik)
def stop_hik(self, event):
"""Shutdown Hikvision subscriptions and subscription thread on exit."""
self.camdata.disconnect()
def start_hik(self, event):
"""Start Hikvision event stream thread."""
self.camdata.start_stream()
@property
def sensors(self):
"""Return list of available sensors and their states."""
return self.camdata.current_event_states
@property
def cam_id(self):
"""Return device id."""
return self.camdata.get_id
@property
def name(self):
"""Return device name."""
return self._name
@property
def type(self):
"""Return device type."""
return self.camdata.get_type
def get_attributes(self, sensor, channel):
"""Return attribute list for sensor/channel."""
return self.camdata.fetch_attributes(sensor, channel)
async_add_entities(
HikvisionBinarySensor(
entry=entry,
sensor_type=sensor_type,
channel=channel_info[1],
)
for sensor_type, channel_list in sensors.items()
for channel_info in channel_list
)
class HikvisionBinarySensor(BinarySensorEntity):
"""Representation of a Hikvision binary sensor."""
_attr_has_entity_name = True
_attr_should_poll = False
def __init__(self, hass, sensor, channel, cam, delay):
"""Initialize the binary_sensor."""
self._hass = hass
self._cam = cam
self._sensor = sensor
def __init__(
self,
entry: HikvisionConfigEntry,
sensor_type: str,
channel: int,
) -> None:
"""Initialize the binary sensor."""
self._data = entry.runtime_data
self._camera = self._data.camera
self._sensor_type = sensor_type
self._channel = channel
if self._cam.type == "NVR":
self._name = f"{self._cam.name} {sensor} {channel}"
# Build unique ID
self._attr_unique_id = f"{self._data.device_id}_{sensor_type}_{channel}"
# Build entity name based on device type
if self._data.device_type == "NVR":
self._attr_name = f"{sensor_type} {channel}"
else:
self._name = f"{self._cam.name} {sensor}"
self._attr_name = sensor_type
self._id = f"{self._cam.cam_id}.{sensor}.{channel}"
# Device info for device registry
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, self._data.device_id)},
name=self._data.device_name,
manufacturer="Hikvision",
model=self._data.device_type,
)
if delay is None:
self._delay = 0
else:
self._delay = delay
# Set device class
self._attr_device_class = DEVICE_CLASS_MAP.get(sensor_type)
self._timer = None
# Callback ID for pyhik
self._callback_id = f"{self._data.device_id}.{sensor_type}.{channel}"
# Register callback function with pyHik
self._cam.camdata.add_update_callback(self._update_callback, self._id)
def _sensor_state(self):
"""Extract sensor state."""
return self._cam.get_attributes(self._sensor, self._channel)[0]
def _sensor_last_update(self):
"""Extract sensor last update time."""
return self._cam.get_attributes(self._sensor, self._channel)[3]
def _get_sensor_attributes(self) -> tuple[bool, Any, Any, Any]:
"""Get sensor attributes from camera."""
return self._camera.fetch_attributes(self._sensor_type, self._channel)
@property
def name(self):
"""Return the name of the Hikvision sensor."""
return self._name
@property
def unique_id(self):
"""Return a unique ID."""
return self._id
@property
def is_on(self):
def is_on(self) -> bool:
"""Return true if sensor is on."""
return self._sensor_state()
return self._get_sensor_attributes()[0]
@property
def device_class(self):
"""Return the class of this sensor, from DEVICE_CLASSES."""
try:
return DEVICE_CLASS_MAP[self._sensor]
except KeyError:
# Sensor must be unknown to us, add as generic
return None
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes."""
attr = {ATTR_LAST_TRIP_TIME: self._sensor_last_update()}
attrs = self._get_sensor_attributes()
return {ATTR_LAST_TRIP_TIME: attrs[3]}
if self._delay != 0:
attr[ATTR_DELAY] = self._delay
async def async_added_to_hass(self) -> None:
"""Register callback when entity is added."""
await super().async_added_to_hass()
return attr
# Register callback with pyhik
self._camera.add_update_callback(self._update_callback, self._callback_id)
def _update_callback(self, msg):
"""Update the sensor's state, if needed."""
_LOGGER.debug("Callback signal from: %s", msg)
if self._delay > 0 and not self.is_on:
# Set timer to wait until updating the state
def _delay_update(now):
"""Timer callback for sensor update."""
_LOGGER.debug(
"%s Called delayed (%ssec) update", self._name, self._delay
)
self.schedule_update_ha_state()
self._timer = None
if self._timer is not None:
self._timer()
self._timer = None
self._timer = track_point_in_utc_time(
self._hass, _delay_update, utcnow() + timedelta(seconds=self._delay)
)
elif self._delay > 0 and self.is_on:
# For delayed sensors kill any callbacks on true events and update
if self._timer is not None:
self._timer()
self._timer = None
self.schedule_update_ha_state()
else:
self.schedule_update_ha_state()
@callback
def _update_callback(self, msg: str) -> None:
"""Update the sensor's state when callback is triggered."""
self.async_write_ha_state()

View File

@@ -0,0 +1,134 @@
"""Config flow for Hikvision integration."""
from __future__ import annotations
import logging
from typing import Any
from pyhik.hikvision import HikCamera
import requests
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_SSL,
CONF_USERNAME,
)
from homeassistant.helpers.typing import ConfigType
from .const import DEFAULT_PORT, DOMAIN
_LOGGER = logging.getLogger(__name__)
class HikvisionConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Hikvision."""
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:
host = user_input[CONF_HOST]
port = user_input[CONF_PORT]
username = user_input[CONF_USERNAME]
password = user_input[CONF_PASSWORD]
ssl = user_input[CONF_SSL]
protocol = "https" if ssl else "http"
url = f"{protocol}://{host}"
try:
camera = await self.hass.async_add_executor_job(
HikCamera, url, port, username, password
)
device_id = camera.get_id()
device_name = camera.get_name
except requests.exceptions.RequestException:
_LOGGER.exception("Error connecting to Hikvision device")
errors["base"] = "cannot_connect"
else:
if device_id is None:
errors["base"] = "cannot_connect"
else:
await self.async_set_unique_id(device_id)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=device_name or host,
data={
CONF_HOST: host,
CONF_PORT: port,
CONF_USERNAME: username,
CONF_PASSWORD: password,
CONF_SSL: ssl,
},
)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Required(CONF_PORT, default=DEFAULT_PORT): int,
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
vol.Required(CONF_SSL, default=False): bool,
}
),
errors=errors,
)
async def async_step_import(self, import_data: ConfigType) -> ConfigFlowResult:
"""Handle import from configuration.yaml."""
host = import_data[CONF_HOST]
port = import_data.get(CONF_PORT, DEFAULT_PORT)
username = import_data[CONF_USERNAME]
password = import_data[CONF_PASSWORD]
ssl = import_data.get(CONF_SSL, False)
name = import_data.get(CONF_NAME)
protocol = "https" if ssl else "http"
url = f"{protocol}://{host}"
try:
camera = await self.hass.async_add_executor_job(
HikCamera, url, port, username, password
)
device_id = camera.get_id()
device_name = camera.get_name
except requests.exceptions.RequestException:
_LOGGER.exception(
"Error connecting to Hikvision device during import, aborting"
)
return self.async_abort(reason="cannot_connect")
if device_id is None:
return self.async_abort(reason="cannot_connect")
await self.async_set_unique_id(device_id)
self._abort_if_unique_id_configured()
_LOGGER.warning(
"Importing Hikvision config from configuration.yaml for %s", host
)
return self.async_create_entry(
title=name or device_name or host,
data={
CONF_HOST: host,
CONF_PORT: port,
CONF_USERNAME: username,
CONF_PASSWORD: password,
CONF_SSL: ssl,
},
)

View File

@@ -0,0 +1,6 @@
"""Constants for the Hikvision integration."""
DOMAIN = "hikvision"
# Default values
DEFAULT_PORT = 80

View File

@@ -2,7 +2,9 @@
"domain": "hikvision",
"name": "Hikvision",
"codeowners": ["@mezz64"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/hikvision",
"integration_type": "device",
"iot_class": "local_push",
"loggers": ["pyhik"],
"quality_scale": "legacy",

View File

@@ -0,0 +1,36 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"password": "[%key:common::config_flow::data::password%]",
"port": "[%key:common::config_flow::data::port%]",
"ssl": "Use SSL",
"username": "[%key:common::config_flow::data::username%]"
},
"data_description": {
"host": "The hostname or IP address of your Hikvision device",
"password": "The password for your Hikvision device",
"port": "The port number for the device (default is 80)",
"ssl": "Enable if your device uses HTTPS",
"username": "The username for your Hikvision device"
},
"description": "Enter your Hikvision device connection details.",
"title": "Set up Hikvision device"
}
}
},
"issues": {
"deprecated_yaml_import_issue": {
"description": "Configuring {integration_title} using YAML is deprecated and the import failed. Please remove the `{domain}` entry from your `configuration.yaml` file and set up the integration manually.",
"title": "YAML import failed"
}
}
}

View File

@@ -108,7 +108,7 @@ class IcloudAccount:
if self.api.requires_2fa:
# Trigger a new log in to ensure the user enters the 2FA code again.
raise PyiCloudFailedLoginException # noqa: TRY301
raise PyiCloudFailedLoginException("2FA Required") # noqa: TRY301
except PyiCloudFailedLoginException:
self.api = None

View File

@@ -261,7 +261,8 @@ class ImprovBLEConfigFlow(ConfigFlow, domain=DOMAIN):
if self._can_identify is None:
try:
self._can_identify = await self._try_call(device.can_identify())
await self._try_call(device.ensure_connected())
self._can_identify = device.can_identify
except AbortFlow as err:
return self.async_abort(reason=err.reason)
if self._can_identify:

View File

@@ -13,5 +13,5 @@
"documentation": "https://www.home-assistant.io/integrations/improv_ble",
"integration_type": "device",
"iot_class": "local_polling",
"requirements": ["py-improv-ble-client==1.0.3"]
"requirements": ["py-improv-ble-client==2.0.1"]
}

View File

@@ -20,5 +20,13 @@
"turn_on": {
"service": "mdi:toggle-switch"
}
},
"triggers": {
"turned_off": {
"trigger": "mdi:toggle-switch-off"
},
"turned_on": {
"trigger": "mdi:toggle-switch"
}
}
}

View File

@@ -1,4 +1,8 @@
{
"common": {
"trigger_behavior_description": "The behavior of the targeted input booleans to trigger on.",
"trigger_behavior_name": "Behavior"
},
"entity_component": {
"_": {
"name": "[%key:component::input_boolean::title%]",
@@ -17,6 +21,15 @@
}
}
},
"selector": {
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
}
}
},
"services": {
"reload": {
"description": "Reloads helpers from the YAML-configuration.",
@@ -35,5 +48,27 @@
"name": "[%key:common::action::turn_on%]"
}
},
"title": "Input boolean"
"title": "Input boolean",
"triggers": {
"turned_off": {
"description": "Triggers after one or more input booleans turn off.",
"fields": {
"behavior": {
"description": "[%key:component::input_boolean::common::trigger_behavior_description%]",
"name": "[%key:component::input_boolean::common::trigger_behavior_name%]"
}
},
"name": "Input boolean turned off"
},
"turned_on": {
"description": "Triggers after one or more input booleans turn on.",
"fields": {
"behavior": {
"description": "[%key:component::input_boolean::common::trigger_behavior_description%]",
"name": "[%key:component::input_boolean::common::trigger_behavior_name%]"
}
},
"name": "Input boolean turned on"
}
}
}

View File

@@ -0,0 +1,17 @@
"""Provides triggers for input booleans."""
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import Trigger, make_entity_target_state_trigger
from . import DOMAIN
TRIGGERS: dict[str, type[Trigger]] = {
"turned_on": make_entity_target_state_trigger(DOMAIN, STATE_ON),
"turned_off": make_entity_target_state_trigger(DOMAIN, STATE_OFF),
}
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
"""Return the triggers for input booleans."""
return TRIGGERS

View File

@@ -0,0 +1,18 @@
.trigger_common: &trigger_common
target:
entity:
domain: input_boolean
fields:
behavior:
required: true
default: any
selector:
select:
options:
- first
- last
- any
translation_key: trigger_behavior
turned_off: *trigger_common
turned_on: *trigger_common

View File

@@ -18,7 +18,7 @@
"iot_class": "local_push",
"loggers": ["pyinsteon", "pypubsub"],
"requirements": [
"pyinsteon==1.6.3",
"pyinsteon==1.6.4",
"insteon-frontend-home-assistant==0.5.0"
],
"single_config_entry": true,

View File

@@ -4,6 +4,7 @@
"codeowners": ["@dgomes"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/ipma",
"integration_type": "service",
"iot_class": "cloud_polling",
"loggers": ["geopy", "pyipma"],
"requirements": ["pyipma==3.0.9"]

View File

@@ -1,15 +1,17 @@
"""Provides triggers for lawn mowers."""
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import Trigger, make_entity_state_trigger
from homeassistant.helpers.trigger import Trigger, make_entity_target_state_trigger
from .const import DOMAIN, LawnMowerActivity
TRIGGERS: dict[str, type[Trigger]] = {
"docked": make_entity_state_trigger(DOMAIN, LawnMowerActivity.DOCKED),
"errored": make_entity_state_trigger(DOMAIN, LawnMowerActivity.ERROR),
"paused_mowing": make_entity_state_trigger(DOMAIN, LawnMowerActivity.PAUSED),
"started_mowing": make_entity_state_trigger(DOMAIN, LawnMowerActivity.MOWING),
"docked": make_entity_target_state_trigger(DOMAIN, LawnMowerActivity.DOCKED),
"errored": make_entity_target_state_trigger(DOMAIN, LawnMowerActivity.ERROR),
"paused_mowing": make_entity_target_state_trigger(DOMAIN, LawnMowerActivity.PAUSED),
"started_mowing": make_entity_target_state_trigger(
DOMAIN, LawnMowerActivity.MOWING
),
}

View File

@@ -2,13 +2,13 @@
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import Trigger, make_entity_state_trigger
from homeassistant.helpers.trigger import Trigger, make_entity_target_state_trigger
from .const import DOMAIN
TRIGGERS: dict[str, type[Trigger]] = {
"turned_off": make_entity_state_trigger(DOMAIN, STATE_OFF),
"turned_on": make_entity_state_trigger(DOMAIN, STATE_ON),
"turned_off": make_entity_target_state_trigger(DOMAIN, STATE_OFF),
"turned_on": make_entity_target_state_trigger(DOMAIN, STATE_ON),
}

View File

@@ -1,13 +1,13 @@
"""Provides triggers for media players."""
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import Trigger, make_conditional_entity_state_trigger
from homeassistant.helpers.trigger import Trigger, make_entity_transition_trigger
from . import MediaPlayerState
from .const import DOMAIN
TRIGGERS: dict[str, type[Trigger]] = {
"stopped_playing": make_conditional_entity_state_trigger(
"stopped_playing": make_entity_transition_trigger(
DOMAIN,
from_states={
MediaPlayerState.BUFFERING,

View File

@@ -1,9 +1,10 @@
{
"domain": "meteo_france",
"name": "M\u00e9t\u00e9o-France",
"name": "Météo-France",
"codeowners": ["@hacf-fr", "@oncleben31", "@Quentame"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/meteo_france",
"integration_type": "service",
"iot_class": "cloud_polling",
"loggers": ["meteofrance_api"],
"requirements": ["meteofrance-api==1.4.0"]

View File

@@ -62,7 +62,7 @@ SENSOR_TYPES: tuple[MeteoFranceSensorEntityDescription, ...] = (
key="pressure",
name="Pressure",
native_unit_of_measurement=UnitOfPressure.HPA,
device_class=SensorDeviceClass.PRESSURE,
device_class=SensorDeviceClass.ATMOSPHERIC_PRESSURE,
state_class=SensorStateClass.MEASUREMENT,
entity_registry_enabled_default=False,
data_path="current_forecast:sea_level",

View File

@@ -7,5 +7,5 @@
"documentation": "https://www.home-assistant.io/integrations/mill",
"iot_class": "local_polling",
"loggers": ["mill", "mill_local"],
"requirements": ["millheater==0.14.1", "mill-local==0.3.0"]
"requirements": ["millheater==0.14.1", "mill-local==0.5.0"]
}

View File

@@ -24,6 +24,7 @@ _PLATFORMS: list[Platform] = [
Platform.TIME,
Platform.SWITCH,
Platform.NUMBER,
Platform.SELECT,
]
PLATFORM_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)

View File

@@ -5,14 +5,18 @@ from __future__ import annotations
from datetime import timedelta
import logging
from pynintendoauth.exceptions import InvalidOAuthConfigurationException
from pynintendoauth.exceptions import (
HttpException,
InvalidOAuthConfigurationException,
InvalidSessionTokenException,
)
from pynintendoparental import Authenticator, NintendoParental
from pynintendoparental.exceptions import NoDevicesFoundException
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
@@ -58,3 +62,13 @@ class NintendoUpdateCoordinator(DataUpdateCoordinator[None]):
translation_domain=DOMAIN,
translation_key="no_devices_found",
) from err
except InvalidSessionTokenException as err:
_LOGGER.debug("Session token invalid, will renew on next update")
raise UpdateFailed from err
except HttpException as err:
if err.error_code == "update_required":
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="update_required",
) from err
raise UpdateFailed(retry_after=900) from err

View File

@@ -0,0 +1,95 @@
"""Nintendo Switch Parental Controls select entity definitions."""
from __future__ import annotations
from collections.abc import Callable, Coroutine
from dataclasses import dataclass
from enum import StrEnum
from typing import Any
from pynintendoparental.enum import DeviceTimerMode
from homeassistant.components.select import SelectEntity, SelectEntityDescription
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import NintendoParentalControlsConfigEntry, NintendoUpdateCoordinator
from .entity import Device, NintendoDevice
PARALLEL_UPDATES = 1
class NintendoParentalSelect(StrEnum):
"""Store keys for Nintendo Parental Controls select entities."""
TIMER_MODE = "timer_mode"
@dataclass(kw_only=True, frozen=True)
class NintendoParentalControlsSelectEntityDescription(SelectEntityDescription):
"""Description for Nintendo Parental Controls select entities."""
get_option: Callable[[Device], DeviceTimerMode | None]
set_option_fn: Callable[[Device, DeviceTimerMode], Coroutine[Any, Any, None]]
options_enum: type[DeviceTimerMode]
SELECT_DESCRIPTIONS: tuple[NintendoParentalControlsSelectEntityDescription, ...] = (
NintendoParentalControlsSelectEntityDescription(
key=NintendoParentalSelect.TIMER_MODE,
translation_key=NintendoParentalSelect.TIMER_MODE,
get_option=lambda device: device.timer_mode,
set_option_fn=lambda device, option: device.set_timer_mode(option),
options_enum=DeviceTimerMode,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: NintendoParentalControlsConfigEntry,
async_add_devices: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the select platform."""
async_add_devices(
NintendoParentalSelectEntity(
coordinator=entry.runtime_data,
device=device,
description=description,
)
for device in entry.runtime_data.api.devices.values()
for description in SELECT_DESCRIPTIONS
)
class NintendoParentalSelectEntity(NintendoDevice, SelectEntity):
"""Nintendo Parental Controls select entity."""
entity_description: NintendoParentalControlsSelectEntityDescription
def __init__(
self,
coordinator: NintendoUpdateCoordinator,
device: Device,
description: NintendoParentalControlsSelectEntityDescription,
) -> None:
"""Initialize the select entity."""
super().__init__(coordinator=coordinator, device=device, key=description.key)
self.entity_description = description
@property
def current_option(self) -> str | None:
"""Return the current selected option."""
option = self.entity_description.get_option(self._device)
return option.name.lower() if option else None
@property
def options(self) -> list[str]:
"""Return a list of available options."""
return [option.name.lower() for option in self.entity_description.options_enum]
async def async_select_option(self, option: str) -> None:
"""Change the selected option."""
enum_option = self.entity_description.options_enum[option.upper()]
await self.entity_description.set_option_fn(self._device, enum_option)
await self.coordinator.async_request_refresh()

View File

@@ -7,7 +7,7 @@ import voluptuous as vol
from homeassistant.const import ATTR_DEVICE_ID
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import config_validation as cv, device_registry as dr
from .const import ATTR_BONUS_TIME, DOMAIN
@@ -56,7 +56,7 @@ async def async_add_bonus_time(call: ServiceCall) -> None:
bonus_time: int = data[ATTR_BONUS_TIME]
device = dr.async_get(call.hass).async_get(device_id)
if device is None:
raise HomeAssistantError(
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="device_not_found",
)
@@ -66,6 +66,10 @@ async def async_add_bonus_time(call: ServiceCall) -> None:
break
nintendo_device_id = _get_nintendo_device_id(device)
if config_entry and nintendo_device_id:
await config_entry.runtime_data.api.devices[nintendo_device_id].add_extra_time(
bonus_time
)
return await config_entry.runtime_data.api.devices[
nintendo_device_id
].add_extra_time(bonus_time)
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="invalid_device",
)

View File

@@ -37,6 +37,15 @@
"name": "Max screentime today"
}
},
"select": {
"timer_mode": {
"name": "Restriction mode",
"state": {
"daily": "Same for all days",
"each_day_of_the_week": "Different for each day"
}
}
},
"sensor": {
"playing_time": {
"name": "Used screen time"
@@ -69,8 +78,14 @@
"device_not_found": {
"message": "Device not found."
},
"invalid_device": {
"message": "The specified device is not a Nintendo device."
},
"no_devices_found": {
"message": "No Nintendo devices found for this account."
},
"update_required": {
"message": "The Nintendo Switch parental controls integration requires an update due to changes in Nintendo's API."
}
},
"services": {

View File

@@ -159,7 +159,8 @@ class OverseerrWebhookManager:
"""Handle webhook."""
data = await request.json()
LOGGER.debug("Received webhook payload: %s", data)
if data["notification_type"].startswith("MEDIA"):
notification_type = data["notification_type"]
if notification_type.startswith(("REQUEST_", "ISSUE_", "MEDIA_")):
await self.entry.runtime_data.async_refresh()
async_dispatcher_send(hass, EVENT_KEY, data)
return HomeAssistantView.json({"message": "ok"})

View File

@@ -22,6 +22,10 @@ REGISTERED_NOTIFICATIONS = (
| NotificationType.REQUEST_AVAILABLE
| NotificationType.REQUEST_PROCESSING_FAILED
| NotificationType.REQUEST_AUTOMATICALLY_APPROVED
| NotificationType.ISSUE_REPORTED
| NotificationType.ISSUE_COMMENTED
| NotificationType.ISSUE_RESOLVED
| NotificationType.ISSUE_REOPENED
)
JSON_PAYLOAD = (
'"{\\"notification_type\\":\\"{{notification_type}}\\",\\"subject\\":\\"{{subject}'

View File

@@ -6,7 +6,6 @@ from python_overseerr import (
OverseerrAuthenticationError,
OverseerrClient,
OverseerrConnectionError,
RequestCount,
)
from yarl import URL
@@ -18,11 +17,12 @@ from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN, LOGGER
from .models import OverseerrData
type OverseerrConfigEntry = ConfigEntry[OverseerrCoordinator]
class OverseerrCoordinator(DataUpdateCoordinator[RequestCount]):
class OverseerrCoordinator(DataUpdateCoordinator[OverseerrData]):
"""Class to manage fetching Overseerr data."""
config_entry: OverseerrConfigEntry
@@ -49,10 +49,12 @@ class OverseerrCoordinator(DataUpdateCoordinator[RequestCount]):
self.url = URL.build(host=host, port=port, scheme="https" if ssl else "http")
self.push = False
async def _async_update_data(self) -> RequestCount:
async def _async_update_data(self) -> OverseerrData:
"""Fetch data from API endpoint."""
try:
return await self.client.get_request_count()
requests = await self.client.get_request_count()
issues = await self.client.get_issue_count()
return OverseerrData(requests=requests, issues=issues)
except OverseerrAuthenticationError as err:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,

View File

@@ -0,0 +1,13 @@
"""Data models for Overseerr integration."""
from dataclasses import dataclass
from python_overseerr import IssueCount, RequestCount
@dataclass
class OverseerrData:
"""Data model for Overseerr coordinator."""
requests: RequestCount
issues: IssueCount

View File

@@ -3,8 +3,6 @@
from collections.abc import Callable
from dataclasses import dataclass
from python_overseerr import RequestCount
from homeassistant.components.sensor import (
SensorEntity,
SensorEntityDescription,
@@ -16,6 +14,7 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import REQUESTS
from .coordinator import OverseerrConfigEntry, OverseerrCoordinator
from .entity import OverseerrEntity
from .models import OverseerrData
PARALLEL_UPDATES = 0
@@ -24,7 +23,7 @@ PARALLEL_UPDATES = 0
class OverseerrSensorEntityDescription(SensorEntityDescription):
"""Describes Overseerr config sensor entity."""
value_fn: Callable[[RequestCount], int]
value_fn: Callable[[OverseerrData], int]
SENSORS: tuple[OverseerrSensorEntityDescription, ...] = (
@@ -32,43 +31,73 @@ SENSORS: tuple[OverseerrSensorEntityDescription, ...] = (
key="total_requests",
native_unit_of_measurement=REQUESTS,
state_class=SensorStateClass.TOTAL,
value_fn=lambda count: count.total,
value_fn=lambda data: data.requests.total,
),
OverseerrSensorEntityDescription(
key="movie_requests",
native_unit_of_measurement=REQUESTS,
state_class=SensorStateClass.TOTAL,
value_fn=lambda count: count.movie,
value_fn=lambda data: data.requests.movie,
),
OverseerrSensorEntityDescription(
key="tv_requests",
native_unit_of_measurement=REQUESTS,
state_class=SensorStateClass.TOTAL,
value_fn=lambda count: count.tv,
value_fn=lambda data: data.requests.tv,
),
OverseerrSensorEntityDescription(
key="pending_requests",
native_unit_of_measurement=REQUESTS,
state_class=SensorStateClass.TOTAL,
value_fn=lambda count: count.pending,
value_fn=lambda data: data.requests.pending,
),
OverseerrSensorEntityDescription(
key="declined_requests",
native_unit_of_measurement=REQUESTS,
state_class=SensorStateClass.TOTAL,
value_fn=lambda count: count.declined,
value_fn=lambda data: data.requests.declined,
),
OverseerrSensorEntityDescription(
key="processing_requests",
native_unit_of_measurement=REQUESTS,
state_class=SensorStateClass.TOTAL,
value_fn=lambda count: count.processing,
value_fn=lambda data: data.requests.processing,
),
OverseerrSensorEntityDescription(
key="available_requests",
native_unit_of_measurement=REQUESTS,
state_class=SensorStateClass.TOTAL,
value_fn=lambda count: count.available,
value_fn=lambda data: data.requests.available,
),
OverseerrSensorEntityDescription(
key="total_issues",
state_class=SensorStateClass.TOTAL,
value_fn=lambda data: data.issues.total,
),
OverseerrSensorEntityDescription(
key="open_issues",
state_class=SensorStateClass.TOTAL,
value_fn=lambda data: data.issues.open,
),
OverseerrSensorEntityDescription(
key="closed_issues",
state_class=SensorStateClass.TOTAL,
value_fn=lambda data: data.issues.closed,
),
OverseerrSensorEntityDescription(
key="video_issues",
state_class=SensorStateClass.TOTAL,
value_fn=lambda data: data.issues.video,
),
OverseerrSensorEntityDescription(
key="audio_issues",
state_class=SensorStateClass.TOTAL,
value_fn=lambda data: data.issues.audio,
),
OverseerrSensorEntityDescription(
key="subtitle_issues",
state_class=SensorStateClass.TOTAL,
value_fn=lambda data: data.issues.subtitles,
),
)

View File

@@ -50,26 +50,62 @@
}
},
"sensor": {
"audio_issues": {
"name": "Audio issues",
"state": {
"measurement": "issues"
}
},
"available_requests": {
"name": "Available requests"
},
"closed_issues": {
"name": "Closed issues",
"state": {
"measurement": "issues"
}
},
"declined_requests": {
"name": "Declined requests"
},
"movie_requests": {
"name": "Movie requests"
},
"open_issues": {
"name": "Open issues",
"state": {
"measurement": "issues"
}
},
"pending_requests": {
"name": "Pending requests"
},
"processing_requests": {
"name": "Processing requests"
},
"subtitle_issues": {
"name": "Subtitle issues",
"state": {
"measurement": "issues"
}
},
"total_issues": {
"name": "Total issues",
"state": {
"measurement": "issues"
}
},
"total_requests": {
"name": "Total requests"
},
"tv_requests": {
"name": "TV requests"
},
"video_issues": {
"name": "Video issues",
"state": {
"measurement": "issues"
}
}
}
},

View File

@@ -20,7 +20,7 @@
"loggers": ["roborock"],
"quality_scale": "silver",
"requirements": [
"python-roborock==3.12.2",
"python-roborock==3.18.0",
"vacuum-map-parser-roborock==0.1.4"
]
}

View File

@@ -13,5 +13,6 @@
"integration_type": "hub",
"iot_class": "local_polling",
"loggers": ["pysqueezebox"],
"quality_scale": "silver",
"requirements": ["pysqueezebox==0.13.0"]
}

View File

@@ -0,0 +1,69 @@
rules:
# Bronze
action-setup:
status: exempt
comment: Integration only has entity_actions, which are setup in the entity async_setup_entry.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow:
status: done
comment: Future enhancements, 1) separate manual and discovery flows, 2) allow for discovery of multiple LMS and selection of one. PR 153958
dependency-transparency: done
docs-actions: done
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup: done
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions:
status: done
config-entry-unloading: done
docs-configuration-parameters: done
docs-installation-parameters: done
entity-unavailable: done
integration-owner: done
log-when-unavailable: done
parallel-updates: done
reauthentication-flow:
status: exempt
comment: Integration doesn't have an auth flow.
test-coverage: done
# Gold
devices: done
diagnostics: todo
discovery-update-info: todo
discovery: todo
docs-data-update: done
docs-examples: todo
docs-known-limitations: done
docs-supported-devices: done
docs-supported-functions: done
docs-troubleshooting: todo
docs-use-cases: done
dynamic-devices: done
entity-category: todo
entity-device-class: done
entity-disabled-by-default:
status: exempt
comment: There aren't any entities that should be disabled by default.
entity-translations: done
exception-translations: todo
icon-translations: todo
reconfiguration-flow: todo
repair-issues: todo
stale-devices: todo
# Platinum
async-dependency: done
inject-websession: todo
strict-typing: todo

View File

@@ -18,6 +18,7 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
from .const import CONF_SERIAL_NUMBER, DOMAIN, MANUFACTURER
from .types import DaliCenterConfigEntry, DaliCenterData
@@ -58,6 +59,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: DaliCenterConfigEntry) -
dev_reg = dr.async_get(hass)
dev_reg.async_get_or_create(
config_entry_id=entry.entry_id,
connections={(CONNECTION_NETWORK_MAC, gw_sn)},
identifiers={(DOMAIN, gw_sn)},
manufacturer=MANUFACTURER,
name=gateway.name,

View File

@@ -18,11 +18,13 @@ from homeassistant.const import (
CONF_PORT,
CONF_USERNAME,
)
from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.selector import (
SelectOptionDict,
SelectSelector,
SelectSelectorConfig,
)
from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
from .const import CONF_SERIAL_NUMBER, DOMAIN
@@ -132,3 +134,15 @@ class DaliCenterConfigFlow(ConfigFlow, domain=DOMAIN):
),
errors=errors,
)
async def async_step_dhcp(
self, discovery_info: DhcpServiceInfo
) -> ConfigFlowResult:
"""Handle DHCP discovery to update existing entries."""
mac_address = format_mac(discovery_info.macaddress)
serial_number = mac_address.replace(":", "").upper()
await self.async_set_unique_id(serial_number)
self._abort_if_unique_id_configured(updates={CONF_HOST: discovery_info.ip})
return self.async_abort(reason="no_dhcp_flow")

View File

@@ -3,6 +3,11 @@
"name": "Sunricher DALI",
"codeowners": ["@niracler"],
"config_flow": true,
"dhcp": [
{
"registered_devices": true
}
],
"documentation": "https://www.home-assistant.io/integrations/sunricher_dali",
"iot_class": "local_push",
"quality_scale": "bronze",

View File

@@ -40,8 +40,10 @@ rules:
# Gold
devices: done
diagnostics: todo
discovery-update-info: todo
discovery: todo
discovery-update-info: done
discovery:
status: exempt
comment: Device has no way to be discovered.
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo

View File

@@ -1,13 +1,13 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"no_dhcp_flow": "No DHCP flow"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"discovery_failed": "Failed to discover Sunricher DALI gateways on the network",
"no_devices_found": "No Sunricher DALI gateways found on the network",
"unknown": "[%key:common::config_flow::error::unknown%]"
"no_devices_found": "No Sunricher DALI gateways found on the network"
},
"step": {
"select_gateway": {

View File

@@ -2,13 +2,13 @@
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import Trigger, make_entity_state_trigger
from homeassistant.helpers.trigger import Trigger, make_entity_target_state_trigger
from .const import DOMAIN
TRIGGERS: dict[str, type[Trigger]] = {
"turned_on": make_entity_state_trigger(DOMAIN, STATE_ON),
"turned_off": make_entity_state_trigger(DOMAIN, STATE_OFF),
"turned_on": make_entity_target_state_trigger(DOMAIN, STATE_ON),
"turned_off": make_entity_target_state_trigger(DOMAIN, STATE_OFF),
}

View File

@@ -63,6 +63,7 @@ from .const import (
ATTR_PASSWORD,
ATTR_QUESTION,
ATTR_REACTION,
ATTR_REPLY_TO_MSGID,
ATTR_RESIZE_KEYBOARD,
ATTR_SHOW_ALERT,
ATTR_STICKER_ID,
@@ -126,21 +127,26 @@ BASE_SERVICE_SCHEMA = vol.Schema(
vol.Optional(ATTR_TIMEOUT): cv.positive_int,
vol.Optional(ATTR_MESSAGE_TAG): cv.string,
vol.Optional(ATTR_MESSAGE_THREAD_ID): vol.Coerce(int),
},
extra=vol.ALLOW_EXTRA,
}
)
SERVICE_SCHEMA_SEND_MESSAGE = vol.All(
cv.deprecated(ATTR_TIMEOUT),
BASE_SERVICE_SCHEMA.extend(
{vol.Required(ATTR_MESSAGE): cv.string, vol.Optional(ATTR_TITLE): cv.string}
{
vol.Required(ATTR_MESSAGE): cv.string,
vol.Optional(ATTR_TITLE): cv.string,
vol.Optional(ATTR_REPLY_TO_MSGID): vol.Coerce(int),
}
),
)
SERVICE_SCHEMA_SEND_CHAT_ACTION = vol.All(
cv.deprecated(ATTR_TIMEOUT),
BASE_SERVICE_SCHEMA.extend(
vol.Schema(
{
vol.Optional(CONF_CONFIG_ENTRY_ID): cv.string,
vol.Optional(ATTR_TARGET): vol.All(cv.ensure_list, [vol.Coerce(int)]),
vol.Required(ATTR_CHAT_ACTION): vol.In(
(
CHAT_ACTION_TYPING,
@@ -156,6 +162,7 @@ SERVICE_SCHEMA_SEND_CHAT_ACTION = vol.All(
CHAT_ACTION_UPLOAD_VIDEO_NOTE,
)
),
vol.Optional(ATTR_MESSAGE_THREAD_ID): vol.Coerce(int),
}
),
)
@@ -169,6 +176,7 @@ SERVICE_SCHEMA_BASE_SEND_FILE = BASE_SERVICE_SCHEMA.extend(
vol.Optional(ATTR_PASSWORD): cv.string,
vol.Optional(ATTR_AUTHENTICATION): cv.string,
vol.Optional(ATTR_VERIFY_SSL): cv.boolean,
vol.Optional(ATTR_REPLY_TO_MSGID): vol.Coerce(int),
}
)
@@ -188,6 +196,7 @@ SERVICE_SCHEMA_SEND_LOCATION = vol.All(
{
vol.Required(ATTR_LONGITUDE): cv.string,
vol.Required(ATTR_LATITUDE): cv.string,
vol.Optional(ATTR_REPLY_TO_MSGID): vol.Coerce(int),
}
),
)
@@ -205,18 +214,25 @@ SERVICE_SCHEMA_SEND_POLL = vol.All(
vol.Optional(ATTR_ALLOWS_MULTIPLE_ANSWERS, default=False): cv.boolean,
vol.Optional(ATTR_DISABLE_NOTIF): cv.boolean,
vol.Optional(ATTR_MESSAGE_THREAD_ID): vol.Coerce(int),
vol.Optional(ATTR_REPLY_TO_MSGID): vol.Coerce(int),
}
),
)
SERVICE_SCHEMA_EDIT_MESSAGE = vol.All(
cv.deprecated(ATTR_TIMEOUT),
SERVICE_SCHEMA_BASE_SEND_FILE.extend(
vol.Schema(
{
vol.Optional(CONF_CONFIG_ENTRY_ID): cv.string,
vol.Optional(ATTR_TITLE): cv.string,
vol.Required(ATTR_MESSAGE): cv.string,
vol.Required(ATTR_MESSAGEID): vol.Any(
cv.positive_int, vol.All(cv.string, "last")
),
vol.Required(ATTR_CHAT_ID): vol.Coerce(int),
vol.Optional(ATTR_PARSER): cv.string,
vol.Optional(ATTR_KEYBOARD_INLINE): cv.ensure_list,
vol.Optional(ATTR_DISABLE_WEB_PREV): cv.boolean,
}
),
)

View File

@@ -783,6 +783,7 @@ class TelegramNotificationService:
None,
chat_id=chat_id,
action=chat_action,
message_thread_id=kwargs.get(ATTR_MESSAGE_THREAD_ID),
context=context,
)
result[chat_id] = is_successful

View File

@@ -1,5 +1,6 @@
"""Support for setting the Transmission BitTorrent client Turtle Mode."""
import asyncio
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any
@@ -12,6 +13,7 @@ from .coordinator import TransmissionConfigEntry, TransmissionDataUpdateCoordina
from .entity import TransmissionEntity
PARALLEL_UPDATES = 0
AFTER_WRITE_SLEEP = 2
@dataclass(frozen=True, kw_only=True)
@@ -70,6 +72,7 @@ class TransmissionSwitch(TransmissionEntity, SwitchEntity):
await self.hass.async_add_executor_job(
self.entity_description.on_func, self.coordinator
)
await asyncio.sleep(AFTER_WRITE_SLEEP)
await self.coordinator.async_request_refresh()
async def async_turn_off(self, **kwargs: Any) -> None:
@@ -77,4 +80,5 @@ class TransmissionSwitch(TransmissionEntity, SwitchEntity):
await self.hass.async_add_executor_job(
self.entity_description.off_func, self.coordinator
)
await asyncio.sleep(AFTER_WRITE_SLEEP)
await self.coordinator.async_request_refresh()

View File

@@ -40,7 +40,7 @@ from .const import (
PLATFORMS,
)
from .data import ProtectData, UFPConfigEntry
from .discovery import async_start_discovery
from .discovery import DATA_UNIFIPROTECT, UniFiProtectRuntimeData, async_start_discovery
from .migrate import async_migrate_data
from .services import async_setup_services
from .utils import (
@@ -64,6 +64,8 @@ CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the UniFi Protect."""
# Initialize domain data structure (setdefault in case discovery already started)
hass.data.setdefault(DATA_UNIFIPROTECT, UniFiProtectRuntimeData())
# Only start discovery once regardless of how many entries they have
async_setup_services(hass)
async_start_discovery(hass)
@@ -79,11 +81,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: UFPConfigEntry) -> bool:
try:
await protect.update()
except NotAuthorized as err:
retry_key = f"{entry.entry_id}_auth"
retries = hass.data.setdefault(DOMAIN, {}).get(retry_key, 0)
domain_data = hass.data.setdefault(DATA_UNIFIPROTECT, UniFiProtectRuntimeData())
retries = domain_data.auth_retries.get(entry.entry_id, 0)
if retries < AUTH_RETRIES:
retries += 1
hass.data[DOMAIN][retry_key] = retries
domain_data.auth_retries[entry.entry_id] = retries
raise ConfigEntryNotReady from err
raise ConfigEntryAuthFailed(err) from err
except (TimeoutError, ClientError, ServerDisconnectedError) as err:

View File

@@ -2,7 +2,7 @@
from __future__ import annotations
from dataclasses import asdict
from dataclasses import asdict, dataclass, field
from datetime import timedelta
import logging
from typing import Any
@@ -13,22 +13,34 @@ from homeassistant import config_entries
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import discovery_flow
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.util.hass_dict import HassKey
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
DISCOVERY = "discovery"
@dataclass
class UniFiProtectRuntimeData:
"""Runtime data stored in hass.data[DOMAIN]."""
auth_retries: dict[str, int] = field(default_factory=dict)
discovery_started: bool = False
# Typed key for hass.data access at DOMAIN level
DATA_UNIFIPROTECT: HassKey[UniFiProtectRuntimeData] = HassKey(DOMAIN)
DISCOVERY_INTERVAL = timedelta(minutes=60)
@callback
def async_start_discovery(hass: HomeAssistant) -> None:
"""Start discovery."""
domain_data = hass.data.setdefault(DOMAIN, {})
if DISCOVERY in domain_data:
domain_data = hass.data.setdefault(DATA_UNIFIPROTECT, UniFiProtectRuntimeData())
if domain_data.discovery_started:
return
domain_data[DISCOVERY] = True
domain_data.discovery_started = True
async def _async_discovery() -> None:
async_trigger_discovery(hass, await async_discover_devices())

View File

@@ -2,12 +2,12 @@
from homeassistant.const import STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import Trigger, make_entity_state_trigger
from homeassistant.helpers.trigger import Trigger, make_entity_target_state_trigger
from .const import DOMAIN
TRIGGERS: dict[str, type[Trigger]] = {
"update_became_available": make_entity_state_trigger(DOMAIN, STATE_ON),
"update_became_available": make_entity_target_state_trigger(DOMAIN, STATE_ON),
}

View File

@@ -1,15 +1,17 @@
"""Provides triggers for vacuum cleaners."""
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import Trigger, make_entity_state_trigger
from homeassistant.helpers.trigger import Trigger, make_entity_target_state_trigger
from .const import DOMAIN, VacuumActivity
TRIGGERS: dict[str, type[Trigger]] = {
"docked": make_entity_state_trigger(DOMAIN, VacuumActivity.DOCKED),
"errored": make_entity_state_trigger(DOMAIN, VacuumActivity.ERROR),
"paused_cleaning": make_entity_state_trigger(DOMAIN, VacuumActivity.PAUSED),
"started_cleaning": make_entity_state_trigger(DOMAIN, VacuumActivity.CLEANING),
"docked": make_entity_target_state_trigger(DOMAIN, VacuumActivity.DOCKED),
"errored": make_entity_target_state_trigger(DOMAIN, VacuumActivity.ERROR),
"paused_cleaning": make_entity_target_state_trigger(DOMAIN, VacuumActivity.PAUSED),
"started_cleaning": make_entity_target_state_trigger(
DOMAIN, VacuumActivity.CLEANING
),
}

View File

@@ -4,10 +4,22 @@ from __future__ import annotations
import logging
from httpx import HTTPStatusError, RequestError, TimeoutException
from pythonxbox.api.client import XboxLiveClient
from homeassistant.config_entries import ConfigSubentry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.helpers.config_entry_oauth2_flow import (
ImplementationUnavailableError,
OAuth2Session,
async_get_config_entry_implementation,
)
from homeassistant.helpers.httpx_client import get_async_client
from .api import AsyncConfigEntryAuth
from .const import DOMAIN
from .coordinator import (
XboxConfigEntry,
@@ -41,34 +53,105 @@ async def async_setup_entry(hass: HomeAssistant, entry: XboxConfigEntry) -> bool
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
await async_migrate_unique_id(hass, entry)
entry.async_on_unload(entry.add_update_listener(_async_update_listener))
return True
async def _async_update_listener(hass: HomeAssistant, entry: XboxConfigEntry) -> None:
"""Handle update."""
await hass.config_entries.async_reload(entry.entry_id)
async def async_unload_entry(hass: HomeAssistant, entry: XboxConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def async_migrate_unique_id(hass: HomeAssistant, entry: XboxConfigEntry) -> bool:
"""Migrate config entry.
async def async_migrate_entry(hass: HomeAssistant, entry: XboxConfigEntry) -> bool:
"""Migrate config entry."""
Migration requires runtime data
"""
if entry.version == 1 and entry.minor_version < 3:
try:
implementation = await async_get_config_entry_implementation(hass, entry)
except ImplementationUnavailableError as e:
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="oauth2_implementation_unavailable",
) from e
session = OAuth2Session(hass, entry, implementation)
async_session = get_async_client(hass)
auth = AsyncConfigEntryAuth(async_session, session)
await auth.refresh_tokens()
client = XboxLiveClient(auth)
if entry.version == 1 and entry.minor_version < 2:
# Migrate unique_id from `xbox` to account xuid and
# change generic entry name to user's gamertag
coordinator = entry.runtime_data.status
xuid = coordinator.client.xuid
gamertag = coordinator.data.presence[xuid].gamertag
if entry.minor_version < 2:
# Migrate unique_id from `xbox` to account xuid and
# change generic entry name to user's gamertag
try:
own = await client.people.get_friends_by_xuid(client.xuid)
except TimeoutException as e:
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="timeout_exception",
) from e
except (RequestError, HTTPStatusError) as e:
_LOGGER.debug("Xbox exception:", exc_info=True)
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="request_exception",
) from e
return hass.config_entries.async_update_entry(
entry,
unique_id=xuid,
title=(gamertag if entry.title == "Home Assistant Cloud" else entry.title),
minor_version=2,
)
hass.config_entries.async_update_entry(
entry,
unique_id=client.xuid,
title=(
own.people[0].gamertag
if entry.title == "Home Assistant Cloud"
else entry.title
),
minor_version=2,
)
if entry.minor_version < 3:
# Migrate favorite friends to friend subentries
try:
friends = await client.people.get_friends_own()
except TimeoutException as e:
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="timeout_exception",
) from e
except (RequestError, HTTPStatusError) as e:
_LOGGER.debug("Xbox exception:", exc_info=True)
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="request_exception",
) from e
dev_reg = dr.async_get(hass)
for friend in friends.people:
if not friend.is_favorite:
continue
subentry = ConfigSubentry(
subentry_type="friend",
title=friend.gamertag,
unique_id=friend.xuid,
data={}, # type: ignore[arg-type]
)
hass.config_entries.async_add_subentry(entry, subentry)
if device := dev_reg.async_get_device({(DOMAIN, friend.xuid)}):
dev_reg.async_update_device(
device.id,
remove_config_entry_id=entry.entry_id,
add_config_subentry_id=subentry.subentry_id,
add_config_entry_id=entry.entry_id,
)
if device := dev_reg.async_get_device({(DOMAIN, "xbox_live")}):
dev_reg.async_update_device(
device.id, new_identifiers={(DOMAIN, client.xuid)}
)
hass.config_entries.async_update_entry(entry, minor_version=3)
hass.config_entries.async_schedule_reload(entry.entry_id)
return True

View File

@@ -5,7 +5,7 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from enum import StrEnum
from typing import Any
from typing import TYPE_CHECKING, Any
from pythonxbox.api.provider.people.models import Person
from pythonxbox.api.provider.titlehub.models import Title
@@ -15,7 +15,7 @@ from homeassistant.components.binary_sensor import (
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import XboxConfigEntry
@@ -112,30 +112,34 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Xbox Live friends."""
xuids_added: set[str] = set()
coordinator = entry.runtime_data.status
@callback
def add_entities() -> None:
nonlocal xuids_added
current_xuids = set(coordinator.data.presence)
if new_xuids := current_xuids - xuids_added:
async_add_entities(
[
XboxBinarySensorEntity(coordinator, xuid, description)
for xuid in new_xuids
for description in SENSOR_DESCRIPTIONS
if check_deprecated_entity(
hass, xuid, description, BINARY_SENSOR_DOMAIN
)
]
if TYPE_CHECKING:
assert entry.unique_id
async_add_entities(
[
XboxBinarySensorEntity(coordinator, entry.unique_id, description)
for description in SENSOR_DESCRIPTIONS
if check_deprecated_entity(
hass, entry.unique_id, description, BINARY_SENSOR_DOMAIN
)
xuids_added |= new_xuids
xuids_added &= current_xuids
]
)
coordinator.async_add_listener(add_entities)
add_entities()
for subentry_id, subentry in entry.subentries.items():
async_add_entities(
[
XboxBinarySensorEntity(coordinator, subentry.unique_id, description)
for description in SENSOR_DESCRIPTIONS
if subentry.unique_id
and check_deprecated_entity(
hass, subentry.unique_id, description, BINARY_SENSOR_DOMAIN
)
and subentry.unique_id in coordinator.data.presence
and subentry.subentry_type == "friend"
],
config_subentry_id=subentry_id,
)
class XboxBinarySensorEntity(XboxBaseEntity, BinarySensorEntity):

View File

@@ -8,11 +8,26 @@ from httpx import AsyncClient
from pythonxbox.api.client import XboxLiveClient
from pythonxbox.authentication.manager import AuthenticationManager
from pythonxbox.authentication.models import OAuth2TokenResponse
import voluptuous as vol
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlowResult
from homeassistant.config_entries import (
SOURCE_REAUTH,
ConfigEntry,
ConfigEntryState,
ConfigFlowResult,
ConfigSubentryFlow,
SubentryFlowResult,
)
from homeassistant.core import callback
from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.helpers.selector import (
SelectOptionDict,
SelectSelector,
SelectSelectorConfig,
)
from .const import DOMAIN
from .const import CONF_XUID, DOMAIN
from .coordinator import XboxConfigEntry
class OAuth2FlowHandler(
@@ -22,7 +37,7 @@ class OAuth2FlowHandler(
DOMAIN = DOMAIN
MINOR_VERSION = 2
MINOR_VERSION = 3
@property
def logger(self) -> logging.Logger:
@@ -35,6 +50,14 @@ class OAuth2FlowHandler(
scopes = ["Xboxlive.signin", "Xboxlive.offline_access"]
return {"scope": " ".join(scopes)}
@classmethod
@callback
def async_get_supported_subentry_types(
cls, config_entry: ConfigEntry
) -> dict[str, type[ConfigSubentryFlow]]:
"""Return subentries supported by this integration."""
return {"friend": FriendSubentryFlowHandler}
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
@@ -65,6 +88,14 @@ class OAuth2FlowHandler(
)
self._abort_if_unique_id_configured()
config_entries = self.hass.config_entries.async_entries(DOMAIN)
for entry in config_entries:
if client.xuid in {
subentry.unique_id for subentry in entry.subentries.values()
}:
return self.async_abort(reason="already_configured_as_subentry")
return self.async_create_entry(title=me.people[0].gamertag, data=data)
async def async_step_reauth(self, _: Mapping[str, Any]) -> ConfigFlowResult:
@@ -78,3 +109,63 @@ class OAuth2FlowHandler(
if user_input is None:
return self.async_show_form(step_id="reauth_confirm")
return await self.async_step_user()
class FriendSubentryFlowHandler(ConfigSubentryFlow):
"""Handle subentry flow for adding a friend."""
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Subentry user flow."""
config_entry: XboxConfigEntry = self._get_entry()
if config_entry.state is not ConfigEntryState.LOADED:
return self.async_abort(reason="config_entry_not_loaded")
client = config_entry.runtime_data.status.client
friends_list = await client.people.get_friends_own()
if user_input is not None:
config_entries = self.hass.config_entries.async_entries(DOMAIN)
if user_input[CONF_XUID] in {entry.unique_id for entry in config_entries}:
return self.async_abort(reason="already_configured_as_entry")
for entry in config_entries:
if user_input[CONF_XUID] in {
subentry.unique_id for subentry in entry.subentries.values()
}:
return self.async_abort(reason="already_configured")
return self.async_create_entry(
title=next(
f.gamertag
for f in friends_list.people
if f.xuid == user_input[CONF_XUID]
),
data={},
unique_id=user_input[CONF_XUID],
)
if not friends_list.people:
return self.async_abort(reason="no_friends")
options = [
SelectOptionDict(
value=friend.xuid,
label=friend.gamertag,
)
for friend in friends_list.people
]
return self.async_show_form(
step_id="user",
data_schema=self.add_suggested_values_to_schema(
vol.Schema(
{
vol.Required(CONF_XUID): SelectSelector(
SelectSelectorConfig(options=options)
)
}
),
user_input,
),
)

View File

@@ -4,3 +4,5 @@ DOMAIN = "xbox"
OAUTH2_AUTHORIZE = "https://login.live.com/oauth20_authorize.srf"
OAUTH2_TOKEN = "https://login.live.com/oauth20_token.srf"
CONF_XUID = "xuid"

View File

@@ -21,7 +21,6 @@ from pythonxbox.api.provider.titlehub.models import Title
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.config_entry_oauth2_flow import (
ImplementationUnavailableError,
OAuth2Session,
@@ -208,14 +207,7 @@ class XboxUpdateCoordinator(DataUpdateCoordinator[XboxData]):
) from e
else:
presence_data = {self.client.xuid: batch.people[0]}
configured_xuids = self.configured_as_entry()
presence_data.update(
{
friend.xuid: friend
for friend in friends.people
if friend.is_favorite and friend.xuid not in configured_xuids
}
)
presence_data.update({friend.xuid: friend for friend in friends.people})
# retrieve title details
for person in presence_data.values():
@@ -260,13 +252,6 @@ class XboxUpdateCoordinator(DataUpdateCoordinator[XboxData]):
else:
self.title_data.pop(person.xuid, None)
person.last_seen_date_time_utc = self.last_seen_timestamp(person)
if (
self.current_friends - (new_friends := set(presence_data))
or not self.current_friends
):
self.remove_stale_devices(new_friends)
self.current_friends = new_friends
return XboxData(new_console_data, presence_data, self.title_data)
def last_seen_timestamp(self, person: Person) -> datetime | None:
@@ -285,25 +270,6 @@ class XboxUpdateCoordinator(DataUpdateCoordinator[XboxData]):
return cur_dt
def remove_stale_devices(self, xuids: set[str]) -> None:
"""Remove stale devices from registry."""
device_reg = dr.async_get(self.hass)
identifiers = (
{(DOMAIN, xuid) for xuid in xuids}
| {(DOMAIN, console.id) for console in self.consoles.result}
| self.configured_as_entry()
)
for device in dr.async_entries_for_config_entry(
device_reg, self.config_entry.entry_id
):
if not set(device.identifiers) & identifiers:
_LOGGER.debug("Removing stale device %s", device.name)
device_reg.async_update_device(
device.id, remove_config_entry_id=self.config_entry.entry_id
)
def configured_as_entry(self) -> set[str]:
"""Get xuids of configured entries."""

View File

@@ -84,7 +84,8 @@ class XboxBaseEntity(CoordinatorEntity[XboxUpdateCoordinator]):
return (
entity_picture
if (fn := self.entity_description.entity_picture_fn) is not None
if self.available
and (fn := self.entity_description.entity_picture_fn) is not None
and (entity_picture := fn(self.data, self.title_info)) is not None
else super().entity_picture
)
@@ -98,6 +99,12 @@ class XboxBaseEntity(CoordinatorEntity[XboxUpdateCoordinator]):
else super().extra_state_attributes
)
@property
def available(self) -> bool:
"""Return True if entity is available."""
return super().available and self.xuid in self.coordinator.data.presence
class XboxConsoleBaseEntity(CoordinatorEntity[XboxUpdateCoordinator]):
"""Console base entity for the Xbox integration."""

View File

@@ -5,12 +5,13 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from enum import StrEnum
from typing import TYPE_CHECKING
from pythonxbox.api.provider.people.models import Person
from pythonxbox.api.provider.titlehub.models import Title
from homeassistant.components.image import ImageEntity, ImageEntityDescription
from homeassistant.core import HomeAssistant, callback
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util import dt as dt_util
@@ -63,30 +64,27 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Xbox images."""
coordinator = config_entry.runtime_data.status
if TYPE_CHECKING:
assert config_entry.unique_id
async_add_entities(
[
XboxImageEntity(hass, coordinator, config_entry.unique_id, description)
for description in IMAGE_DESCRIPTIONS
]
)
xuids_added: set[str] = set()
@callback
def add_entities() -> None:
"""Add image entities."""
nonlocal xuids_added
current_xuids = set(coordinator.data.presence)
if new_xuids := current_xuids - xuids_added:
async_add_entities(
[
XboxImageEntity(hass, coordinator, xuid, description)
for xuid in new_xuids
for description in IMAGE_DESCRIPTIONS
]
)
xuids_added |= new_xuids
xuids_added &= current_xuids
coordinator.async_add_listener(add_entities)
add_entities()
for subentry_id, subentry in config_entry.subentries.items():
async_add_entities(
[
XboxImageEntity(hass, coordinator, subentry.unique_id, description)
for description in IMAGE_DESCRIPTIONS
if subentry.unique_id
and subentry.unique_id in coordinator.data.presence
and subentry.subentry_type == "friend"
],
config_subentry_id=subentry_id,
)
class XboxImageEntity(XboxBaseEntity, ImageEntity):
@@ -113,11 +111,12 @@ class XboxImageEntity(XboxBaseEntity, ImageEntity):
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
url = self.entity_description.image_url_fn(self.data, self.title_info)
if self.available:
url = self.entity_description.image_url_fn(self.data, self.title_info)
if url != self._attr_image_url:
self._attr_image_url = url
self._cached_image = None
self._attr_image_last_updated = dt_util.utcnow()
if url != self._attr_image_url:
self._attr_image_url = url
self._cached_image = None
self._attr_image_last_updated = dt_util.utcnow()
super()._handle_coordinator_update()

View File

@@ -6,7 +6,7 @@ from collections.abc import Callable
from dataclasses import dataclass
from datetime import UTC, datetime
from enum import StrEnum
from typing import Any
from typing import TYPE_CHECKING, Any
from pythonxbox.api.provider.people.models import Person
from pythonxbox.api.provider.smartglass.models import SmartglassConsole, StorageDevice
@@ -21,7 +21,7 @@ from homeassistant.components.sensor import (
SensorStateClass,
)
from homeassistant.const import CONF_NAME, UnitOfInformation
from homeassistant.core import HomeAssistant, callback
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
@@ -253,28 +253,32 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Xbox Live friends."""
xuids_added: set[str] = set()
coordinator = config_entry.runtime_data.status
@callback
def add_entities() -> None:
nonlocal xuids_added
current_xuids = set(coordinator.data.presence)
if new_xuids := current_xuids - xuids_added:
async_add_entities(
[
XboxSensorEntity(coordinator, xuid, description)
for xuid in new_xuids
for description in SENSOR_DESCRIPTIONS
if check_deprecated_entity(hass, xuid, description, SENSOR_DOMAIN)
]
if TYPE_CHECKING:
assert config_entry.unique_id
async_add_entities(
[
XboxSensorEntity(coordinator, config_entry.unique_id, description)
for description in SENSOR_DESCRIPTIONS
if check_deprecated_entity(
hass, config_entry.unique_id, description, SENSOR_DOMAIN
)
xuids_added |= new_xuids
xuids_added &= current_xuids
coordinator.async_add_listener(add_entities)
add_entities()
]
)
for subentry_id, subentry in config_entry.subentries.items():
async_add_entities(
[
XboxSensorEntity(coordinator, subentry.unique_id, description)
for description in SENSOR_DESCRIPTIONS
if subentry.unique_id
and check_deprecated_entity(
hass, subentry.unique_id, description, SENSOR_DOMAIN
)
and subentry.unique_id in coordinator.data.presence
and subentry.subentry_type == "friend"
],
config_subentry_id=subentry_id,
)
consoles_coordinator = config_entry.runtime_data.consoles

View File

@@ -2,6 +2,7 @@
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"already_configured_as_subentry": "This account is already configured as a sub-entry. Please remove the existing sub-entry before adding it.",
"authorize_url_timeout": "[%key:common::config_flow::abort::oauth2_authorize_url_timeout%]",
"missing_configuration": "[%key:common::config_flow::abort::oauth2_missing_configuration%]",
"oauth_error": "[%key:common::config_flow::abort::oauth2_error%]",
@@ -34,6 +35,36 @@
}
}
},
"config_subentries": {
"friend": {
"abort": {
"already_configured": "Already configured as a friend in this or another account.",
"already_configured_as_entry": "Already configured as a service. This account cannot be added as a friend.",
"config_entry_not_loaded": "Cannot add friend accounts when the main account is disabled or not loaded.",
"no_friends": "Looks like your friend list is empty right now. Add friends on Xbox Network first."
},
"entry_type": "Friend",
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"initiate_flow": {
"user": "Add friend"
},
"step": {
"user": {
"data": {
"xuid": "Gamertag"
},
"data_description": {
"xuid": "Select a friend from your friend list to track their online status."
},
"description": "Track the online status of an Xbox Network friend.",
"title": "Friend online status"
}
}
}
},
"entity": {
"binary_sensor": {
"has_game_pass": {

View File

@@ -278,6 +278,7 @@ FLOWS = {
"harmony",
"heos",
"here_travel_time",
"hikvision",
"hisense_aehw4a1",
"hive",
"hko",

View File

@@ -829,6 +829,10 @@ DHCP: Final[list[dict[str, str | bool]]] = [
"hostname": "my[45]50*",
"macaddress": "001E0C*",
},
{
"domain": "sunricher_dali",
"registered_devices": True,
},
{
"domain": "tado",
"hostname": "tado*",

View File

@@ -2714,8 +2714,8 @@
"name": "Hikvision",
"integrations": {
"hikvision": {
"integration_type": "hub",
"config_flow": false,
"integration_type": "device",
"config_flow": true,
"iot_class": "local_push",
"name": "Hikvision"
},
@@ -3136,7 +3136,7 @@
},
"ipma": {
"name": "Instituto Portugu\u00eas do Mar e Atmosfera (IPMA)",
"integration_type": "hub",
"integration_type": "service",
"config_flow": true,
"iot_class": "cloud_polling"
},
@@ -3938,7 +3938,7 @@
},
"meteo_france": {
"name": "M\u00e9t\u00e9o-France",
"integration_type": "hub",
"integration_type": "service",
"config_flow": true,
"iot_class": "cloud_polling"
},

View File

@@ -430,8 +430,8 @@ class EntityTriggerBase(Trigger):
)
class EntityStateTriggerBase(EntityTriggerBase):
"""Trigger for entity state changes."""
class EntityTargetStateTriggerBase(EntityTriggerBase):
"""Trigger for entity state changes to a specific state."""
_to_state: str
@@ -440,8 +440,8 @@ class EntityStateTriggerBase(EntityTriggerBase):
return state.state == self._to_state
class ConditionalEntityStateTriggerBase(EntityTriggerBase):
"""Class for entity state changes where the from state is restricted."""
class EntityTransitionTriggerBase(EntityTriggerBase):
"""Trigger for entity state changes between specific states."""
_from_states: set[str]
_to_states: set[str]
@@ -458,8 +458,24 @@ class ConditionalEntityStateTriggerBase(EntityTriggerBase):
return state.state in self._to_states
class EntityStateAttributeTriggerBase(EntityTriggerBase):
"""Trigger for entity state attribute changes."""
class EntityOriginStateTriggerBase(EntityTriggerBase):
"""Trigger for entity state changes from a specific state."""
_from_state: str
def is_valid_transition(self, from_state: State, to_state: State) -> bool:
"""Check if the origin state matches the expected one and that the state changed."""
return (
from_state.state == self._from_state and to_state.state != self._from_state
)
def is_valid_state(self, state: State) -> bool:
"""Check if the new state is not the same as the expected origin state."""
return state.state != self._from_state
class EntityTargetStateAttributeTriggerBase(EntityTriggerBase):
"""Trigger for entity state attribute changes to a specific state."""
_attribute: str
_attribute_to_state: str
@@ -478,12 +494,12 @@ class EntityStateAttributeTriggerBase(EntityTriggerBase):
return state.attributes.get(self._attribute) == self._attribute_to_state
def make_entity_state_trigger(
def make_entity_target_state_trigger(
domain: str, to_state: str
) -> type[EntityStateTriggerBase]:
"""Create an entity state trigger class."""
) -> type[EntityTargetStateTriggerBase]:
"""Create a trigger for entity state changes to a specific state."""
class CustomTrigger(EntityStateTriggerBase):
class CustomTrigger(EntityTargetStateTriggerBase):
"""Trigger for entity state changes."""
_domain = domain
@@ -492,12 +508,12 @@ def make_entity_state_trigger(
return CustomTrigger
def make_conditional_entity_state_trigger(
def make_entity_transition_trigger(
domain: str, *, from_states: set[str], to_states: set[str]
) -> type[ConditionalEntityStateTriggerBase]:
"""Create a conditional entity state trigger class."""
) -> type[EntityTransitionTriggerBase]:
"""Create a trigger for entity state changes between specific states."""
class CustomTrigger(ConditionalEntityStateTriggerBase):
class CustomTrigger(EntityTransitionTriggerBase):
"""Trigger for conditional entity state changes."""
_domain = domain
@@ -507,12 +523,26 @@ def make_conditional_entity_state_trigger(
return CustomTrigger
def make_entity_state_attribute_trigger(
domain: str, attribute: str, to_state: str
) -> type[EntityStateAttributeTriggerBase]:
"""Create an entity state attribute trigger class."""
def make_entity_origin_state_trigger(
domain: str, *, from_state: str
) -> type[EntityOriginStateTriggerBase]:
"""Create a trigger for entity state changes from a specific state."""
class CustomTrigger(EntityStateAttributeTriggerBase):
class CustomTrigger(EntityOriginStateTriggerBase):
"""Trigger for entity "from state" changes."""
_domain = domain
_from_state = from_state
return CustomTrigger
def make_entity_target_state_attribute_trigger(
domain: str, attribute: str, to_state: str
) -> type[EntityTargetStateAttributeTriggerBase]:
"""Create a trigger for entity state attribute changes to a specific state."""
class CustomTrigger(EntityTargetStateAttributeTriggerBase):
"""Trigger for entity state changes."""
_domain = domain
@@ -1105,6 +1135,5 @@ async def async_get_all_descriptions(
description["target"] = target
new_descriptions_cache[missing_trigger] = description
hass.data[TRIGGER_DESCRIPTION_CACHE] = new_descriptions_cache
return new_descriptions_cache

10
requirements_all.txt generated
View File

@@ -860,7 +860,7 @@ egauge-async==0.4.0
eheimdigital==1.4.0
# homeassistant.components.ekeybionyx
ekey-bionyxpy==1.0.0
ekey-bionyxpy==1.0.1
# homeassistant.components.electric_kiwi
electrickiwi-api==0.9.14
@@ -1482,7 +1482,7 @@ micloud==0.5
microBeesPy==0.3.5
# homeassistant.components.mill
mill-local==0.3.0
mill-local==0.5.0
# homeassistant.components.mill
millheater==0.14.1
@@ -1810,7 +1810,7 @@ py-dactyl==2.0.4
py-dormakaba-dkey==1.0.6
# homeassistant.components.improv_ble
py-improv-ble-client==1.0.3
py-improv-ble-client==2.0.1
# homeassistant.components.madvr
py-madvr2==1.6.40
@@ -2097,7 +2097,7 @@ pyialarm==2.2.0
pyicloud==2.2.0
# homeassistant.components.insteon
pyinsteon==1.6.3
pyinsteon==1.6.4
# homeassistant.components.intesishome
pyintesishome==1.8.0
@@ -2575,7 +2575,7 @@ python-rabbitair==0.0.8
python-ripple-api==0.0.3
# homeassistant.components.roborock
python-roborock==3.12.2
python-roborock==3.18.0
# homeassistant.components.smarttub
python-smarttub==0.0.46

View File

@@ -760,7 +760,7 @@ egauge-async==0.4.0
eheimdigital==1.4.0
# homeassistant.components.ekeybionyx
ekey-bionyxpy==1.0.0
ekey-bionyxpy==1.0.1
# homeassistant.components.electric_kiwi
electrickiwi-api==0.9.14
@@ -1289,7 +1289,7 @@ micloud==0.5
microBeesPy==0.3.5
# homeassistant.components.mill
mill-local==0.3.0
mill-local==0.5.0
# homeassistant.components.mill
millheater==0.14.1
@@ -1550,7 +1550,7 @@ py-dactyl==2.0.4
py-dormakaba-dkey==1.0.6
# homeassistant.components.improv_ble
py-improv-ble-client==1.0.3
py-improv-ble-client==2.0.1
# homeassistant.components.madvr
py-madvr2==1.6.40
@@ -1582,6 +1582,9 @@ pyDuotecno==2024.10.1
# homeassistant.components.electrasmart
pyElectra==1.2.4
# homeassistant.components.hikvision
pyHik==0.3.2
# homeassistant.components.homee
pyHomee==1.3.8
@@ -1771,7 +1774,7 @@ pyialarm==2.2.0
pyicloud==2.2.0
# homeassistant.components.insteon
pyinsteon==1.6.3
pyinsteon==1.6.4
# homeassistant.components.ipma
pyipma==3.0.9
@@ -2156,7 +2159,7 @@ python-pooldose==0.8.1
python-rabbitair==0.0.8
# homeassistant.components.roborock
python-roborock==3.12.2
python-roborock==3.18.0
# homeassistant.components.smarttub
python-smarttub==0.0.46

View File

@@ -905,7 +905,6 @@ INTEGRATIONS_WITHOUT_QUALITY_SCALE_FILE = [
"splunk",
"spotify",
"sql",
"squeezebox",
"srp_energy",
"ssdp",
"starline",
@@ -1177,7 +1176,6 @@ INTEGRATIONS_WITHOUT_SCALE = [
"aten_pe",
"atome",
"august",
"autarco",
"aurora",
"aurora_abb_powerone",
"aussie_broadband",
@@ -1926,7 +1924,6 @@ INTEGRATIONS_WITHOUT_SCALE = [
"splunk",
"spotify",
"sql",
"squeezebox",
"srp_energy",
"ssdp",
"starline",

View File

@@ -168,11 +168,6 @@ FORBIDDEN_PACKAGE_EXCEPTIONS: dict[str, dict[str, set[str]]] = {
# influxdb-client > setuptools
"influxdb-client": {"setuptools"}
},
"insteon": {
# https://github.com/pyinsteon/pyinsteon/issues/430
# pyinsteon > pyserial-asyncio
"pyinsteon": {"pyserial-asyncio"}
},
"izone": {"python-izone": {"async-timeout"}},
"keba": {
# https://github.com/jsbronder/asyncio-dgram/issues/20

View File

@@ -72,7 +72,7 @@ async def _entry(hass: HomeAssistant, filter_schema: dict[str, Any], entry) -> N
assert await async_setup_component(
hass, DOMAIN, {DOMAIN: {CONF_FILTER: filter_schema}}
)
assert entry.state == ConfigEntryState.LOADED
assert entry.state is ConfigEntryState.LOADED
# Clear the component_loaded event from the queue.
async_fire_time_changed(
@@ -87,7 +87,7 @@ async def mock_entry_with_one_event(
hass: HomeAssistant, entry_managed
) -> MockConfigEntry:
"""Use the entry and add a single test event to the queue."""
assert entry_managed.state == ConfigEntryState.LOADED
assert entry_managed.state is ConfigEntryState.LOADED
hass.states.async_set("sensor.test", STATE_ON)
return entry_managed

View File

@@ -106,10 +106,10 @@ async def test_unload_entry(
this verifies that the unload, calls async_stop, which calls async_send and
shuts down the hub.
"""
assert entry_managed.state == ConfigEntryState.LOADED
assert entry_managed.state is ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(entry_managed.entry_id)
mock_managed_streaming.assert_not_called()
assert entry_managed.state == ConfigEntryState.NOT_LOADED
assert entry_managed.state is ConfigEntryState.NOT_LOADED
@pytest.mark.freeze_time("2024-01-01 00:00:00")
@@ -261,4 +261,4 @@ async def test_connection(
entry.add_to_hass(hass)
mock_execute_query.side_effect = sideeffect
await hass.config_entries.async_setup(entry.entry_id)
assert entry.state == ConfigEntryState.SETUP_ERROR
assert entry.state is ConfigEntryState.SETUP_ERROR

View File

@@ -22,12 +22,12 @@ async def test_load_unload_config_entry(
"""Test loading and unloading the integration."""
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state == ConfigEntryState.LOADED
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 == ConfigEntryState.NOT_LOADED # type: ignore[comparison-overlap]
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED # type: ignore[comparison-overlap]
async def test_setup_entry_invalid_auth(

View File

@@ -22,13 +22,13 @@ async def test_setup_entry(
) -> None:
"""Test async_setup_entry."""
assert mock_config_entry.state == ConfigEntryState.NOT_LOADED
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
# Load entry
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state == ConfigEntryState.LOADED
assert mock_config_entry.state is ConfigEntryState.LOADED
# Check that the device has been registered properly
device = device_registry.async_get_device(
@@ -57,13 +57,13 @@ async def test_setup_entry_failed(
"", (ServerTimeoutError(), TimeoutError())
)
assert mock_config_entry.state == ConfigEntryState.NOT_LOADED
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
# Load entry
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state == ConfigEntryState.SETUP_RETRY
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
# Ensure that the connection has been checked, API client correctly closed
# and WebSocket connection has not been initialized
@@ -80,12 +80,12 @@ async def test_unload_entry(
"""Test unload_entry."""
# Load entry
assert mock_config_entry.state == ConfigEntryState.NOT_LOADED
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state == ConfigEntryState.LOADED
assert mock_config_entry.state is ConfigEntryState.LOADED
assert hasattr(mock_config_entry, "runtime_data")
# Unload entry
@@ -97,4 +97,4 @@ async def test_unload_entry(
# Ensure that the entry is not loaded and has been removed from hass
assert not hasattr(mock_config_entry, "runtime_data")
assert mock_config_entry.state == ConfigEntryState.NOT_LOADED
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED

View File

@@ -148,6 +148,22 @@ def mock_stream_source_fixture() -> Generator[AsyncMock]:
yield mock_stream_source
@pytest.fixture(name="mock_create_stream")
def mock_create_stream_fixture() -> Generator[Mock]:
"""Fixture to mock create_stream and prevent real stream threads."""
mock_stream = Mock()
mock_stream.add_provider = Mock()
mock_stream.start = AsyncMock()
mock_stream.endpoint_url = Mock(return_value="http://home.assistant/playlist.m3u8")
mock_stream.set_update_callback = Mock()
mock_stream.available = True
with patch(
"homeassistant.components.camera.create_stream",
return_value=mock_stream,
):
yield mock_stream
@pytest.fixture
async def mock_test_webrtc_cameras(hass: HomeAssistant) -> None:
"""Initialize test WebRTC cameras with native RTC support."""

View File

@@ -346,20 +346,14 @@ async def test_websocket_stream_no_source(
@pytest.mark.usefixtures("mock_camera", "mock_stream")
async def test_websocket_camera_stream(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
hass: HomeAssistant, hass_ws_client: WebSocketGenerator, mock_create_stream: Mock
) -> None:
"""Test camera/stream websocket command."""
await async_setup_component(hass, "camera", {})
with (
patch(
"homeassistant.components.camera.Stream.endpoint_url",
return_value="http://home.assistant/playlist.m3u8",
) as mock_stream_view_url,
patch(
"homeassistant.components.demo.camera.DemoCamera.stream_source",
return_value="http://example.com",
),
with patch(
"homeassistant.components.demo.camera.DemoCamera.stream_source",
return_value="http://example.com",
):
# Request playlist through WebSocket
client = await hass_ws_client(hass)
@@ -369,7 +363,7 @@ async def test_websocket_camera_stream(
msg = await client.receive_json()
# Assert WebSocket response
assert mock_stream_view_url.called
assert mock_create_stream.endpoint_url.called
assert msg["id"] == 6
assert msg["type"] == TYPE_RESULT
assert msg["success"]
@@ -505,21 +499,18 @@ async def test_play_stream_service_no_source(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("mock_camera", "mock_stream")
async def test_handle_play_stream_service(hass: HomeAssistant) -> None:
async def test_handle_play_stream_service(
hass: HomeAssistant, mock_create_stream: Mock
) -> None:
"""Test camera play_stream service."""
await async_process_ha_core_config(
hass,
{"external_url": "https://example.com"},
)
await async_setup_component(hass, "media_player", {})
with (
patch(
"homeassistant.components.camera.Stream.endpoint_url",
) as mock_request_stream,
patch(
"homeassistant.components.demo.camera.DemoCamera.stream_source",
return_value="http://example.com",
),
with patch(
"homeassistant.components.demo.camera.DemoCamera.stream_source",
return_value="http://example.com",
):
# Call service
await hass.services.async_call(
@@ -533,17 +524,14 @@ async def test_handle_play_stream_service(hass: HomeAssistant) -> None:
)
# So long as we request the stream, the rest should be covered
# by the play_media service tests.
assert mock_request_stream.called
assert mock_create_stream.endpoint_url.called
@pytest.mark.usefixtures("mock_stream")
async def test_no_preload_stream(hass: HomeAssistant) -> None:
async def test_no_preload_stream(hass: HomeAssistant, mock_create_stream: Mock) -> None:
"""Test camera preload preference."""
demo_settings = camera.DynamicStreamSettings()
with (
patch(
"homeassistant.components.camera.Stream.endpoint_url",
) as mock_request_stream,
patch(
"homeassistant.components.camera.prefs.CameraPreferences.get_dynamic_stream_settings",
return_value=demo_settings,
@@ -557,15 +545,14 @@ async def test_no_preload_stream(hass: HomeAssistant) -> None:
await async_setup_component(hass, "camera", {DOMAIN: {"platform": "demo"}})
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done()
assert not mock_request_stream.called
assert not mock_create_stream.endpoint_url.called
@pytest.mark.usefixtures("mock_stream")
async def test_preload_stream(hass: HomeAssistant) -> None:
async def test_preload_stream(hass: HomeAssistant, mock_create_stream: Mock) -> None:
"""Test camera preload preference."""
demo_settings = camera.DynamicStreamSettings(preload_stream=True)
with (
patch("homeassistant.components.camera.create_stream") as mock_create_stream,
patch(
"homeassistant.components.camera.prefs.CameraPreferences.get_dynamic_stream_settings",
return_value=demo_settings,
@@ -575,14 +562,13 @@ async def test_preload_stream(hass: HomeAssistant) -> None:
return_value="http://example.com",
),
):
mock_create_stream.return_value.start = AsyncMock()
assert await async_setup_component(
hass, "camera", {DOMAIN: {"platform": "demo"}}
)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done()
assert mock_create_stream.called
assert mock_create_stream.start.called
@pytest.mark.usefixtures("mock_camera")
@@ -694,25 +680,16 @@ async def test_state_streaming(hass: HomeAssistant) -> None:
assert demo_camera.state == camera.CameraState.STREAMING
@pytest.mark.usefixtures("mock_camera", "mock_stream")
@pytest.mark.usefixtures("mock_camera", "mock_stream", "mock_create_stream")
async def test_stream_unavailable(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
hass: HomeAssistant, hass_ws_client: WebSocketGenerator, mock_create_stream: Mock
) -> None:
"""Camera state."""
await async_setup_component(hass, "camera", {})
with (
patch(
"homeassistant.components.camera.Stream.endpoint_url",
return_value="http://home.assistant/playlist.m3u8",
),
patch(
"homeassistant.components.demo.camera.DemoCamera.stream_source",
return_value="http://example.com",
),
patch(
"homeassistant.components.camera.Stream.set_update_callback",
) as mock_update_callback,
with patch(
"homeassistant.components.demo.camera.DemoCamera.stream_source",
return_value="http://example.com",
):
# Request playlist through WebSocket. We just want to create the stream
# but don't care about the result.
@@ -721,26 +698,22 @@ async def test_stream_unavailable(
{"id": 10, "type": "camera/stream", "entity_id": "camera.demo_camera"}
)
await client.receive_json()
assert mock_update_callback.called
assert mock_create_stream.set_update_callback.called
# Simulate the stream going unavailable
callback = mock_update_callback.call_args.args[0]
with patch(
"homeassistant.components.camera.Stream.available", new_callable=lambda: False
):
callback()
await hass.async_block_till_done()
callback = mock_create_stream.set_update_callback.call_args.args[0]
mock_create_stream.available = False
callback()
await hass.async_block_till_done()
demo_camera = hass.states.get("camera.demo_camera")
assert demo_camera is not None
assert demo_camera.state == STATE_UNAVAILABLE
# Simulate stream becomes available
with patch(
"homeassistant.components.camera.Stream.available", new_callable=lambda: True
):
callback()
await hass.async_block_till_done()
mock_create_stream.available = True
callback()
await hass.async_block_till_done()
demo_camera = hass.states.get("camera.demo_camera")
assert demo_camera is not None

View File

@@ -0,0 +1,231 @@
"""Test device_tracker trigger."""
from collections.abc import Generator
from unittest.mock import patch
import pytest
from homeassistant.const import (
ATTR_LABEL_ID,
CONF_ENTITY_ID,
STATE_HOME,
STATE_NOT_HOME,
)
from homeassistant.core import HomeAssistant, ServiceCall
from tests.components import (
StateDescription,
arm_trigger,
parametrize_target_entities,
parametrize_trigger_states,
set_or_remove_state,
target_entities,
)
STATE_WORK_ZONE = "work"
@pytest.fixture(autouse=True, name="stub_blueprint_populate")
def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder."""
@pytest.fixture(name="enable_experimental_triggers_conditions")
def enable_experimental_triggers_conditions() -> Generator[None]:
"""Enable experimental triggers and conditions."""
with patch(
"homeassistant.components.labs.async_is_preview_feature_enabled",
return_value=True,
):
yield
@pytest.fixture
async def target_device_trackers(hass: HomeAssistant) -> list[str]:
"""Create multiple device_trackers entities associated with different targets."""
return (await target_entities(hass, "device_tracker"))["included"]
@pytest.mark.parametrize(
"trigger_key",
["device_tracker.entered_home", "device_tracker.left_home"],
)
async def test_device_tracker_triggers_gated_by_labs_flag(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, trigger_key: str
) -> None:
"""Test the device_tracker triggers are gated by the labs flag."""
await arm_trigger(hass, trigger_key, None, {ATTR_LABEL_ID: "test_label"})
assert (
"Unnamed automation failed to setup triggers and has been disabled: Trigger "
f"'{trigger_key}' requires the experimental 'New triggers and conditions' "
"feature to be enabled in Home Assistant Labs settings (feature flag: "
"'new_triggers_conditions')"
) in caplog.text
@pytest.mark.usefixtures("enable_experimental_triggers_conditions")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("device_tracker"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="device_tracker.entered_home",
target_states=[STATE_HOME],
other_states=[STATE_NOT_HOME, STATE_WORK_ZONE],
),
*parametrize_trigger_states(
trigger="device_tracker.left_home",
target_states=[STATE_NOT_HOME, STATE_WORK_ZONE],
other_states=[STATE_HOME],
),
],
)
async def test_device_tracker_home_trigger_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_device_trackers: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the device_tracker home triggers when any device_tracker changes to a specific state."""
other_entity_ids = set(target_device_trackers) - {entity_id}
# Set all device_trackers, including the tested device_tracker, to the initial state
for eid in target_device_trackers:
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Check that changing other device_trackers also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * state["count"]
service_calls.clear()
@pytest.mark.usefixtures("enable_experimental_triggers_conditions")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("device_tracker"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="device_tracker.entered_home",
target_states=[STATE_HOME],
other_states=[STATE_NOT_HOME, STATE_WORK_ZONE],
),
*parametrize_trigger_states(
trigger="device_tracker.left_home",
target_states=[STATE_NOT_HOME, STATE_WORK_ZONE],
other_states=[STATE_HOME],
),
],
)
async def test_device_tracker_state_trigger_behavior_first(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_device_trackers: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the device_tracker home triggers when the first device_tracker changes to a specific state."""
other_entity_ids = set(target_device_trackers) - {entity_id}
# Set all device_trackers, including the tested device_tracker, to the initial state
for eid in target_device_trackers:
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Triggering other device_trackers should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
@pytest.mark.usefixtures("enable_experimental_triggers_conditions")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("device_tracker"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="device_tracker.entered_home",
target_states=[STATE_HOME],
other_states=[STATE_NOT_HOME, STATE_WORK_ZONE],
),
*parametrize_trigger_states(
trigger="device_tracker.left_home",
target_states=[STATE_NOT_HOME, STATE_WORK_ZONE],
other_states=[STATE_HOME],
),
],
)
async def test_device_tracker_state_trigger_behavior_last(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_device_trackers: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the device_tracker home triggers when the last device_tracker changes to a specific state."""
other_entity_ids = set(target_device_trackers) - {entity_id}
# Set all device_trackers, including the tested device_tracker, to the initial state
for eid in target_device_trackers:
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
included_state = state["included"]
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()

View File

@@ -18,10 +18,10 @@ async def test_setup(
"""Test entry setup without any exceptions."""
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
assert mock_entry.state == ConfigEntryState.LOADED
assert mock_entry.state is ConfigEntryState.LOADED
# Unload
await hass.config_entries.async_unload(mock_entry.entry_id)
assert mock_entry.state == ConfigEntryState.NOT_LOADED
assert mock_entry.state is ConfigEntryState.NOT_LOADED
async def test_setup_connect_error(
@@ -33,4 +33,4 @@ async def test_setup_connect_error(
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
# Ensure is not ready
assert mock_entry.state == ConfigEntryState.SETUP_RETRY
assert mock_entry.state is ConfigEntryState.SETUP_RETRY

View File

@@ -3,10 +3,17 @@
from typing import Any
from unittest.mock import AsyncMock
import jwt
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
TEST_CREDENTIALS = {CONF_EMAIL: "test@test.com", CONF_PASSWORD: "SomePassword"}
TEST_ACCESS_JWT = jwt.encode({"sub": "some-uuid"}, key="secret")
async def setup_integration(hass: HomeAssistant, entry: MockConfigEntry) -> None:
"""Set up the homelink integration for testing."""

View File

@@ -9,6 +9,8 @@ import pytest
from homeassistant.components.gentex_homelink import DOMAIN
from . import TEST_ACCESS_JWT
from tests.common import MockConfigEntry
@@ -21,7 +23,7 @@ def mock_srp_auth() -> Generator[AsyncMock]:
instance = mock_srp_auth.return_value
instance.async_get_access_token.return_value = {
"AuthenticationResult": {
"AccessToken": "access",
"AccessToken": TEST_ACCESS_JWT,
"RefreshToken": "refresh",
"TokenType": "bearer",
"ExpiresIn": 3600,
@@ -60,6 +62,8 @@ def mock_device() -> AsyncMock:
def mock_config_entry() -> MockConfigEntry:
"""Mock setup entry."""
return MockConfigEntry(
unique_id="some-uuid",
version=1,
domain=DOMAIN,
data={
"auth_implementation": "gentex_homelink",

View File

@@ -7,10 +7,13 @@ import pytest
from homeassistant.components.gentex_homelink.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from . import TEST_ACCESS_JWT, TEST_CREDENTIALS, setup_integration
from tests.common import MockConfigEntry
async def test_full_flow(
hass: HomeAssistant, mock_srp_auth: AsyncMock, mock_setup_entry: AsyncMock
@@ -26,13 +29,13 @@ async def test_full_flow(
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_EMAIL: "test@test.com", CONF_PASSWORD: "SomePassword"},
user_input=TEST_CREDENTIALS,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {
"auth_implementation": "gentex_homelink",
"token": {
"access_token": "access",
"access_token": TEST_ACCESS_JWT,
"refresh_token": "refresh",
"expires_in": 3600,
"token_type": "bearer",
@@ -40,6 +43,31 @@ async def test_full_flow(
},
}
assert result["title"] == "SRPAuth"
assert result["result"].unique_id == "some-uuid"
async def test_unique_configurations(
hass: HomeAssistant,
mock_srp_auth: AsyncMock,
mock_setup_entry: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Check full flow."""
await setup_integration(hass, mock_config_entry)
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 not result["errors"]
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=TEST_CREDENTIALS,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
@@ -69,7 +97,7 @@ async def test_exceptions(
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_EMAIL: "test@test.com", CONF_PASSWORD: "SomePassword"},
user_input=TEST_CREDENTIALS,
)
assert result["type"] is FlowResultType.FORM
@@ -79,6 +107,6 @@ async def test_exceptions(
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_EMAIL: "test@test.com", CONF_PASSWORD: "SomePassword"},
user_input=TEST_CREDENTIALS,
)
assert result["type"] is FlowResultType.CREATE_ENTRY

View File

@@ -96,7 +96,7 @@ async def _test_setup_and_signaling(
config_entries = hass.config_entries.async_entries(DOMAIN)
assert len(config_entries) == 1
config_entry = config_entries[0]
assert config_entry.state == ConfigEntryState.LOADED
assert config_entry.state is ConfigEntryState.LOADED
after_setup_fn()
receive_message_callback = Mock(spec_set=WebRTCSendMessage)
@@ -183,7 +183,7 @@ async def _test_setup_and_signaling(
await hass.config_entries.async_unload(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state == ConfigEntryState.NOT_LOADED
assert config_entry.state is ConfigEntryState.NOT_LOADED
assert teardown.call_count == 2
@@ -625,7 +625,7 @@ async def test_setup_with_setup_entry_error(
await hass.async_block_till_done(wait_background_tasks=True)
config_entries = hass.config_entries.async_entries(DOMAIN)
assert len(config_entries) == 1
assert config_entries[0].state == ConfigEntryState.SETUP_ERROR
assert config_entries[0].state is ConfigEntryState.SETUP_ERROR
assert expected_log_message in caplog.text

View File

@@ -0,0 +1,14 @@
"""Common test tools for the Hikvision integration."""
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Set up the Hikvision integration for testing."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()

View File

@@ -0,0 +1,94 @@
"""Common fixtures for the Hikvision tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from homeassistant.components.hikvision.const import DOMAIN
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
CONF_SSL,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from . import setup_integration
from tests.common import MockConfigEntry
TEST_HOST = "192.168.1.100"
TEST_PORT = 80
TEST_USERNAME = "admin"
TEST_PASSWORD = "password123"
TEST_DEVICE_ID = "DS-2CD2142FWD-I20170101AAAA"
TEST_DEVICE_NAME = "Front Camera"
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.hikvision.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
title=TEST_DEVICE_NAME,
domain=DOMAIN,
version=1,
minor_version=1,
data={
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_SSL: False,
},
unique_id=TEST_DEVICE_ID,
)
@pytest.fixture
def mock_hikcamera() -> Generator[MagicMock]:
"""Return a mocked HikCamera."""
with (
patch(
"homeassistant.components.hikvision.HikCamera",
autospec=True,
) as hikcamera_mock,
patch(
"homeassistant.components.hikvision.config_flow.HikCamera",
new=hikcamera_mock,
),
):
camera = hikcamera_mock.return_value
camera.get_id.return_value = TEST_DEVICE_ID
camera.get_name = TEST_DEVICE_NAME
camera.get_type = "Camera"
camera.current_event_states = {
"Motion": [(True, 1)],
"Line Crossing": [(False, 1)],
}
camera.fetch_attributes.return_value = (
False,
None,
None,
"2024-01-01T00:00:00Z",
)
yield hikcamera_mock
@pytest.fixture
async def init_integration(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_hikcamera: MagicMock
) -> MockConfigEntry:
"""Set up the Hikvision integration for testing."""
await setup_integration(hass, mock_config_entry)
return mock_config_entry

View File

@@ -0,0 +1,101 @@
# serializer version: 1
# name: test_all_entities[binary_sensor.front_camera_line_crossing-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.front_camera_line_crossing',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.MOTION: 'motion'>,
'original_icon': None,
'original_name': 'Line Crossing',
'platform': 'hikvision',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'DS-2CD2142FWD-I20170101AAAA_Line Crossing_1',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[binary_sensor.front_camera_line_crossing-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'motion',
'friendly_name': 'Front Camera Line Crossing',
'last_tripped_time': '2024-01-01T00:00:00Z',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.front_camera_line_crossing',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_entities[binary_sensor.front_camera_motion-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.front_camera_motion',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.MOTION: 'motion'>,
'original_icon': None,
'original_name': 'Motion',
'platform': 'hikvision',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'DS-2CD2142FWD-I20170101AAAA_Motion_1',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[binary_sensor.front_camera_motion-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'motion',
'friendly_name': 'Front Camera Motion',
'last_tripped_time': '2024-01-01T00:00:00Z',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.front_camera_motion',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---

View File

@@ -0,0 +1,300 @@
"""Test Hikvision binary sensors."""
from unittest.mock import MagicMock
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from homeassistant.components.hikvision.const import DOMAIN
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_LAST_TRIP_TIME,
CONF_HOST,
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_SSL,
CONF_USERNAME,
STATE_OFF,
)
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
from homeassistant.helpers import (
device_registry as dr,
entity_registry as er,
issue_registry as ir,
)
from homeassistant.setup import async_setup_component
from . import setup_integration
from .conftest import (
TEST_DEVICE_ID,
TEST_DEVICE_NAME,
TEST_HOST,
TEST_PASSWORD,
TEST_PORT,
TEST_USERNAME,
)
from tests.common import MockConfigEntry, snapshot_platform
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_all_entities(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hikcamera: MagicMock,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test all binary sensor entities."""
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_binary_sensors_created(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hikcamera: MagicMock,
) -> None:
"""Test binary sensors are created for each event type."""
await setup_integration(hass, mock_config_entry)
# Check Motion sensor (camera type doesn't include channel in name)
state = hass.states.get("binary_sensor.front_camera_motion")
assert state is not None
assert state.state == STATE_OFF
assert state.attributes.get(ATTR_DEVICE_CLASS) == BinarySensorDeviceClass.MOTION
assert ATTR_LAST_TRIP_TIME in state.attributes
# Check Line Crossing sensor
state = hass.states.get("binary_sensor.front_camera_line_crossing")
assert state is not None
assert state.state == STATE_OFF
assert state.attributes.get(ATTR_DEVICE_CLASS) == BinarySensorDeviceClass.MOTION
async def test_binary_sensor_device_info(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hikcamera: MagicMock,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test binary sensors are linked to device."""
await setup_integration(hass, mock_config_entry)
device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, TEST_DEVICE_ID)}
)
assert device_entry is not None
assert device_entry.name == TEST_DEVICE_NAME
assert device_entry.manufacturer == "Hikvision"
assert device_entry.model == "Camera"
async def test_binary_sensor_callback_registered(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hikcamera: MagicMock,
) -> None:
"""Test that callback is registered with pyhik."""
await setup_integration(hass, mock_config_entry)
# Verify callback was registered for each sensor
assert mock_hikcamera.return_value.add_update_callback.call_count == 2
async def test_binary_sensor_no_sensors(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hikcamera: MagicMock,
) -> None:
"""Test setup when device has no sensors."""
mock_hikcamera.return_value.current_event_states = None
await setup_integration(hass, mock_config_entry)
# No binary sensors should be created
states = hass.states.async_entity_ids("binary_sensor")
assert len(states) == 0
async def test_binary_sensor_nvr_device(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hikcamera: MagicMock,
) -> None:
"""Test binary sensor naming for NVR devices."""
mock_hikcamera.return_value.get_type = "NVR"
mock_hikcamera.return_value.current_event_states = {
"Motion": [(True, 1), (False, 2)],
}
await setup_integration(hass, mock_config_entry)
# NVR sensors should include channel number in name
state = hass.states.get("binary_sensor.front_camera_motion_1")
assert state is not None
state = hass.states.get("binary_sensor.front_camera_motion_2")
assert state is not None
async def test_binary_sensor_state_on(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hikcamera: MagicMock,
) -> None:
"""Test binary sensor state when on."""
mock_hikcamera.return_value.fetch_attributes.return_value = (
True,
None,
None,
"2024-01-01T12:00:00Z",
)
await setup_integration(hass, mock_config_entry)
state = hass.states.get("binary_sensor.front_camera_motion")
assert state is not None
assert state.state == "on"
async def test_binary_sensor_device_class_unknown(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hikcamera: MagicMock,
) -> None:
"""Test binary sensor with unknown device class."""
mock_hikcamera.return_value.current_event_states = {
"Unknown Event": [(False, 1)],
}
await setup_integration(hass, mock_config_entry)
state = hass.states.get("binary_sensor.front_camera_unknown_event")
assert state is not None
assert state.attributes.get(ATTR_DEVICE_CLASS) is None
async def test_yaml_import_creates_deprecation_issue(
hass: HomeAssistant,
mock_hikcamera: MagicMock,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test YAML import creates deprecation issue."""
await async_setup_component(
hass,
"binary_sensor",
{
"binary_sensor": {
"platform": DOMAIN,
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_SSL: False,
}
},
)
await hass.async_block_till_done()
# Check that deprecation issue was created in homeassistant domain
issue = issue_registry.async_get_issue(
HOMEASSISTANT_DOMAIN, f"deprecated_yaml_{DOMAIN}"
)
assert issue is not None
assert issue.severity == ir.IssueSeverity.WARNING
async def test_yaml_import_with_name(
hass: HomeAssistant,
mock_hikcamera: MagicMock,
) -> None:
"""Test YAML import uses custom name for config entry."""
await async_setup_component(
hass,
"binary_sensor",
{
"binary_sensor": {
"platform": DOMAIN,
CONF_NAME: "Custom Camera Name",
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_SSL: False,
}
},
)
await hass.async_block_till_done()
# Check that the config entry was created with the custom name
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
assert entries[0].title == "Custom Camera Name"
async def test_yaml_import_abort_creates_issue(
hass: HomeAssistant,
mock_hikcamera: MagicMock,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test YAML import creates issue when import is aborted."""
mock_hikcamera.return_value.get_id.return_value = None
await async_setup_component(
hass,
"binary_sensor",
{
"binary_sensor": {
"platform": DOMAIN,
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_SSL: False,
}
},
)
await hass.async_block_till_done()
# Check that import failure issue was created
issue = issue_registry.async_get_issue(
DOMAIN, "deprecated_yaml_import_issue_cannot_connect"
)
assert issue is not None
assert issue.severity == ir.IssueSeverity.WARNING
async def test_binary_sensor_update_callback(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hikcamera: MagicMock,
) -> None:
"""Test binary sensor state updates via callback."""
await setup_integration(hass, mock_config_entry)
state = hass.states.get("binary_sensor.front_camera_motion")
assert state is not None
assert state.state == STATE_OFF
# Simulate state change via callback
mock_hikcamera.return_value.fetch_attributes.return_value = (
True,
None,
None,
"2024-01-01T12:00:00Z",
)
# Get the registered callback and call it
add_callback_call = mock_hikcamera.return_value.add_update_callback.call_args_list[
0
]
callback_func = add_callback_call[0][0]
callback_func("motion detected")
# Verify state was updated
state = hass.states.get("binary_sensor.front_camera_motion")
assert state is not None
assert state.state == "on"

View File

@@ -0,0 +1,312 @@
"""Test the Hikvision config flow."""
from unittest.mock import AsyncMock, MagicMock
import requests
from homeassistant.components.hikvision.const import DOMAIN
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
CONF_SSL,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .conftest import (
TEST_DEVICE_ID,
TEST_DEVICE_NAME,
TEST_HOST,
TEST_PASSWORD,
TEST_PORT,
TEST_USERNAME,
)
from tests.common import MockConfigEntry
async def test_form(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_hikcamera: MagicMock,
) -> None:
"""Test we get the form and can create entry."""
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"],
{
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_SSL: False,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == TEST_DEVICE_NAME
assert result["result"].unique_id == TEST_DEVICE_ID
assert result["data"] == {
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_SSL: False,
}
async def test_form_cannot_connect(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_hikcamera: MagicMock,
) -> None:
"""Test we handle cannot connect error and can recover."""
mock_hikcamera.return_value.get_id.return_value = None
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_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_SSL: False,
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "cannot_connect"}
# Recover from error
mock_hikcamera.return_value.get_id.return_value = TEST_DEVICE_ID
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_SSL: False,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["result"].unique_id == TEST_DEVICE_ID
async def test_form_exception(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_hikcamera: MagicMock,
) -> None:
"""Test we handle exception during connection and can recover."""
mock_hikcamera.side_effect = requests.exceptions.RequestException(
"Connection failed"
)
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_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_SSL: False,
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "cannot_connect"}
# Recover from error
mock_hikcamera.side_effect = None
mock_hikcamera.return_value.get_id.return_value = TEST_DEVICE_ID
mock_hikcamera.return_value.get_name = TEST_DEVICE_NAME
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_SSL: False,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["result"].unique_id == TEST_DEVICE_ID
async def test_form_already_configured(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_hikcamera: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test we handle already configured devices."""
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_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_SSL: False,
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_import_flow(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_hikcamera: MagicMock,
) -> None:
"""Test YAML import flow creates config entry."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_SSL: False,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == TEST_DEVICE_NAME
assert result["result"].unique_id == TEST_DEVICE_ID
assert result["data"] == {
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_SSL: False,
}
async def test_import_flow_with_defaults(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_hikcamera: MagicMock,
) -> None:
"""Test YAML import flow uses default values."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_HOST: TEST_HOST,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["result"].unique_id == TEST_DEVICE_ID
# Default port (80) and SSL (False) should be used
assert result["data"][CONF_PORT] == 80
assert result["data"][CONF_SSL] is False
async def test_import_flow_cannot_connect(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_hikcamera: MagicMock,
) -> None:
"""Test YAML import flow aborts on connection error."""
mock_hikcamera.side_effect = requests.exceptions.RequestException(
"Connection failed"
)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_SSL: False,
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
async def test_import_flow_no_device_id(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_hikcamera: MagicMock,
) -> None:
"""Test YAML import flow aborts when device_id is None."""
mock_hikcamera.return_value.get_id.return_value = None
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_SSL: False,
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
async def test_import_flow_already_configured(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_hikcamera: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test YAML import flow aborts when device is already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_SSL: False,
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"

View File

@@ -0,0 +1,62 @@
"""Test Hikvision integration setup and unload."""
from unittest.mock import MagicMock
import requests
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from . import setup_integration
from tests.common import MockConfigEntry
async def test_setup_and_unload_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hikcamera: MagicMock,
) -> None:
"""Test successful setup and unload of config entry."""
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
mock_hikcamera.return_value.start_stream.assert_called_once()
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
mock_hikcamera.return_value.disconnect.assert_called_once()
async def test_setup_entry_connection_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hikcamera: MagicMock,
) -> None:
"""Test setup fails on connection error."""
mock_hikcamera.side_effect = requests.exceptions.RequestException(
"Connection failed"
)
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_setup_entry_no_device_id(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hikcamera: MagicMock,
) -> None:
"""Test setup fails when device_id is None."""
mock_hikcamera.return_value.get_id.return_value = None
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

Some files were not shown because too many files have changed in this diff Show More