mirror of
https://github.com/home-assistant/core.git
synced 2026-09-06 13:32:08 +01:00
Add Kiosker service platform (#171094)
This commit is contained in:
Martin Claesson
@@ -2,8 +2,14 @@
|
||||
|
||||
from homeassistant.const import Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
from .const import DOMAIN
|
||||
from .coordinator import KioskerConfigEntry, KioskerDataUpdateCoordinator
|
||||
from .services import async_setup_services
|
||||
|
||||
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
|
||||
|
||||
_PLATFORMS: list[Platform] = [
|
||||
Platform.BINARY_SENSOR,
|
||||
@@ -13,6 +19,12 @@ _PLATFORMS: list[Platform] = [
|
||||
]
|
||||
|
||||
|
||||
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
"""Set up the Kiosker integration."""
|
||||
async_setup_services(hass)
|
||||
return True
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: KioskerConfigEntry) -> bool:
|
||||
"""Set up Kiosker from a config entry."""
|
||||
|
||||
|
||||
@@ -8,3 +8,16 @@ POLL_INTERVAL = 15
|
||||
DEFAULT_SSL = False
|
||||
DEFAULT_SSL_VERIFY = False
|
||||
REFRESH_DELAY = 0.5
|
||||
|
||||
# Service attribute keys
|
||||
ATTR_URL = "url"
|
||||
ATTR_VISIBLE = "visible"
|
||||
ATTR_TEXT = "text"
|
||||
ATTR_BACKGROUND = "background"
|
||||
ATTR_FOREGROUND = "foreground"
|
||||
ATTR_EXPIRE = "expire"
|
||||
ATTR_DISMISSIBLE = "dismissible"
|
||||
ATTR_BUTTON_BACKGROUND = "button_background"
|
||||
ATTR_BUTTON_FOREGROUND = "button_foreground"
|
||||
ATTR_BUTTON_TEXT = "button_text"
|
||||
ATTR_SOUND = "sound"
|
||||
|
||||
@@ -65,5 +65,13 @@
|
||||
"default": "mdi:power-sleep"
|
||||
}
|
||||
}
|
||||
},
|
||||
"services": {
|
||||
"navigate_url": {
|
||||
"service": "mdi:web"
|
||||
},
|
||||
"set_blackout": {
|
||||
"service": "mdi:monitor-off"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,23 +1,17 @@
|
||||
rules:
|
||||
# Bronze
|
||||
action-setup:
|
||||
status: exempt
|
||||
comment: Integration does not register custom actions
|
||||
action-setup: done
|
||||
appropriate-polling: done
|
||||
brands: done
|
||||
common-modules: done
|
||||
config-flow-test-coverage: done
|
||||
config-flow: done
|
||||
dependency-transparency: done
|
||||
docs-actions:
|
||||
status: exempt
|
||||
comment: Integration does not provide custom actions to document
|
||||
docs-actions: done
|
||||
docs-high-level-description: done
|
||||
docs-installation-instructions: done
|
||||
docs-removal-instructions: done
|
||||
entity-event-setup:
|
||||
status: exempt
|
||||
comment: Integration is polling-only and does not subscribe to external events
|
||||
entity-event-setup: done
|
||||
entity-unique-id: done
|
||||
has-entity-name: done
|
||||
runtime-data: done
|
||||
@@ -26,9 +20,7 @@ rules:
|
||||
unique-config-entry: done
|
||||
|
||||
# Silver
|
||||
action-exceptions:
|
||||
status: exempt
|
||||
comment: Integration does not provide custom actions
|
||||
action-exceptions: done
|
||||
config-entry-unloading: done
|
||||
docs-configuration-parameters: done
|
||||
docs-installation-parameters: done
|
||||
|
||||
@@ -0,0 +1,168 @@
|
||||
"""Services for the Kiosker integration."""
|
||||
|
||||
from collections.abc import Awaitable, Callable, Coroutine
|
||||
import functools
|
||||
from typing import Any
|
||||
|
||||
from kiosker import (
|
||||
AuthenticationError,
|
||||
BadRequestError,
|
||||
Blackout,
|
||||
ConnectionError,
|
||||
IPAuthenticationError,
|
||||
TLSVerificationError,
|
||||
)
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.const import ATTR_DEVICE_ID, ATTR_ICON
|
||||
from homeassistant.core import HomeAssistant, ServiceCall, ServiceResponse, callback
|
||||
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
|
||||
from homeassistant.helpers import (
|
||||
config_validation as cv,
|
||||
device_registry as dr,
|
||||
selector,
|
||||
)
|
||||
|
||||
from .const import (
|
||||
ATTR_BACKGROUND,
|
||||
ATTR_BUTTON_BACKGROUND,
|
||||
ATTR_BUTTON_FOREGROUND,
|
||||
ATTR_BUTTON_TEXT,
|
||||
ATTR_DISMISSIBLE,
|
||||
ATTR_EXPIRE,
|
||||
ATTR_FOREGROUND,
|
||||
ATTR_SOUND,
|
||||
ATTR_TEXT,
|
||||
ATTR_URL,
|
||||
ATTR_VISIBLE,
|
||||
DOMAIN,
|
||||
)
|
||||
from .coordinator import KioskerDataUpdateCoordinator
|
||||
|
||||
NAVIGATE_URL_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(ATTR_DEVICE_ID): str,
|
||||
vol.Required(ATTR_URL): str,
|
||||
}
|
||||
)
|
||||
|
||||
SET_BLACKOUT_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(ATTR_DEVICE_ID): str,
|
||||
vol.Optional(ATTR_VISIBLE, default=True): cv.boolean,
|
||||
vol.Optional(ATTR_TEXT): str,
|
||||
vol.Optional(ATTR_BACKGROUND, default=[0, 0, 0]): selector.ColorRGBSelector(),
|
||||
vol.Optional(
|
||||
ATTR_FOREGROUND, default=[255, 255, 255]
|
||||
): selector.ColorRGBSelector(),
|
||||
vol.Optional(ATTR_ICON): str,
|
||||
vol.Optional(ATTR_EXPIRE, default=60): vol.All(
|
||||
vol.Coerce(int), vol.Range(min=0, max=100000)
|
||||
),
|
||||
vol.Optional(ATTR_DISMISSIBLE, default=False): cv.boolean,
|
||||
vol.Optional(
|
||||
ATTR_BUTTON_BACKGROUND, default=[255, 255, 255]
|
||||
): selector.ColorRGBSelector(),
|
||||
vol.Optional(
|
||||
ATTR_BUTTON_FOREGROUND, default=[0, 0, 0]
|
||||
): selector.ColorRGBSelector(),
|
||||
vol.Optional(ATTR_BUTTON_TEXT): str,
|
||||
vol.Optional(ATTR_SOUND): str,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def handle_kiosker_api_errors(
|
||||
func: Callable[[ServiceCall], Awaitable[None]],
|
||||
) -> Callable[[ServiceCall], Coroutine[Any, Any, ServiceResponse]]:
|
||||
"""Decorator to handle Kiosker API errors consistently across all service calls."""
|
||||
|
||||
@functools.wraps(func)
|
||||
async def wrapper(call: ServiceCall) -> ServiceResponse:
|
||||
try:
|
||||
await func(call)
|
||||
except ConnectionError as ex:
|
||||
raise HomeAssistantError(f"Unable to connect to Kiosker: {ex}") from ex
|
||||
except AuthenticationError as ex:
|
||||
raise ServiceValidationError(
|
||||
"Authentication failed. Check your API token."
|
||||
) from ex
|
||||
except IPAuthenticationError as ex:
|
||||
raise ServiceValidationError(
|
||||
"IP authentication failed. Check your IP whitelist."
|
||||
) from ex
|
||||
except TLSVerificationError as ex:
|
||||
raise ServiceValidationError(f"TLS verification failed: {ex}") from ex
|
||||
except BadRequestError as ex:
|
||||
raise ServiceValidationError(f"Bad request: {ex}") from ex
|
||||
else:
|
||||
return None
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
async def _get_coordinator(
|
||||
call: ServiceCall,
|
||||
) -> KioskerDataUpdateCoordinator:
|
||||
"""Get the coordinator for the targeted device."""
|
||||
registry = dr.async_get(call.hass)
|
||||
device_id: str = call.data[ATTR_DEVICE_ID]
|
||||
device = registry.async_get(device_id)
|
||||
|
||||
if device:
|
||||
for entry_id in device.config_entries:
|
||||
entry = call.hass.config_entries.async_get_entry(entry_id)
|
||||
if entry and entry.domain == DOMAIN:
|
||||
if entry.state != ConfigEntryState.LOADED:
|
||||
raise HomeAssistantError(f"{entry.title} is not loaded")
|
||||
return entry.runtime_data
|
||||
|
||||
raise ServiceValidationError(f"No {DOMAIN} devices found in targeted selection")
|
||||
|
||||
|
||||
def _rgb_to_hex(rgb: list[int]) -> str:
|
||||
"""Convert an [r, g, b] list to a hex color string."""
|
||||
return f"#{rgb[0]:02x}{rgb[1]:02x}{rgb[2]:02x}"
|
||||
|
||||
|
||||
@handle_kiosker_api_errors
|
||||
async def navigate_url(call: ServiceCall) -> None:
|
||||
"""Navigate to a URL on the Kiosker device."""
|
||||
coordinator = await _get_coordinator(call)
|
||||
await call.hass.async_add_executor_job(
|
||||
coordinator.api.navigate_url, call.data[ATTR_URL]
|
||||
)
|
||||
|
||||
|
||||
@handle_kiosker_api_errors
|
||||
async def set_blackout(call: ServiceCall) -> None:
|
||||
"""Set blackout mode on the Kiosker device."""
|
||||
blackout = Blackout(
|
||||
visible=call.data[ATTR_VISIBLE],
|
||||
text=call.data.get(ATTR_TEXT),
|
||||
background=_rgb_to_hex(call.data[ATTR_BACKGROUND]),
|
||||
foreground=_rgb_to_hex(call.data[ATTR_FOREGROUND]),
|
||||
icon=call.data.get(ATTR_ICON),
|
||||
expire=call.data[ATTR_EXPIRE],
|
||||
dismissible=call.data[ATTR_DISMISSIBLE],
|
||||
buttonBackground=_rgb_to_hex(call.data[ATTR_BUTTON_BACKGROUND]),
|
||||
buttonForeground=_rgb_to_hex(call.data[ATTR_BUTTON_FOREGROUND]),
|
||||
buttonText=call.data.get(ATTR_BUTTON_TEXT),
|
||||
sound=call.data.get(ATTR_SOUND),
|
||||
)
|
||||
|
||||
coordinator = await _get_coordinator(call)
|
||||
await call.hass.async_add_executor_job(coordinator.api.blackout_set, blackout)
|
||||
await coordinator.async_request_refresh()
|
||||
|
||||
|
||||
@callback
|
||||
def async_setup_services(hass: HomeAssistant) -> None:
|
||||
"""Set up the services for the Kiosker integration."""
|
||||
hass.services.async_register(
|
||||
DOMAIN, "navigate_url", navigate_url, schema=NAVIGATE_URL_SCHEMA
|
||||
)
|
||||
hass.services.async_register(
|
||||
DOMAIN, "set_blackout", set_blackout, schema=SET_BLACKOUT_SCHEMA
|
||||
)
|
||||
@@ -0,0 +1,62 @@
|
||||
navigate_url:
|
||||
fields:
|
||||
device_id:
|
||||
required: true
|
||||
selector:
|
||||
device:
|
||||
integration: kiosker
|
||||
url:
|
||||
required: true
|
||||
selector:
|
||||
text:
|
||||
|
||||
set_blackout:
|
||||
fields:
|
||||
device_id:
|
||||
required: true
|
||||
selector:
|
||||
device:
|
||||
integration: kiosker
|
||||
visible:
|
||||
default: true
|
||||
selector:
|
||||
boolean:
|
||||
text:
|
||||
selector:
|
||||
text:
|
||||
background:
|
||||
default: [0, 0, 0]
|
||||
selector:
|
||||
color_rgb:
|
||||
foreground:
|
||||
default: [255, 255, 255]
|
||||
selector:
|
||||
color_rgb:
|
||||
icon:
|
||||
selector:
|
||||
text:
|
||||
expire:
|
||||
default: 60
|
||||
selector:
|
||||
number:
|
||||
min: 0
|
||||
max: 3600
|
||||
unit_of_measurement: seconds
|
||||
dismissible:
|
||||
default: false
|
||||
selector:
|
||||
boolean:
|
||||
button_background:
|
||||
default: [255, 255, 255]
|
||||
selector:
|
||||
color_rgb:
|
||||
button_foreground:
|
||||
default: [0, 0, 0]
|
||||
selector:
|
||||
color_rgb:
|
||||
button_text:
|
||||
selector:
|
||||
text:
|
||||
sound:
|
||||
selector:
|
||||
text:
|
||||
@@ -104,5 +104,75 @@
|
||||
"name": "Disable screensaver"
|
||||
}
|
||||
}
|
||||
},
|
||||
"services": {
|
||||
"navigate_url": {
|
||||
"description": "Navigate to a specific URL",
|
||||
"fields": {
|
||||
"device_id": {
|
||||
"description": "The Kiosker device to control",
|
||||
"name": "Device"
|
||||
},
|
||||
"url": {
|
||||
"description": "The URL to navigate to",
|
||||
"name": "URL"
|
||||
}
|
||||
},
|
||||
"name": "Navigate to URL"
|
||||
},
|
||||
"set_blackout": {
|
||||
"description": "Set blackout screen with custom message",
|
||||
"fields": {
|
||||
"background": {
|
||||
"description": "Background color in rgb format",
|
||||
"name": "Background color"
|
||||
},
|
||||
"button_background": {
|
||||
"description": "Background color of the dismiss button in rgb format",
|
||||
"name": "Button background color"
|
||||
},
|
||||
"button_foreground": {
|
||||
"description": "Text color of the dismiss button in rgb format",
|
||||
"name": "Button foreground color"
|
||||
},
|
||||
"button_text": {
|
||||
"description": "Text to display on the dismiss button",
|
||||
"name": "Button text"
|
||||
},
|
||||
"device_id": {
|
||||
"description": "The Kiosker device to control",
|
||||
"name": "Device"
|
||||
},
|
||||
"dismissible": {
|
||||
"description": "Whether the blackout can be dismissed by user interaction",
|
||||
"name": "Dismissible"
|
||||
},
|
||||
"expire": {
|
||||
"description": "Time in seconds before the blackout expires",
|
||||
"name": "Expire time"
|
||||
},
|
||||
"foreground": {
|
||||
"description": "Text color in rgb format",
|
||||
"name": "Foreground color"
|
||||
},
|
||||
"icon": {
|
||||
"description": "Icon to display (SF Symbols name)",
|
||||
"name": "Icon"
|
||||
},
|
||||
"sound": {
|
||||
"description": "Sound to play when blackout is displayed (SystemSoundID, e.g., 1007)",
|
||||
"name": "Sound"
|
||||
},
|
||||
"text": {
|
||||
"description": "Text to display on blackout screen",
|
||||
"name": "Text"
|
||||
},
|
||||
"visible": {
|
||||
"description": "Whether the blackout is visible",
|
||||
"name": "Visible"
|
||||
}
|
||||
},
|
||||
"name": "Set blackout"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
# serializer version: 1
|
||||
# name: test_navigate_url
|
||||
_Call(
|
||||
tuple(
|
||||
'https://example.com',
|
||||
),
|
||||
dict({
|
||||
}),
|
||||
)
|
||||
# ---
|
||||
# name: test_set_blackout[all_fields]
|
||||
_Call(
|
||||
tuple(
|
||||
dict({
|
||||
'background': '#000000',
|
||||
'buttonBackground': '#ff0000',
|
||||
'buttonForeground': '#00ff00',
|
||||
'buttonText': 'Dismiss',
|
||||
'dismissible': True,
|
||||
'expire': 30,
|
||||
'foreground': '#ffffff',
|
||||
'icon': 'star',
|
||||
'sound': '1007',
|
||||
'text': 'Hello World',
|
||||
'visible': True,
|
||||
}),
|
||||
),
|
||||
dict({
|
||||
}),
|
||||
)
|
||||
# ---
|
||||
# name: test_set_blackout[defaults]
|
||||
_Call(
|
||||
tuple(
|
||||
dict({
|
||||
'background': '#000000',
|
||||
'buttonBackground': '#ffffff',
|
||||
'buttonForeground': '#000000',
|
||||
'buttonText': None,
|
||||
'dismissible': False,
|
||||
'expire': 60,
|
||||
'foreground': '#ffffff',
|
||||
'icon': None,
|
||||
'sound': None,
|
||||
'text': None,
|
||||
'visible': True,
|
||||
}),
|
||||
),
|
||||
dict({
|
||||
}),
|
||||
)
|
||||
# ---
|
||||
@@ -0,0 +1,261 @@
|
||||
"""Test the Kiosker services."""
|
||||
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from kiosker import (
|
||||
AuthenticationError,
|
||||
BadRequestError,
|
||||
Blackout,
|
||||
ConnectionError,
|
||||
IPAuthenticationError,
|
||||
ScreensaverState,
|
||||
TLSVerificationError,
|
||||
)
|
||||
import pytest
|
||||
from syrupy.assertion import SnapshotAssertion
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.kiosker.const import (
|
||||
ATTR_BACKGROUND,
|
||||
ATTR_BUTTON_BACKGROUND,
|
||||
ATTR_BUTTON_FOREGROUND,
|
||||
ATTR_BUTTON_TEXT,
|
||||
ATTR_DISMISSIBLE,
|
||||
ATTR_EXPIRE,
|
||||
ATTR_FOREGROUND,
|
||||
ATTR_SOUND,
|
||||
ATTR_TEXT,
|
||||
ATTR_URL,
|
||||
ATTR_VISIBLE,
|
||||
DOMAIN,
|
||||
)
|
||||
from homeassistant.const import ATTR_DEVICE_ID, ATTR_ICON, Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
|
||||
from . import setup_integration
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
KIOSKER_DEVICE_ID = "A98BE1CE-5FE7-4A8D-B2C3-123456789ABC"
|
||||
|
||||
|
||||
async def _setup(
|
||||
hass: HomeAssistant,
|
||||
mock_kiosker_api: MagicMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
mock_kiosker_api.screensaver_get_state.return_value = ScreensaverState(
|
||||
visible=True, disabled=False
|
||||
)
|
||||
mock_kiosker_api.blackout_get.return_value = Blackout(visible=False)
|
||||
with patch("homeassistant.components.kiosker._PLATFORMS", [Platform.BUTTON]):
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
|
||||
async def test_navigate_url(
|
||||
hass: HomeAssistant,
|
||||
mock_kiosker_api: MagicMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
device_registry: dr.DeviceRegistry,
|
||||
snapshot: SnapshotAssertion,
|
||||
) -> None:
|
||||
"""Test navigate_url service calls the API with the correct URL."""
|
||||
await _setup(hass, mock_kiosker_api, mock_config_entry)
|
||||
|
||||
device = device_registry.async_get_device(identifiers={(DOMAIN, KIOSKER_DEVICE_ID)})
|
||||
assert device is not None
|
||||
|
||||
await hass.services.async_call(
|
||||
DOMAIN,
|
||||
"navigate_url",
|
||||
{ATTR_DEVICE_ID: device.id, ATTR_URL: "https://example.com"},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
assert mock_kiosker_api.navigate_url.call_args == snapshot
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"service_data",
|
||||
[
|
||||
pytest.param(
|
||||
{
|
||||
ATTR_VISIBLE: True,
|
||||
ATTR_TEXT: "Hello World",
|
||||
ATTR_BACKGROUND: [0, 0, 0],
|
||||
ATTR_FOREGROUND: [255, 255, 255],
|
||||
ATTR_ICON: "star",
|
||||
ATTR_EXPIRE: 30,
|
||||
ATTR_DISMISSIBLE: True,
|
||||
ATTR_BUTTON_BACKGROUND: [255, 0, 0],
|
||||
ATTR_BUTTON_FOREGROUND: [0, 255, 0],
|
||||
ATTR_BUTTON_TEXT: "Dismiss",
|
||||
ATTR_SOUND: "1007",
|
||||
},
|
||||
id="all_fields",
|
||||
),
|
||||
pytest.param(
|
||||
{},
|
||||
id="defaults",
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_set_blackout(
|
||||
hass: HomeAssistant,
|
||||
mock_kiosker_api: MagicMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
device_registry: dr.DeviceRegistry,
|
||||
snapshot: SnapshotAssertion,
|
||||
service_data: dict[str, Any],
|
||||
) -> None:
|
||||
"""Test set_blackout service builds the correct Blackout object."""
|
||||
await _setup(hass, mock_kiosker_api, mock_config_entry)
|
||||
|
||||
device = device_registry.async_get_device(identifiers={(DOMAIN, KIOSKER_DEVICE_ID)})
|
||||
assert device is not None
|
||||
|
||||
await hass.services.async_call(
|
||||
DOMAIN,
|
||||
"set_blackout",
|
||||
{ATTR_DEVICE_ID: device.id, **service_data},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
assert mock_kiosker_api.blackout_set.call_args == snapshot
|
||||
|
||||
|
||||
async def test_service_entry_not_loaded(
|
||||
hass: HomeAssistant,
|
||||
mock_kiosker_api: MagicMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
device_registry: dr.DeviceRegistry,
|
||||
) -> None:
|
||||
"""Test services raise HomeAssistantError when the config entry is not loaded."""
|
||||
await _setup(hass, mock_kiosker_api, mock_config_entry)
|
||||
|
||||
device = device_registry.async_get_device(identifiers={(DOMAIN, KIOSKER_DEVICE_ID)})
|
||||
assert device is not None
|
||||
|
||||
await hass.config_entries.async_unload(mock_config_entry.entry_id)
|
||||
|
||||
with pytest.raises(HomeAssistantError, match="is not loaded"):
|
||||
await hass.services.async_call(
|
||||
DOMAIN,
|
||||
"navigate_url",
|
||||
{ATTR_DEVICE_ID: device.id, ATTR_URL: "https://example.com"},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("service", "extra_data"),
|
||||
[
|
||||
pytest.param(
|
||||
"navigate_url",
|
||||
{},
|
||||
id="navigate_url_missing_url",
|
||||
),
|
||||
pytest.param(
|
||||
"set_blackout",
|
||||
{ATTR_BACKGROUND: [0, 0]},
|
||||
id="set_blackout_rgb_too_short",
|
||||
),
|
||||
pytest.param(
|
||||
"set_blackout",
|
||||
{ATTR_BACKGROUND: [0, 0, 300]},
|
||||
id="set_blackout_rgb_out_of_range",
|
||||
),
|
||||
pytest.param(
|
||||
"set_blackout",
|
||||
{ATTR_EXPIRE: "not_a_number"},
|
||||
id="set_blackout_expire_non_integer",
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_schema_rejects_invalid_input(
|
||||
hass: HomeAssistant,
|
||||
mock_kiosker_api: MagicMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
device_registry: dr.DeviceRegistry,
|
||||
service: str,
|
||||
extra_data: dict[str, Any],
|
||||
) -> None:
|
||||
"""Test that invalid service data is rejected by schema validation."""
|
||||
await _setup(hass, mock_kiosker_api, mock_config_entry)
|
||||
|
||||
device = device_registry.async_get_device(identifiers={(DOMAIN, KIOSKER_DEVICE_ID)})
|
||||
assert device is not None
|
||||
|
||||
with pytest.raises(vol.Invalid):
|
||||
await hass.services.async_call(
|
||||
DOMAIN,
|
||||
service,
|
||||
{ATTR_DEVICE_ID: device.id, **extra_data},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
mock_kiosker_api.navigate_url.assert_not_called()
|
||||
mock_kiosker_api.blackout_set.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("exception", "expected"),
|
||||
[
|
||||
pytest.param(ConnectionError, HomeAssistantError, id="connection_error"),
|
||||
pytest.param(AuthenticationError, ServiceValidationError, id="auth_error"),
|
||||
pytest.param(IPAuthenticationError, ServiceValidationError, id="ip_auth_error"),
|
||||
pytest.param(TLSVerificationError, ServiceValidationError, id="tls_error"),
|
||||
pytest.param(BadRequestError, ServiceValidationError, id="bad_request"),
|
||||
],
|
||||
)
|
||||
async def test_api_errors_are_wrapped(
|
||||
hass: HomeAssistant,
|
||||
mock_kiosker_api: MagicMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
device_registry: dr.DeviceRegistry,
|
||||
exception: type[Exception],
|
||||
expected: type[Exception],
|
||||
) -> None:
|
||||
"""Test that kiosker API exceptions are translated to HA exceptions."""
|
||||
await _setup(hass, mock_kiosker_api, mock_config_entry)
|
||||
|
||||
device = device_registry.async_get_device(identifiers={(DOMAIN, KIOSKER_DEVICE_ID)})
|
||||
assert device is not None
|
||||
|
||||
mock_kiosker_api.navigate_url.side_effect = exception
|
||||
|
||||
with pytest.raises(expected):
|
||||
await hass.services.async_call(
|
||||
DOMAIN,
|
||||
"navigate_url",
|
||||
{ATTR_DEVICE_ID: device.id, ATTR_URL: "https://example.com"},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
|
||||
async def test_service_non_kiosker_device(
|
||||
hass: HomeAssistant,
|
||||
mock_kiosker_api: MagicMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
device_registry: dr.DeviceRegistry,
|
||||
) -> None:
|
||||
"""Test services raise ServiceValidationError when targeting a non-kiosker device."""
|
||||
await _setup(hass, mock_kiosker_api, mock_config_entry)
|
||||
|
||||
other_config_entry = MockConfigEntry(domain="other_domain")
|
||||
other_config_entry.add_to_hass(hass)
|
||||
other_device = device_registry.async_get_or_create(
|
||||
config_entry_id=other_config_entry.entry_id,
|
||||
identifiers={("other_domain", "other_device")},
|
||||
)
|
||||
|
||||
with pytest.raises(ServiceValidationError, match=f"No {DOMAIN} devices"):
|
||||
await hass.services.async_call(
|
||||
DOMAIN,
|
||||
"navigate_url",
|
||||
{ATTR_DEVICE_ID: other_device.id, ATTR_URL: "https://example.com"},
|
||||
blocking=True,
|
||||
)
|
||||
Reference in New Issue
Block a user