From b2b25ca28cb012ec30eedae5f5d3a94a99434234 Mon Sep 17 00:00:00 2001 From: Robert Resch Date: Fri, 16 Jan 2026 10:06:30 +0100 Subject: [PATCH] Revert "Add SmartThings media-player audio notifications" (#161049) --- homeassistant/components/smartthings/audio.py | 265 --------- .../components/smartthings/manifest.json | 2 +- .../components/smartthings/media_player.py | 42 -- tests/components/smartthings/__init__.py | 37 -- .../snapshots/test_media_player.ambr | 20 +- tests/components/smartthings/test_audio.py | 531 ------------------ .../smartthings/test_media_player.py | 179 +----- 7 files changed, 12 insertions(+), 1064 deletions(-) delete mode 100644 homeassistant/components/smartthings/audio.py delete mode 100644 tests/components/smartthings/test_audio.py diff --git a/homeassistant/components/smartthings/audio.py b/homeassistant/components/smartthings/audio.py deleted file mode 100644 index d45435a8da2..00000000000 --- a/homeassistant/components/smartthings/audio.py +++ /dev/null @@ -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 diff --git a/homeassistant/components/smartthings/manifest.json b/homeassistant/components/smartthings/manifest.json index 4961611583b..2490404e41f 100644 --- a/homeassistant/components/smartthings/manifest.json +++ b/homeassistant/components/smartthings/manifest.json @@ -3,7 +3,7 @@ "name": "SmartThings", "codeowners": ["@joostlek"], "config_flow": true, - "dependencies": ["application_credentials", "http", "ffmpeg"], + "dependencies": ["application_credentials"], "dhcp": [ { "hostname": "st*", diff --git a/homeassistant/components/smartthings/media_player.py b/homeassistant/components/smartthings/media_player.py index 02fa4c787e5..335e8255ae4 100644 --- a/homeassistant/components/smartthings/media_player.py +++ b/homeassistant/components/smartthings/media_player.py @@ -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.""" diff --git a/tests/components/smartthings/__init__.py b/tests/components/smartthings/__init__.py index 35b2269c9b2..3395f7f4673 100644 --- a/tests/components/smartthings/__init__.py +++ b/tests/components/smartthings/__init__.py @@ -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() diff --git a/tests/components/smartthings/snapshots/test_media_player.ambr b/tests/components/smartthings/snapshots/test_media_player.ambr index c098edb01b7..9e11b4e283c 100644 --- a/tests/components/smartthings/snapshots/test_media_player.ambr +++ b/tests/components/smartthings/snapshots/test_media_player.ambr @@ -37,7 +37,7 @@ 'platform': 'smartthings', 'previous_unique_id': None, 'suggested_object_id': None, - 'supported_features': , + 'supported_features': , 'translation_key': None, 'unique_id': 'afcf3b91-0000-1111-2222-ddff2a0a6577_main', 'unit_of_measurement': None, @@ -59,7 +59,7 @@ 'HDMI2', 'digital', ]), - 'supported_features': , + 'supported_features': , 'volume_level': 0.01, }), 'context': , @@ -101,7 +101,7 @@ 'platform': 'smartthings', 'previous_unique_id': None, 'suggested_object_id': None, - 'supported_features': , + 'supported_features': , 'translation_key': None, 'unique_id': 'c9276e43-fe3c-88c3-1dcc-2eb79e292b8c_main', 'unit_of_measurement': None, @@ -115,7 +115,7 @@ 'is_volume_muted': False, 'repeat': , 'shuffle': False, - 'supported_features': , + 'supported_features': , 'volume_level': 0.52, }), 'context': , @@ -157,7 +157,7 @@ 'platform': 'smartthings', 'previous_unique_id': None, 'suggested_object_id': None, - 'supported_features': , + 'supported_features': , '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': , + 'supported_features': , 'volume_level': 0.15, }), 'context': , @@ -213,7 +213,7 @@ 'platform': 'smartthings', 'previous_unique_id': None, 'suggested_object_id': None, - 'supported_features': , + 'supported_features': , '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': , + 'supported_features': , 'volume_level': 0.17, }), 'context': , @@ -270,7 +270,7 @@ 'platform': 'smartthings', 'previous_unique_id': None, 'suggested_object_id': None, - 'supported_features': , + 'supported_features': , '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': , + 'supported_features': , }), 'context': , 'entity_id': 'media_player.soundbar', diff --git a/tests/components/smartthings/test_audio.py b/tests/components/smartthings/test_audio.py deleted file mode 100644 index a4310c90804..00000000000 --- a/tests/components/smartthings/test_audio.py +++ /dev/null @@ -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) diff --git a/tests/components/smartthings/test_media_player.py b/tests/components/smartthings/test_media_player.py index 84c13c5485e..0fb53e642d4 100644 --- a/tests/components/smartthings/test_media_player.py +++ b/tests/components/smartthings/test_media_player.py @@ -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,