Compare commits

..

7 Commits

Author SHA1 Message Date
Paul Bottein
0063dc81d3 Copilot suggestions 2026-03-21 19:09:13 +01:00
Paul Bottein
7463bb79dd Remove expiration 2026-03-21 19:07:13 +01:00
Paul Bottein
d17b681477 Add time sync button to Matter integration 2026-03-21 18:57:50 +01:00
Ingmar Stein
c6c5661b4b Add Identify button to Velux integration (#163893)
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-21 18:20:02 +01:00
Joost Lekkerkerker
d0154e5019 Add stick cleaner fixture to SmartThings (#166121) 2026-03-21 16:57:26 +01:00
Joost Lekkerkerker
16fb7ed21e Bump TRMNL to platinum (#166066) 2026-03-21 06:49:50 +01:00
tronikos
d0a751abe4 Bump python-google-weather-api to 0.0.6 (#166085) 2026-03-20 17:02:22 -07:00
21 changed files with 1668 additions and 419 deletions

View File

@@ -1,12 +1,10 @@
"""HTTP view that converts audio from a URL to a preferred format."""
import asyncio
from collections import defaultdict, deque
import contextlib
from collections import defaultdict
from dataclasses import dataclass, field
from http import HTTPStatus
import logging
import re
import secrets
from typing import Final
@@ -24,12 +22,6 @@ from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
_MAX_CONVERSIONS_PER_DEVICE: Final[int] = 2
_MAX_STDERR_LINES: Final[int] = 64
_PROC_WAIT_TIMEOUT: Final[int] = 5
_STDERR_DRAIN_TIMEOUT: Final[int] = 1
_SENSITIVE_QUERY_PARAMS: Final[re.Pattern[str]] = re.compile(
r"(?<=[?&])(authSig|token|key|password|secret)=[^&\s]+", re.IGNORECASE
)
@callback
@@ -223,10 +215,8 @@ class FFmpegConvertResponse(web.StreamResponse):
assert proc.stdout is not None
assert proc.stderr is not None
stderr_lines: deque[str] = deque(maxlen=_MAX_STDERR_LINES)
stderr_task = self.hass.async_create_background_task(
self._collect_ffmpeg_stderr(proc, stderr_lines),
"ESPHome media proxy dump stderr",
self._dump_ffmpeg_stderr(proc), "ESPHome media proxy dump stderr"
)
try:
@@ -245,80 +235,33 @@ class FFmpegConvertResponse(web.StreamResponse):
if request.transport:
request.transport.abort()
raise # don't log error
except Exception:
except:
_LOGGER.exception("Unexpected error during ffmpeg conversion")
raise
finally:
# Allow conversion info to be removed
self.convert_info.is_finished = True
# Ensure subprocess and stderr cleanup run even if this task
# is cancelled (e.g., during shutdown)
try:
# Terminate hangs, so kill is used
if proc.returncode is None:
proc.kill()
# stop dumping ffmpeg stderr task
stderr_task.cancel()
# Wait for process to exit so returncode is set
await asyncio.wait_for(proc.wait(), timeout=_PROC_WAIT_TIMEOUT)
# Let stderr collector finish draining
if not stderr_task.done():
try:
await asyncio.wait_for(
stderr_task, timeout=_STDERR_DRAIN_TIMEOUT
)
except TimeoutError:
stderr_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await stderr_task
except TimeoutError:
_LOGGER.warning(
"Timed out waiting for ffmpeg process to exit for device %s",
self.device_id,
)
stderr_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await stderr_task
except asyncio.CancelledError:
# Kill the process if we were interrupted
if proc.returncode is None:
proc.kill()
stderr_task.cancel()
raise
if proc.returncode is not None and proc.returncode > 0:
_LOGGER.error(
"FFmpeg conversion failed for device %s (return code %s):\n%s",
self.device_id,
proc.returncode,
"\n".join(
_SENSITIVE_QUERY_PARAMS.sub(r"\1=REDACTED", line)
for line in stderr_lines
),
)
# Terminate hangs, so kill is used
if proc.returncode is None:
proc.kill()
# Close connection by writing EOF unless already closing
if request.transport and not request.transport.is_closing():
with contextlib.suppress(ConnectionResetError, RuntimeError, OSError):
await writer.write_eof()
await writer.write_eof()
async def _collect_ffmpeg_stderr(
async def _dump_ffmpeg_stderr(
self,
proc: asyncio.subprocess.Process,
stderr_lines: deque[str],
) -> None:
"""Collect stderr output from ffmpeg for error reporting."""
assert proc.stdout is not None
assert proc.stderr is not None
while self.hass.is_running and (chunk := await proc.stderr.readline()):
line = chunk.decode(errors="replace").rstrip()
stderr_lines.append(line)
_LOGGER.debug(
"ffmpeg[%s] output: %s",
proc.pid,
_SENSITIVE_QUERY_PARAMS.sub(r"\1=REDACTED", line),
)
_LOGGER.debug("ffmpeg[%s] output: %s", proc.pid, chunk.decode().rstrip())
class FFmpegProxyView(HomeAssistantView):

View File

@@ -8,5 +8,5 @@
"iot_class": "cloud_polling",
"loggers": ["google_weather_api"],
"quality_scale": "bronze",
"requirements": ["python-google-weather-api==0.0.4"]
"requirements": ["python-google-weather-api==0.0.6"]
}

View File

@@ -4,9 +4,11 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any
from chip.clusters import Objects as clusters
from chip.clusters.Types import NullValue
from homeassistant.components.button import (
ButtonDeviceClass,
@@ -17,6 +19,7 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util import dt as dt_util
from .entity import MatterEntity, MatterEntityDescription
from .helpers import get_matter
@@ -52,6 +55,67 @@ class MatterCommandButton(MatterEntity, ButtonEntity):
await self.send_device_command(self.entity_description.command())
# CHIP epoch: 2000-01-01 00:00:00 UTC
CHIP_EPOCH = datetime(2000, 1, 1, tzinfo=UTC)
class MatterTimeSyncButton(MatterEntity, ButtonEntity):
"""Button to synchronize time to a Matter device."""
entity_description: MatterButtonEntityDescription
async def async_press(self) -> None:
"""Sync Home Assistant time to the Matter device."""
now = dt_util.utcnow()
tz = dt_util.get_default_time_zone()
delta = now - CHIP_EPOCH
utc_us = (
(delta.days * 86400 * 1_000_000)
+ (delta.seconds * 1_000_000)
+ delta.microseconds
)
# Compute timezone and DST offsets
local_now = now.astimezone(tz)
utc_offset_delta = local_now.utcoffset()
utc_offset = int(utc_offset_delta.total_seconds()) if utc_offset_delta else 0
dst_offset_delta = local_now.dst()
dst_offset = int(dst_offset_delta.total_seconds()) if dst_offset_delta else 0
standard_offset = utc_offset - dst_offset
# 1. Set timezone
await self.send_device_command(
clusters.TimeSynchronization.Commands.SetTimeZone(
timeZone=[
clusters.TimeSynchronization.Structs.TimeZoneStruct(
offset=standard_offset, validAt=0, name=str(tz)
)
]
)
)
# 2. Set DST offset
await self.send_device_command(
clusters.TimeSynchronization.Commands.SetDSTOffset(
DSTOffset=[
clusters.TimeSynchronization.Structs.DSTOffsetStruct(
offset=dst_offset,
validStarting=0,
validUntil=NullValue,
)
]
)
)
# 3. Set UTC time
await self.send_device_command(
clusters.TimeSynchronization.Commands.SetUTCTime(
UTCTime=utc_us,
granularity=clusters.TimeSynchronization.Enums.GranularityEnum.kMicrosecondsGranularity,
)
)
# Discovery schema(s) to map Matter Attributes to HA entities
DISCOVERY_SCHEMAS = [
MatterDiscoverySchema(
@@ -169,4 +233,16 @@ DISCOVERY_SCHEMAS = [
value_contains=clusters.WaterHeaterManagement.Commands.CancelBoost.command_id,
allow_multi=True, # Also used in water_heater
),
MatterDiscoverySchema(
platform=Platform.BUTTON,
entity_description=MatterButtonEntityDescription(
key="TimeSynchronizationSyncTimeButton",
translation_key="sync_time",
entity_category=EntityCategory.CONFIG,
),
entity_class=MatterTimeSyncButton,
required_attributes=(clusters.TimeSynchronization.Attributes.UTCTime,),
allow_multi=True,
allow_none_value=True,
),
]

View File

@@ -20,6 +20,9 @@
},
"stop": {
"default": "mdi:stop"
},
"sync_time": {
"default": "mdi:clock-check-outline"
}
},
"fan": {

View File

@@ -141,6 +141,9 @@
},
"stop": {
"name": "[%key:common::action::stop%]"
},
"sync_time": {
"name": "Sync time"
}
},
"climate": {

View File

@@ -6,6 +6,6 @@
"documentation": "https://www.home-assistant.io/integrations/trmnl",
"integration_type": "hub",
"iot_class": "cloud_polling",
"quality_scale": "bronze",
"quality_scale": "platinum",
"requirements": ["trmnl==0.1.1"]
}

View File

@@ -2,7 +2,7 @@
from __future__ import annotations
from pyvlx import PyVLX, PyVLXException
from pyvlx import Node, PyVLX, PyVLXException
from homeassistant.components.button import ButtonDeviceClass, ButtonEntity
from homeassistant.const import EntityCategory
@@ -13,6 +13,7 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import VeluxConfigEntry
from .const import DOMAIN
from .entity import VeluxEntity, wrap_pyvlx_call_exceptions
PARALLEL_UPDATES = 1
@@ -23,9 +24,32 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up button entities for the Velux integration."""
async_add_entities(
[VeluxGatewayRebootButton(config_entry.entry_id, config_entry.runtime_data)]
entities: list[ButtonEntity] = [
VeluxGatewayRebootButton(config_entry.entry_id, config_entry.runtime_data)
]
entities.extend(
VeluxIdentifyButton(node, config_entry.entry_id)
for node in config_entry.runtime_data.nodes
if isinstance(node, Node)
)
async_add_entities(entities)
class VeluxIdentifyButton(VeluxEntity, ButtonEntity):
"""Representation of a Velux identify button."""
_attr_device_class = ButtonDeviceClass.IDENTIFY
_attr_entity_category = EntityCategory.DIAGNOSTIC
def __init__(self, node: Node, config_entry_id: str) -> None:
"""Initialize the Velux identify button."""
super().__init__(node, config_entry_id)
self._attr_unique_id = f"{self._attr_unique_id}_identify"
@wrap_pyvlx_call_exceptions
async def async_press(self) -> None:
"""Identify the physical device."""
await self.node.wink()
class VeluxGatewayRebootButton(ButtonEntity):

2
requirements_all.txt generated
View File

@@ -2593,7 +2593,7 @@ python-gitlab==1.6.0
python-google-drive-api==0.1.0
# homeassistant.components.google_weather
python-google-weather-api==0.0.4
python-google-weather-api==0.0.6
# homeassistant.components.analytics_insights
python-homeassistant-analytics==0.9.0

View File

@@ -2198,7 +2198,7 @@ python-fullykiosk==0.0.15
python-google-drive-api==0.1.0
# homeassistant.components.google_weather
python-google-weather-api==0.0.4
python-google-weather-api==0.0.6
# homeassistant.components.analytics_insights
python-homeassistant-analytics==0.9.0

View File

@@ -1,13 +1,11 @@
"""Tests for ffmpeg proxy view."""
import asyncio
from collections.abc import Generator
from http import HTTPStatus
import io
import logging
import os
import tempfile
from unittest.mock import AsyncMock, MagicMock, patch
from unittest.mock import patch
from urllib.request import pathname2url
import wave
@@ -16,17 +14,12 @@ import mutagen
import pytest
from homeassistant.components import esphome
from homeassistant.components.esphome.ffmpeg_proxy import (
_MAX_STDERR_LINES,
async_create_proxy_url,
)
from homeassistant.components.esphome.ffmpeg_proxy import async_create_proxy_url
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.typing import ClientSessionGenerator
FFMPEG_PROXY = "homeassistant.components.esphome.ffmpeg_proxy"
@pytest.fixture(name="wav_file_length")
def wav_file_length_fixture() -> int:
@@ -126,7 +119,6 @@ async def test_proxy_view(
async def test_ffmpeg_file_doesnt_exist(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test ffmpeg conversion with a file that doesn't exist."""
device_id = "1234"
@@ -144,327 +136,6 @@ async def test_ffmpeg_file_doesnt_exist(
mp3_data = await req.content.read()
assert not mp3_data
# ffmpeg failure should be logged at error level
assert "FFmpeg conversion failed for device" in caplog.text
assert device_id in caplog.text
async def test_ffmpeg_error_stderr_truncated(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test that ffmpeg stderr output is truncated in error logs."""
device_id = "1234"
await async_setup_component(hass, esphome.DOMAIN, {esphome.DOMAIN: {}})
client = await hass_client()
total_lines = _MAX_STDERR_LINES + 50
stderr_lines_data = [f"stderr line {i}\n".encode() for i in range(total_lines)] + [
b""
]
async def _stdout_read(_size: int = -1) -> bytes:
"""Yield to event loop so stderr collector can run, then return EOF."""
await asyncio.sleep(0)
return b""
mock_proc = AsyncMock()
mock_proc.stdout.read = _stdout_read
mock_proc.stderr.readline = AsyncMock(side_effect=stderr_lines_data)
mock_proc.returncode = 1
with patch("asyncio.create_subprocess_exec", return_value=mock_proc):
url = async_create_proxy_url(hass, device_id, "dummy-input", media_format="mp3")
req = await client.get(url)
assert req.status == HTTPStatus.OK
await req.content.read()
# Should log an error with stderr content
assert "FFmpeg conversion failed for device" in caplog.text
# Find the error message to verify truncation.
# We can't just check caplog.text because lines beyond the limit
# are still present at debug level from _collect_ffmpeg_stderr.
error_message = next(
r.message
for r in caplog.records
if r.levelno >= logging.ERROR and "FFmpeg conversion failed" in r.message
)
total_lines = _MAX_STDERR_LINES + 50
# The last _MAX_STDERR_LINES lines should be present
for i in range(total_lines - _MAX_STDERR_LINES, total_lines):
assert f"stderr line {i}" in error_message
# Early lines that were evicted should not be in the error log
assert "stderr line 0" not in error_message
async def test_ffmpeg_error_redacts_sensitive_urls(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test that sensitive query params are redacted in error logs."""
device_id = "1234"
await async_setup_component(hass, esphome.DOMAIN, {esphome.DOMAIN: {}})
client = await hass_client()
sensitive_url = (
"https://example.com/api/tts?authSig=secret123&token=abc456&other=keep"
)
stderr_lines_data = [
f"Error opening input file {sensitive_url}\n".encode(),
b"",
]
async def _stdout_read(_size: int = -1) -> bytes:
await asyncio.sleep(0)
return b""
mock_proc = AsyncMock()
mock_proc.stdout.read = _stdout_read
mock_proc.stderr.readline = AsyncMock(side_effect=stderr_lines_data)
mock_proc.returncode = 1
with patch("asyncio.create_subprocess_exec", return_value=mock_proc):
url = async_create_proxy_url(hass, device_id, "dummy-input", media_format="mp3")
req = await client.get(url)
assert req.status == HTTPStatus.OK
await req.content.read()
error_message = next(
r.message
for r in caplog.records
if r.levelno >= logging.ERROR and "FFmpeg conversion failed" in r.message
)
assert "authSig=REDACTED" in error_message
assert "token=REDACTED" in error_message
assert "secret123" not in error_message
assert "abc456" not in error_message
assert "other=keep" in error_message
async def test_ffmpeg_stderr_drain_timeout(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test that stderr drain timeout is handled gracefully."""
device_id = "1234"
await async_setup_component(hass, esphome.DOMAIN, {esphome.DOMAIN: {}})
client = await hass_client()
never_finish: asyncio.Future[bytes] = asyncio.get_running_loop().create_future()
call_count = 0
async def _slow_stderr_readline() -> bytes:
nonlocal call_count
call_count += 1
if call_count == 1:
return b"first error line\n"
# Block forever on second call so the drain times out
return await never_finish
async def _stdout_read(_size: int = -1) -> bytes:
await asyncio.sleep(0)
return b""
mock_proc = AsyncMock()
mock_proc.stdout.read = _stdout_read
mock_proc.stderr.readline = _slow_stderr_readline
mock_proc.returncode = 1
with (
patch("asyncio.create_subprocess_exec", return_value=mock_proc),
patch(f"{FFMPEG_PROXY}._STDERR_DRAIN_TIMEOUT", 0),
):
url = async_create_proxy_url(hass, device_id, "dummy-input", media_format="mp3")
req = await client.get(url)
assert req.status == HTTPStatus.OK
await req.content.read()
assert "FFmpeg conversion failed for device" in caplog.text
assert "first error line" in caplog.text
async def test_ffmpeg_proc_wait_timeout(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test that proc.wait() timeout is handled gracefully."""
device_id = "1234"
await async_setup_component(hass, esphome.DOMAIN, {esphome.DOMAIN: {}})
client = await hass_client()
async def _stdout_read(_size: int = -1) -> bytes:
await asyncio.sleep(0)
return b""
async def _proc_wait() -> None:
# Block forever so wait_for times out
await asyncio.Future()
mock_proc = AsyncMock()
mock_proc.stdout.read = _stdout_read
mock_proc.stderr.readline = AsyncMock(return_value=b"")
mock_proc.returncode = None
mock_proc.kill = MagicMock()
mock_proc.wait = _proc_wait
with (
patch("asyncio.create_subprocess_exec", return_value=mock_proc),
patch(f"{FFMPEG_PROXY}._PROC_WAIT_TIMEOUT", 0),
patch(f"{FFMPEG_PROXY}._STDERR_DRAIN_TIMEOUT", 0),
):
url = async_create_proxy_url(hass, device_id, "dummy-input", media_format="mp3")
req = await client.get(url)
assert req.status == HTTPStatus.OK
await req.content.read()
assert "Timed out waiting for ffmpeg process to exit" in caplog.text
async def test_ffmpeg_cleanup_on_cancellation(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
) -> None:
"""Test that ffmpeg process is killed when task is cancelled during cleanup."""
device_id = "1234"
await async_setup_component(hass, esphome.DOMAIN, {esphome.DOMAIN: {}})
client = await hass_client()
async def _stdout_read(_size: int = -1) -> bytes:
await asyncio.sleep(0)
return b""
async def _proc_wait() -> None:
# Simulate cancellation during proc.wait()
raise asyncio.CancelledError
mock_kill = MagicMock()
mock_proc = AsyncMock()
mock_proc.stdout.read = _stdout_read
mock_proc.stderr.readline = AsyncMock(return_value=b"")
mock_proc.returncode = None
mock_proc.kill = mock_kill
mock_proc.wait = _proc_wait
with patch("asyncio.create_subprocess_exec", return_value=mock_proc):
url = async_create_proxy_url(hass, device_id, "dummy-input", media_format="mp3")
req = await client.get(url)
assert req.status == HTTPStatus.OK
with pytest.raises(client_exceptions.ClientPayloadError):
await req.content.read()
# proc.kill should have been called (once in the initial check, once in the
# CancelledError handler)
assert mock_kill.call_count >= 1
async def test_ffmpeg_unexpected_exception(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test that unexpected exceptions during ffmpeg conversion are logged."""
device_id = "1234"
await async_setup_component(hass, esphome.DOMAIN, {esphome.DOMAIN: {}})
client = await hass_client()
async def _stdout_read_error(_size: int = -1) -> bytes:
raise RuntimeError("unexpected read error")
mock_proc = AsyncMock()
mock_proc.stdout.read = _stdout_read_error
mock_proc.stderr.readline = AsyncMock(return_value=b"")
mock_proc.returncode = 0
with patch("asyncio.create_subprocess_exec", return_value=mock_proc):
url = async_create_proxy_url(hass, device_id, "dummy-input", media_format="mp3")
req = await client.get(url)
assert req.status == HTTPStatus.OK
await req.content.read()
assert "Unexpected error during ffmpeg conversion" in caplog.text
async def test_max_conversions_kills_running_process(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test that exceeding max conversions kills a running ffmpeg process."""
device_id = "1234"
await async_setup_component(hass, esphome.DOMAIN, {esphome.DOMAIN: {}})
client = await hass_client()
stdout_futures: list[asyncio.Future[bytes]] = []
mock_kills: list[MagicMock] = []
procs_started = asyncio.Event()
proc_count = 0
def _make_mock_proc(*_args: object, **_kwargs: object) -> AsyncMock:
"""Create a mock ffmpeg process that blocks on stdout read."""
nonlocal proc_count
future: asyncio.Future[bytes] = hass.loop.create_future()
stdout_futures.append(future)
kill = MagicMock()
mock_kills.append(kill)
async def _stdout_read(_size: int = -1) -> bytes:
return await future
mock = AsyncMock()
mock.stdout.read = _stdout_read
mock.stderr.readline = AsyncMock(return_value=b"")
mock.returncode = None
mock.kill = kill
proc_count += 1
if proc_count >= 2:
procs_started.set()
return mock
with patch(
"asyncio.create_subprocess_exec",
side_effect=_make_mock_proc,
):
url1 = async_create_proxy_url(hass, device_id, "url1", media_format="mp3")
url2 = async_create_proxy_url(hass, device_id, "url2", media_format="mp3")
# Start both HTTP requests — each spawns an ffmpeg process that blocks
task1 = hass.async_create_task(client.get(url1))
task2 = hass.async_create_task(client.get(url2))
# Wait until both ffmpeg processes have been created
await procs_started.wait()
assert len(mock_kills) == 2
# Creating a third conversion should kill the oldest running process
async_create_proxy_url(hass, device_id, "url3", media_format="mp3")
assert "Stopping existing ffmpeg process" in caplog.text
mock_kills[0].assert_called_once()
# Unblock stdout reads so background tasks can finish
for future in stdout_futures:
if not future.done():
future.set_result(b"")
await task1
await task2
async def test_lingering_process(
hass: HomeAssistant,

View File

@@ -1325,6 +1325,56 @@
'state': 'unknown',
})
# ---
# name: test_buttons[eve_shutter][button.eve_shutter_switch_20eci1701_sync_time-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'button.eve_shutter_switch_20eci1701_sync_time',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Sync time',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Sync time',
'platform': 'matter',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'sync_time',
'unique_id': '00000000000004D2-0000000000000094-MatterNodeDevice-0-TimeSynchronizationSyncTimeButton-56-0',
'unit_of_measurement': None,
})
# ---
# name: test_buttons[eve_shutter][button.eve_shutter_switch_20eci1701_sync_time-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Eve Shutter Switch 20ECI1701 Sync time',
}),
'context': <ANY>,
'entity_id': 'button.eve_shutter_switch_20eci1701_sync_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_buttons[eve_thermo_v4][button.eve_thermo_20ebp1701_identify-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -1376,6 +1426,56 @@
'state': 'unknown',
})
# ---
# name: test_buttons[eve_thermo_v4][button.eve_thermo_20ebp1701_sync_time-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'button.eve_thermo_20ebp1701_sync_time',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Sync time',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Sync time',
'platform': 'matter',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'sync_time',
'unique_id': '00000000000004D2-0000000000000021-MatterNodeDevice-0-TimeSynchronizationSyncTimeButton-56-0',
'unit_of_measurement': None,
})
# ---
# name: test_buttons[eve_thermo_v4][button.eve_thermo_20ebp1701_sync_time-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Eve Thermo 20EBP1701 Sync time',
}),
'context': <ANY>,
'entity_id': 'button.eve_thermo_20ebp1701_sync_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_buttons[eve_thermo_v5][button.eve_thermo_20ecd1701_identify-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -1427,6 +1527,56 @@
'state': 'unknown',
})
# ---
# name: test_buttons[eve_thermo_v5][button.eve_thermo_20ecd1701_sync_time-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'button.eve_thermo_20ecd1701_sync_time',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Sync time',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Sync time',
'platform': 'matter',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'sync_time',
'unique_id': '00000000000004D2-000000000000000C-MatterNodeDevice-0-TimeSynchronizationSyncTimeButton-56-0',
'unit_of_measurement': None,
})
# ---
# name: test_buttons[eve_thermo_v5][button.eve_thermo_20ecd1701_sync_time-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Eve Thermo 20ECD1701 Sync time',
}),
'context': <ANY>,
'entity_id': 'button.eve_thermo_20ecd1701_sync_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_buttons[eve_weather_sensor][button.eve_weather_identify_1-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -1935,6 +2085,56 @@
'state': 'unknown',
})
# ---
# name: test_buttons[ikea_air_quality_monitor][button.alpstuga_air_quality_monitor_sync_time-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'button.alpstuga_air_quality_monitor_sync_time',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Sync time',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Sync time',
'platform': 'matter',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'sync_time',
'unique_id': '00000000000004D2-0000000000000025-MatterNodeDevice-0-TimeSynchronizationSyncTimeButton-56-0',
'unit_of_measurement': None,
})
# ---
# name: test_buttons[ikea_air_quality_monitor][button.alpstuga_air_quality_monitor_sync_time-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'ALPSTUGA air quality monitor Sync time',
}),
'context': <ANY>,
'entity_id': 'button.alpstuga_air_quality_monitor_sync_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_buttons[inovelli_vtm30][button.white_series_onoff_switch_identify_load_control-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -2845,6 +3045,56 @@
'state': 'unknown',
})
# ---
# name: test_buttons[mock_leak_sensor][button.water_leak_detector_sync_time-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'button.water_leak_detector_sync_time',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Sync time',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Sync time',
'platform': 'matter',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'sync_time',
'unique_id': '00000000000004D2-0000000000000020-MatterNodeDevice-0-TimeSynchronizationSyncTimeButton-56-0',
'unit_of_measurement': None,
})
# ---
# name: test_buttons[mock_leak_sensor][button.water_leak_detector_sync_time-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Water Leak Detector Sync time',
}),
'context': <ANY>,
'entity_id': 'button.water_leak_detector_sync_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_buttons[mock_lock][button.mock_lock_identify-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -4211,6 +4461,56 @@
'state': 'unknown',
})
# ---
# name: test_buttons[silabs_light_switch][button.light_switch_example_sync_time-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'button.light_switch_example_sync_time',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Sync time',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Sync time',
'platform': 'matter',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'sync_time',
'unique_id': '00000000000004D2-000000000000008E-MatterNodeDevice-0-TimeSynchronizationSyncTimeButton-56-0',
'unit_of_measurement': None,
})
# ---
# name: test_buttons[silabs_light_switch][button.light_switch_example_sync_time-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Light switch example Sync time',
}),
'context': <ANY>,
'entity_id': 'button.light_switch_example_sync_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_buttons[silabs_refrigerator][button.refrigerator_identify-entry]
EntityRegistryEntrySnapshot({
'aliases': list([

View File

@@ -1,8 +1,10 @@
"""Test Matter switches."""
"""Test Matter buttons."""
from datetime import UTC, datetime
from unittest.mock import MagicMock, call
from chip.clusters import Objects as clusters
from chip.clusters.Types import NullValue
from matter_server.client.models.node import MatterNode
import pytest
from syrupy.assertion import SnapshotAssertion
@@ -10,6 +12,7 @@ from syrupy.assertion import SnapshotAssertion
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.util import dt as dt_util
from .common import snapshot_matter_entities
@@ -107,3 +110,82 @@ async def test_smoke_detector_self_test(
endpoint_id=1,
command=clusters.SmokeCoAlarm.Commands.SelfTestRequest(),
)
@pytest.mark.freeze_time("2025-06-15T12:00:00+00:00")
@pytest.mark.parametrize("node_fixture", ["ikea_air_quality_monitor"])
async def test_time_sync_button(
hass: HomeAssistant,
matter_client: MagicMock,
matter_node: MatterNode,
) -> None:
"""Test button entity is created for a Matter TimeSynchronization Cluster."""
entity_id = "button.alpstuga_air_quality_monitor_sync_time"
state = hass.states.get(entity_id)
assert state
assert state.attributes["friendly_name"] == "ALPSTUGA air quality monitor Sync time"
# test press action
await hass.services.async_call(
"button",
"press",
{
"entity_id": entity_id,
},
blocking=True,
)
assert matter_client.send_device_command.call_count == 3
# Compute expected values based on HA's configured timezone
chip_epoch = datetime(2000, 1, 1, tzinfo=UTC)
frozen_now = datetime(2025, 6, 15, 12, 0, 0, tzinfo=UTC)
delta = frozen_now - chip_epoch
expected_utc_us = (
(delta.days * 86400 * 1_000_000)
+ (delta.seconds * 1_000_000)
+ delta.microseconds
)
ha_tz = dt_util.get_default_time_zone()
local_now = frozen_now.astimezone(ha_tz)
utc_offset_delta = local_now.utcoffset()
utc_offset = int(utc_offset_delta.total_seconds()) if utc_offset_delta else 0
dst_offset_delta = local_now.dst()
dst_offset = int(dst_offset_delta.total_seconds()) if dst_offset_delta else 0
standard_offset = utc_offset - dst_offset
# Verify SetTimeZone command
assert matter_client.send_device_command.call_args_list[0] == call(
node_id=matter_node.node_id,
endpoint_id=0,
command=clusters.TimeSynchronization.Commands.SetTimeZone(
timeZone=[
clusters.TimeSynchronization.Structs.TimeZoneStruct(
offset=standard_offset,
validAt=0,
name=str(ha_tz),
)
]
),
)
# Verify SetDSTOffset command
assert matter_client.send_device_command.call_args_list[1] == call(
node_id=matter_node.node_id,
endpoint_id=0,
command=clusters.TimeSynchronization.Commands.SetDSTOffset(
DSTOffset=[
clusters.TimeSynchronization.Structs.DSTOffsetStruct(
offset=dst_offset,
validStarting=0,
validUntil=NullValue,
)
]
),
)
# Verify SetUTCTime command
assert matter_client.send_device_command.call_args_list[2] == call(
node_id=matter_node.node_id,
endpoint_id=0,
command=clusters.TimeSynchronization.Commands.SetUTCTime(
UTCTime=expected_utc_us,
granularity=clusters.TimeSynchronization.Enums.GranularityEnum.kMicrosecondsGranularity,
),
)

View File

@@ -74,6 +74,7 @@ DEVICE_FIXTURES = [
"da_wm_dw_01011",
"da_rvc_normal_000001",
"da_rvc_map_01011",
"da_vc_stick_01001",
"da_ks_microwave_0101x",
"da_ks_cooktop_000001",
"da_ks_cooktop_31001",

View File

@@ -0,0 +1,426 @@
{
"components": {
"station": {
"samsungce.cleanStationStickStatus": {
"status": {
"value": "attached",
"timestamp": "2026-03-21T15:25:21.619Z"
}
},
"custom.disabledCapabilities": {
"disabledCapabilities": {
"value": ["samsungce.cleanStationUvCleaning"],
"timestamp": "2026-03-21T14:44:08.042Z"
}
},
"samsungce.cleanStationUvCleaning": {
"operatingState": {
"value": "ready",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"lastFinishedTime": {
"value": null
},
"uvcIntensive": {
"value": null
},
"operationTime": {
"value": null
},
"remainingTime": {
"value": null
}
},
"samsungce.stickCleanerDustBag": {
"supportedStatus": {
"value": ["full", "normal"],
"timestamp": "2026-03-21T14:44:06.339Z"
},
"usage": {
"value": 343,
"timestamp": "2026-03-21T15:25:35.951Z"
},
"status": {
"value": "normal",
"timestamp": "2026-03-21T14:44:06.339Z"
}
}
},
"main": {
"custom.disabledComponents": {
"disabledComponents": {
"value": [],
"timestamp": "2026-03-21T14:44:06.339Z"
}
},
"powerConsumptionReport": {
"powerConsumption": {
"value": {
"energy": 4,
"deltaEnergy": 3,
"power": 0,
"powerEnergy": 0.0,
"persistedEnergy": 0,
"energySaved": 0,
"start": "2026-03-21T15:35:31Z",
"end": "2026-03-21T15:41:41Z"
},
"timestamp": "2026-03-21T15:41:41.889Z"
}
},
"samsungce.stickCleanerStatus": {
"operatingState": {
"value": "ready",
"timestamp": "2026-03-21T15:25:35.978Z"
}
},
"refresh": {},
"samsungce.notification": {
"supportedActionSettings": {
"value": [
{
"action": "stop",
"supportedSettings": ["on", "off"]
}
],
"timestamp": "2026-03-21T14:44:06.339Z"
},
"actionSetting": {
"value": {
"stop": {
"setting": "on"
}
},
"timestamp": "2026-03-21T14:50:31.556Z"
},
"supportedContexts": {
"value": ["incomingCall", "messageReceived"],
"timestamp": "2026-03-21T14:44:06.339Z"
},
"supportCustomContent": {
"value": false,
"timestamp": "2026-03-21T14:44:06.339Z"
}
},
"samsungce.stickCleanerDustbinStatus": {
"operatingState": {
"value": "ready",
"timestamp": "2026-03-21T15:25:35.978Z"
},
"lastEmptiedTime": {
"value": "2026-03-21T15:25:00Z",
"timestamp": "2026-03-21T15:25:35.978Z"
}
},
"battery": {
"quantity": {
"value": null
},
"battery": {
"value": 80,
"unit": "%",
"timestamp": "2026-03-21T15:41:41.889Z"
},
"type": {
"value": null
}
},
"execute": {
"data": {
"value": null
}
},
"samsungce.deviceIdentification": {
"micomAssayCode": {
"value": "50025842",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"modelName": {
"value": null
},
"serialNumber": {
"value": null
},
"serialNumberExtra": {
"value": null
},
"releaseCountry": {
"value": null
},
"modelClassificationCode": {
"value": "80030200001711000802000000000000",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"description": {
"value": "A-VSWW-TP1-23-VS9700",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"releaseYear": {
"value": null
},
"binaryId": {
"value": "A-VSWW-TP1-23-VS9700",
"timestamp": "2026-03-21T15:41:41.888Z"
}
},
"sec.wifiConfiguration": {
"autoReconnection": {
"value": true,
"timestamp": "2026-03-21T14:44:06.339Z"
},
"minVersion": {
"value": "1.0",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"supportedWiFiFreq": {
"value": ["2.4G"],
"timestamp": "2026-03-21T14:44:06.339Z"
},
"supportedAuthType": {
"value": ["OPEN", "WEP", "WPA-PSK", "WPA2-PSK", "SAE"],
"timestamp": "2026-03-21T14:44:06.339Z"
},
"protocolType": {
"value": ["helper_hotspot", "ble_ocf"],
"timestamp": "2026-03-21T14:44:06.339Z"
}
},
"samsungce.selfCheck": {
"result": {
"value": "failed",
"timestamp": "2026-03-21T15:25:48.370Z"
},
"supportedActions": {
"value": ["start", "cancel"],
"timestamp": "2026-03-21T14:44:06.339Z"
},
"progress": {
"value": 100,
"unit": "%",
"timestamp": "2026-03-21T15:25:48.370Z"
},
"errors": {
"value": [
{
"code": "DA_VCS_E_001"
}
],
"timestamp": "2026-03-21T15:25:48.370Z"
},
"status": {
"value": "ready",
"timestamp": "2026-03-21T15:25:48.370Z"
}
},
"samsungce.softwareVersion": {
"versions": {
"value": [
{
"id": "0",
"swType": "Software",
"versionNumber": "25051400",
"description": "Version"
},
{
"id": "1",
"swType": "Firmware",
"versionNumber": "00258B23110300",
"description": "Version"
},
{
"id": "2",
"swType": "Firmware",
"versionNumber": "00253B23070700",
"description": "Version"
}
],
"timestamp": "2026-03-21T14:50:31.808Z"
}
},
"ocf": {
"st": {
"value": null
},
"mndt": {
"value": null
},
"mnfv": {
"value": "A-VSWW-TP1-23-VS9700_51250514",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"mnhw": {
"value": "Realtek",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"di": {
"value": "e1f93c0c-6fe0-c65a-a314-c8f7b163c86b",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"mnsl": {
"value": "http://www.samsung.com",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"dmv": {
"value": "1.2.1",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"n": {
"value": "[vacuum] Samsung",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"mnmo": {
"value": "A-VSWW-TP1-23-VS9700|50025842|80030200001711000802000000000000",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"vid": {
"value": "DA-VC-STICK-01001",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"mnmn": {
"value": "Samsung Electronics",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"mnml": {
"value": "http://www.samsung.com",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"mnpv": {
"value": "SYSTEM 2.0",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"mnos": {
"value": "TizenRT 4.0",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"pi": {
"value": "e1f93c0c-6fe0-c65a-a314-c8f7b163c86b",
"timestamp": "2026-03-21T14:50:07.585Z"
},
"icv": {
"value": "core.1.1.0",
"timestamp": "2026-03-21T14:50:07.585Z"
}
},
"samsungce.stickCleanerStickStatus": {
"mode": {
"value": "none",
"timestamp": "2026-03-21T15:25:06.593Z"
},
"status": {
"value": "charging",
"timestamp": "2026-03-21T15:25:22.052Z"
},
"bleConnectionState": {
"value": "connected",
"timestamp": "2026-03-21T14:50:29.775Z"
}
},
"custom.disabledCapabilities": {
"disabledCapabilities": {
"value": ["sec.wifiConfiguration"],
"timestamp": "2026-03-21T14:44:06.339Z"
}
},
"samsungce.driverVersion": {
"versionNumber": {
"value": 25040101,
"timestamp": "2026-03-21T14:44:06.339Z"
}
},
"samsungce.softwareUpdate": {
"targetModule": {
"value": {},
"timestamp": "2026-03-21T14:44:08.245Z"
},
"otnDUID": {
"value": "BDCPH4AI7GMCS",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"lastUpdatedDate": {
"value": null
},
"availableModules": {
"value": [],
"timestamp": "2026-03-21T14:44:06.339Z"
},
"newVersionAvailable": {
"value": false,
"timestamp": "2026-03-21T14:44:06.339Z"
},
"operatingState": {
"value": "none",
"timestamp": "2026-03-21T14:44:08.245Z"
},
"progress": {
"value": 0,
"unit": "%",
"timestamp": "2026-03-21T14:50:09.045Z"
}
},
"sec.diagnosticsInformation": {
"logType": {
"value": ["errCode", "dump"],
"timestamp": "2026-03-21T14:44:06.339Z"
},
"endpoint": {
"value": "SSM",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"minVersion": {
"value": "3.0",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"signinPermission": {
"value": null
},
"setupId": {
"value": "VS2",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"protocolType": {
"value": "ble_ocf",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"tsId": {
"value": "DA01",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"mnId": {
"value": "0AJT",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"dumpType": {
"value": "file",
"timestamp": "2026-03-21T14:44:06.339Z"
}
},
"custom.deviceReportStateConfiguration": {
"reportStateRealtimePeriod": {
"value": null
},
"reportStateRealtime": {
"value": {
"state": "enabled",
"duration": 10,
"unit": "minute"
},
"timestamp": "2026-03-21T14:44:42.985Z"
},
"reportStatePeriod": {
"value": "enabled",
"timestamp": "2026-03-21T14:44:06.339Z"
}
},
"samsungce.lamp": {
"brightnessLevel": {
"value": "on",
"timestamp": "2026-03-21T14:44:06.339Z"
},
"supportedBrightnessLevel": {
"value": ["on", "off"],
"timestamp": "2026-03-21T14:44:06.339Z"
}
}
}
}
}

View File

@@ -0,0 +1,168 @@
{
"items": [
{
"deviceId": "e1f93c0c-6fe0-c65a-a314-c8f7b163c86b",
"name": "[vacuum] Samsung",
"label": "Stick vacuum",
"manufacturerName": "Samsung Electronics",
"presentationId": "DA-VC-STICK-01001",
"deviceManufacturerCode": "Samsung Electronics",
"locationId": "03f25476-ce87-4f94-b153-03d40451dee0",
"ownerId": "62619912-9710-ee72-bdf7-6e3910560913",
"roomId": "2f820695-73c1-4d43-8ee9-7c6a07feeb9a",
"deviceTypeName": "x.com.st.d.stickcleaner",
"components": [
{
"id": "main",
"label": "main",
"capabilities": [
{
"id": "ocf",
"version": 1
},
{
"id": "execute",
"version": 1
},
{
"id": "powerConsumptionReport",
"version": 1
},
{
"id": "refresh",
"version": 1
},
{
"id": "samsungce.deviceIdentification",
"version": 1
},
{
"id": "samsungce.driverVersion",
"version": 1
},
{
"id": "samsungce.stickCleanerStickStatus",
"version": 1
},
{
"id": "battery",
"version": 1
},
{
"id": "samsungce.lamp",
"version": 1
},
{
"id": "samsungce.notification",
"version": 1
},
{
"id": "samsungce.selfCheck",
"version": 1
},
{
"id": "samsungce.stickCleanerDustbinStatus",
"version": 1
},
{
"id": "samsungce.softwareUpdate",
"version": 1
},
{
"id": "samsungce.softwareVersion",
"version": 1
},
{
"id": "samsungce.stickCleanerStatus",
"version": 1
},
{
"id": "custom.deviceReportStateConfiguration",
"version": 1
},
{
"id": "custom.disabledComponents",
"version": 1
},
{
"id": "custom.disabledCapabilities",
"version": 1
},
{
"id": "sec.diagnosticsInformation",
"version": 1
},
{
"id": "sec.wifiConfiguration",
"version": 1
}
],
"categories": [
{
"name": "StickVacuumCleaner",
"categoryType": "manufacturer"
}
],
"optional": false
},
{
"id": "station",
"label": "station",
"capabilities": [
{
"id": "samsungce.stickCleanerDustBag",
"version": 1
},
{
"id": "samsungce.cleanStationStickStatus",
"version": 1
},
{
"id": "samsungce.cleanStationUvCleaning",
"version": 1
},
{
"id": "custom.disabledCapabilities",
"version": 1
}
],
"categories": [
{
"name": "Other",
"categoryType": "manufacturer"
}
],
"optional": false
}
],
"createTime": "2026-03-21T14:43:55.855Z",
"profile": {
"id": "21c15481-d69b-34a9-86a6-bcdb478a68cb"
},
"ocf": {
"ocfDeviceType": "x.com.st.d.stickcleaner",
"name": "[vacuum] Samsung",
"specVersion": "core.1.1.0",
"verticalDomainSpecVersion": "1.2.1",
"manufacturerName": "Samsung Electronics",
"modelNumber": "A-VSWW-TP1-23-VS9700|50025842|80030200001711000802000000000000",
"platformVersion": "SYSTEM 2.0",
"platformOS": "TizenRT 4.0",
"hwVersion": "Realtek",
"firmwareVersion": "A-VSWW-TP1-23-VS9700_51250514",
"vendorId": "DA-VC-STICK-01001",
"vendorResourceClientServerVersion": "Realtek Release 250514",
"lastSignupTime": "2026-03-21T14:43:55.796850354Z",
"transferCandidate": true,
"additionalAuthCodeRequired": false,
"modelCode": "VS28C9784QK/WA"
},
"type": "OCF",
"restrictionTier": 0,
"allowed": null,
"executionContext": "CLOUD",
"relationships": []
}
],
"_links": {}
}

View File

@@ -1239,6 +1239,37 @@
'via_device_id': None,
})
# ---
# name: test_devices[da_vc_stick_01001]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'config_entries_subentries': <ANY>,
'configuration_url': 'https://account.smartthings.com',
'connections': set({
}),
'disabled_by': None,
'entry_type': None,
'hw_version': 'Realtek',
'id': <ANY>,
'identifiers': set({
tuple(
'smartthings',
'e1f93c0c-6fe0-c65a-a314-c8f7b163c86b',
),
}),
'labels': set({
}),
'manufacturer': 'Samsung Electronics',
'model': 'A-VSWW-TP1-23-VS9700',
'model_id': 'VS28C9784QK/WA',
'name': 'Stick vacuum',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': None,
'sw_version': 'A-VSWW-TP1-23-VS9700_51250514',
'via_device_id': None,
})
# ---
# name: test_devices[da_wm_dw_000001]
DeviceRegistryEntrySnapshot({
'area_id': 'theater',

View File

@@ -851,6 +851,65 @@
'state': 'medium',
})
# ---
# name: test_all_entities[da_vc_stick_01001][select.stick_vacuum_lamp-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'options': list([
'on',
'off',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'select',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'select.stick_vacuum_lamp',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Lamp',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Lamp',
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'lamp',
'unique_id': 'e1f93c0c-6fe0-c65a-a314-c8f7b163c86b_main_samsungce.lamp_brightnessLevel_brightnessLevel',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[da_vc_stick_01001][select.stick_vacuum_lamp-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Stick vacuum Lamp',
'options': list([
'on',
'off',
]),
}),
'context': <ANY>,
'entity_id': 'select.stick_vacuum_lamp',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_all_entities[da_wm_dw_000001][select.dishwasher-entry]
EntityRegistryEntrySnapshot({
'aliases': list([

View File

@@ -13340,6 +13340,350 @@
'state': 'room',
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_battery-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.stick_vacuum_battery',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Battery',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.BATTERY: 'battery'>,
'original_icon': None,
'original_name': 'Battery',
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'e1f93c0c-6fe0-c65a-a314-c8f7b163c86b_main_battery_battery_battery',
'unit_of_measurement': '%',
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_battery-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'battery',
'friendly_name': 'Stick vacuum Battery',
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'sensor.stick_vacuum_battery',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '80',
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_energy-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.stick_vacuum_energy',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Energy',
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.ENERGY: 'energy'>,
'original_icon': None,
'original_name': 'Energy',
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'e1f93c0c-6fe0-c65a-a314-c8f7b163c86b_main_powerConsumptionReport_powerConsumption_energy_meter',
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_energy-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'energy',
'friendly_name': 'Stick vacuum Energy',
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
}),
'context': <ANY>,
'entity_id': 'sensor.stick_vacuum_energy',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.004',
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_energy_difference-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.TOTAL: 'total'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.stick_vacuum_energy_difference',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Energy difference',
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.ENERGY: 'energy'>,
'original_icon': None,
'original_name': 'Energy difference',
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'energy_difference',
'unique_id': 'e1f93c0c-6fe0-c65a-a314-c8f7b163c86b_main_powerConsumptionReport_powerConsumption_deltaEnergy_meter',
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_energy_difference-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'energy',
'friendly_name': 'Stick vacuum Energy difference',
'state_class': <SensorStateClass.TOTAL: 'total'>,
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
}),
'context': <ANY>,
'entity_id': 'sensor.stick_vacuum_energy_difference',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.003',
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_energy_saved-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.stick_vacuum_energy_saved',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Energy saved',
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.ENERGY: 'energy'>,
'original_icon': None,
'original_name': 'Energy saved',
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'energy_saved',
'unique_id': 'e1f93c0c-6fe0-c65a-a314-c8f7b163c86b_main_powerConsumptionReport_powerConsumption_energySaved_meter',
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_energy_saved-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'energy',
'friendly_name': 'Stick vacuum Energy saved',
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
}),
'context': <ANY>,
'entity_id': 'sensor.stick_vacuum_energy_saved',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.0',
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_power-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.stick_vacuum_power',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Power',
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.POWER: 'power'>,
'original_icon': None,
'original_name': 'Power',
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'e1f93c0c-6fe0-c65a-a314-c8f7b163c86b_main_powerConsumptionReport_powerConsumption_power_meter',
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_power-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'power',
'friendly_name': 'Stick vacuum Power',
'power_consumption_end': '2026-03-21T15:41:41Z',
'power_consumption_start': '2026-03-21T15:35:31Z',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
}),
'context': <ANY>,
'entity_id': 'sensor.stick_vacuum_power',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_power_energy-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.TOTAL: 'total'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.stick_vacuum_power_energy',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Power energy',
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.ENERGY: 'energy'>,
'original_icon': None,
'original_name': 'Power energy',
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'power_energy',
'unique_id': 'e1f93c0c-6fe0-c65a-a314-c8f7b163c86b_main_powerConsumptionReport_powerConsumption_powerEnergy_meter',
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
})
# ---
# name: test_all_entities[da_vc_stick_01001][sensor.stick_vacuum_power_energy-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'energy',
'friendly_name': 'Stick vacuum Power energy',
'state_class': <SensorStateClass.TOTAL: 'total'>,
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
}),
'context': <ANY>,
'entity_id': 'sensor.stick_vacuum_power_energy',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.0',
})
# ---
# name: test_all_entities[da_wm_dw_000001][sensor.dishwasher_completion_time-entry]
EntityRegistryEntrySnapshot({
'aliases': list([

View File

@@ -75,6 +75,7 @@ def mock_window() -> AsyncMock:
window.is_opening = False
window.is_closing = False
window.position = MagicMock(position_percent=30, closed=False)
window.wink = AsyncMock()
window.pyvlx = MagicMock()
return window
@@ -213,7 +214,6 @@ def mock_pyvlx(
mock_blind,
mock_window,
mock_exterior_heating,
mock_cover_type,
]
pyvlx.scenes = [mock_scene]

View File

@@ -1,5 +1,5 @@
# serializer version: 1
# name: test_button_snapshot[button.klf_200_gateway_restart-entry]
# name: test_button_snapshot[mock_window][button.klf_200_gateway_restart-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
@@ -36,7 +36,7 @@
'unit_of_measurement': None,
})
# ---
# name: test_button_snapshot[button.klf_200_gateway_restart-state]
# name: test_button_snapshot[mock_window][button.klf_200_gateway_restart-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'restart',
@@ -50,3 +50,54 @@
'state': 'unknown',
})
# ---
# name: test_button_snapshot[mock_window][button.test_window_identify-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'button.test_window_identify',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Identify',
'options': dict({
}),
'original_device_class': <ButtonDeviceClass.IDENTIFY: 'identify'>,
'original_icon': None,
'original_name': 'Identify',
'platform': 'velux',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '123456789_identify',
'unit_of_measurement': None,
})
# ---
# name: test_button_snapshot[mock_window][button.test_window_identify-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'identify',
'friendly_name': 'Test Window Identify',
}),
'context': <ANY>,
'entity_id': 'button.test_window_identify',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---

View File

@@ -23,6 +23,7 @@ def platform() -> Platform:
@pytest.mark.usefixtures("setup_integration")
@pytest.mark.parametrize("mock_pyvlx", ["mock_window"], indirect=True)
async def test_button_snapshot(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
@@ -38,18 +39,33 @@ async def test_button_snapshot(
mock_config_entry.entry_id,
)
# Get the button entity setup and test device association
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
assert len(entity_entries) == 1
entry = entity_entries[0]
assert len(entity_entries) == 2
assert entry.device_id is not None
device_entry = device_registry.async_get(entry.device_id)
assert device_entry is not None
assert (DOMAIN, f"gateway_{mock_config_entry.entry_id}") in device_entry.identifiers
assert device_entry.via_device_id is None
# Check Reboot button is associated with the gateway device
reboot_entry = next(
e for e in entity_entries if e.entity_id == "button.klf_200_gateway_restart"
)
assert reboot_entry.device_id is not None
gateway_device = device_registry.async_get(reboot_entry.device_id)
assert gateway_device is not None
assert (
DOMAIN,
f"gateway_{mock_config_entry.entry_id}",
) in gateway_device.identifiers
assert gateway_device.via_device_id is None
# Check Identify button is associated with the node device via the gateway
identify_entry = next(
e for e in entity_entries if e.entity_id == "button.test_window_identify"
)
assert identify_entry.device_id is not None
node_device = device_registry.async_get(identify_entry.device_id)
assert node_device is not None
assert (DOMAIN, "123456789") in node_device.identifiers
assert node_device.via_device_id == gateway_device.id
@pytest.mark.usefixtures("setup_integration")
@@ -98,3 +114,54 @@ async def test_button_press_failure(
# Verify the reboot method was called
mock_pyvlx.reboot_gateway.assert_called_once()
@pytest.mark.usefixtures("setup_integration")
@pytest.mark.parametrize("mock_pyvlx", ["mock_window"], indirect=True)
async def test_identify_button_press_success(
hass: HomeAssistant,
mock_window: AsyncMock,
) -> None:
"""Test successful identify button press."""
entity_id = "button.test_window_identify"
# Press the button
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
# Verify the wink method was called
mock_window.wink.assert_awaited_once()
@pytest.mark.usefixtures("setup_integration")
@pytest.mark.parametrize("mock_pyvlx", ["mock_window"], indirect=True)
async def test_identify_button_press_failure(
hass: HomeAssistant,
mock_window: AsyncMock,
) -> None:
"""Test identify button press failure handling."""
entity_id = "button.test_window_identify"
# Mock wink failure
mock_window.wink.side_effect = PyVLXException("Connection failed")
# Press the button and expect HomeAssistantError
with pytest.raises(
HomeAssistantError,
match='Failed to communicate with Velux device: <PyVLXException description="Connection failed" />',
):
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
# Verify the wink method was called
mock_window.wink.assert_awaited_once()