Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-02-15 07:36:16 +00:00
Code Activity

Revert "Add SmartThings media-player audio notifications" (#161049)

Browse Source
This commit is contained in:
Robert Resch
2026-01-16 10:06:30 +01:00
committed by GitHub
parent ad9efab16a
commit b2b25ca28c
7 changed files with 12 additions and 1064 deletions
Download Patch File Download Diff File Expand all files Collapse all files

265
homeassistant/components/smartthings/audio.py
View File

@@ -1,265 +0,0 @@
"""Audio helper for SmartThings audio notifications."""
from __future__ import annotations
import asyncio
import contextlib
from dataclasses import dataclass
from datetime import timedelta
import logging
import secrets
from aiohttp import hdrs, web
from homeassistant.components import ffmpeg
from homeassistant.components.http import HomeAssistantView
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.network import NoURLAvailableError, get_url
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
PCM_SAMPLE_RATE = 24000
PCM_SAMPLE_WIDTH = 2
PCM_CHANNELS = 1
PCM_MIME = "audio/L16"
PCM_EXTENSION = ".pcm"
WARNING_DURATION_SECONDS = 40
FFMPEG_MAX_DURATION_SECONDS = 10 * 60
TRANSCODE_TIMEOUT_SECONDS = WARNING_DURATION_SECONDS + 10
_TRUNCATION_EPSILON = 1 / PCM_SAMPLE_RATE
ENTRY_TTL = timedelta(minutes=5)
MAX_STORED_ENTRIES = 4 # Limit the number of cached notifications.
PCM_FRAME_BYTES = PCM_SAMPLE_WIDTH * PCM_CHANNELS
DATA_AUDIO_MANAGER = "audio_manager"
class SmartThingsAudioError(HomeAssistantError):
"""Error raised when SmartThings audio preparation fails."""
@dataclass
class _AudioEntry:
"""Stored PCM audio entry."""
pcm: bytes
created: float
expires: float
class SmartThingsAudioManager(HomeAssistantView):
"""Manage PCM proxy URLs for SmartThings audio notifications."""
url = "/api/smartthings/audio/{token}"
name = "api:smartthings:audio"
requires_auth = False
def __init__(self, hass: HomeAssistant) -> None:
"""Initialize the manager."""
self.hass = hass
self._entries: dict[str, _AudioEntry] = {}
self._cleanup_handle: asyncio.TimerHandle | None = None
async def async_prepare_notification(self, source_url: str) -> str:
"""Generate an externally accessible PCM URL for SmartThings."""
pcm, duration, truncated = await self._transcode_to_pcm(source_url)
if not pcm:
raise SmartThingsAudioError("Converted audio is empty")
if truncated:
_LOGGER.warning(
"SmartThings audio notification truncated to %s seconds (output length %.1fs); longer sources may be cut off",
FFMPEG_MAX_DURATION_SECONDS,
duration,
)
elif duration > WARNING_DURATION_SECONDS:
_LOGGER.warning(
"SmartThings audio notification is %.1fs; playback over %s seconds may be cut off",
duration,
WARNING_DURATION_SECONDS,
)
token = secrets.token_urlsafe(
16
) # Shorter tokens avoid playback issues in some devices.
now = self.hass.loop.time()
entry = _AudioEntry(
pcm=pcm,
created=now,
expires=now + ENTRY_TTL.total_seconds(),
)
self._cleanup(now)
while token in self._entries:
token = secrets.token_urlsafe(16)
self._entries[token] = entry
while len(self._entries) > MAX_STORED_ENTRIES:
dropped_token = next(iter(self._entries))
self._entries.pop(dropped_token, None)
_LOGGER.debug(
"Dropped oldest SmartThings audio token %s to cap cache",
dropped_token,
)
self._schedule_cleanup()
path = f"/api/smartthings/audio/{token}{PCM_EXTENSION}"
try:
base_url = get_url(
self.hass,
allow_internal=True,
allow_external=True,
allow_cloud=True,
prefer_external=False, # Prevent NAT loopback failures; may break non-local access for devices outside the LAN.
prefer_cloud=True,
)
except NoURLAvailableError as err:
self._entries.pop(token, None)
self._schedule_cleanup()
raise SmartThingsAudioError(
"SmartThings audio notifications require an accessible Home Assistant URL"
) from err
return f"{base_url}{path}"
async def get(self, request: web.Request, token: str) -> web.StreamResponse:
"""Serve a PCM audio response."""
token = token.removesuffix(PCM_EXTENSION)
now = self.hass.loop.time()
self._cleanup(now)
self._schedule_cleanup()
entry = self._entries.get(token)
if entry is None:
raise web.HTTPNotFound
_LOGGER.debug("Serving SmartThings audio token=%s to %s", token, request.remote)
response = web.Response(body=entry.pcm, content_type=PCM_MIME)
response.headers[hdrs.CACHE_CONTROL] = "no-store"
response.headers[hdrs.ACCEPT_RANGES] = "none"
response.headers[hdrs.CONTENT_DISPOSITION] = (
f'inline; filename="{token}{PCM_EXTENSION}"'
)
return response
async def _transcode_to_pcm(self, source_url: str) -> tuple[bytes, float, bool]:
"""Use ffmpeg to convert the source media to 24kHz mono PCM."""
manager = ffmpeg.get_ffmpeg_manager(self.hass)
command = [
manager.binary,
"-hide_banner",
"-loglevel",
"error",
"-nostdin",
"-i",
source_url,
"-ac",
str(PCM_CHANNELS),
"-ar",
str(PCM_SAMPLE_RATE),
"-c:a",
"pcm_s16le",
"-t",
str(FFMPEG_MAX_DURATION_SECONDS),
"-f",
"s16le",
"pipe:1",
]
try:
process = await asyncio.create_subprocess_exec(
*command,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
except FileNotFoundError as err:
raise SmartThingsAudioError(
"FFmpeg is required for SmartThings audio notifications"
) from err
try:
stdout, stderr = await asyncio.wait_for(
process.communicate(), timeout=TRANSCODE_TIMEOUT_SECONDS
)
except TimeoutError:
_LOGGER.warning(
"FFmpeg timed out after %s seconds while converting SmartThings audio from %s",
TRANSCODE_TIMEOUT_SECONDS,
source_url,
)
with contextlib.suppress(ProcessLookupError):
process.kill()
stdout, stderr = await process.communicate()
if process.returncode != 0:
message = stderr.decode().strip() or "unknown error"
_LOGGER.error(
"FFmpeg failed to convert SmartThings audio from %s: %s",
source_url,
message,
)
raise SmartThingsAudioError(
"Unable to convert audio to PCM for SmartThings"
)
if not stdout:
return b"", 0.0, False
frame_count, remainder = divmod(len(stdout), PCM_FRAME_BYTES)
if remainder:
_LOGGER.debug(
"SmartThings audio conversion produced misaligned PCM: dropping %s extra byte(s)",
remainder,
)
stdout = stdout[: len(stdout) - remainder]
frame_count = len(stdout) // PCM_FRAME_BYTES
if frame_count == 0:
return b"", 0.0, False
duration = frame_count / PCM_SAMPLE_RATE
truncated = duration >= (FFMPEG_MAX_DURATION_SECONDS - _TRUNCATION_EPSILON)
return stdout, duration, truncated
@callback
def _schedule_cleanup(self) -> None:
"""Schedule the next cleanup based on entry expiry."""
if self._cleanup_handle is not None:
self._cleanup_handle.cancel()
self._cleanup_handle = None
if not self._entries:
return
next_expiry = min(entry.expires for entry in self._entries.values())
delay = max(0.0, next_expiry - self.hass.loop.time())
self._cleanup_handle = self.hass.loop.call_later(delay, self._cleanup_callback)
@callback
def _cleanup_callback(self) -> None:
"""Run a cleanup pass."""
self._cleanup_handle = None
now = self.hass.loop.time()
self._cleanup(now)
self._schedule_cleanup()
def _cleanup(self, now: float) -> None:
"""Remove expired entries."""
expired = [
token for token, entry in self._entries.items() if entry.expires <= now
]
for token in expired:
self._entries.pop(token, None)
async def async_get_audio_manager(hass: HomeAssistant) -> SmartThingsAudioManager:
"""Return the shared SmartThings audio manager."""
domain_data = hass.data.setdefault(DOMAIN, {})
if (manager := domain_data.get(DATA_AUDIO_MANAGER)) is None:
manager = SmartThingsAudioManager(hass)
hass.http.register_view(manager)
domain_data[DATA_AUDIO_MANAGER] = manager
return manager

2
homeassistant/components/smartthings/manifest.json
View File

@@ -3,7 +3,7 @@
"name": "SmartThings",
"codeowners": ["@joostlek"],
"config_flow": true,
"dependencies": ["application_credentials", "http", "ffmpeg"],
"dependencies": ["application_credentials"],
"dhcp": [
{
"hostname": "st*",

42
homeassistant/components/smartthings/media_player.py
View File

@@ -6,22 +6,17 @@ from typing import Any
from pysmartthings import Attribute, Capability, Category, Command, SmartThings
from homeassistant.components import media_source
from homeassistant.components.media_player import (
MediaPlayerDeviceClass,
MediaPlayerEntity,
MediaPlayerEntityFeature,
MediaPlayerState,
MediaType,
RepeatMode,
async_process_play_media_url,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import FullDevice, SmartThingsConfigEntry
from .audio import SmartThingsAudioError, async_get_audio_manager
from .const import MAIN
from .entity import SmartThingsEntity
@@ -89,7 +84,6 @@ class SmartThingsMediaPlayer(SmartThingsEntity, MediaPlayerEntity):
Capability.AUDIO_MUTE,
Capability.AUDIO_TRACK_DATA,
Capability.AUDIO_VOLUME,
Capability.AUDIO_NOTIFICATION,
Capability.MEDIA_INPUT_SOURCE,
Capability.MEDIA_PLAYBACK,
Capability.MEDIA_PLAYBACK_REPEAT,
@@ -134,8 +128,6 @@ class SmartThingsMediaPlayer(SmartThingsEntity, MediaPlayerEntity):
flags |= MediaPlayerEntityFeature.SHUFFLE_SET
if self.supports_capability(Capability.MEDIA_PLAYBACK_REPEAT):
flags |= MediaPlayerEntityFeature.REPEAT_SET
if self.supports_capability(Capability.AUDIO_NOTIFICATION):
flags |= MediaPlayerEntityFeature.PLAY_MEDIA
return flags
async def async_turn_off(self, **kwargs: Any) -> None:
@@ -241,40 +233,6 @@ class SmartThingsMediaPlayer(SmartThingsEntity, MediaPlayerEntity):
argument=HA_REPEAT_MODE_TO_SMARTTHINGS[repeat],
)
async def async_play_media(
self, media_type: MediaType | str, media_id: str, **kwargs: Any
) -> None:
"""Play media using SmartThings audio notifications."""
if not self.supports_capability(Capability.AUDIO_NOTIFICATION):
raise HomeAssistantError("Device does not support audio notifications")
if media_type not in (MediaType.MUSIC,):
raise HomeAssistantError(
"Unsupported media type for SmartThings audio notification"
)
if media_source.is_media_source_id(media_id):
play_item = await media_source.async_resolve_media(
self.hass, media_id, self.entity_id
)
media_id = async_process_play_media_url(self.hass, play_item.url)
else:
media_id = async_process_play_media_url(self.hass, media_id)
audio_manager = await async_get_audio_manager(self.hass)
try:
proxy_url = await audio_manager.async_prepare_notification(media_id)
except SmartThingsAudioError as err:
raise HomeAssistantError(str(err)) from err
command = Command("playTrackAndResume")
await self.execute_device_command(
Capability.AUDIO_NOTIFICATION,
command,
argument=[proxy_url],
)
@property
def media_title(self) -> str | None:
"""Title of current playing media."""

37
tests/components/smartthings/__init__.py
View File

@@ -1,7 +1,5 @@
"""Tests for the SmartThings integration."""
import sys
import types
from typing import Any
from unittest.mock import AsyncMock
@@ -92,38 +90,3 @@ async def trigger_health_update(
if call[0][0] == device_id:
call[0][1](event)
await hass.async_block_till_done()
def ensure_haffmpeg_stubs() -> None:
"""Ensure haffmpeg stubs are available for SmartThings tests."""
if "haffmpeg" in sys.modules:
return
haffmpeg_module = types.ModuleType("haffmpeg")
haffmpeg_core_module = types.ModuleType("haffmpeg.core")
haffmpeg_tools_module = types.ModuleType("haffmpeg.tools")
class _StubHAFFmpeg: ...
class _StubFFVersion:
def __init__(self, bin_path: str | None = None) -> None:
self.bin_path = bin_path
async def get_version(self) -> str:
return "4.0.0"
class _StubImageFrame: ...
haffmpeg_core_module.HAFFmpeg = _StubHAFFmpeg
haffmpeg_tools_module.IMAGE_JPEG = b""
haffmpeg_tools_module.FFVersion = _StubFFVersion
haffmpeg_tools_module.ImageFrame = _StubImageFrame
haffmpeg_module.core = haffmpeg_core_module
haffmpeg_module.tools = haffmpeg_tools_module
sys.modules["haffmpeg"] = haffmpeg_module
sys.modules["haffmpeg.core"] = haffmpeg_core_module
sys.modules["haffmpeg.tools"] = haffmpeg_tools_module
ensure_haffmpeg_stubs()

20
tests/components/smartthings/snapshots/test_media_player.ambr
View File

@@ -37,7 +37,7 @@
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <MediaPlayerEntityFeature: 24461>,
'supported_features': <MediaPlayerEntityFeature: 23949>,
'translation_key': None,
'unique_id': 'afcf3b91-0000-1111-2222-ddff2a0a6577_main',
'unit_of_measurement': None,
@@ -59,7 +59,7 @@
'HDMI2',
'digital',
]),
'supported_features': <MediaPlayerEntityFeature: 24461>,
'supported_features': <MediaPlayerEntityFeature: 23949>,
'volume_level': 0.01,
}),
'context': <ANY>,
@@ -101,7 +101,7 @@
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <MediaPlayerEntityFeature: 318989>,
'supported_features': <MediaPlayerEntityFeature: 318477>,
'translation_key': None,
'unique_id': 'c9276e43-fe3c-88c3-1dcc-2eb79e292b8c_main',
'unit_of_measurement': None,
@@ -115,7 +115,7 @@
'is_volume_muted': False,
'repeat': <RepeatMode.OFF: 'off'>,
'shuffle': False,
'supported_features': <MediaPlayerEntityFeature: 318989>,
'supported_features': <MediaPlayerEntityFeature: 318477>,
'volume_level': 0.52,
}),
'context': <ANY>,
@@ -157,7 +157,7 @@
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <MediaPlayerEntityFeature: 22029>,
'supported_features': <MediaPlayerEntityFeature: 21517>,
'translation_key': None,
'unique_id': 'c85fced9-c474-4a47-93c2-037cc7829536_main',
'unit_of_measurement': None,
@@ -171,7 +171,7 @@
'is_volume_muted': False,
'media_artist': 'David Guetta',
'media_title': 'Forever Young',
'supported_features': <MediaPlayerEntityFeature: 22029>,
'supported_features': <MediaPlayerEntityFeature: 21517>,
'volume_level': 0.15,
}),
'context': <ANY>,
@@ -213,7 +213,7 @@
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <MediaPlayerEntityFeature: 22413>,
'supported_features': <MediaPlayerEntityFeature: 21901>,
'translation_key': None,
'unique_id': '0d94e5db-8501-2355-eb4f-214163702cac_main',
'unit_of_measurement': None,
@@ -228,7 +228,7 @@
'media_artist': '',
'media_title': '',
'source': 'HDMI1',
'supported_features': <MediaPlayerEntityFeature: 22413>,
'supported_features': <MediaPlayerEntityFeature: 21901>,
'volume_level': 0.17,
}),
'context': <ANY>,
@@ -270,7 +270,7 @@
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <MediaPlayerEntityFeature: 1932>,
'supported_features': <MediaPlayerEntityFeature: 1420>,
'translation_key': None,
'unique_id': 'a75cb1e1-03fd-3c77-ca9f-d4e56c4096c6_main',
'unit_of_measurement': None,
@@ -281,7 +281,7 @@
'attributes': ReadOnlyDict({
'device_class': 'speaker',
'friendly_name': 'Soundbar',
'supported_features': <MediaPlayerEntityFeature: 1932>,
'supported_features': <MediaPlayerEntityFeature: 1420>,
}),
'context': <ANY>,
'entity_id': 'media_player.soundbar',

531
tests/components/smartthings/test_audio.py
View File

@@ -1,531 +0,0 @@
"""Tests for SmartThings audio helper."""
from __future__ import annotations
import asyncio
import logging
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
from urllib.parse import urlsplit
import pytest
from homeassistant.components.smartthings.audio import (
FFMPEG_MAX_DURATION_SECONDS,
MAX_STORED_ENTRIES,
PCM_CHANNELS,
PCM_MIME,
PCM_SAMPLE_RATE,
PCM_SAMPLE_WIDTH,
TRANSCODE_TIMEOUT_SECONDS,
WARNING_DURATION_SECONDS,
SmartThingsAudioError,
async_get_audio_manager,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.network import NoURLAvailableError
from tests.typing import ClientSessionGenerator
class _FakeProcess:
"""Async subprocess stand-in that provides communicate."""
def __init__(self, stdout: bytes, stderr: bytes, returncode: int) -> None:
self._stdout = stdout
self._stderr = stderr
self.returncode = returncode
self.killed = False
async def communicate(self) -> tuple[bytes, bytes]:
return self._stdout, self._stderr
def kill(self) -> None:
self.killed = True
def _build_pcm(
duration_seconds: float = 1.0,
*,
sample_rate: int = PCM_SAMPLE_RATE,
sample_width: int = PCM_SAMPLE_WIDTH,
channels: int = PCM_CHANNELS,
) -> bytes:
"""Generate silent raw PCM bytes for testing."""
frame_count = int(sample_rate * duration_seconds)
return b"\x00" * frame_count * sample_width * channels
async def test_prepare_notification_creates_url(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
) -> None:
"""Ensure PCM proxy URLs are generated and served."""
hass.config.external_url = "https://example.com"
manager = await async_get_audio_manager(hass)
pcm_bytes = _build_pcm()
with patch.object(
manager, "_transcode_to_pcm", AsyncMock(return_value=(pcm_bytes, 1.0, False))
):
url = await manager.async_prepare_notification("https://example.com/source.mp3")
parsed = urlsplit(url)
assert parsed.path.endswith(".pcm")
assert not parsed.query
client = await hass_client_no_auth()
response = await client.get(parsed.path)
assert response.status == 200
assert response.headers["Content-Type"] == PCM_MIME
assert response.headers["Cache-Control"] == "no-store"
body = await response.read()
assert body == pcm_bytes
@pytest.mark.asyncio
async def test_prepare_notification_uses_internal_url_when_external_missing(
hass: HomeAssistant,
) -> None:
"""Fallback to the internal URL if no external URL is available."""
hass.config.external_url = None
hass.config.internal_url = "http://homeassistant.local:8123"
manager = await async_get_audio_manager(hass)
pcm_bytes = _build_pcm()
with patch.object(
manager, "_transcode_to_pcm", AsyncMock(return_value=(pcm_bytes, 1.0, False))
):
url = await manager.async_prepare_notification("https://example.com/source.mp3")
parsed = urlsplit(url)
assert parsed.scheme == "http"
assert parsed.netloc == "homeassistant.local:8123"
assert parsed.path.endswith(".pcm")
@pytest.mark.asyncio
async def test_prepare_notification_requires_accessible_url(
hass: HomeAssistant,
) -> None:
"""Fail if neither external nor internal URLs are available."""
hass.config.external_url = None
hass.config.internal_url = None
manager = await async_get_audio_manager(hass)
pcm_bytes = _build_pcm()
with (
patch.object(
manager,
"_transcode_to_pcm",
AsyncMock(return_value=(pcm_bytes, 1.0, False)),
),
patch(
"homeassistant.components.smartthings.audio.get_url",
side_effect=NoURLAvailableError,
) as mock_get_url,
pytest.raises(SmartThingsAudioError),
):
await manager.async_prepare_notification("https://example.com/source.mp3")
assert mock_get_url.called
# Stored entry should be cleaned up after failure so subsequent requests
# don't leak memory or serve stale audio.
assert not manager._entries
async def test_audio_view_returns_404_for_unknown_token(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
) -> None:
"""Unknown tokens should return 404."""
await async_get_audio_manager(hass)
client = await hass_client_no_auth()
response = await client.get("/api/smartthings/audio/invalid-token.pcm")
assert response.status == 404
@pytest.mark.asyncio
async def test_prepare_notification_raises_when_transcode_empty(
hass: HomeAssistant,
) -> None:
"""Transcoding empty audio results in an error."""
hass.config.external_url = "https://example.com"
manager = await async_get_audio_manager(hass)
with (
patch.object(
manager, "_transcode_to_pcm", AsyncMock(return_value=(b"", 0.0, False))
),
pytest.raises(SmartThingsAudioError, match="Converted audio is empty"),
):
await manager.async_prepare_notification("https://example.com/source.mp3")
@pytest.mark.asyncio
async def test_prepare_notification_warns_when_duration_exceeds_max(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Warn when transcoded audio exceeds the SmartThings duration limit."""
hass.config.external_url = "https://example.com"
manager = await async_get_audio_manager(hass)
pcm_bytes = b"pcm"
caplog.set_level(logging.WARNING)
with patch.object(
manager,
"_transcode_to_pcm",
AsyncMock(return_value=(pcm_bytes, FFMPEG_MAX_DURATION_SECONDS + 1.0, True)),
):
await manager.async_prepare_notification("https://example.com/source.mp3")
assert any("truncated" in record.message for record in caplog.records)
@pytest.mark.asyncio
async def test_prepare_notification_warns_when_duration_exceeds_warning(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Warn when transcoded audio exceeds the SmartThings warning threshold."""
hass.config.external_url = "https://example.com"
manager = await async_get_audio_manager(hass)
pcm_bytes = _build_pcm(duration_seconds=WARNING_DURATION_SECONDS + 1)
caplog.set_level(logging.WARNING)
with patch.object(
manager,
"_transcode_to_pcm",
AsyncMock(return_value=(pcm_bytes, WARNING_DURATION_SECONDS + 1.0, False)),
):
await manager.async_prepare_notification("https://example.com/source.mp3")
assert any(
"playback over" in record.message and "truncated" not in record.message
for record in caplog.records
)
@pytest.mark.asyncio
async def test_prepare_notification_regenerates_token_on_collision(
hass: HomeAssistant,
) -> None:
"""Regenerate tokens when a collision is detected."""
hass.config.external_url = "https://example.com"
manager = await async_get_audio_manager(hass)
pcm_bytes = _build_pcm()
with (
patch.object(
manager,
"_transcode_to_pcm",
AsyncMock(return_value=(pcm_bytes, 1.0, False)),
),
patch(
"homeassistant.components.smartthings.audio.secrets.token_urlsafe",
side_effect=["dup", "dup", "unique"],
),
):
url1 = await manager.async_prepare_notification(
"https://example.com/source.mp3"
)
url2 = await manager.async_prepare_notification(
"https://example.com/source.mp3"
)
assert urlsplit(url1).path.endswith("/dup.pcm")
assert urlsplit(url2).path.endswith("/unique.pcm")
@pytest.mark.asyncio
async def test_prepare_notification_schedules_cleanup(
hass: HomeAssistant,
) -> None:
"""Ensure cached entries are scheduled for cleanup."""
hass.config.external_url = "https://example.com"
manager = await async_get_audio_manager(hass)
pcm_bytes = _build_pcm()
with patch.object(
manager,
"_transcode_to_pcm",
AsyncMock(return_value=(pcm_bytes, 1.0, False)),
):
await manager.async_prepare_notification("https://example.com/source.mp3")
assert manager._cleanup_handle is not None
for entry in manager._entries.values():
entry.expires = 0
manager._cleanup_callback()
assert not manager._entries
@pytest.mark.asyncio
async def test_prepare_notification_caps_entry_count(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Ensure cached entries are capped."""
hass.config.external_url = "https://example.com"
manager = await async_get_audio_manager(hass)
pcm_bytes = _build_pcm()
caplog.set_level(logging.DEBUG)
with patch.object(
manager,
"_transcode_to_pcm",
AsyncMock(return_value=(pcm_bytes, 1.0, False)),
):
for _ in range(MAX_STORED_ENTRIES + 2):
await manager.async_prepare_notification("https://example.com/source.mp3")
assert len(manager._entries) == MAX_STORED_ENTRIES
assert any(
"Dropped oldest SmartThings audio token" in record.message
for record in caplog.records
)
@pytest.mark.asyncio
async def test_transcode_to_pcm_handles_missing_ffmpeg(
hass: HomeAssistant,
) -> None:
"""Raise friendly error when ffmpeg is unavailable."""
manager = await async_get_audio_manager(hass)
with (
patch(
"homeassistant.components.smartthings.audio.ffmpeg.get_ffmpeg_manager",
return_value=SimpleNamespace(binary="ffmpeg"),
),
patch(
"homeassistant.components.smartthings.audio.asyncio.create_subprocess_exec",
side_effect=FileNotFoundError,
),
pytest.raises(SmartThingsAudioError, match="FFmpeg is required"),
):
await manager._transcode_to_pcm("https://example.com/source.mp3")
@pytest.mark.asyncio
async def test_transcode_to_pcm_handles_process_failure(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Raise when ffmpeg reports an error."""
manager = await async_get_audio_manager(hass)
caplog.set_level(logging.ERROR)
fake_process = _FakeProcess(stdout=b"", stderr=b"boom", returncode=1)
with (
patch(
"homeassistant.components.smartthings.audio.ffmpeg.get_ffmpeg_manager",
return_value=SimpleNamespace(binary="ffmpeg"),
),
patch(
"homeassistant.components.smartthings.audio.asyncio.create_subprocess_exec",
AsyncMock(return_value=fake_process),
),
pytest.raises(SmartThingsAudioError, match="Unable to convert"),
):
await manager._transcode_to_pcm("https://example.com/source.mp3")
assert any("FFmpeg failed" in record.message for record in caplog.records)
@pytest.mark.asyncio
async def test_transcode_to_pcm_times_out_and_kills_process(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Kill ffmpeg when the transcode times out."""
manager = await async_get_audio_manager(hass)
fake_process = _FakeProcess(stdout=b"\x00\x00", stderr=b"", returncode=0)
caplog.set_level(logging.WARNING)
with (
patch(
"homeassistant.components.smartthings.audio.ffmpeg.get_ffmpeg_manager",
return_value=SimpleNamespace(binary="ffmpeg"),
),
patch(
"homeassistant.components.smartthings.audio.asyncio.create_subprocess_exec",
AsyncMock(return_value=fake_process),
),
patch(
"homeassistant.components.smartthings.audio.asyncio.wait_for",
side_effect=TimeoutError,
),
):
pcm, duration, truncated = await manager._transcode_to_pcm(
"https://example.com/source.mp3"
)
assert fake_process.killed is True
assert pcm == b"\x00\x00"
assert duration == pytest.approx(1 / PCM_SAMPLE_RATE)
assert truncated is False
assert any("FFmpeg timed out" in record.message for record in caplog.records)
@pytest.mark.asyncio
async def test_transcode_to_pcm_returns_empty_audio(
hass: HomeAssistant,
) -> None:
"""Return empty payload when ffmpeg produced nothing."""
manager = await async_get_audio_manager(hass)
fake_process = _FakeProcess(stdout=b"", stderr=b"", returncode=0)
with (
patch(
"homeassistant.components.smartthings.audio.ffmpeg.get_ffmpeg_manager",
return_value=SimpleNamespace(binary="ffmpeg"),
),
patch(
"homeassistant.components.smartthings.audio.asyncio.create_subprocess_exec",
AsyncMock(return_value=fake_process),
) as mock_exec,
):
pcm, duration, truncated = await manager._transcode_to_pcm(
"https://example.com/source.mp3"
)
assert pcm == b""
assert duration == 0.0
assert truncated is False
mock_exec.assert_awaited_once()
@pytest.mark.asyncio
async def test_transcode_to_pcm_enforces_duration_cap(
hass: HomeAssistant,
) -> None:
"""Ensure ffmpeg is instructed to limit duration and timeout is enforced."""
manager = await async_get_audio_manager(hass)
pcm_bytes = _build_pcm(duration_seconds=FFMPEG_MAX_DURATION_SECONDS)
fake_process = _FakeProcess(stdout=pcm_bytes, stderr=b"", returncode=0)
timeouts: list[float] = []
original_wait_for = asyncio.wait_for
async def _wait_for(awaitable, timeout):
timeouts.append(timeout)
return await original_wait_for(awaitable, timeout)
mock_exec = AsyncMock(return_value=fake_process)
with (
patch(
"homeassistant.components.smartthings.audio.ffmpeg.get_ffmpeg_manager",
return_value=SimpleNamespace(binary="ffmpeg"),
),
patch(
"homeassistant.components.smartthings.audio.asyncio.create_subprocess_exec",
mock_exec,
),
patch(
"homeassistant.components.smartthings.audio.asyncio.wait_for",
new=_wait_for,
),
):
pcm, duration, truncated = await manager._transcode_to_pcm(
"https://example.com/source.mp3"
)
command = list(mock_exec.await_args.args)
assert "-t" in command
assert command[command.index("-t") + 1] == str(FFMPEG_MAX_DURATION_SECONDS)
assert timeouts == [TRANSCODE_TIMEOUT_SECONDS]
assert pcm == pcm_bytes
assert duration == pytest.approx(FFMPEG_MAX_DURATION_SECONDS)
assert truncated is True
async def test_transcode_to_pcm_logs_misaligned_pcm(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Log debug output when ffmpeg output contains a partial frame."""
manager = await async_get_audio_manager(hass)
caplog.set_level(logging.DEBUG)
pcm_bytes = _build_pcm() + b"\xaa"
fake_process = _FakeProcess(stdout=pcm_bytes, stderr=b"", returncode=0)
with (
patch(
"homeassistant.components.smartthings.audio.ffmpeg.get_ffmpeg_manager",
return_value=SimpleNamespace(binary="ffmpeg"),
),
patch(
"homeassistant.components.smartthings.audio.asyncio.create_subprocess_exec",
AsyncMock(return_value=fake_process),
),
):
pcm, duration, truncated = await manager._transcode_to_pcm(
"https://example.com/source.mp3"
)
assert pcm == _build_pcm()
assert duration > 0
assert truncated is False
assert any("misaligned PCM" in record.message for record in caplog.records)
@pytest.mark.asyncio
async def test_transcode_to_pcm_drops_partial_frame_payload(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Drop audio entirely when ffmpeg returns fewer bytes than a full frame."""
manager = await async_get_audio_manager(hass)
caplog.set_level(logging.DEBUG)
fake_process = _FakeProcess(stdout=b"\x00", stderr=b"", returncode=0)
with (
patch(
"homeassistant.components.smartthings.audio.ffmpeg.get_ffmpeg_manager",
return_value=SimpleNamespace(binary="ffmpeg"),
),
patch(
"homeassistant.components.smartthings.audio.asyncio.create_subprocess_exec",
AsyncMock(return_value=fake_process),
),
):
pcm, duration, truncated = await manager._transcode_to_pcm(
"https://example.com/source.mp3"
)
assert pcm == b""
assert duration == 0.0
assert truncated is False
assert any("misaligned PCM" in record.message for record in caplog.records)

179
tests/components/smartthings/test_media_player.py
View File

@@ -1,7 +1,6 @@
"""Test for the SmartThings media player platform."""
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock
from pysmartthings import Attribute, Capability, Command, Status
from pysmartthings.models import HealthStatus
@@ -10,19 +9,14 @@ from syrupy.assertion import SnapshotAssertion
from homeassistant.components.media_player import (
ATTR_INPUT_SOURCE,
ATTR_MEDIA_CONTENT_ID,
ATTR_MEDIA_CONTENT_TYPE,
ATTR_MEDIA_REPEAT,
ATTR_MEDIA_SHUFFLE,
ATTR_MEDIA_VOLUME_LEVEL,
ATTR_MEDIA_VOLUME_MUTED,
DOMAIN as MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
SERVICE_SELECT_SOURCE,
MediaType,
RepeatMode,
)
from homeassistant.components.smartthings.audio import SmartThingsAudioError
from homeassistant.components.smartthings.const import MAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
@@ -45,7 +39,6 @@ from homeassistant.const import (
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from . import (
@@ -205,176 +198,6 @@ async def test_volume_down(
)
@pytest.mark.parametrize("device_fixture", ["hw_q80r_soundbar"])
async def test_play_media_notification(
hass: HomeAssistant,
devices: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test playing media via SmartThings audio notification."""
await setup_integration(hass, mock_config_entry)
manager = AsyncMock()
manager.async_prepare_notification.return_value = "https://example.com/audio.pcm"
with (
patch(
"homeassistant.components.smartthings.media_player.async_get_audio_manager",
AsyncMock(return_value=manager),
),
patch(
"homeassistant.components.smartthings.media_player.async_process_play_media_url",
return_value="https://example.com/source.mp3",
),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: "media_player.soundbar",
ATTR_MEDIA_CONTENT_TYPE: MediaType.MUSIC,
ATTR_MEDIA_CONTENT_ID: "https://example.com/source.mp3",
},
blocking=True,
)
expected_command = Command("playTrackAndResume")
devices.execute_device_command.assert_called_once_with(
"afcf3b91-0000-1111-2222-ddff2a0a6577",
Capability.AUDIO_NOTIFICATION,
expected_command,
MAIN,
argument=["https://example.com/audio.pcm"],
)
@pytest.mark.parametrize("device_fixture", ["hw_q80r_soundbar"])
async def test_play_media_requires_audio_notification_capability(
hass: HomeAssistant,
devices: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Expect an error if the device lacks audio notification support."""
devices.get_device_status.return_value[MAIN].pop(
Capability.AUDIO_NOTIFICATION, None
)
await setup_integration(hass, mock_config_entry)
entity = hass.data["entity_components"][MEDIA_PLAYER_DOMAIN].get_entity(
"media_player.soundbar"
)
assert entity is not None
with pytest.raises(
HomeAssistantError, match="Device does not support audio notifications"
):
await entity.async_play_media(MediaType.MUSIC, "https://example.com/source.mp3")
@pytest.mark.parametrize("device_fixture", ["hw_q80r_soundbar"])
async def test_play_media_rejects_unsupported_media_type(
hass: HomeAssistant,
devices: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Unsupported media types should raise an error."""
await setup_integration(hass, mock_config_entry)
entity = hass.data["entity_components"][MEDIA_PLAYER_DOMAIN].get_entity(
"media_player.soundbar"
)
assert entity is not None
with pytest.raises(
HomeAssistantError, match="Unsupported media type for SmartThings audio"
):
await entity.async_play_media(
MediaType.TVSHOW, "https://example.com/source.mp3"
)
@pytest.mark.parametrize("device_fixture", ["hw_q80r_soundbar"])
async def test_play_media_uses_media_source_resolution(
hass: HomeAssistant,
devices: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Media source IDs are resolved and processed before playback."""
await setup_integration(hass, mock_config_entry)
manager = AsyncMock()
manager.async_prepare_notification.return_value = "https://example.com/audio.pcm"
with (
patch(
"homeassistant.components.smartthings.media_player.async_get_audio_manager",
AsyncMock(return_value=manager),
),
patch(
"homeassistant.components.smartthings.media_player.async_process_play_media_url",
return_value="https://example.com/processed.mp3",
) as mock_process,
patch(
"homeassistant.components.smartthings.media_player.media_source.is_media_source_id",
return_value=True,
) as mock_is_media,
patch(
"homeassistant.components.smartthings.media_player.media_source.async_resolve_media",
AsyncMock(
return_value=SimpleNamespace(url="https://example.com/from_source")
),
) as mock_resolve,
):
entity = hass.data["entity_components"][MEDIA_PLAYER_DOMAIN].get_entity(
"media_player.soundbar"
)
assert entity is not None
await entity.async_play_media(MediaType.MUSIC, "media-source://foo")
mock_is_media.assert_called_once()
mock_resolve.assert_called_once()
mock_process.assert_called_with(hass, "https://example.com/from_source")
devices.execute_device_command.assert_called_once()
@pytest.mark.parametrize("device_fixture", ["hw_q80r_soundbar"])
async def test_play_media_wraps_audio_errors(
hass: HomeAssistant,
devices: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""SmartThings audio errors propagate as HomeAssistantError."""
await setup_integration(hass, mock_config_entry)
manager = AsyncMock()
manager.async_prepare_notification.side_effect = SmartThingsAudioError("boom")
entity = hass.data["entity_components"][MEDIA_PLAYER_DOMAIN].get_entity(
"media_player.soundbar"
)
assert entity is not None
with (
patch(
"homeassistant.components.smartthings.media_player.async_get_audio_manager",
AsyncMock(return_value=manager),
),
patch(
"homeassistant.components.smartthings.media_player.async_process_play_media_url",
return_value="https://example.com/source.mp3",
),
pytest.raises(HomeAssistantError, match="boom"),
):
await entity.async_play_media(MediaType.MUSIC, "https://example.com/source.mp3")
@pytest.mark.parametrize("device_fixture", ["hw_q80r_soundbar"])
async def test_media_play(
hass: HomeAssistant,