Compare commits

..

5 Commits

Author SHA1 Message Date
Erik
7d4b004806 Refactor tests 2025-12-04 09:21:22 +01:00
Erik
9ea532d0ec Update tests 2025-12-04 08:47:40 +01:00
Erik
f0840af408 Update strings 2025-12-04 08:31:47 +01:00
Erik Montnemery
355435754a Merge branch 'dev' into add_trigger_media_player_muted 2025-12-04 08:05:22 +01:00
Erik
9ff93fc1ee Add trigger media_player.muted 2025-11-27 07:58:11 +01:00
104 changed files with 763 additions and 3036 deletions

View File

@@ -27,6 +27,7 @@
"charliermarsh.ruff",
"ms-python.pylint",
"ms-python.vscode-pylance",
"visualstudioexptteam.vscodeintellicode",
"redhat.vscode-yaml",
"esbenp.prettier-vscode",
"GitHub.vscode-pull-request-github",

4
CODEOWNERS generated
View File

@@ -571,8 +571,6 @@ build.json @home-assistant/supervisor
/tests/components/generic_hygrostat/ @Shulyaka
/homeassistant/components/geniushub/ @manzanotti
/tests/components/geniushub/ @manzanotti
/homeassistant/components/gentex_homelink/ @niaexa @ryanjones-gentex
/tests/components/gentex_homelink/ @niaexa @ryanjones-gentex
/homeassistant/components/geo_json_events/ @exxamalte
/tests/components/geo_json_events/ @exxamalte
/homeassistant/components/geo_location/ @home-assistant/core
@@ -1805,8 +1803,6 @@ build.json @home-assistant/supervisor
/tests/components/weatherflow_cloud/ @jeeftor
/homeassistant/components/weatherkit/ @tjhorner
/tests/components/weatherkit/ @tjhorner
/homeassistant/components/web_rtc/ @home-assistant/core
/tests/components/web_rtc/ @home-assistant/core
/homeassistant/components/webdav/ @jpbede
/tests/components/webdav/ @jpbede
/homeassistant/components/webhook/ @home-assistant/core

View File

@@ -30,7 +30,6 @@ STEP_USER_DATA_SCHEMA = vol.Schema(
vol.Required(CONF_PASSWORD): selector.TextSelector(
selector.TextSelectorConfig(type=selector.TextSelectorType.PASSWORD)
),
vol.Required(CONF_ACCOUNT_NUMBER): selector.TextSelector(),
}
)
@@ -69,19 +68,34 @@ class AnglianWaterConfigFlow(ConfigFlow, domain=DOMAIN):
self.hass,
cookie_jar=CookieJar(quote_cookie=False),
),
account_number=user_input[CONF_ACCOUNT_NUMBER],
account_number=user_input.get(CONF_ACCOUNT_NUMBER),
)
)
if isinstance(validation_response, BaseAuth):
await self.async_set_unique_id(user_input[CONF_ACCOUNT_NUMBER])
account_number = (
user_input.get(CONF_ACCOUNT_NUMBER)
or validation_response.account_number
)
await self.async_set_unique_id(account_number)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=user_input[CONF_ACCOUNT_NUMBER],
title=account_number,
data={
**user_input,
CONF_ACCESS_TOKEN: validation_response.refresh_token,
CONF_ACCOUNT_NUMBER: account_number,
},
)
if validation_response == "smart_meter_unavailable":
return self.async_show_form(
step_id="user",
data_schema=STEP_USER_DATA_SCHEMA.extend(
{
vol.Required(CONF_ACCOUNT_NUMBER): selector.TextSelector(),
}
),
errors={"base": validation_response},
)
errors["base"] = validation_response
return self.async_show_form(

View File

@@ -5,7 +5,6 @@
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/anglian_water",
"iot_class": "cloud_polling",
"loggers": ["pyanglianwater"],
"quality_scale": "bronze",
"requirements": ["pyanglianwater==2.1.0"]
}

View File

@@ -20,7 +20,7 @@ from aiohttp import hdrs, web
import attr
from propcache.api import cached_property, under_cached_property
import voluptuous as vol
from webrtc_models import RTCIceCandidateInit
from webrtc_models import RTCIceCandidateInit, RTCIceServer
from homeassistant.components import websocket_api
from homeassistant.components.http import KEY_AUTHENTICATED, HomeAssistantView
@@ -37,7 +37,6 @@ from homeassistant.components.stream import (
Stream,
create_stream,
)
from homeassistant.components.web_rtc import async_get_ice_servers
from homeassistant.components.websocket_api import ActiveConnection
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
@@ -85,6 +84,7 @@ from .prefs import (
get_dynamic_camera_stream_settings,
)
from .webrtc import (
DATA_ICE_SERVERS,
CameraWebRTCProvider,
WebRTCAnswer, # noqa: F401
WebRTCCandidate, # noqa: F401
@@ -93,6 +93,7 @@ from .webrtc import (
WebRTCMessage, # noqa: F401
WebRTCSendMessage,
async_get_supported_provider,
async_register_ice_servers,
async_register_webrtc_provider, # noqa: F401
async_register_ws,
)
@@ -399,6 +400,20 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
SERVICE_RECORD, CAMERA_SERVICE_RECORD, async_handle_record_service
)
@callback
def get_ice_servers() -> list[RTCIceServer]:
if hass.config.webrtc.ice_servers:
return hass.config.webrtc.ice_servers
return [
RTCIceServer(
urls=[
"stun:stun.home-assistant.io:3478",
"stun:stun.home-assistant.io:80",
]
),
]
async_register_ice_servers(hass, get_ice_servers)
return True
@@ -716,7 +731,11 @@ class Camera(Entity, cached_properties=CACHED_PROPERTIES_WITH_ATTR_):
"""Return the WebRTC client configuration and extend it with the registered ice servers."""
config = self._async_get_webrtc_client_configuration()
ice_servers = async_get_ice_servers(self.hass)
ice_servers = [
server
for servers in self.hass.data.get(DATA_ICE_SERVERS, [])
for server in servers()
]
config.configuration.ice_servers.extend(ice_servers)
return config

View File

@@ -3,7 +3,7 @@
"name": "Camera",
"after_dependencies": ["media_player"],
"codeowners": ["@home-assistant/core"],
"dependencies": ["http", "web_rtc"],
"dependencies": ["http"],
"documentation": "https://www.home-assistant.io/integrations/camera",
"integration_type": "entity",
"quality_scale": "internal",

View File

@@ -4,7 +4,7 @@ from __future__ import annotations
from abc import ABC, abstractmethod
import asyncio
from collections.abc import Awaitable, Callable
from collections.abc import Awaitable, Callable, Iterable
from dataclasses import asdict, dataclass, field
from functools import cache, partial, wraps
import logging
@@ -12,7 +12,12 @@ from typing import TYPE_CHECKING, Any
from mashumaro import MissingField
import voluptuous as vol
from webrtc_models import RTCConfiguration, RTCIceCandidate, RTCIceCandidateInit
from webrtc_models import (
RTCConfiguration,
RTCIceCandidate,
RTCIceCandidateInit,
RTCIceServer,
)
from homeassistant.components import websocket_api
from homeassistant.core import HomeAssistant, callback
@@ -33,6 +38,9 @@ _LOGGER = logging.getLogger(__name__)
DATA_WEBRTC_PROVIDERS: HassKey[set[CameraWebRTCProvider]] = HassKey(
"camera_webrtc_providers"
)
DATA_ICE_SERVERS: HassKey[list[Callable[[], Iterable[RTCIceServer]]]] = HassKey(
"camera_webrtc_ice_servers"
)
_WEBRTC = "WebRTC"
@@ -359,3 +367,21 @@ async def async_get_supported_provider(
return provider
return None
@callback
def async_register_ice_servers(
hass: HomeAssistant,
get_ice_server_fn: Callable[[], Iterable[RTCIceServer]],
) -> Callable[[], None]:
"""Register a ICE server.
The registering integration is responsible to implement caching if needed.
"""
servers = hass.data.setdefault(DATA_ICE_SERVERS, [])
def remove() -> None:
servers.remove(get_ice_server_fn)
servers.append(get_ice_server_fn)
return remove

View File

@@ -19,8 +19,8 @@ from homeassistant.components.alexa import (
errors as alexa_errors,
smart_home as alexa_smart_home,
)
from homeassistant.components.camera import async_register_ice_servers
from homeassistant.components.google_assistant import smart_home as ga
from homeassistant.components.web_rtc import async_register_ice_servers
from homeassistant.const import __version__ as HA_VERSION
from homeassistant.core import Context, HassJob, HomeAssistant, callback
from homeassistant.helpers.aiohttp_client import SERVER_SOFTWARE

View File

@@ -8,7 +8,7 @@
"google_assistant"
],
"codeowners": ["@home-assistant/cloud"],
"dependencies": ["auth", "http", "repairs", "webhook", "web_rtc"],
"dependencies": ["auth", "http", "repairs", "webhook"],
"documentation": "https://www.home-assistant.io/integrations/cloud",
"integration_type": "system",
"iot_class": "cloud_push",

View File

@@ -15,9 +15,7 @@ from .coordinator import (
PLATFORMS: list[Platform] = [
Platform.BINARY_SENSOR,
Platform.DEVICE_TRACKER,
Platform.LIGHT,
Platform.SENSOR,
Platform.SWITCH,
]

View File

@@ -4,14 +4,6 @@
"pet": {
"default": "mdi:paw"
}
},
"switch": {
"energy_saving": {
"default": "mdi:sleep",
"state": {
"off": "mdi:sleep-off"
}
}
}
}
}

View File

@@ -1,93 +0,0 @@
"""Light platform for fressnapf_tracker."""
from typing import TYPE_CHECKING, Any
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ColorMode,
LightEntity,
LightEntityDescription,
)
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import FressnapfTrackerConfigEntry
from .const import DOMAIN
from .entity import FressnapfTrackerEntity
LIGHT_ENTITY_DESCRIPTION = LightEntityDescription(
translation_key="led",
entity_category=EntityCategory.CONFIG,
key="led_brightness_value",
)
async def async_setup_entry(
hass: HomeAssistant,
entry: FressnapfTrackerConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Fressnapf Tracker lights."""
async_add_entities(
FressnapfTrackerLight(coordinator, LIGHT_ENTITY_DESCRIPTION)
for coordinator in entry.runtime_data
if coordinator.data.led_activatable is not None
and coordinator.data.led_activatable.has_led
and coordinator.data.tracker_settings.features.flash_light
)
class FressnapfTrackerLight(FressnapfTrackerEntity, LightEntity):
"""Fressnapf Tracker light."""
_attr_color_mode: ColorMode = ColorMode.BRIGHTNESS
_attr_supported_color_modes: set[ColorMode] = {ColorMode.BRIGHTNESS}
@property
def brightness(self) -> int:
"""Return the brightness of this light between 0..255."""
if TYPE_CHECKING:
# The entity is not created if led_brightness_value is None
assert self.coordinator.data.led_brightness_value is not None
return int(round((self.coordinator.data.led_brightness_value / 100) * 255))
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the device."""
self.raise_if_not_activatable()
brightness = kwargs.get(ATTR_BRIGHTNESS, 255)
brightness = int((brightness / 255) * 100)
await self.coordinator.client.set_led_brightness(brightness)
await self.coordinator.async_request_refresh()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the device."""
await self.coordinator.client.set_led_brightness(0)
await self.coordinator.async_request_refresh()
def raise_if_not_activatable(self) -> None:
"""Raise error with reasoning if light is not activatable."""
if TYPE_CHECKING:
# The entity is not created if led_activatable is None
assert self.coordinator.data.led_activatable is not None
error_type: str | None = None
if not self.coordinator.data.led_activatable.seen_recently:
error_type = "not_seen_recently"
elif not self.coordinator.data.led_activatable.not_charging:
error_type = "charging"
elif not self.coordinator.data.led_activatable.nonempty_battery:
error_type = "low_battery"
if error_type is not None:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key=error_type,
)
@property
def is_on(self) -> bool:
"""Return true if device is on."""
if self.coordinator.data.led_brightness_value is not None:
return self.coordinator.data.led_brightness_value > 0
return False

View File

@@ -45,28 +45,5 @@
}
}
}
},
"entity": {
"light": {
"led": {
"name": "Flashlight"
}
},
"switch": {
"energy_saving": {
"name": "Sleep mode"
}
}
},
"exceptions": {
"charging": {
"message": "The flashlight cannot be activated while charging."
},
"low_battery": {
"message": "The flashlight cannot be activated due to low battery."
},
"not_seen_recently": {
"message": "The flashlight cannot be activated when the tracker has not moved recently."
}
}
}

View File

@@ -1,58 +0,0 @@
"""Switch platform for Fressnapf Tracker."""
from typing import TYPE_CHECKING, Any
from homeassistant.components.switch import (
SwitchDeviceClass,
SwitchEntity,
SwitchEntityDescription,
)
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import FressnapfTrackerConfigEntry
from .entity import FressnapfTrackerEntity
SWITCH_ENTITY_DESCRIPTION = SwitchEntityDescription(
translation_key="energy_saving",
entity_category=EntityCategory.CONFIG,
device_class=SwitchDeviceClass.SWITCH,
key="energy_saving",
)
async def async_setup_entry(
hass: HomeAssistant,
entry: FressnapfTrackerConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Fressnapf Tracker switches."""
async_add_entities(
FressnapfTrackerSwitch(coordinator, SWITCH_ENTITY_DESCRIPTION)
for coordinator in entry.runtime_data
if coordinator.data.tracker_settings.features.energy_saving_mode
)
class FressnapfTrackerSwitch(FressnapfTrackerEntity, SwitchEntity):
"""Fressnapf Tracker switch."""
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the device."""
await self.coordinator.client.set_energy_saving(True)
await self.coordinator.async_request_refresh()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the device."""
await self.coordinator.client.set_energy_saving(False)
await self.coordinator.async_request_refresh()
@property
def is_on(self) -> bool:
"""Return true if device is on."""
if TYPE_CHECKING:
# The entity is not created if energy_saving is None
assert self.coordinator.data.energy_saving is not None
return self.coordinator.data.energy_saving.value == 1

View File

@@ -1,58 +0,0 @@
"""The homelink integration."""
from __future__ import annotations
from homelink.mqtt_provider import MQTTProvider
from homeassistant.const import EVENT_HOMEASSISTANT_STOP, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import aiohttp_client, config_entry_oauth2_flow
from . import oauth2
from .const import DOMAIN
from .coordinator import HomeLinkConfigEntry, HomeLinkCoordinator, HomeLinkData
PLATFORMS: list[Platform] = [Platform.EVENT]
async def async_setup_entry(hass: HomeAssistant, entry: HomeLinkConfigEntry) -> bool:
"""Set up homelink from a config entry."""
auth_implementation = oauth2.SRPAuthImplementation(hass, DOMAIN)
config_entry_oauth2_flow.async_register_implementation(
hass, DOMAIN, auth_implementation
)
implementation = (
await config_entry_oauth2_flow.async_get_config_entry_implementation(
hass, entry
)
)
session = config_entry_oauth2_flow.OAuth2Session(hass, entry, implementation)
authenticated_session = oauth2.AsyncConfigEntryAuth(
aiohttp_client.async_get_clientsession(hass), session
)
provider = MQTTProvider(authenticated_session)
coordinator = HomeLinkCoordinator(hass, provider, entry)
entry.async_on_unload(
hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_STOP, coordinator.async_on_unload
)
)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = HomeLinkData(
provider=provider, coordinator=coordinator, last_update_id=None
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: HomeLinkConfigEntry) -> bool:
"""Unload a config entry."""
await entry.runtime_data.coordinator.async_on_unload(None)
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

View File

@@ -1,14 +0,0 @@
"""application_credentials platform for the gentex homelink integration."""
from homeassistant.components.application_credentials import ClientCredential
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_oauth2_flow
from . import oauth2
async def async_get_auth_implementation(
hass: HomeAssistant, auth_domain: str, _credential: ClientCredential
) -> config_entry_oauth2_flow.AbstractOAuth2Implementation:
"""Return custom SRPAuth implementation."""
return oauth2.SRPAuthImplementation(hass, auth_domain)

View File

@@ -1,66 +0,0 @@
"""Config flow for homelink."""
import logging
from typing import Any
import botocore.exceptions
from homelink.auth.srp_auth import SRPAuth
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.helpers.config_entry_oauth2_flow import AbstractOAuth2FlowHandler
from .const import DOMAIN
from .oauth2 import SRPAuthImplementation
_LOGGER = logging.getLogger(__name__)
class SRPFlowHandler(AbstractOAuth2FlowHandler, domain=DOMAIN):
"""Config flow to handle homelink OAuth2 authentication."""
DOMAIN = DOMAIN
def __init__(self) -> None:
"""Set up the flow handler."""
super().__init__()
self.flow_impl = SRPAuthImplementation(self.hass, DOMAIN)
@property
def logger(self):
"""Get the logger."""
return _LOGGER
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> config_entries.ConfigFlowResult:
"""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(
srp_auth.async_get_access_token,
user_input[CONF_EMAIL],
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:
self.external_data = {"tokens": tokens}
return await self.async_step_creation()
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{vol.Required(CONF_EMAIL): str, vol.Required(CONF_PASSWORD): str}
),
errors=errors,
)

View File

@@ -1,7 +0,0 @@
"""Constants for the homelink integration."""
DOMAIN = "gentex_homelink"
OAUTH2_TOKEN = "https://auth.homelinkcloud.com/oauth2/token"
POLLING_INTERVAL = 5
EVENT_PRESSED = "Pressed"

View File

@@ -1,113 +0,0 @@
"""Makes requests to the state server and stores the resulting data so that the buttons can access it."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from functools import partial
import logging
from typing import TYPE_CHECKING, TypedDict
from homelink.model.device import Device
from homelink.mqtt_provider import MQTTProvider
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.util.ssl import get_default_context
if TYPE_CHECKING:
from .event import HomeLinkEventEntity
_LOGGER = logging.getLogger(__name__)
type HomeLinkConfigEntry = ConfigEntry[HomeLinkData]
type EventCallback = Callable[[HomeLinkEventData], None]
@dataclass
class HomeLinkData:
"""Class for HomeLink integration runtime data."""
provider: MQTTProvider
coordinator: HomeLinkCoordinator
last_update_id: str | None
class HomeLinkEventData(TypedDict):
"""Data for a single event."""
requestId: str
timestamp: int
class HomeLinkMQTTMessage(TypedDict):
"""HomeLink MQTT Event message."""
type: str
data: dict[str, HomeLinkEventData] # Each key is a button id
class HomeLinkCoordinator:
"""HomeLink integration coordinator."""
def __init__(
self,
hass: HomeAssistant,
provider: MQTTProvider,
config_entry: HomeLinkConfigEntry,
) -> None:
"""Initialize my coordinator."""
self.hass = hass
self.config_entry = config_entry
self.provider = provider
self.device_data: list[Device] = []
self.buttons: list[HomeLinkEventEntity] = []
self._listeners: dict[str, EventCallback] = {}
@callback
def async_add_event_listener(
self, update_callback: EventCallback, target_event_id: str
) -> Callable[[], None]:
"""Listen for updates."""
self._listeners[target_event_id] = update_callback
return partial(self.__async_remove_listener_internal, target_event_id)
def __async_remove_listener_internal(self, listener_id: str):
del self._listeners[listener_id]
@callback
def async_handle_state_data(self, data: dict[str, HomeLinkEventData]):
"""Notify listeners."""
for button_id, event in data.items():
if listener := self._listeners.get(button_id):
listener(event)
async def async_config_entry_first_refresh(self) -> None:
"""Refresh data for the first time when a config entry is setup."""
await self._async_setup()
async def async_on_unload(self, _event):
"""Disconnect and unregister when unloaded."""
await self.provider.disable()
async def _async_setup(self) -> None:
"""Set up the coordinator."""
await self.provider.enable(get_default_context())
await self.discover_devices()
self.provider.listen(self.on_message)
async def discover_devices(self):
"""Discover devices and build the Entities."""
self.device_data = await self.provider.discover()
def on_message(
self: HomeLinkCoordinator, _topic: str, message: HomeLinkMQTTMessage
):
"MQTT Callback function."
if message["type"] == "state":
self.hass.add_job(self.async_handle_state_data, message["data"])
if message["type"] == "requestSync":
self.hass.add_job(
self.hass.config_entries.async_reload,
self.config_entry.entry_id,
)

View File

@@ -1,83 +0,0 @@
"""Platform for Event integration."""
from __future__ import annotations
from homeassistant.components.event import EventDeviceClass, EventEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN, EVENT_PRESSED
from .coordinator import HomeLinkCoordinator, HomeLinkEventData
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Add the entities for the binary sensor."""
coordinator = config_entry.runtime_data.coordinator
for device in coordinator.device_data:
buttons = [
HomeLinkEventEntity(b.id, b.name, device.id, device.name, coordinator)
for b in device.buttons
]
coordinator.buttons.extend(buttons)
async_add_entities(coordinator.buttons)
# Updates are centralized by the coordinator.
PARALLEL_UPDATES = 0
class HomeLinkEventEntity(EventEntity):
"""Event Entity."""
_attr_has_entity_name = True
_attr_event_types = [EVENT_PRESSED]
_attr_device_class = EventDeviceClass.BUTTON
def __init__(
self,
id: str,
param_name: str,
device_id: str,
device_name: str,
coordinator: HomeLinkCoordinator,
) -> None:
"""Initialize the event entity."""
self.id: str = id
self._attr_name: str = param_name
self._attr_unique_id: str = id
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, device_id)},
name=device_name,
)
self.coordinator = coordinator
self.last_request_id: str | None = None
async def async_added_to_hass(self) -> None:
"""When entity is added to hass."""
await super().async_added_to_hass()
self.async_on_remove(
self.coordinator.async_add_event_listener(
self._handle_event_data_update, self.id
)
)
@callback
def _handle_event_data_update(self, update_data: HomeLinkEventData) -> None:
"""Update this button."""
if update_data["requestId"] != self.last_request_id:
self._trigger_event(EVENT_PRESSED)
self.last_request_id = update_data["requestId"]
self.async_write_ha_state()
async def async_update(self):
"""Request early polling. Left intentionally blank because it's not possible in this implementation."""

View File

@@ -1,11 +0,0 @@
{
"domain": "gentex_homelink",
"name": "HomeLink",
"codeowners": ["@niaexa", "@ryanjones-gentex"],
"config_flow": true,
"dependencies": ["application_credentials"],
"documentation": "https://www.home-assistant.io/integrations/gentex_homelink",
"iot_class": "cloud_push",
"quality_scale": "bronze",
"requirements": ["homelink-integration-api==0.0.1"]
}

View File

@@ -1,114 +0,0 @@
"""API for homelink bound to Home Assistant OAuth."""
from json import JSONDecodeError
import logging
import time
from typing import cast
from aiohttp import ClientError, ClientSession
from homelink.auth.abstract_auth import AbstractAuth
from homelink.settings import COGNITO_CLIENT_ID
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import OAUTH2_TOKEN
_LOGGER = logging.getLogger(__name__)
class SRPAuthImplementation(config_entry_oauth2_flow.AbstractOAuth2Implementation):
"""Base class to abstract OAuth2 authentication."""
def __init__(self, hass: HomeAssistant, domain) -> None:
"""Initialize the SRP Auth implementation."""
self.hass = hass
self._domain = domain
self.client_id = COGNITO_CLIENT_ID
@property
def name(self) -> str:
"""Name of the implementation."""
return "SRPAuth"
@property
def domain(self) -> str:
"""Domain that is providing the implementation."""
return self._domain
async def async_generate_authorize_url(self, flow_id: str) -> str:
"""Left intentionally blank because the auth is handled by SRP."""
return ""
async def async_resolve_external_data(self, external_data) -> dict:
"""Format the token from the source appropriately for HomeAssistant."""
tokens = external_data["tokens"]
new_token = {}
new_token["access_token"] = tokens["AuthenticationResult"]["AccessToken"]
new_token["refresh_token"] = tokens["AuthenticationResult"]["RefreshToken"]
new_token["token_type"] = tokens["AuthenticationResult"]["TokenType"]
new_token["expires_in"] = tokens["AuthenticationResult"]["ExpiresIn"]
new_token["expires_at"] = (
time.time() + tokens["AuthenticationResult"]["ExpiresIn"]
)
return new_token
async def _token_request(self, data: dict) -> dict:
"""Make a token request."""
session = async_get_clientsession(self.hass)
data["client_id"] = self.client_id
_LOGGER.debug("Sending token request to %s", OAUTH2_TOKEN)
resp = await session.post(OAUTH2_TOKEN, data=data)
if resp.status >= 400:
try:
error_response = await resp.json()
except (ClientError, JSONDecodeError):
error_response = {}
error_code = error_response.get("error", "unknown")
error_description = error_response.get(
"error_description", "unknown error"
)
_LOGGER.error(
"Token request for %s failed (%s): %s",
self.domain,
error_code,
error_description,
)
resp.raise_for_status()
return cast(dict, await resp.json())
async def _async_refresh_token(self, token: dict) -> dict:
"""Refresh tokens."""
new_token = await self._token_request(
{
"grant_type": "refresh_token",
"client_id": self.client_id,
"refresh_token": token["refresh_token"],
}
)
return {**token, **new_token}
class AsyncConfigEntryAuth(AbstractAuth):
"""Provide homelink authentication tied to an OAuth2 based config entry."""
def __init__(
self,
websession: ClientSession,
oauth_session: config_entry_oauth2_flow.OAuth2Session,
) -> None:
"""Initialize homelink auth."""
super().__init__(websession)
self._oauth_session = oauth_session
async def async_get_access_token(self) -> str:
"""Return a valid access token."""
if not self._oauth_session.valid_token:
await self._oauth_session.async_ensure_token_valid()
return self._oauth_session.token["access_token"]

View File

@@ -1,76 +0,0 @@
rules:
# Bronze
action-setup:
status: exempt
comment: Integration does not register any service actions
appropriate-polling:
status: exempt
comment: Integration does not poll
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: Integration does not register any service actions
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup: done
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions:
status: exempt
comment: Integration does not register any service actions
config-entry-unloading: done
docs-configuration-parameters: done
docs-installation-parameters: done
entity-unavailable: todo
integration-owner: done
log-when-unavailable: todo
parallel-updates: done
reauthentication-flow: todo
test-coverage: todo
# Gold
devices: done
diagnostics: todo
discovery-update-info:
status: exempt
comment: It is not necessary to update IP addresses of devices or services in this Integration
discovery: todo
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
docs-supported-functions: todo
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices: done
entity-category: todo
entity-device-class: todo
entity-disabled-by-default:
status: exempt
comment: Entities are not noisy and are expected to be enabled by default
entity-translations:
status: exempt
comment: Entity properties are user-defined, and therefore cannot be translated
exception-translations: todo
icon-translations:
status: exempt
comment: Entities in this integration do not use icons, and therefore do not require translation
reconfiguration-flow: todo
repair-issues: todo
stale-devices: done
# Platinum
async-dependency: done
inject-websession: done
strict-typing: todo

View File

@@ -1,38 +0,0 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"authorize_url_timeout": "[%key:common::config_flow::abort::oauth2_authorize_url_timeout%]",
"missing_configuration": "[%key:common::config_flow::abort::oauth2_missing_configuration%]",
"no_url_available": "[%key:common::config_flow::abort::oauth2_no_url_available%]",
"oauth_error": "[%key:common::config_flow::abort::oauth2_error%]",
"oauth_failed": "[%key:common::config_flow::abort::oauth2_failed%]",
"oauth_timeout": "[%key:common::config_flow::abort::oauth2_timeout%]",
"oauth_unauthorized": "[%key:common::config_flow::abort::oauth2_unauthorized%]",
"user_rejected_authorize": "[%key:common::config_flow::abort::oauth2_user_rejected_authorize%]"
},
"create_entry": {
"default": "[%key:common::config_flow::create_entry::authenticated%]"
},
"error": {
"srp_auth_failed": "Error authenticating HomeLink account",
"unknown": "An unknown error occurred. Please try again later"
},
"step": {
"pick_implementation": {
"title": "[%key:common::config_flow::title::oauth2_pick_implementation%]"
},
"user": {
"data": {
"email": "[%key:common::config_flow::data::email%]",
"password": "[%key:common::config_flow::data::password%]"
},
"data_description": {
"email": "Email address associated with your HomeLink account",
"password": "Password associated with your HomeLink account"
}
}
}
}
}

View File

@@ -106,6 +106,9 @@
}
},
"triggers": {
"muted": {
"trigger": "mdi:volume-mute"
},
"stopped_playing": {
"trigger": "mdi:stop"
}

View File

@@ -380,6 +380,16 @@
},
"title": "Media player",
"triggers": {
"muted": {
"description": "Triggers after one or more media players are muted.",
"fields": {
"behavior": {
"description": "[%key:component::media_player::common::trigger_behavior_description%]",
"name": "[%key:component::media_player::common::trigger_behavior_name%]"
}
},
"name": "Media player muted"
},
"stopped_playing": {
"description": "Triggers after one or more media players stop playing media.",
"fields": {

View File

@@ -1,12 +1,35 @@
"""Provides triggers for media players."""
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import Trigger, make_conditional_entity_state_trigger
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers.trigger import (
EntityTriggerBase,
Trigger,
make_conditional_entity_state_trigger,
)
from . import MediaPlayerState
from . import ATTR_MEDIA_VOLUME_LEVEL, ATTR_MEDIA_VOLUME_MUTED, MediaPlayerState
from .const import DOMAIN
class MediaPlayerMutedTrigger(EntityTriggerBase):
"""Class for media player muted triggers."""
_domain: str = DOMAIN
def is_muted(self, state: State) -> bool:
"""Check if the media player is muted."""
return (
state.attributes.get(ATTR_MEDIA_VOLUME_MUTED) is True
or state.attributes.get(ATTR_MEDIA_VOLUME_LEVEL) == 0
)
def is_to_state(self, state: State) -> bool:
"""Check if the state matches the target state."""
return self.is_muted(state)
TRIGGERS: dict[str, type[Trigger]] = {
"muted": MediaPlayerMutedTrigger,
"stopped_playing": make_conditional_entity_state_trigger(
DOMAIN,
from_states={

View File

@@ -1,4 +1,4 @@
stopped_playing:
.trigger_common: &trigger_common
target:
entity:
domain: media_player
@@ -13,3 +13,6 @@ stopped_playing:
- first
- last
- any
muted: *trigger_common
stopped_playing: *trigger_common

View File

@@ -81,9 +81,6 @@ async def async_setup_entry(
SERVICE_PUBLISH,
SERVICE_PUBLISH_SCHEMA,
"publish",
description_placeholders={
"markdown_guide_url": "https://www.markdownguide.org/basic-syntax/"
},
)

View File

@@ -345,7 +345,7 @@
"name": "Icon URL"
},
"markdown": {
"description": "Enable Markdown formatting for the message body. See the Markdown guide for syntax details: {markdown_guide_url}.",
"description": "Enable Markdown formatting for the message body. See the Markdown guide for syntax details: https://www.markdownguide.org/basic-syntax/.",
"name": "Format as Markdown"
},
"message": {

View File

@@ -25,7 +25,6 @@ from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.event import track_point_in_utc_time
from homeassistant.helpers.typing import ConfigType
from homeassistant.util import dt as dt_util
from homeassistant.util.async_ import run_callback_threadsafe
_LOGGER = logging.getLogger(__name__)
@@ -102,18 +101,7 @@ def setup(hass: HomeAssistant, config: ConfigType) -> bool:
except OSError:
_LOGGER.error("Pilight send failed for %s", str(message_data))
def _register_service() -> None:
hass.services.async_register(
DOMAIN,
SERVICE_NAME,
send_code,
schema=RF_CODE_SCHEMA,
description_placeholders={
"pilight_protocols_docs_url": "https://manual.pilight.org/protocols/index.html"
},
)
run_callback_threadsafe(hass.loop, _register_service).result()
hass.services.register(DOMAIN, SERVICE_NAME, send_code, schema=RF_CODE_SCHEMA)
# Publish received codes on the HA event bus
# A whitelist of codes to be published in the event bus

View File

@@ -4,7 +4,7 @@
"description": "Sends RF code to Pilight device.",
"fields": {
"protocol": {
"description": "Protocol that Pilight recognizes. See {pilight_protocols_docs_url} for supported protocols and additional parameters that each protocol supports.",
"description": "Protocol that Pilight recognizes. See https://manual.pilight.org/protocols/index.html for supported protocols and additional parameters that each protocol supports.",
"name": "Protocol"
}
},

View File

@@ -54,12 +54,9 @@ from .const import (
)
from .coordinator import RainMachineDataUpdateCoordinator
API_URL_REFERENCE = (
"https://rainmachine.docs.apiary.io/#reference/weather-services/parserdata/post"
)
DEFAULT_SSL = True
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.BUTTON,
@@ -458,15 +455,7 @@ async def async_setup_entry( # noqa: C901
):
if hass.services.has_service(DOMAIN, service_name):
continue
hass.services.async_register(
DOMAIN,
service_name,
method,
schema=schema,
description_placeholders={
"api_url": API_URL_REFERENCE,
},
)
hass.services.async_register(DOMAIN, service_name, method, schema=schema)
return True

View File

@@ -128,7 +128,7 @@
"name": "Push flow meter data"
},
"push_weather_data": {
"description": "Sends weather data from Home Assistant to the RainMachine device.\nLocal Weather Push service should be enabled from Settings > Weather > Developer tab for RainMachine to consider the values being sent. Units must be sent in metric; no conversions are performed by the integration.\nSee details of RainMachine API here: {api_url}",
"description": "Sends weather data from Home Assistant to the RainMachine device.\nLocal Weather Push service should be enabled from Settings > Weather > Developer tab for RainMachine to consider the values being sent. Units must be sent in metric; no conversions are performed by the integration.\nSee details of RainMachine API here: https://rainmachine.docs.apiary.io/#reference/weather-services/parserdata/post.",
"fields": {
"condition": {
"description": "Current weather condition code (WNUM).",

View File

@@ -20,5 +20,5 @@
"iot_class": "local_push",
"loggers": ["reolink_aio"],
"quality_scale": "platinum",
"requirements": ["reolink-aio==0.17.1"]
"requirements": ["reolink-aio==0.17.0"]
}

View File

@@ -72,6 +72,7 @@ class StarlinkUpdateCoordinator(DataUpdateCoordinator[StarlinkData]):
def _get_starlink_data(self) -> StarlinkData:
"""Retrieve Starlink data."""
context = self.channel_context
status = status_data(context)
location = location_data(context)
sleep = get_sleep_config(context)
status, obstruction, alert = status_data(context)

View File

@@ -28,7 +28,6 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.util.dt import now
from homeassistant.util.variance import ignore_variance
from .coordinator import StarlinkConfigEntry, StarlinkData
from .entity import StarlinkEntity
@@ -92,10 +91,6 @@ class StarlinkAccumulationSensor(StarlinkSensorEntity, RestoreSensor):
self._attr_native_value = last_native_value
uptime_to_stable_datetime = ignore_variance(
lambda value: now() - timedelta(seconds=value), timedelta(minutes=1)
)
SENSORS: tuple[StarlinkSensorEntityDescription, ...] = (
StarlinkSensorEntityDescription(
key="ping",
@@ -155,7 +150,9 @@ SENSORS: tuple[StarlinkSensorEntityDescription, ...] = (
translation_key="last_restart",
device_class=SensorDeviceClass.TIMESTAMP,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda data: uptime_to_stable_datetime(data.status["uptime"]),
value_fn=lambda data: (
now() - timedelta(seconds=data.status["uptime"], milliseconds=-500)
).replace(microsecond=0),
entity_class=StarlinkSensorEntity,
),
StarlinkSensorEntityDescription(

View File

@@ -1,7 +1,7 @@
"""Helpers for template integration."""
from collections.abc import Callable
from enum import StrEnum
from enum import Enum
import hashlib
import itertools
import logging
@@ -33,7 +33,6 @@ from homeassistant.helpers.entity_platform import (
async_get_platforms,
)
from homeassistant.helpers.issue_registry import IssueSeverity
from homeassistant.helpers.script_variables import ScriptVariables
from homeassistant.helpers.singleton import singleton
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import yaml as yaml_util
@@ -191,12 +190,12 @@ def async_create_template_tracking_entities(
async_add_entities(entities)
def _format_template(value: Any, field: str | None = None) -> Any:
def _format_template(value: Any) -> Any:
if isinstance(value, template.Template):
return value.template
if isinstance(value, StrEnum):
return value.value
if isinstance(value, Enum):
return value.name
if isinstance(value, (int, float, str, bool)):
return value
@@ -208,13 +207,14 @@ def format_migration_config(
config: ConfigType | list[ConfigType], depth: int = 0
) -> ConfigType | list[ConfigType]:
"""Recursive method to format templates as strings from ConfigType."""
types = (dict, list)
if depth > 9:
raise RecursionError
if isinstance(config, list):
items = []
for item in config:
if isinstance(item, (dict, list)):
if isinstance(item, types):
if len(item) > 0:
items.append(format_migration_config(item, depth + 1))
else:
@@ -223,18 +223,9 @@ def format_migration_config(
formatted_config = {}
for field, value in config.items():
if isinstance(value, dict):
if isinstance(value, types):
if len(value) > 0:
formatted_config[field] = format_migration_config(value, depth + 1)
elif isinstance(value, list):
if len(value) > 0:
formatted_config[field] = format_migration_config(value, depth + 1)
else:
formatted_config[field] = []
elif isinstance(value, ScriptVariables):
formatted_config[field] = format_migration_config(
value.as_dict(), depth + 1
)
else:
formatted_config[field] = _format_template(value)

View File

@@ -325,9 +325,6 @@ def async_setup_services(hass: HomeAssistant) -> None:
vol.Required(ATTR_TOU_SETTINGS): dict,
}
),
description_placeholders={
"time_of_use_url": "https://developer.tesla.com/docs/fleet-api#time_of_use_settings"
},
)
async def add_charge_schedule(call: ServiceCall) -> None:

View File

@@ -1358,7 +1358,7 @@
"name": "Energy Site"
},
"tou_settings": {
"description": "See {time_use_url} for details.",
"description": "See https://developer.tesla.com/docs/fleet-api#time_of_use_settings for details.",
"name": "Settings"
}
},

View File

@@ -208,18 +208,6 @@ def _get_temperature_wrappers(
device, DPCode.TEMP_SET_F, prefer_function=True
)
# If there is a temp unit convert dpcode, override empty units
if (
temp_unit_convert := DPCodeEnumWrapper.find_dpcode(
device, DPCode.TEMP_UNIT_CONVERT
)
) is not None:
for wrapper in (temp_current, temp_current_f, temp_set, temp_set_f):
if wrapper is not None and not wrapper.type_information.unit:
wrapper.type_information.unit = temp_unit_convert.read_device_status(
device
)
# Get wrappers for celsius and fahrenheit
# We need to check the unit of measurement
current_celsius = _get_temperature_wrapper(

View File

@@ -35,8 +35,8 @@ from .models import (
DPCodeEnumWrapper,
DPCodeIntegerWrapper,
DPCodeJsonWrapper,
IntegerTypeData,
)
from .type_information import IntegerTypeInformation
from .util import remap_value
@@ -138,24 +138,24 @@ class _ColorTempWrapper(DPCodeIntegerWrapper):
)
DEFAULT_H_TYPE = IntegerTypeInformation(
DEFAULT_H_TYPE = IntegerTypeData(
dpcode=DPCode.COLOUR_DATA_HSV, min=1, scale=0, max=360, step=1
)
DEFAULT_S_TYPE = IntegerTypeInformation(
DEFAULT_S_TYPE = IntegerTypeData(
dpcode=DPCode.COLOUR_DATA_HSV, min=1, scale=0, max=255, step=1
)
DEFAULT_V_TYPE = IntegerTypeInformation(
DEFAULT_V_TYPE = IntegerTypeData(
dpcode=DPCode.COLOUR_DATA_HSV, min=1, scale=0, max=255, step=1
)
DEFAULT_H_TYPE_V2 = IntegerTypeInformation(
DEFAULT_H_TYPE_V2 = IntegerTypeData(
dpcode=DPCode.COLOUR_DATA_HSV, min=1, scale=0, max=360, step=1
)
DEFAULT_S_TYPE_V2 = IntegerTypeInformation(
DEFAULT_S_TYPE_V2 = IntegerTypeData(
dpcode=DPCode.COLOUR_DATA_HSV, min=1, scale=0, max=1000, step=1
)
DEFAULT_V_TYPE_V2 = IntegerTypeInformation(
DEFAULT_V_TYPE_V2 = IntegerTypeData(
dpcode=DPCode.COLOUR_DATA_HSV, min=1, scale=0, max=1000, step=1
)
@@ -578,15 +578,15 @@ def _get_color_data_wrapper(
if function_data := json_loads_object(
cast(str, color_data_wrapper.type_information.type_data)
):
color_data_wrapper.h_type = IntegerTypeInformation(
color_data_wrapper.h_type = IntegerTypeData(
dpcode=color_data_wrapper.dpcode,
**cast(dict, function_data["h"]),
)
color_data_wrapper.s_type = IntegerTypeInformation(
color_data_wrapper.s_type = IntegerTypeData(
dpcode=color_data_wrapper.dpcode,
**cast(dict, function_data["s"]),
)
color_data_wrapper.v_type = IntegerTypeInformation(
color_data_wrapper.v_type = IntegerTypeData(
dpcode=color_data_wrapper.dpcode,
**cast(dict, function_data["v"]),
)

View File

@@ -3,19 +3,15 @@
from __future__ import annotations
import base64
from typing import Any, Self
from dataclasses import dataclass
from typing import Any, Literal, Self, cast, overload
from tuya_sharing import CustomerDevice
from homeassistant.util.json import json_loads
from homeassistant.util.json import json_loads, json_loads_object
from .const import LOGGER, DPType
from .type_information import (
EnumTypeInformation,
IntegerTypeInformation,
TypeInformation,
find_dpcode,
)
from .util import parse_dptype, remap_value
# Dictionary to track logged warnings to avoid spamming logs
# Keyed by device ID
@@ -36,6 +32,139 @@ def _should_log_warning(device_id: str, warning_key: str) -> bool:
return True
@dataclass(kw_only=True)
class TypeInformation:
"""Type information.
As provided by the SDK, from `device.function` / `device.status_range`.
"""
dpcode: str
type_data: str | None = None
@classmethod
def from_json(cls, dpcode: str, type_data: str) -> Self | None:
"""Load JSON string and return a TypeInformation object."""
return cls(dpcode=dpcode, type_data=type_data)
@dataclass(kw_only=True)
class IntegerTypeData(TypeInformation):
"""Integer Type Data."""
min: int
max: int
scale: int
step: int
unit: str | None = None
@property
def max_scaled(self) -> float:
"""Return the max scaled."""
return self.scale_value(self.max)
@property
def min_scaled(self) -> float:
"""Return the min scaled."""
return self.scale_value(self.min)
@property
def step_scaled(self) -> float:
"""Return the step scaled."""
return self.step / (10**self.scale)
def scale_value(self, value: int) -> float:
"""Scale a value."""
return value / (10**self.scale)
def scale_value_back(self, value: float) -> int:
"""Return raw value for scaled."""
return round(value * (10**self.scale))
def remap_value_to(
self,
value: float,
to_min: float = 0,
to_max: float = 255,
reverse: bool = False,
) -> float:
"""Remap a value from this range to a new range."""
return remap_value(value, self.min, self.max, to_min, to_max, reverse)
def remap_value_from(
self,
value: float,
from_min: float = 0,
from_max: float = 255,
reverse: bool = False,
) -> float:
"""Remap a value from its current range to this range."""
return remap_value(value, from_min, from_max, self.min, self.max, reverse)
@classmethod
def from_json(cls, dpcode: str, type_data: str) -> Self | None:
"""Load JSON string and return a IntegerTypeData object."""
if not (parsed := cast(dict[str, Any] | None, json_loads_object(type_data))):
return None
return cls(
dpcode=dpcode,
type_data=type_data,
min=int(parsed["min"]),
max=int(parsed["max"]),
scale=int(parsed["scale"]),
step=int(parsed["step"]),
unit=parsed.get("unit"),
)
@dataclass(kw_only=True)
class BitmapTypeInformation(TypeInformation):
"""Bitmap type information."""
label: list[str]
@classmethod
def from_json(cls, dpcode: str, type_data: str) -> Self | None:
"""Load JSON string and return a BitmapTypeInformation object."""
if not (parsed := json_loads_object(type_data)):
return None
return cls(
dpcode=dpcode,
type_data=type_data,
**cast(dict[str, list[str]], parsed),
)
@dataclass(kw_only=True)
class EnumTypeData(TypeInformation):
"""Enum Type Data."""
range: list[str]
@classmethod
def from_json(cls, dpcode: str, type_data: str) -> Self | None:
"""Load JSON string and return a EnumTypeData object."""
if not (parsed := json_loads_object(type_data)):
return None
return cls(
dpcode=dpcode,
type_data=type_data,
**cast(dict[str, list[str]], parsed),
)
_TYPE_INFORMATION_MAPPINGS: dict[DPType, type[TypeInformation]] = {
DPType.BITMAP: BitmapTypeInformation,
DPType.BOOLEAN: TypeInformation,
DPType.ENUM: EnumTypeData,
DPType.INTEGER: IntegerTypeData,
DPType.JSON: TypeInformation,
DPType.RAW: TypeInformation,
DPType.STRING: TypeInformation,
}
class DeviceWrapper:
"""Base device wrapper."""
@@ -174,8 +303,8 @@ class DPCodeJsonWrapper(DPCodeTypeInformationWrapper[TypeInformation]):
return json_loads(raw_value)
class DPCodeEnumWrapper(DPCodeTypeInformationWrapper[EnumTypeInformation]):
"""Simple wrapper for EnumTypeInformation values."""
class DPCodeEnumWrapper(DPCodeTypeInformationWrapper[EnumTypeData]):
"""Simple wrapper for EnumTypeData values."""
DPTYPE = DPType.ENUM
@@ -213,12 +342,12 @@ class DPCodeEnumWrapper(DPCodeTypeInformationWrapper[EnumTypeInformation]):
)
class DPCodeIntegerWrapper(DPCodeTypeInformationWrapper[IntegerTypeInformation]):
"""Simple wrapper for IntegerTypeInformation values."""
class DPCodeIntegerWrapper(DPCodeTypeInformationWrapper[IntegerTypeData]):
"""Simple wrapper for IntegerTypeData values."""
DPTYPE = DPType.INTEGER
def __init__(self, dpcode: str, type_information: IntegerTypeInformation) -> None:
def __init__(self, dpcode: str, type_information: IntegerTypeData) -> None:
"""Init DPCodeIntegerWrapper."""
super().__init__(dpcode, type_information)
self.native_unit = type_information.unit
@@ -285,3 +414,82 @@ class DPCodeBitmapBitWrapper(DPCodeWrapper):
type_information.dpcode, type_information.label.index(bitmap_key)
)
return None
@overload
def find_dpcode(
device: CustomerDevice,
dpcodes: str | tuple[str, ...] | None,
*,
prefer_function: bool = False,
dptype: Literal[DPType.BITMAP],
) -> BitmapTypeInformation | None: ...
@overload
def find_dpcode(
device: CustomerDevice,
dpcodes: str | tuple[str, ...] | None,
*,
prefer_function: bool = False,
dptype: Literal[DPType.ENUM],
) -> EnumTypeData | None: ...
@overload
def find_dpcode(
device: CustomerDevice,
dpcodes: str | tuple[str, ...] | None,
*,
prefer_function: bool = False,
dptype: Literal[DPType.INTEGER],
) -> IntegerTypeData | None: ...
@overload
def find_dpcode(
device: CustomerDevice,
dpcodes: str | tuple[str, ...] | None,
*,
prefer_function: bool = False,
dptype: Literal[DPType.BOOLEAN, DPType.JSON, DPType.RAW],
) -> TypeInformation | None: ...
def find_dpcode(
device: CustomerDevice,
dpcodes: str | tuple[str, ...] | None,
*,
prefer_function: bool = False,
dptype: DPType,
) -> TypeInformation | None:
"""Find type information for a matching DP code available for this device."""
if not (type_information_cls := _TYPE_INFORMATION_MAPPINGS.get(dptype)):
raise NotImplementedError(f"find_dpcode not supported for {dptype}")
if dpcodes is None:
return None
if not isinstance(dpcodes, tuple):
dpcodes = (dpcodes,)
lookup_tuple = (
(device.function, device.status_range)
if prefer_function
else (device.status_range, device.function)
)
for dpcode in dpcodes:
for device_specs in lookup_tuple:
if (
(current_definition := device_specs.get(dpcode))
and parse_dptype(current_definition.type) is dptype
and (
type_information := type_information_cls.from_json(
dpcode=dpcode, type_data=current_definition.values
)
)
):
return type_information
return None

View File

@@ -25,7 +25,7 @@ from .const import (
DPCode,
)
from .entity import TuyaEntity
from .models import DPCodeIntegerWrapper
from .models import DPCodeIntegerWrapper, IntegerTypeData
NUMBERS: dict[DeviceCategory, tuple[NumberEntityDescription, ...]] = {
DeviceCategory.BH: (
@@ -483,6 +483,8 @@ async def async_setup_entry(
class TuyaNumberEntity(TuyaEntity, NumberEntity):
"""Tuya Number Entity."""
_number: IntegerTypeData | None = None
def __init__(
self,
device: CustomerDevice,

View File

@@ -46,12 +46,12 @@ from .models import (
DPCodeJsonWrapper,
DPCodeTypeInformationWrapper,
DPCodeWrapper,
EnumTypeData,
)
from .raw_data_models import ElectricityData
from .type_information import EnumTypeInformation
class _WindDirectionWrapper(DPCodeTypeInformationWrapper[EnumTypeInformation]):
class _WindDirectionWrapper(DPCodeTypeInformationWrapper[EnumTypeData]):
"""Custom DPCode Wrapper for converting enum to wind direction."""
DPTYPE = DPType.ENUM

View File

@@ -1,225 +0,0 @@
"""Type information classes for the Tuya integration."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Literal, Self, cast, overload
from tuya_sharing import CustomerDevice
from homeassistant.util.json import json_loads_object
from .const import DPType
from .util import parse_dptype, remap_value
@dataclass(kw_only=True)
class TypeInformation:
"""Type information.
As provided by the SDK, from `device.function` / `device.status_range`.
"""
dpcode: str
type_data: str | None = None
@classmethod
def from_json(cls, dpcode: str, type_data: str) -> Self | None:
"""Load JSON string and return a TypeInformation object."""
return cls(dpcode=dpcode, type_data=type_data)
@dataclass(kw_only=True)
class BitmapTypeInformation(TypeInformation):
"""Bitmap type information."""
label: list[str]
@classmethod
def from_json(cls, dpcode: str, type_data: str) -> Self | None:
"""Load JSON string and return a BitmapTypeInformation object."""
if not (parsed := json_loads_object(type_data)):
return None
return cls(
dpcode=dpcode,
type_data=type_data,
**cast(dict[str, list[str]], parsed),
)
@dataclass(kw_only=True)
class EnumTypeInformation(TypeInformation):
"""Enum type information."""
range: list[str]
@classmethod
def from_json(cls, dpcode: str, type_data: str) -> Self | None:
"""Load JSON string and return an EnumTypeInformation object."""
if not (parsed := json_loads_object(type_data)):
return None
return cls(
dpcode=dpcode,
type_data=type_data,
**cast(dict[str, list[str]], parsed),
)
@dataclass(kw_only=True)
class IntegerTypeInformation(TypeInformation):
"""Integer type information."""
min: int
max: int
scale: int
step: int
unit: str | None = None
@property
def max_scaled(self) -> float:
"""Return the max scaled."""
return self.scale_value(self.max)
@property
def min_scaled(self) -> float:
"""Return the min scaled."""
return self.scale_value(self.min)
@property
def step_scaled(self) -> float:
"""Return the step scaled."""
return self.step / (10**self.scale)
def scale_value(self, value: int) -> float:
"""Scale a value."""
return value / (10**self.scale)
def scale_value_back(self, value: float) -> int:
"""Return raw value for scaled."""
return round(value * (10**self.scale))
def remap_value_to(
self,
value: float,
to_min: float = 0,
to_max: float = 255,
reverse: bool = False,
) -> float:
"""Remap a value from this range to a new range."""
return remap_value(value, self.min, self.max, to_min, to_max, reverse)
def remap_value_from(
self,
value: float,
from_min: float = 0,
from_max: float = 255,
reverse: bool = False,
) -> float:
"""Remap a value from its current range to this range."""
return remap_value(value, from_min, from_max, self.min, self.max, reverse)
@classmethod
def from_json(cls, dpcode: str, type_data: str) -> Self | None:
"""Load JSON string and return an IntegerTypeInformation object."""
if not (parsed := cast(dict[str, Any] | None, json_loads_object(type_data))):
return None
return cls(
dpcode=dpcode,
type_data=type_data,
min=int(parsed["min"]),
max=int(parsed["max"]),
scale=int(parsed["scale"]),
step=int(parsed["step"]),
unit=parsed.get("unit"),
)
_TYPE_INFORMATION_MAPPINGS: dict[DPType, type[TypeInformation]] = {
DPType.BITMAP: BitmapTypeInformation,
DPType.BOOLEAN: TypeInformation,
DPType.ENUM: EnumTypeInformation,
DPType.INTEGER: IntegerTypeInformation,
DPType.JSON: TypeInformation,
DPType.RAW: TypeInformation,
DPType.STRING: TypeInformation,
}
@overload
def find_dpcode(
device: CustomerDevice,
dpcodes: str | tuple[str, ...] | None,
*,
prefer_function: bool = False,
dptype: Literal[DPType.BITMAP],
) -> BitmapTypeInformation | None: ...
@overload
def find_dpcode(
device: CustomerDevice,
dpcodes: str | tuple[str, ...] | None,
*,
prefer_function: bool = False,
dptype: Literal[DPType.ENUM],
) -> EnumTypeInformation | None: ...
@overload
def find_dpcode(
device: CustomerDevice,
dpcodes: str | tuple[str, ...] | None,
*,
prefer_function: bool = False,
dptype: Literal[DPType.INTEGER],
) -> IntegerTypeInformation | None: ...
@overload
def find_dpcode(
device: CustomerDevice,
dpcodes: str | tuple[str, ...] | None,
*,
prefer_function: bool = False,
dptype: Literal[DPType.BOOLEAN, DPType.JSON, DPType.RAW],
) -> TypeInformation | None: ...
def find_dpcode(
device: CustomerDevice,
dpcodes: str | tuple[str, ...] | None,
*,
prefer_function: bool = False,
dptype: DPType,
) -> TypeInformation | None:
"""Find type information for a matching DP code available for this device."""
if not (type_information_cls := _TYPE_INFORMATION_MAPPINGS.get(dptype)):
raise NotImplementedError(f"find_dpcode not supported for {dptype}")
if dpcodes is None:
return None
if not isinstance(dpcodes, tuple):
dpcodes = (dpcodes,)
lookup_tuple = (
(device.function, device.status_range)
if prefer_function
else (device.status_range, device.function)
)
for dpcode in dpcodes:
for device_specs in lookup_tuple:
if (
(current_definition := device_specs.get(dpcode))
and parse_dptype(current_definition.type) is dptype
and (
type_information := type_information_cls.from_json(
dpcode=dpcode, type_data=current_definition.values
)
)
):
return type_information
return None

View File

@@ -64,9 +64,9 @@ async def async_setup_entry(
coordinator = hass.data[DOMAIN][VS_COORDINATOR]
@callback
def discover(devices: list[VeSyncBaseDevice]) -> None:
def discover(devices):
"""Add new devices to platform."""
_setup_entities(devices, async_add_entities, coordinator)
_setup_entities(devices, async_add_entities)
config_entry.async_on_unload(
async_dispatcher_connect(hass, VS_DISCOVERY.format(VS_DEVICES), discover)
@@ -78,11 +78,7 @@ async def async_setup_entry(
@callback
def _setup_entities(
devices: list[VeSyncBaseDevice],
async_add_entities: AddConfigEntryEntitiesCallback,
coordinator: VeSyncDataCoordinator,
) -> None:
def _setup_entities(devices, async_add_entities, coordinator):
"""Add entity."""
async_add_entities(
(

View File

@@ -50,7 +50,7 @@ async def async_setup_entry(
coordinator = hass.data[DOMAIN][VS_COORDINATOR]
@callback
def discover(devices: list[VeSyncBaseDevice]) -> None:
def discover(devices):
"""Add new devices to platform."""
_setup_entities(devices, async_add_entities, coordinator)
@@ -66,9 +66,9 @@ async def async_setup_entry(
@callback
def _setup_entities(
devices: list[VeSyncBaseDevice],
async_add_entities: AddConfigEntryEntitiesCallback,
async_add_entities,
coordinator: VeSyncDataCoordinator,
) -> None:
):
"""Check if device is fan and add entity."""
async_add_entities(

View File

@@ -53,7 +53,7 @@ async def async_setup_entry(
coordinator = hass.data[DOMAIN][VS_COORDINATOR]
@callback
def discover(devices: list[VeSyncBaseDevice]) -> None:
def discover(devices):
"""Add new devices to platform."""
_setup_entities(devices, async_add_entities, coordinator)
@@ -73,7 +73,7 @@ def _setup_entities(
devices: list[VeSyncBaseDevice],
async_add_entities: AddConfigEntryEntitiesCallback,
coordinator: VeSyncDataCoordinator,
) -> None:
):
"""Add humidifier entities."""
async_add_entities(VeSyncHumidifierHA(dev, coordinator) for dev in devices)

View File

@@ -38,7 +38,7 @@ async def async_setup_entry(
coordinator = hass.data[DOMAIN][VS_COORDINATOR]
@callback
def discover(devices: list[VeSyncBaseDevice]) -> None:
def discover(devices):
"""Add new devices to platform."""
_setup_entities(devices, async_add_entities, coordinator)
@@ -54,9 +54,9 @@ async def async_setup_entry(
@callback
def _setup_entities(
devices: list[VeSyncBaseDevice],
async_add_entities: AddConfigEntryEntitiesCallback,
async_add_entities,
coordinator: VeSyncDataCoordinator,
) -> None:
):
"""Check if device is a light and add entity."""
entities: list[VeSyncBaseLightHA] = []
for dev in devices:

View File

@@ -61,7 +61,7 @@ async def async_setup_entry(
coordinator = hass.data[DOMAIN][VS_COORDINATOR]
@callback
def discover(devices: list[VeSyncBaseDevice]) -> None:
def discover(devices):
"""Add new devices to platform."""
_setup_entities(devices, async_add_entities, coordinator)
@@ -79,7 +79,7 @@ def _setup_entities(
devices: list[VeSyncBaseDevice],
async_add_entities: AddConfigEntryEntitiesCallback,
coordinator: VeSyncDataCoordinator,
) -> None:
):
"""Add number entities."""
async_add_entities(

View File

@@ -115,7 +115,7 @@ async def async_setup_entry(
coordinator = hass.data[DOMAIN][VS_COORDINATOR]
@callback
def discover(devices: list[VeSyncBaseDevice]) -> None:
def discover(devices):
"""Add new devices to platform."""
_setup_entities(devices, async_add_entities, coordinator)
@@ -133,7 +133,7 @@ def _setup_entities(
devices: list[VeSyncBaseDevice],
async_add_entities: AddConfigEntryEntitiesCallback,
coordinator: VeSyncDataCoordinator,
) -> None:
):
"""Add select entities."""
async_add_entities(

View File

@@ -162,7 +162,7 @@ async def async_setup_entry(
coordinator = hass.data[DOMAIN][VS_COORDINATOR]
@callback
def discover(devices: list[VeSyncBaseDevice]) -> None:
def discover(devices):
"""Add new devices to platform."""
_setup_entities(devices, async_add_entities, coordinator)
@@ -180,7 +180,7 @@ def _setup_entities(
devices: list[VeSyncBaseDevice],
async_add_entities: AddConfigEntryEntitiesCallback,
coordinator: VeSyncDataCoordinator,
) -> None:
):
"""Check if device is online and add entity."""
async_add_entities(

View File

@@ -77,7 +77,7 @@ async def async_setup_entry(
coordinator = hass.data[DOMAIN][VS_COORDINATOR]
@callback
def discover(devices: list[VeSyncBaseDevice]) -> None:
def discover(devices):
"""Add new devices to platform."""
_setup_entities(devices, async_add_entities, coordinator)
@@ -93,9 +93,9 @@ async def async_setup_entry(
@callback
def _setup_entities(
devices: list[VeSyncBaseDevice],
async_add_entities: AddConfigEntryEntitiesCallback,
async_add_entities,
coordinator: VeSyncDataCoordinator,
) -> None:
):
"""Check if device is online and add entity."""
async_add_entities(
VeSyncSwitchEntity(dev, description, coordinator)

View File

@@ -22,7 +22,7 @@ async def async_setup_entry(
coordinator = hass.data[DOMAIN][VS_COORDINATOR]
@callback
def discover(devices: list[VeSyncBaseDevice]) -> None:
def discover(devices):
"""Add new devices to platform."""
_setup_entities(devices, async_add_entities, coordinator)

View File

@@ -1,138 +0,0 @@
"""The WebRTC integration."""
from __future__ import annotations
from collections.abc import Callable, Iterable
from typing import Any
import voluptuous as vol
from webrtc_models import RTCIceServer
from homeassistant.components import websocket_api
from homeassistant.const import CONF_URL, CONF_USERNAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.core_config import (
CONF_CREDENTIAL,
CONF_ICE_SERVERS,
validate_stun_or_turn_url,
)
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType
from homeassistant.util.hass_dict import HassKey
__all__ = [
"async_get_ice_servers",
"async_register_ice_servers",
]
DOMAIN = "web_rtc"
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Required(CONF_ICE_SERVERS): vol.All(
cv.ensure_list,
[
vol.Schema(
{
vol.Required(CONF_URL): vol.All(
cv.ensure_list, [validate_stun_or_turn_url]
),
vol.Optional(CONF_USERNAME): cv.string,
vol.Optional(CONF_CREDENTIAL): cv.string,
}
)
],
)
}
)
},
extra=vol.ALLOW_EXTRA,
)
DATA_ICE_SERVERS_USER: HassKey[Iterable[RTCIceServer]] = HassKey(
"web_rtc_ice_servers_user"
)
DATA_ICE_SERVERS: HassKey[list[Callable[[], Iterable[RTCIceServer]]]] = HassKey(
"web_rtc_ice_servers"
)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the WebRTC integration."""
servers = [
RTCIceServer(
server[CONF_URL],
server.get(CONF_USERNAME),
server.get(CONF_CREDENTIAL),
)
for server in config.get(DOMAIN, {}).get(CONF_ICE_SERVERS, [])
]
if servers:
hass.data[DATA_ICE_SERVERS_USER] = servers
hass.data[DATA_ICE_SERVERS] = []
websocket_api.async_register_command(hass, ws_ice_servers)
return True
@callback
def async_register_ice_servers(
hass: HomeAssistant,
get_ice_server_fn: Callable[[], Iterable[RTCIceServer]],
) -> Callable[[], None]:
"""Register an ICE server.
The registering integration is responsible to implement caching if needed.
"""
servers = hass.data[DATA_ICE_SERVERS]
def remove() -> None:
servers.remove(get_ice_server_fn)
servers.append(get_ice_server_fn)
return remove
@callback
def async_get_ice_servers(hass: HomeAssistant) -> list[RTCIceServer]:
"""Return all registered ICE servers."""
servers: list[RTCIceServer] = []
if hass.config.webrtc.ice_servers:
servers.extend(hass.config.webrtc.ice_servers)
if DATA_ICE_SERVERS_USER in hass.data:
servers.extend(hass.data[DATA_ICE_SERVERS_USER])
if not servers:
servers = [
RTCIceServer(
urls=[
"stun:stun.home-assistant.io:3478",
"stun:stun.home-assistant.io:80",
]
),
]
for gen_servers in hass.data[DATA_ICE_SERVERS]:
servers.extend(gen_servers())
return servers
@websocket_api.websocket_command(
{
"type": "web_rtc/ice_servers",
}
)
@callback
def ws_ice_servers(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
"""Handle get WebRTC ICE servers websocket command."""
ice_servers = [server.to_dict() for server in async_get_ice_servers(hass)]
connection.send_result(msg["id"], ice_servers)

View File

@@ -1,8 +0,0 @@
{
"domain": "web_rtc",
"name": "WebRTC",
"codeowners": ["@home-assistant/core"],
"documentation": "https://www.home-assistant.io/integrations/web_rtc",
"integration_type": "system",
"quality_scale": "internal"
}

View File

@@ -2,7 +2,6 @@
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
import logging
from typing import Any, Self
@@ -25,14 +24,9 @@ from homeassistant.helpers.trigger import (
async_get_all_descriptions as async_get_all_trigger_descriptions,
)
from homeassistant.helpers.typing import ConfigType
from homeassistant.util.hass_dict import HassKey
_LOGGER = logging.getLogger(__name__)
FLATTENED_SERVICE_DESCRIPTIONS_CACHE: HassKey[
tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]]]
] = HassKey("websocket_automation_flat_service_description_cache")
@dataclass(slots=True, kw_only=True)
class _EntityFilter:
@@ -142,7 +136,7 @@ def _async_get_automation_components_for_target(
hass: HomeAssistant,
target_selection: ConfigType,
expand_group: bool,
component_descriptions: Mapping[str, Mapping[str, Any] | None],
component_descriptions: dict[str, dict[str, Any] | None],
) -> set[str]:
"""Get automation components (triggers/conditions/services) for a target.
@@ -223,29 +217,12 @@ async def async_get_services_for_target(
) -> set[str]:
"""Get services for a target."""
descriptions = await async_get_all_service_descriptions(hass)
def get_flattened_service_descriptions() -> dict[str, dict[str, Any]]:
"""Get flattened service descriptions, with caching."""
if FLATTENED_SERVICE_DESCRIPTIONS_CACHE in hass.data:
cached_descriptions, cached_flattened_descriptions = hass.data[
FLATTENED_SERVICE_DESCRIPTIONS_CACHE
]
# If the descriptions are the same, return the cached flattened version
if cached_descriptions is descriptions:
return cached_flattened_descriptions
# Flatten dicts to be keyed by domain.name to match trigger/condition format
flattened_descriptions = {
f"{domain}.{service_name}": desc
for domain, services in descriptions.items()
for service_name, desc in services.items()
}
hass.data[FLATTENED_SERVICE_DESCRIPTIONS_CACHE] = (
descriptions,
flattened_descriptions,
)
return flattened_descriptions
# Flatten dicts to be keyed by domain.name to match trigger/condition format
descriptions_flatten = {
f"{domain}.{service_name}": desc
for domain, services in descriptions.items()
for service_name, desc in services.items()
}
return _async_get_automation_components_for_target(
hass, target_selector, expand_group, get_flattened_service_descriptions()
hass, target_selector, expand_group, descriptions_flatten
)

View File

@@ -19,8 +19,6 @@ from pythonxbox.api.provider.smartglass.models import (
from homeassistant.components.media_player import BrowseMedia, MediaClass, MediaType
from .entity import to_https
class MediaTypeDetails(NamedTuple):
"""Details for media type."""
@@ -153,5 +151,5 @@ def _find_media_image(images: list[Image]) -> str | None:
if match := next(
(image for image in images if image.image_purpose == purpose), None
):
return to_https(match.uri)
return f"https:{match.uri}" if match.uri.startswith("/") else match.uri
return None

View File

@@ -151,15 +151,6 @@ def check_deprecated_entity(
return False
def to_https(image_url: str) -> str:
"""Convert image URLs to secure URLs."""
url = URL(image_url)
if url.host == "images-eds.xboxlive.com":
url = url.with_host("images-eds-ssl.xboxlive.com")
return str(url.with_scheme("https"))
def profile_pic(person: Person, _: Title | None = None) -> str | None:
"""Return the gamer pic."""
@@ -169,4 +160,9 @@ def profile_pic(person: Person, _: Title | None = None) -> str | None:
# to point to the correct image, with the correct domain and certificate.
# We need to also remove the 'mode=Padding' query because with it,
# it results in an error 400.
return str(URL(to_https(person.display_pic_raw)).without_query_params("mode"))
url = URL(person.display_pic_raw)
if url.host == "images-eds.xboxlive.com":
url = url.with_host("images-eds-ssl.xboxlive.com").with_scheme("https")
query = dict(url.query)
query.pop("mode", None)
return str(url.with_query(query))

View File

@@ -22,7 +22,6 @@ from homeassistant.util import dt as dt_util
from .binary_sensor import profile_pic
from .const import DOMAIN
from .coordinator import XboxConfigEntry
from .entity import to_https
_LOGGER = logging.getLogger(__name__)
@@ -656,6 +655,6 @@ def game_thumbnail(images: list[Image]) -> str | None:
(i for i in images if i.type == img_type),
None,
):
return to_https(match.url)
return match.url
return None

View File

@@ -34,7 +34,6 @@ from .entity import (
XboxBaseEntity,
XboxBaseEntityDescription,
check_deprecated_entity,
to_https,
)
PARALLEL_UPDATES = 0
@@ -143,8 +142,8 @@ def title_logo(_: Person, title: Title | None) -> str | None:
"""Get the game logo."""
return (
next((to_https(i.url) for i in title.images if i.type == "Tile"), None)
or next((to_https(i.url) for i in title.images if i.type == "Logo"), None)
next((i.url for i in title.images if i.type == "Tile"), None)
or next((i.url for i in title.images if i.type == "Logo"), None)
if title and title.images
else None
)

View File

@@ -44,11 +44,6 @@ GATEWAY_SETTINGS = vol.Schema(
}
)
ERROR_STEP_PLACEHOLDERS = {
"tutorial_url": "https://www.domoticz.com/wiki/Xiaomi_Gateway_(Aqara)#Adding_the_Xiaomi_Gateway_to_Domoticz",
"invalid_host_url": "https://www.home-assistant.io/integrations/xiaomi_aqara/#connection-problem",
}
class XiaomiAqaraFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle a Xiaomi Aqara config flow."""
@@ -71,12 +66,7 @@ class XiaomiAqaraFlowHandler(ConfigFlow, domain=DOMAIN):
if (self.host is None and self.sid is None) or errors:
schema = GATEWAY_CONFIG_HOST
return self.async_show_form(
step_id="user",
data_schema=schema,
errors=errors,
description_placeholders=ERROR_STEP_PLACEHOLDERS,
)
return self.async_show_form(step_id="user", data_schema=schema, errors=errors)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
@@ -159,10 +149,7 @@ class XiaomiAqaraFlowHandler(ConfigFlow, domain=DOMAIN):
)
return self.async_show_form(
step_id="select",
data_schema=select_schema,
errors=errors,
description_placeholders=ERROR_STEP_PLACEHOLDERS,
step_id="select", data_schema=select_schema, errors=errors
)
async def async_step_zeroconf(
@@ -249,8 +236,5 @@ class XiaomiAqaraFlowHandler(ConfigFlow, domain=DOMAIN):
errors[CONF_KEY] = "invalid_key"
return self.async_show_form(
step_id="settings",
data_schema=GATEWAY_SETTINGS,
errors=errors,
description_placeholders=ERROR_STEP_PLACEHOLDERS,
step_id="settings", data_schema=GATEWAY_SETTINGS, errors=errors
)

View File

@@ -7,7 +7,7 @@
},
"error": {
"discovery_error": "Failed to discover a Xiaomi Aqara Gateway, try using the IP of the device running Home Assistant as interface",
"invalid_host": "Invalid hostname or IP address, see {invalid_host_url}",
"invalid_host": "Invalid hostname or IP address, see https://www.home-assistant.io/integrations/xiaomi_aqara/#connection-problem",
"invalid_interface": "Invalid network interface",
"invalid_key": "Invalid Gateway key",
"invalid_mac": "Invalid MAC address"
@@ -25,7 +25,7 @@
"key": "The key of your Gateway",
"name": "Name of the Gateway"
},
"description": "The key (password) can be retrieved using this tutorial: {tutorial_url}. If the key is not provided only sensors will be accessible",
"description": "The key (password) can be retrieved using this tutorial: https://www.domoticz.com/wiki/Xiaomi_Gateway_(Aqara)#Adding_the_Xiaomi_Gateway_to_Domoticz. If the key is not provided only sensors will be accessible",
"title": "Optional settings"
},
"user": {

View File

@@ -380,12 +380,7 @@ def _async_setup_services(hass: HomeAssistant):
SERVICE_SET_MODE, SERVICE_SCHEMA_SET_MODE, "async_set_mode"
)
platform.async_register_entity_service(
SERVICE_START_FLOW,
SERVICE_SCHEMA_START_FLOW,
_async_start_flow,
description_placeholders={
"flow_objects_urls": "https://yeelight.readthedocs.io/en/stable/yeelight.html#flow-objects"
},
SERVICE_START_FLOW, SERVICE_SCHEMA_START_FLOW, _async_start_flow
)
platform.async_register_entity_service(
SERVICE_SET_COLOR_SCENE, SERVICE_SCHEMA_SET_COLOR_SCENE, _async_set_color_scene
@@ -402,9 +397,6 @@ def _async_setup_services(hass: HomeAssistant):
SERVICE_SET_COLOR_FLOW_SCENE,
SERVICE_SCHEMA_SET_COLOR_FLOW_SCENE,
_async_set_color_flow_scene,
description_placeholders={
"examples_url": "https://yeelight.readthedocs.io/en/stable/flow.html"
},
)
platform.async_register_entity_service(
SERVICE_SET_AUTO_DELAY_OFF_SCENE,

View File

@@ -102,7 +102,7 @@
"name": "Count"
},
"transitions": {
"description": "Array of transitions, for desired effect. Examples {examples_url}.",
"description": "Array of transitions, for desired effect. Examples https://yeelight.readthedocs.io/en/stable/flow.html.",
"name": "Transitions"
}
},
@@ -171,7 +171,7 @@
"name": "Set music mode"
},
"start_flow": {
"description": "Starts a custom flow, using transitions from {flow_objects_urls}.",
"description": "Starts a custom flow, using transitions from https://yeelight.readthedocs.io/en/stable/yeelight.html#flow-objects.",
"fields": {
"action": {
"description": "[%key:component::yeelight::services::set_color_flow_scene::fields::action::description%]",

View File

@@ -306,9 +306,6 @@ class ZWaveServices:
has_at_least_one_node,
),
),
description_placeholders={
"api_docs_url": "https://zwave-js.github.io/node-zwave-js/#/api/CCs/index"
},
)
self._hass.services.async_register(

View File

@@ -400,11 +400,11 @@
"name": "Entity ID(s)"
},
"method_name": {
"description": "The name of the API method to call. Refer to the Z-Wave Command Class API documentation ({api_docs_url}) for available methods.",
"description": "The name of the API method to call. Refer to the Z-Wave Command Class API documentation (https://zwave-js.github.io/node-zwave-js/#/api/CCs/index) for available methods.",
"name": "Method name"
},
"parameters": {
"description": "A list of parameters to pass to the API method. Refer to the Z-Wave Command Class API documentation ({api_docs_url}) for parameters.",
"description": "A list of parameters to pass to the API method. Refer to the Z-Wave Command Class API documentation (https://zwave-js.github.io/node-zwave-js/#/api/CCs/index) for parameters.",
"name": "Parameters"
}
},

View File

@@ -3384,7 +3384,10 @@ class ConfigFlow(ConfigEntryBaseFlow):
last_step: bool | None = None,
preview: str | None = None,
) -> ConfigFlowResult:
"""Return the definition of a form to gather user input."""
"""Return the definition of a form to gather user input.
The step_id parameter is deprecated and will be removed in a future release.
"""
if self.source == SOURCE_REAUTH and "entry_id" in self.context:
# If the integration does not provide a name for the reauth title,
# we append it to the description placeholders.

View File

@@ -249,7 +249,7 @@ def _validate_currency(data: Any) -> Any:
raise
def validate_stun_or_turn_url(value: Any) -> str:
def _validate_stun_or_turn_url(value: Any) -> str:
"""Validate an URL."""
url_in = str(value)
url = urlparse(url_in)
@@ -331,7 +331,7 @@ CORE_CONFIG_SCHEMA = vol.All(
vol.Schema(
{
vol.Required(CONF_URL): vol.All(
cv.ensure_list, [validate_stun_or_turn_url]
cv.ensure_list, [_validate_stun_or_turn_url]
),
vol.Optional(CONF_USERNAME): cv.string,
vol.Optional(CONF_CREDENTIAL): cv.string,

View File

@@ -9,7 +9,6 @@ APPLICATION_CREDENTIALS = [
"ekeybionyx",
"electric_kiwi",
"fitbit",
"gentex_homelink",
"geocaching",
"google",
"google_assistant_sdk",

View File

@@ -239,7 +239,6 @@ FLOWS = {
"gdacs",
"generic",
"geniushub",
"gentex_homelink",
"geo_json_events",
"geocaching",
"geofency",

View File

@@ -2295,12 +2295,6 @@
"config_flow": true,
"iot_class": "local_polling"
},
"gentex_homelink": {
"name": "HomeLink",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_push"
},
"geo_json_events": {
"name": "GeoJSON",
"integration_type": "service",

5
requirements_all.txt generated
View File

@@ -1209,9 +1209,6 @@ home-assistant-frontend==20251203.0
# homeassistant.components.conversation
home-assistant-intents==2025.12.2
# homeassistant.components.gentex_homelink
homelink-integration-api==0.0.1
# homeassistant.components.homematicip_cloud
homematicip==2.4.0
@@ -2726,7 +2723,7 @@ renault-api==0.5.1
renson-endura-delta==1.7.2
# homeassistant.components.reolink
reolink-aio==0.17.1
reolink-aio==0.17.0
# homeassistant.components.idteck_prox
rfk101py==0.0.1

View File

@@ -1067,9 +1067,6 @@ home-assistant-frontend==20251203.0
# homeassistant.components.conversation
home-assistant-intents==2025.12.2
# homeassistant.components.gentex_homelink
homelink-integration-api==0.0.1
# homeassistant.components.homematicip_cloud
homematicip==2.4.0
@@ -2283,7 +2280,7 @@ renault-api==0.5.1
renson-endura-delta==1.7.2
# homeassistant.components.reolink
reolink-aio==0.17.1
reolink-aio==0.17.0
# homeassistant.components.rflink
rflink==0.0.67

View File

@@ -115,7 +115,6 @@ NO_IOT_CLASS = [
"tag",
"timer",
"trace",
"web_rtc",
"webhook",
"websocket_api",
"zone",

View File

@@ -2196,7 +2196,6 @@ NO_QUALITY_SCALE = [
"timer",
"trace",
"usage_prediction",
"web_rtc",
"webhook",
"websocket_api",
"zone",

View File

@@ -40,7 +40,6 @@ async def test_full_flow(
user_input={
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
CONF_ACCOUNT_NUMBER: ACCOUNT_NUMBER,
},
)
@@ -75,7 +74,6 @@ async def test_already_configured(
user_input={
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
CONF_ACCOUNT_NUMBER: ACCOUNT_NUMBER,
},
)
@@ -109,7 +107,6 @@ async def test_auth_recover_exception(
user_input={
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
CONF_ACCOUNT_NUMBER: ACCOUNT_NUMBER,
},
)
@@ -126,7 +123,6 @@ async def test_auth_recover_exception(
user_input={
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
CONF_ACCOUNT_NUMBER: ACCOUNT_NUMBER,
},
)
@@ -168,7 +164,6 @@ async def test_account_recover_exception(
user_input={
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
CONF_ACCOUNT_NUMBER: ACCOUNT_NUMBER,
},
)

View File

@@ -8,6 +8,7 @@ import pytest
from webrtc_models import RTCIceCandidate, RTCIceCandidateInit, RTCIceServer
from homeassistant.components.camera import (
DATA_ICE_SERVERS,
Camera,
CameraWebRTCProvider,
StreamType,
@@ -16,10 +17,10 @@ from homeassistant.components.camera import (
WebRTCError,
WebRTCMessage,
WebRTCSendMessage,
async_register_ice_servers,
async_register_webrtc_provider,
get_camera_from_entity_id,
)
from homeassistant.components.web_rtc import async_register_ice_servers
from homeassistant.components.websocket_api import TYPE_RESULT
from homeassistant.core import HomeAssistant, callback
from homeassistant.core_config import async_process_ha_core_config
@@ -100,6 +101,89 @@ async def test_async_register_webrtc_provider_camera_not_loaded(
async_register_webrtc_provider(hass, SomeTestProvider())
@pytest.mark.usefixtures("mock_test_webrtc_cameras")
async def test_async_register_ice_server(
hass: HomeAssistant,
) -> None:
"""Test registering an ICE server."""
# Clear any existing ICE servers
hass.data[DATA_ICE_SERVERS].clear()
called = 0
@callback
def get_ice_servers() -> list[RTCIceServer]:
nonlocal called
called += 1
return [
RTCIceServer(urls="stun:example.com"),
RTCIceServer(urls="turn:example.com"),
]
unregister = async_register_ice_servers(hass, get_ice_servers)
assert not called
camera = get_camera_from_entity_id(hass, "camera.async")
config = camera.async_get_webrtc_client_configuration()
assert config.configuration.ice_servers == [
RTCIceServer(urls="stun:example.com"),
RTCIceServer(urls="turn:example.com"),
]
assert called == 1
# register another ICE server
called_2 = 0
@callback
def get_ice_servers_2() -> list[RTCIceServer]:
nonlocal called_2
called_2 += 1
return [
RTCIceServer(
urls=["stun:example2.com", "turn:example2.com"],
username="user",
credential="pass",
)
]
unregister_2 = async_register_ice_servers(hass, get_ice_servers_2)
config = camera.async_get_webrtc_client_configuration()
assert config.configuration.ice_servers == [
RTCIceServer(urls="stun:example.com"),
RTCIceServer(urls="turn:example.com"),
RTCIceServer(
urls=["stun:example2.com", "turn:example2.com"],
username="user",
credential="pass",
),
]
assert called == 2
assert called_2 == 1
# unregister the first ICE server
unregister()
config = camera.async_get_webrtc_client_configuration()
assert config.configuration.ice_servers == [
RTCIceServer(
urls=["stun:example2.com", "turn:example2.com"],
username="user",
credential="pass",
),
]
assert called == 2
assert called_2 == 2
# unregister the second ICE server
unregister_2()
config = camera.async_get_webrtc_client_configuration()
assert config.configuration.ice_servers == []
@pytest.mark.usefixtures("mock_test_webrtc_cameras")
async def test_ws_get_client_config(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator

View File

@@ -21,7 +21,7 @@
## Active Integrations
Built-in integrations: 22
Built-in integrations: 21
Custom integrations: 1
<details><summary>Built-in integrations</summary>
@@ -48,7 +48,6 @@
stt | Speech-to-text (STT)
system_health | System Health
tts | Text-to-speech (TTS)
web_rtc | WebRTC
webhook | Webhook
</details>
@@ -123,7 +122,7 @@
## Active Integrations
Built-in integrations: 22
Built-in integrations: 21
Custom integrations: 0
<details><summary>Built-in integrations</summary>
@@ -150,7 +149,6 @@
stt | Speech-to-text (STT)
system_health | System Health
tts | Text-to-speech (TTS)
web_rtc | WebRTC
webhook | Webhook
</details>

View File

@@ -5,9 +5,6 @@ from unittest.mock import AsyncMock, MagicMock, patch
from fressnapftracker import (
Device,
EnergySaving,
LedActivatable,
LedBrightness,
PhoneVerificationResponse,
Position,
SmsCodeResponse,
@@ -33,33 +30,6 @@ MOCK_USER_ID = 12345
MOCK_ACCESS_TOKEN = "mock_access_token"
MOCK_SERIAL_NUMBER = "ABC123456"
MOCK_DEVICE_TOKEN = "mock_device_token"
MOCK_TRACKER = Tracker(
name="Fluffy",
battery=85,
charging=False,
position=Position(
lat=52.520008,
lng=13.404954,
accuracy=10,
timestamp="2024-01-15T12:00:00Z",
),
tracker_settings=TrackerSettings(
generation="GPS Tracker 2.0",
features=TrackerFeatures(
flash_light=True, energy_saving_mode=True, live_tracking=True
),
),
led_brightness=LedBrightness(status="ok", value=50),
energy_saving=EnergySaving(status="ok", value=1),
deep_sleep=None,
led_activatable=LedActivatable(
has_led=True,
seen_recently=True,
nonempty_battery=True,
not_charging=True,
overall=True,
),
)
@pytest.fixture
@@ -72,6 +42,29 @@ def mock_setup_entry() -> Generator[AsyncMock]:
yield mock_setup_entry
@pytest.fixture
def mock_tracker() -> Tracker:
"""Create a mock Tracker object."""
return Tracker(
name="Fluffy",
battery=85,
charging=False,
position=Position(
lat=52.520008,
lng=13.404954,
accuracy=10,
timestamp="2024-01-15T12:00:00Z",
),
tracker_settings=TrackerSettings(
generation="GPS Tracker 2.0",
features=TrackerFeatures(live_tracking=True),
),
led_brightness=None,
deep_sleep=None,
led_activatable=None,
)
@pytest.fixture
def mock_tracker_no_position() -> Tracker:
"""Create a mock Tracker object without position."""
@@ -129,15 +122,13 @@ def mock_auth_client(mock_device: Device) -> Generator[MagicMock]:
@pytest.fixture
def mock_api_client() -> Generator[MagicMock]:
def mock_api_client(mock_tracker: Tracker) -> Generator[MagicMock]:
"""Mock the ApiClient."""
with patch(
"homeassistant.components.fressnapf_tracker.coordinator.ApiClient"
) as mock_api_client:
client = mock_api_client.return_value
client.get_tracker = AsyncMock(return_value=MOCK_TRACKER)
client.set_led_brightness = AsyncMock(return_value=None)
client.set_energy_saving = AsyncMock(return_value=None)
client.get_tracker = AsyncMock(return_value=mock_tracker)
yield client

View File

@@ -1,59 +0,0 @@
# serializer version: 1
# name: test_state_entity_device_snapshots[light.fluffy_flashlight-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'supported_color_modes': list([
<ColorMode.BRIGHTNESS: 'brightness'>,
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'light',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'light.fluffy_flashlight',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Flashlight',
'platform': 'fressnapf_tracker',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'led',
'unique_id': 'ABC123456_led_brightness_value',
'unit_of_measurement': None,
})
# ---
# name: test_state_entity_device_snapshots[light.fluffy_flashlight-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'brightness': 128,
'color_mode': <ColorMode.BRIGHTNESS: 'brightness'>,
'friendly_name': 'Fluffy Flashlight',
'supported_color_modes': list([
<ColorMode.BRIGHTNESS: 'brightness'>,
]),
'supported_features': <LightEntityFeature: 0>,
}),
'context': <ANY>,
'entity_id': 'light.fluffy_flashlight',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---

View File

@@ -1,50 +0,0 @@
# serializer version: 1
# name: test_state_entity_device_snapshots[switch.fluffy_sleep_mode-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'switch.fluffy_sleep_mode',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SwitchDeviceClass.SWITCH: 'switch'>,
'original_icon': None,
'original_name': 'Sleep mode',
'platform': 'fressnapf_tracker',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'energy_saving',
'unique_id': 'ABC123456_energy_saving',
'unit_of_measurement': None,
})
# ---
# name: test_state_entity_device_snapshots[switch.fluffy_sleep_mode-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'switch',
'friendly_name': 'Fluffy Sleep mode',
}),
'context': <ANY>,
'entity_id': 'switch.fluffy_sleep_mode',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---

View File

@@ -1,174 +0,0 @@
"""Test the Fressnapf Tracker light platform."""
from collections.abc import AsyncGenerator
from unittest.mock import MagicMock, patch
from fressnapftracker import Tracker, TrackerFeatures, TrackerSettings
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
DOMAIN as LIGHT_DOMAIN,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
)
from homeassistant.const import ATTR_ENTITY_ID, STATE_ON, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry, snapshot_platform
TRACKER_NO_LED = Tracker(
name="Fluffy",
battery=0,
charging=False,
position=None,
tracker_settings=TrackerSettings(
generation="GPS Tracker 2.0",
features=TrackerFeatures(flash_light=False, live_tracking=True),
),
)
@pytest.fixture(autouse=True)
async def platforms() -> AsyncGenerator[None]:
"""Return the platforms to be loaded for this test."""
with patch(
"homeassistant.components.fressnapf_tracker.PLATFORMS", [Platform.LIGHT]
):
yield
@pytest.mark.usefixtures("init_integration")
async def test_state_entity_device_snapshots(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
) -> None:
"""Test light entity is created correctly."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.usefixtures("mock_auth_client")
async def test_not_added_when_no_led(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
mock_api_client: MagicMock,
) -> None:
"""Test light entity is created correctly."""
mock_api_client.get_tracker.return_value = TRACKER_NO_LED
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
assert len(entity_entries) == 0
@pytest.mark.usefixtures("init_integration")
async def test_turn_on(
hass: HomeAssistant,
mock_api_client: MagicMock,
) -> None:
"""Test turning the light on."""
entity_id = "light.fluffy_flashlight"
state = hass.states.get(entity_id)
assert state is not None
assert state.state == STATE_ON
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_api_client.set_led_brightness.assert_called_once_with(100)
@pytest.mark.usefixtures("init_integration")
async def test_turn_on_with_brightness(
hass: HomeAssistant,
mock_api_client: MagicMock,
) -> None:
"""Test turning the light on with brightness."""
entity_id = "light.fluffy_flashlight"
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS: 128},
blocking=True,
)
# 128/255 * 100 = 50
mock_api_client.set_led_brightness.assert_called_once_with(50)
@pytest.mark.usefixtures("init_integration")
async def test_turn_off(
hass: HomeAssistant,
mock_api_client: MagicMock,
) -> None:
"""Test turning the light off."""
entity_id = "light.fluffy_flashlight"
state = hass.states.get(entity_id)
assert state is not None
assert state.state == STATE_ON
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_api_client.set_led_brightness.assert_called_once_with(0)
@pytest.mark.parametrize(
"activatable_parameter",
[
"seen_recently",
"nonempty_battery",
"not_charging",
],
)
@pytest.mark.usefixtures("mock_auth_client")
async def test_turn_on_led_not_activatable(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_api_client: MagicMock,
activatable_parameter: str,
) -> None:
"""Test turning on the light when LED is not activatable raises."""
setattr(
mock_api_client.get_tracker.return_value.led_activatable,
activatable_parameter,
False,
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entity_id = "light.fluffy_flashlight"
with pytest.raises(HomeAssistantError, match="The flashlight cannot be activated"):
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_api_client.set_led_brightness.assert_not_called()

View File

@@ -1,114 +0,0 @@
"""Test the Fressnapf Tracker switch platform."""
from collections.abc import AsyncGenerator
from unittest.mock import MagicMock, patch
from fressnapftracker import Tracker, TrackerFeatures, TrackerSettings
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.switch import (
DOMAIN as SWITCH_DOMAIN,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
)
from homeassistant.const import ATTR_ENTITY_ID, STATE_ON, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry, snapshot_platform
TRACKER_NO_ENERGY_SAVING_MODE = Tracker(
name="Fluffy",
battery=0,
charging=False,
position=None,
tracker_settings=TrackerSettings(
generation="GPS Tracker 2.0",
features=TrackerFeatures(energy_saving_mode=False),
),
)
@pytest.fixture(autouse=True)
async def platforms() -> AsyncGenerator[None]:
"""Return the platforms to be loaded for this test."""
with patch(
"homeassistant.components.fressnapf_tracker.PLATFORMS", [Platform.SWITCH]
):
yield
@pytest.mark.usefixtures("init_integration")
async def test_state_entity_device_snapshots(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
) -> None:
"""Test switch entity is created correctly."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.usefixtures("mock_auth_client")
async def test_not_added_when_no_energy_saving_mode(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
mock_api_client: MagicMock,
) -> None:
"""Test switch entity is created correctly."""
mock_api_client.get_tracker.return_value = TRACKER_NO_ENERGY_SAVING_MODE
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
assert len(entity_entries) == 0
@pytest.mark.usefixtures("init_integration")
async def test_turn_on(
hass: HomeAssistant,
mock_api_client: MagicMock,
) -> None:
"""Test turning the switch on."""
entity_id = "switch.fluffy_sleep_mode"
state = hass.states.get(entity_id)
assert state is not None
assert state.state == STATE_ON
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_api_client.set_energy_saving.assert_called_once_with(True)
@pytest.mark.usefixtures("init_integration")
async def test_turn_off(
hass: HomeAssistant,
mock_api_client: MagicMock,
) -> None:
"""Test turning the switch off."""
entity_id = "switch.fluffy_sleep_mode"
state = hass.states.get(entity_id)
assert state is not None
assert state.state == STATE_ON
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_api_client.set_energy_saving.assert_called_once_with(False)

View File

@@ -1 +0,0 @@
"""Tests for the homelink integration."""

View File

@@ -1,91 +0,0 @@
"""Test the homelink config flow."""
from unittest.mock import patch
import botocore.exceptions
from homeassistant import config_entries
from homeassistant.components.gentex_homelink.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
async def test_show_user_form(hass: HomeAssistant) -> None:
"""Test that the user set up form is served."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
assert result["step_id"] == "user"
assert result["type"] == FlowResultType.FORM
async def test_full_flow(hass: HomeAssistant) -> None:
"""Check full flow."""
with patch(
"homeassistant.components.gentex_homelink.config_flow.SRPAuth"
) as MockSRPAuth:
instance = MockSRPAuth.return_value
instance.async_get_access_token.return_value = {
"AuthenticationResult": {
"AccessToken": "access",
"RefreshToken": "refresh",
"TokenType": "bearer",
"ExpiresIn": 3600,
}
}
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"email": "test@test.com", "password": "SomePassword"},
)
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["data"]
assert result["data"]["token"]
assert result["data"]["token"]["access_token"] == "access"
assert result["data"]["token"]["refresh_token"] == "refresh"
assert result["data"]["token"]["expires_in"] == 3600
assert result["data"]["token"]["expires_at"]
async def test_boto_error(hass: HomeAssistant) -> None:
"""Test exceptions from boto are handled correctly."""
with patch(
"homeassistant.components.gentex_homelink.config_flow.SRPAuth"
) as MockSRPAuth:
instance = MockSRPAuth.return_value
instance.async_get_access_token.side_effect = botocore.exceptions.ClientError(
{"Error": {}}, "Some operation"
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"email": "test@test.com", "password": "SomePassword"},
)
assert len(hass.config_entries.async_entries(DOMAIN)) == 0
async def test_generic_error(hass: HomeAssistant) -> None:
"""Test exceptions from boto are handled correctly."""
with patch(
"homeassistant.components.gentex_homelink.config_flow.SRPAuth"
) as MockSRPAuth:
instance = MockSRPAuth.return_value
instance.async_get_access_token.side_effect = Exception("Some error")
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"email": "test@test.com", "password": "SomePassword"},
)
assert len(hass.config_entries.async_entries(DOMAIN)) == 0

View File

@@ -1,160 +0,0 @@
"""Tests for the homelink coordinator."""
import asyncio
import time
from unittest.mock import patch
from homelink.model.button import Button
from homelink.model.device import Device
import pytest
from homeassistant.components.gentex_homelink import async_setup_entry
from homeassistant.components.gentex_homelink.const import EVENT_PRESSED
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
import homeassistant.helpers.entity_registry as er
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import ClientSessionGenerator
DOMAIN = "gentex_homelink"
deviceInst = Device(id="TestDevice", name="TestDevice")
deviceInst.buttons = [
Button(id="Button 1", name="Button 1", device=deviceInst),
Button(id="Button 2", name="Button 2", device=deviceInst),
Button(id="Button 3", name="Button 3", device=deviceInst),
]
async def test_get_state_updates(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test state updates.
Tests that get_state calls are called by home assistant, and the homeassistant components respond appropriately to the data returned.
"""
with patch(
"homeassistant.components.gentex_homelink.MQTTProvider", autospec=True
) as MockProvider:
instance = MockProvider.return_value
instance.discover.return_value = [deviceInst]
config_entry = MockConfigEntry(
domain=DOMAIN,
unique_id=None,
version=1,
data={
"auth_implementation": "gentex_homelink",
"token": {"expires_at": time.time() + 10000, "access_token": ""},
"last_update_id": None,
},
state=ConfigEntryState.LOADED,
)
config_entry.add_to_hass(hass)
result = await async_setup_entry(hass, config_entry)
# Assert configuration worked without errors
assert result
provider = config_entry.runtime_data.provider
state_data = {
"type": "state",
"data": {
"Button 1": {"requestId": "rid1", "timestamp": time.time()},
"Button 2": {"requestId": "rid2", "timestamp": time.time()},
"Button 3": {"requestId": "rid3", "timestamp": time.time()},
},
}
# Test successful setup and first data fetch. The buttons should be unknown at the start
await hass.async_block_till_done(wait_background_tasks=True)
states = hass.states.async_all()
assert states, "No states were loaded"
assert all(state != STATE_UNAVAILABLE for state in states), (
"At least one state was not initialized as STATE_UNAVAILABLE"
)
buttons_unknown = [s.state == "unknown" for s in states]
assert buttons_unknown and all(buttons_unknown), (
"At least one button state was not initialized to unknown"
)
provider.listen.mock_calls[0].args[0](None, state_data)
await hass.async_block_till_done(wait_background_tasks=True)
await asyncio.gather(*asyncio.all_tasks() - {asyncio.current_task()})
await asyncio.sleep(0.01)
states = hass.states.async_all()
assert all(state != STATE_UNAVAILABLE for state in states), (
"Some button became unavailable"
)
buttons_pressed = [s.attributes["event_type"] == EVENT_PRESSED for s in states]
assert buttons_pressed and all(buttons_pressed), (
"At least one button was not pressed"
)
async def test_request_sync(hass: HomeAssistant) -> None:
"""Test that the config entry is reloaded when a requestSync request is sent."""
updatedDeviceInst = Device(id="TestDevice", name="TestDevice")
updatedDeviceInst.buttons = [
Button(id="Button 1", name="New Button 1", device=deviceInst),
Button(id="Button 2", name="New Button 2", device=deviceInst),
Button(id="Button 3", name="New Button 3", device=deviceInst),
]
with patch(
"homeassistant.components.gentex_homelink.MQTTProvider", autospec=True
) as MockProvider:
instance = MockProvider.return_value
instance.discover.side_effect = [[deviceInst], [updatedDeviceInst]]
config_entry = MockConfigEntry(
domain=DOMAIN,
unique_id=None,
version=1,
data={
"auth_implementation": "gentex_homelink",
"token": {"expires_at": time.time() + 10000, "access_token": ""},
"last_update_id": None,
},
state=ConfigEntryState.LOADED,
)
config_entry.add_to_hass(hass)
result = await async_setup_entry(hass, config_entry)
# Assert configuration worked without errors
assert result
# Check to see if the correct buttons names were loaded
comp = er.async_get(hass)
button_names = {"Button 1", "Button 2", "Button 3"}
registered_button_names = {b.original_name for b in comp.entities.values()}
assert button_names == registered_button_names, (
"Expect button names to be correct for the initial config"
)
provider = config_entry.runtime_data.provider
coordinator = config_entry.runtime_data.coordinator
with patch.object(
coordinator.hass.config_entries, "async_reload"
) as async_reload_mock:
# Mimic request sync event
state_data = {
"type": "requestSync",
}
# async reload should not be called yet
async_reload_mock.assert_not_called()
# Send the request sync
provider.listen.mock_calls[0].args[0](None, state_data)
# Wait for the request to be processed
await hass.async_block_till_done(wait_background_tasks=True)
await asyncio.gather(*asyncio.all_tasks() - {asyncio.current_task()})
await asyncio.sleep(0.01)
# Now async reload should have been called
async_reload_mock.assert_called()

View File

@@ -1,77 +0,0 @@
"""Test that the devices and entities are correctly configured."""
from unittest.mock import patch
from homelink.model.button import Button
from homelink.model.device import Device
import pytest
from homeassistant.components.gentex_homelink import async_setup_entry
from homeassistant.components.gentex_homelink.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
import homeassistant.helpers.device_registry as dr
import homeassistant.helpers.entity_registry as er
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import ClientSessionGenerator
CLIENT_ID = "1234"
CLIENT_SECRET = "5678"
TEST_CONFIG_ENTRY_ID = "ABC123"
"""Mock classes for testing."""
deviceInst = Device(id="TestDevice", name="TestDevice")
deviceInst.buttons = [
Button(id="1", name="Button 1", device=deviceInst),
Button(id="2", name="Button 2", device=deviceInst),
Button(id="3", name="Button 3", device=deviceInst),
]
@pytest.fixture
async def test_setup_config(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Setup config entry."""
with patch(
"homeassistant.components.gentex_homelink.MQTTProvider", autospec=True
) as MockProvider:
instance = MockProvider.return_value
instance.discover.return_value = [deviceInst]
config_entry = MockConfigEntry(
domain=DOMAIN,
unique_id=None,
version=1,
data={"auth_implementation": "gentex_homelink"},
state=ConfigEntryState.LOADED,
)
config_entry.add_to_hass(hass)
result = await async_setup_entry(hass, config_entry)
# Assert configuration worked without errors
assert result
async def test_device_registered(hass: HomeAssistant, test_setup_config) -> None:
"""Check if a device is registered."""
# Assert we got a device with the test ID
device_registry = dr.async_get(hass)
device = device_registry.async_get_device([(DOMAIN, "TestDevice")])
assert device
assert device.name == "TestDevice"
def test_entities_registered(hass: HomeAssistant, test_setup_config) -> None:
"""Check if the entities are registered."""
comp = er.async_get(hass)
button_names = {"Button 1", "Button 2", "Button 3"}
registered_button_names = {b.original_name for b in comp.entities.values()}
assert button_names == registered_button_names

View File

@@ -1,32 +0,0 @@
"""Test that the integration is initialized correctly."""
from unittest.mock import patch
from homeassistant.components import gentex_homelink
from homeassistant.components.gentex_homelink.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
async def test_load_unload_entry(
hass: HomeAssistant,
) -> None:
"""Test the entry can be loaded and unloaded."""
with patch("homeassistant.components.gentex_homelink.MQTTProvider", autospec=True):
entry = MockConfigEntry(
domain=DOMAIN,
unique_id=None,
version=1,
data={"auth_implementation": "gentex_homelink"},
)
entry.add_to_hass(hass)
assert await async_setup_component(hass, DOMAIN, {}) is True, (
"Component is not set up"
)
assert await gentex_homelink.async_unload_entry(hass, entry), (
"Component not unloaded"
)

View File

@@ -5,7 +5,11 @@ from unittest.mock import patch
import pytest
from homeassistant.components.media_player import MediaPlayerState
from homeassistant.components.media_player import (
ATTR_MEDIA_VOLUME_LEVEL,
ATTR_MEDIA_VOLUME_MUTED,
MediaPlayerState,
)
from homeassistant.const import ATTR_LABEL_ID, CONF_ENTITY_ID
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.setup import async_setup_component
@@ -60,6 +64,52 @@ async def test_media_player_triggers_gated_by_labs_flag(
) in caplog.text
def parametrize_muted_trigger_states() -> list[tuple[str, list[StateDescription]]]:
"""Parametrize states and expected service call counts.
Returns a list of tuples with (trigger, initial_state, list of states), where
states is a list of tuples (state to set, expected service call count).
"""
trigger = "media_player.muted"
return parametrize_trigger_states(
trigger=trigger,
target_states=[
# States with muted attribute
(MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_MUTED: True}),
# States with volume attribute
(MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_LEVEL: 0}),
# States with muted and volume attribute
(
MediaPlayerState.PLAYING,
{ATTR_MEDIA_VOLUME_LEVEL: 0, ATTR_MEDIA_VOLUME_MUTED: True},
),
(
MediaPlayerState.PLAYING,
{ATTR_MEDIA_VOLUME_LEVEL: 0, ATTR_MEDIA_VOLUME_MUTED: False},
),
(
MediaPlayerState.PLAYING,
{ATTR_MEDIA_VOLUME_LEVEL: 1, ATTR_MEDIA_VOLUME_MUTED: True},
),
],
other_states=[
# States with muted attribute
(MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_MUTED: False}),
(MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_MUTED: None}),
(MediaPlayerState.PLAYING, {}), # Missing attribute
# States with volume attribute
(MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_LEVEL: 1}),
(MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_LEVEL: None}),
(MediaPlayerState.PLAYING, {}), # Missing attribute
# States with muted and volume attribute
(
MediaPlayerState.PLAYING,
{ATTR_MEDIA_VOLUME_LEVEL: 1, ATTR_MEDIA_VOLUME_MUTED: False},
),
],
)
@pytest.mark.usefixtures("enable_experimental_triggers_conditions")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
@@ -122,6 +172,56 @@ async def test_media_player_state_trigger_behavior_any(
service_calls.clear()
@pytest.mark.usefixtures("enable_experimental_triggers_conditions")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("media_player"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_muted_trigger_states(),
],
)
async def test_media_player_state_attribute_trigger_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_media_players: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the media player state trigger fires when any media player state changes to a specific state."""
await async_setup_component(hass, "media player", {})
other_entity_ids = set(target_media_players) - {entity_id}
# Set all media players, including the tested media player, to the initial state
for eid in target_media_players:
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 if changing other media players 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"),
@@ -183,6 +283,60 @@ async def test_media_player_state_trigger_behavior_first(
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("media_player"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_muted_trigger_states(),
],
)
async def test_media_player_state_attribute_trigger_behavior_first(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_media_players: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the media player state trigger fires when the first media player state changes to a specific state."""
await async_setup_component(hass, "media_player", {})
other_entity_ids = set(target_media_players) - {entity_id}
# Set all media players, including the tested media player, to the initial state
for eid in target_media_players:
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 media players 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"),
@@ -241,3 +395,51 @@ async def test_media_player_state_trigger_behavior_last(
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
@pytest.mark.usefixtures("enable_experimental_triggers_conditions")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("media_player"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_muted_trigger_states(),
],
)
async def test_media_player_state_attribute_trigger_behavior_last(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_media_players: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the media player state trigger fires when the last media player state changes to a specific state."""
await async_setup_component(hass, "media_player", {})
other_entity_ids = set(target_media_players) - {entity_id}
# Set all media players, including the tested media player, to the initial state
for eid in target_media_players:
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

@@ -9,6 +9,11 @@ SETUP_ENTRY_PATCHER = patch(
"homeassistant.components.starlink.async_setup_entry", return_value=True
)
STATUS_DATA_SUCCESS_PATCHER = patch(
"homeassistant.components.starlink.coordinator.status_data",
return_value=json.loads(load_fixture("status_data_success.json", "starlink")),
)
LOCATION_DATA_SUCCESS_PATCHER = patch(
"homeassistant.components.starlink.coordinator.location_data",
return_value=json.loads(load_fixture("location_data_success.json", "starlink")),
@@ -19,12 +24,6 @@ SLEEP_DATA_SUCCESS_PATCHER = patch(
return_value=json.loads(load_fixture("sleep_data_success.json", "starlink")),
)
STATUS_DATA_TARGET = "homeassistant.components.starlink.coordinator.status_data"
STATUS_DATA_FIXTURE = json.loads(load_fixture("status_data_success.json", "starlink"))
STATUS_DATA_SUCCESS_PATCHER = patch(
STATUS_DATA_TARGET, return_value=STATUS_DATA_FIXTURE
)
HISTORY_STATS_SUCCESS_PATCHER = patch(
"homeassistant.components.starlink.coordinator.history_stats",
return_value=json.loads(load_fixture("history_stats_success.json", "starlink")),

View File

@@ -1,31 +1,20 @@
"""Tests Starlink integration init/unload."""
from copy import deepcopy
from datetime import datetime, timedelta
from unittest.mock import patch
from freezegun import freeze_time
from homeassistant.components.starlink.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_IP_ADDRESS
from homeassistant.core import HomeAssistant, State
from homeassistant.util import dt as dt_util
from .patchers import (
HISTORY_STATS_SUCCESS_PATCHER,
LOCATION_DATA_SUCCESS_PATCHER,
SLEEP_DATA_SUCCESS_PATCHER,
STATUS_DATA_FIXTURE,
STATUS_DATA_SUCCESS_PATCHER,
STATUS_DATA_TARGET,
)
from tests.common import (
MockConfigEntry,
async_fire_time_changed,
mock_restore_cache_with_extra_data,
)
from tests.common import MockConfigEntry, mock_restore_cache_with_extra_data
async def test_successful_entry(hass: HomeAssistant) -> None:
@@ -36,9 +25,9 @@ async def test_successful_entry(hass: HomeAssistant) -> None:
)
with (
STATUS_DATA_SUCCESS_PATCHER,
LOCATION_DATA_SUCCESS_PATCHER,
SLEEP_DATA_SUCCESS_PATCHER,
STATUS_DATA_SUCCESS_PATCHER,
HISTORY_STATS_SUCCESS_PATCHER,
):
entry.add_to_hass(hass)
@@ -59,9 +48,9 @@ async def test_unload_entry(hass: HomeAssistant) -> None:
)
with (
STATUS_DATA_SUCCESS_PATCHER,
LOCATION_DATA_SUCCESS_PATCHER,
SLEEP_DATA_SUCCESS_PATCHER,
STATUS_DATA_SUCCESS_PATCHER,
HISTORY_STATS_SUCCESS_PATCHER,
):
entry.add_to_hass(hass)
@@ -76,7 +65,7 @@ async def test_unload_entry(hass: HomeAssistant) -> None:
async def test_restore_cache_with_accumulation(hass: HomeAssistant) -> None:
"""Test Starlink accumulation."""
"""Test configuring Starlink."""
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_IP_ADDRESS: "1.2.3.4:0000"},
@@ -100,9 +89,9 @@ async def test_restore_cache_with_accumulation(hass: HomeAssistant) -> None:
)
with (
STATUS_DATA_SUCCESS_PATCHER,
LOCATION_DATA_SUCCESS_PATCHER,
SLEEP_DATA_SUCCESS_PATCHER,
STATUS_DATA_SUCCESS_PATCHER,
HISTORY_STATS_SUCCESS_PATCHER,
):
entry.add_to_hass(hass)
@@ -123,62 +112,3 @@ async def test_restore_cache_with_accumulation(hass: HomeAssistant) -> None:
await entry.runtime_data.async_refresh()
assert hass.states.get(entity_id).state == str(1 + 0.01572462736977)
async def test_last_restart_state(hass: HomeAssistant) -> None:
"""Test Starlink last restart state."""
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_IP_ADDRESS: "1.2.3.4:0000"},
)
entity_id = "sensor.starlink_last_restart"
utc_now = datetime.fromisoformat("2025-10-22T13:31:29+00:00")
with (
LOCATION_DATA_SUCCESS_PATCHER,
SLEEP_DATA_SUCCESS_PATCHER,
STATUS_DATA_SUCCESS_PATCHER,
HISTORY_STATS_SUCCESS_PATCHER,
):
with freeze_time(utc_now):
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == "2025-10-13T06:09:11+00:00"
with patch.object(entry.runtime_data, "always_update", return_value=True):
status_data = deepcopy(STATUS_DATA_FIXTURE)
status_data[0]["uptime"] = 804144
with (
freeze_time(utc_now + timedelta(seconds=5)),
patch(STATUS_DATA_TARGET, return_value=status_data),
):
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=5))
await hass.async_block_till_done(wait_background_tasks=True)
assert hass.states.get(entity_id).state == "2025-10-13T06:09:11+00:00"
status_data[0]["uptime"] = 804134
with (
freeze_time(utc_now + timedelta(seconds=10)),
patch(STATUS_DATA_TARGET, return_value=status_data),
):
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done(wait_background_tasks=True)
assert hass.states.get(entity_id).state == "2025-10-13T06:09:11+00:00"
status_data[0]["uptime"] = 100
with (
freeze_time(utc_now + timedelta(seconds=15)),
patch(STATUS_DATA_TARGET, return_value=status_data),
):
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=15))
await hass.async_block_till_done(wait_background_tasks=True)
assert hass.states.get(entity_id).state == "2025-10-22T13:30:04+00:00"

View File

@@ -600,270 +600,6 @@ async def test_legacy_deprecation(
assert "platform: template" not in issue.translation_placeholders["config"]
@pytest.mark.parametrize(
("domain", "config", "strings_to_check"),
[
(
"light",
{
"light": {
"platform": "template",
"lights": {
"garage_light_template": {
"friendly_name": "Garage Light Template",
"min_mireds_template": 153,
"max_mireds_template": 500,
"turn_on": [],
"turn_off": [],
"set_temperature": [],
"set_hs": [],
"set_level": [],
}
},
},
},
[
"turn_on: []",
"turn_off: []",
"set_temperature: []",
"set_hs: []",
"set_level: []",
],
),
(
"switch",
{
"switch": {
"platform": "template",
"switches": {
"my_switch": {
"friendly_name": "Switch Template",
"turn_on": [],
"turn_off": [],
}
},
},
},
[
"turn_on: []",
"turn_off: []",
],
),
(
"light",
{
"light": [
{
"platform": "template",
"lights": {
"atrium_lichterkette": {
"unique_id": "atrium_lichterkette",
"friendly_name": "Atrium Lichterkette",
"value_template": "{{ states('input_boolean.atrium_lichterkette_power') }}",
"level_template": "{% if is_state('input_boolean.atrium_lichterkette_power', 'off') %}\n 0\n{% else %}\n {{ states('input_number.atrium_lichterkette_brightness') | int * (255 / state_attr('input_number.atrium_lichterkette_brightness', 'max') | int) }}\n{% endif %}",
"effect_list_template": "{{ state_attr('input_select.atrium_lichterkette_mode', 'options') }}",
"effect_template": "'{{ states('input_select.atrium_lichterkette_mode')}}'",
"turn_on": [
{
"service": "button.press",
"target": {
"entity_id": "button.esphome_web_28a814_lichterkette_on"
},
},
{
"service": "input_boolean.turn_on",
"target": {
"entity_id": "input_boolean.atrium_lichterkette_power"
},
},
],
"turn_off": [
{
"service": "button.press",
"target": {
"entity_id": "button.esphome_web_28a814_lichterkette_off"
},
},
{
"service": "input_boolean.turn_off",
"target": {
"entity_id": "input_boolean.atrium_lichterkette_power"
},
},
],
"set_level": [
{
"variables": {
"scaled": "{{ (brightness / (255 / state_attr('input_number.atrium_lichterkette_brightness', 'max'))) | round | int }}",
"diff": "{{ scaled | int - states('input_number.atrium_lichterkette_brightness') | int }}",
"direction": "{{ 'dim' if diff | int < 0 else 'bright' }}",
}
},
{
"repeat": {
"count": "{{ diff | int | abs }}",
"sequence": [
{
"service": "button.press",
"target": {
"entity_id": "button.esphome_web_28a814_lichterkette_{{ direction }}"
},
},
{"delay": {"milliseconds": 500}},
],
}
},
{
"service": "input_number.set_value",
"data": {
"value": "{{ scaled }}",
"entity_id": "input_number.atrium_lichterkette_brightness",
},
},
],
"set_effect": [
{
"service": "button.press",
"target": {
"entity_id": "button.esphome_web_28a814_lichterkette_{{ effect }}"
},
}
],
}
},
}
]
},
[
"scaled: ",
"diff: ",
"direction: ",
],
),
(
"cover",
{
"cover": [
{
"platform": "template",
"covers": {
"large_garage_door": {
"device_class": "garage",
"friendly_name": "Large Garage Door",
"value_template": "{% if is_state('binary_sensor.large_garage_door', 'off') %}\n closed\n{% elif is_state('timer.large_garage_opening_timer', 'active') %}\n opening\n{% elif is_state('timer.large_garage_closing_timer', 'active') %} \n closing\n{% elif is_state('binary_sensor.large_garage_door', 'on') %}\n open\n{% endif %}\n",
"open_cover": [
{
"condition": "state",
"entity_id": "binary_sensor.large_garage_door",
"state": "off",
},
{
"action": "switch.turn_on",
"target": {
"entity_id": "switch.garage_door_relay_1"
},
},
{
"action": "timer.start",
"entity_id": "timer.large_garage_opening_timer",
},
],
"close_cover": [
{
"condition": "state",
"entity_id": "binary_sensor.large_garage_door",
"state": "on",
},
{
"action": "switch.turn_on",
"target": {
"entity_id": "switch.garage_door_relay_1"
},
},
{
"action": "timer.start",
"entity_id": "timer.large_garage_closing_timer",
},
],
"stop_cover": [
{
"action": "switch.turn_on",
"target": {
"entity_id": "switch.garage_door_relay_1"
},
},
{
"action": "timer.cancel",
"entity_id": "timer.large_garage_opening_timer",
},
{
"action": "timer.cancel",
"entity_id": "timer.large_garage_closing_timer",
},
],
}
},
}
]
},
["device_class: garage"],
),
(
"binary_sensor",
{
"binary_sensor": {
"platform": "template",
"sensors": {
"motion_sensor": {
"friendly_name": "Motion Sensor",
"device_class": "motion",
"value_template": "{{ is_state('sensor.motion_detector', 'on') }}",
}
},
},
},
["device_class: motion"],
),
(
"sensor",
{
"sensor": {
"platform": "template",
"sensors": {
"some_sensor": {
"friendly_name": "Sensor",
"device_class": "timestamp",
"value_template": "{{ now().isoformat() }}",
}
},
},
},
["device_class: timestamp"],
),
],
)
async def test_legacy_deprecation_with_unique_objects(
hass: HomeAssistant,
domain: str,
config: dict,
strings_to_check: list[str],
issue_registry: ir.IssueRegistry,
) -> None:
"""Test legacy configuration raises issue and unique objects are properly converted to valid configurations."""
await async_setup_component(hass, domain, config)
await hass.async_block_till_done()
assert len(issue_registry.issues) == 1
issue = next(iter(issue_registry.issues.values()))
assert issue.domain == "template"
assert issue.severity == ir.IssueSeverity.WARNING
assert issue.translation_placeholders is not None
for string in strings_to_check:
assert string in issue.translation_placeholders["config"]
@pytest.mark.parametrize(
("domain", "config"),
[

View File

@@ -182,11 +182,7 @@ async def _create_device(hass: HomeAssistant, mock_device_code: str) -> Customer
key: DeviceFunction(
code=key,
type=value["type"],
values=(
values
if isinstance(values := value["value"], str)
else json_dumps(values)
),
values=json_dumps(value["value"]),
)
for key, value in details["function"].items()
}
@@ -194,11 +190,7 @@ async def _create_device(hass: HomeAssistant, mock_device_code: str) -> Customer
key: DeviceStatusRange(
code=key,
type=value["type"],
values=(
values
if isinstance(values := value["value"], str)
else json_dumps(values)
),
values=json_dumps(value["value"]),
)
for key, value in details["status_range"].items()
}

View File

@@ -1,53 +0,0 @@
{
"endpoint": "https://apigw.tuyaeu.com",
"mqtt_connected": true,
"disabled_by": null,
"disabled_polling": false,
"name": "ITC-308-WIFI Thermostat",
"category": "wk",
"product_id": "B0eP8qYAdpUo4yR9",
"product_name": "ITC-308-WIFI Thermostat",
"online": true,
"sub": false,
"time_zone": "+01:00",
"active_time": "2022-02-08T10:49:39+00:00",
"create_time": "2022-02-08T10:49:39+00:00",
"update_time": "2022-02-08T10:49:39+00:00",
"function": {
"temp_unit_convert": {
"type": "Enum",
"value": "{\"range\":[\"c\",\"f\"]}"
},
"temp_set": {
"type": "Integer",
"value": "{\"unit\":\"\",\"min\":-400,\"max\":2120,\"scale\":1,\"step\":5}"
}
},
"status_range": {
"temp_unit_convert": {
"type": "Enum",
"value": "{\"range\":[\"c\",\"f\"]}"
},
"temp_current": {
"type": "Integer",
"value": "{\"unit\":\"\u2103\",\"min\":-500,\"max\":1200,\"scale\":1,\"step\":10}"
},
"temp_set": {
"type": "Integer",
"value": "{\"unit\":\"\",\"min\":-400,\"max\":2120,\"scale\":1,\"step\":5}"
},
"temp_current_f": {
"type": "Integer",
"value": "{\"unit\":\"\u2109\",\"min\":-500,\"max\":2480,\"scale\":1,\"step\":10}"
}
},
"status": {
"temp_unit_convert": "c",
"temp_current": 340,
"temp_set": 350,
"temp_current_f": 932
},
"set_up": true,
"support_local": true,
"warnings": null
}

View File

@@ -654,68 +654,6 @@
'state': 'off',
})
# ---
# name: test_platform_setup_and_discovery[climate.itc_308_wifi_thermostat-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'hvac_modes': list([
]),
'max_temp': 212.0,
'min_temp': -40.0,
'target_temp_step': 0.5,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'climate',
'entity_category': None,
'entity_id': 'climate.itc_308_wifi_thermostat',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'tuya',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <ClimateEntityFeature: 1>,
'translation_key': None,
'unique_id': 'tuya.9Ry4oUpdAYq8Pe0Bkw',
'unit_of_measurement': None,
})
# ---
# name: test_platform_setup_and_discovery[climate.itc_308_wifi_thermostat-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'current_temperature': 34.0,
'friendly_name': 'ITC-308-WIFI Thermostat',
'hvac_modes': list([
]),
'max_temp': 212.0,
'min_temp': -40.0,
'supported_features': <ClimateEntityFeature: 1>,
'target_temp_step': 0.5,
'temperature': 35.0,
}),
'context': <ANY>,
'entity_id': 'climate.itc_308_wifi_thermostat',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_platform_setup_and_discovery[climate.kabinet-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
@@ -1512,15 +1450,6 @@
'target_temp_step': 1.0,
})
# ---
# name: test_us_customary_system[climate.itc_308_wifi_thermostat]
ReadOnlyDict({
'current_temperature': 93,
'max_temp': 414,
'min_temp': -40,
'target_temp_step': 0.5,
'temperature': 95,
})
# ---
# name: test_us_customary_system[climate.kabinet]
ReadOnlyDict({
'current_temperature': 67,

View File

@@ -1301,37 +1301,6 @@
'via_device_id': None,
})
# ---
# name: test_device_registry[9Ry4oUpdAYq8Pe0Bkw]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'config_entries_subentries': <ANY>,
'configuration_url': None,
'connections': set({
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'tuya',
'9Ry4oUpdAYq8Pe0Bkw',
),
}),
'labels': set({
}),
'manufacturer': 'Tuya',
'model': 'ITC-308-WIFI Thermostat',
'model_id': 'B0eP8qYAdpUo4yR9',
'name': 'ITC-308-WIFI Thermostat',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': None,
'sw_version': None,
'via_device_id': None,
})
# ---
# name: test_device_registry[9c1vlsxoscm]
DeviceRegistryEntrySnapshot({
'area_id': None,

View File

@@ -1 +0,0 @@
"""Tests for the WebRTC integration."""

View File

@@ -1,250 +0,0 @@
"""Test the WebRTC integration."""
from webrtc_models import RTCIceServer
from homeassistant.components.web_rtc import (
async_get_ice_servers,
async_register_ice_servers,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.core_config import async_process_ha_core_config
from homeassistant.setup import async_setup_component
from tests.typing import WebSocketGenerator
async def test_async_setup(hass: HomeAssistant) -> None:
"""Test setting up the web_rtc integration."""
assert await async_setup_component(hass, "web_rtc", {})
await hass.async_block_till_done()
# Verify default ICE servers are registered
ice_servers = async_get_ice_servers(hass)
assert len(ice_servers) == 1
assert ice_servers[0].urls == [
"stun:stun.home-assistant.io:3478",
"stun:stun.home-assistant.io:80",
]
async def test_async_setup_custom_ice_servers_core(hass: HomeAssistant) -> None:
"""Test setting up web_rtc with custom ICE servers in config."""
await async_process_ha_core_config(
hass,
{"webrtc": {"ice_servers": [{"url": "stun:custom_stun_server:3478"}]}},
)
assert await async_setup_component(hass, "web_rtc", {})
await hass.async_block_till_done()
ice_servers = async_get_ice_servers(hass)
assert len(ice_servers) == 1
assert ice_servers[0].urls == ["stun:custom_stun_server:3478"]
async def test_async_setup_custom_ice_servers_integration(hass: HomeAssistant) -> None:
"""Test setting up web_rtc with custom ICE servers in config."""
assert await async_setup_component(
hass,
"web_rtc",
{
"web_rtc": {
"ice_servers": [
{"url": "stun:custom_stun_server:3478"},
{
"url": "stun:custom_stun_server:3478",
"credential": "mock-credential",
},
{
"url": "stun:custom_stun_server:3478",
"username": "mock-username",
},
{
"url": "stun:custom_stun_server:3478",
"credential": "mock-credential",
"username": "mock-username",
},
]
}
},
)
await hass.async_block_till_done()
ice_servers = async_get_ice_servers(hass)
assert ice_servers == [
RTCIceServer(
urls=["stun:custom_stun_server:3478"],
),
RTCIceServer(
urls=["stun:custom_stun_server:3478"],
credential="mock-credential",
),
RTCIceServer(
urls=["stun:custom_stun_server:3478"],
username="mock-username",
),
RTCIceServer(
urls=["stun:custom_stun_server:3478"],
username="mock-username",
credential="mock-credential",
),
]
async def test_async_setup_custom_ice_servers_core_and_integration(
hass: HomeAssistant,
) -> None:
"""Test setting up web_rtc with custom ICE servers in config."""
await async_process_ha_core_config(
hass,
{"webrtc": {"ice_servers": [{"url": "stun:custom_stun_server_core:3478"}]}},
)
assert await async_setup_component(
hass,
"web_rtc",
{
"web_rtc": {
"ice_servers": [{"url": "stun:custom_stun_server_integration:3478"}]
}
},
)
await hass.async_block_till_done()
ice_servers = async_get_ice_servers(hass)
assert ice_servers == [
RTCIceServer(
urls=["stun:custom_stun_server_core:3478"],
),
RTCIceServer(
urls=["stun:custom_stun_server_integration:3478"],
),
]
async def test_async_register_ice_servers(hass: HomeAssistant) -> None:
"""Test registering ICE servers."""
assert await async_setup_component(hass, "web_rtc", {})
await hass.async_block_till_done()
default_servers = async_get_ice_servers(hass)
called = 0
@callback
def get_ice_servers() -> list[RTCIceServer]:
nonlocal called
called += 1
return [
RTCIceServer(urls="stun:example.com"),
RTCIceServer(urls="turn:example.com"),
]
unregister = async_register_ice_servers(hass, get_ice_servers)
assert called == 0
# Getting ice servers should call the callback
ice_servers = async_get_ice_servers(hass)
assert called == 1
assert ice_servers == [
*default_servers,
RTCIceServer(urls="stun:example.com"),
RTCIceServer(urls="turn:example.com"),
]
# Unregister and verify servers are removed
unregister()
ice_servers = async_get_ice_servers(hass)
assert ice_servers == default_servers
async def test_multiple_ice_server_registrations(hass: HomeAssistant) -> None:
"""Test registering multiple ICE server providers."""
assert await async_setup_component(hass, "web_rtc", {})
await hass.async_block_till_done()
default_servers = async_get_ice_servers(hass)
@callback
def get_ice_servers_1() -> list[RTCIceServer]:
return [RTCIceServer(urls="stun:server1.com")]
@callback
def get_ice_servers_2() -> list[RTCIceServer]:
return [
RTCIceServer(
urls=["stun:server2.com", "turn:server2.com"],
username="user",
credential="pass",
)
]
unregister_1 = async_register_ice_servers(hass, get_ice_servers_1)
unregister_2 = async_register_ice_servers(hass, get_ice_servers_2)
ice_servers = async_get_ice_servers(hass)
assert ice_servers == [
*default_servers,
RTCIceServer(urls="stun:server1.com"),
RTCIceServer(
urls=["stun:server2.com", "turn:server2.com"],
username="user",
credential="pass",
),
]
# Unregister first provider
unregister_1()
ice_servers = async_get_ice_servers(hass)
assert ice_servers == [
*default_servers,
RTCIceServer(
urls=["stun:server2.com", "turn:server2.com"],
username="user",
credential="pass",
),
]
# Unregister second provider
unregister_2()
ice_servers = async_get_ice_servers(hass)
assert ice_servers == default_servers
async def test_ws_ice_servers_with_registered_servers(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""Test WebSocket ICE servers endpoint with registered servers."""
assert await async_setup_component(hass, "web_rtc", {})
await hass.async_block_till_done()
@callback
def get_ice_server() -> list[RTCIceServer]:
return [
RTCIceServer(
urls=["stun:example2.com", "turn:example2.com"],
username="user",
credential="pass",
)
]
async_register_ice_servers(hass, get_ice_server)
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "web_rtc/ice_servers"})
msg = await client.receive_json()
# Assert WebSocket response includes registered ICE servers
assert msg["type"] == "result"
assert msg["success"]
assert msg["result"] == [
{
"urls": [
"stun:stun.home-assistant.io:3478",
"stun:stun.home-assistant.io:80",
]
},
{
"urls": ["stun:example2.com", "turn:example2.com"],
"username": "user",
"credential": "pass",
},
]

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