Add metadata support to Snapcast media players (#132283)

Co-authored-by: Joostlek <joostlek@outlook.com>

homeassistant/components/snapcast/media_player.py

@@ -12,9 +12,11 @@ import voluptuous as vol
 
 from homeassistant.components.media_player import (
     DOMAIN as MEDIA_PLAYER_DOMAIN,
+    MediaPlayerDeviceClass,
     MediaPlayerEntity,
     MediaPlayerEntityFeature,
     MediaPlayerState,
+    MediaType,
 )
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_HOST, CONF_PORT
@@ -180,6 +182,8 @@ class SnapcastBaseDevice(SnapcastCoordinatorEntity, MediaPlayerEntity):
         | MediaPlayerEntityFeature.VOLUME_SET
         | MediaPlayerEntityFeature.SELECT_SOURCE
     )
+    _attr_media_content_type = MediaType.MUSIC
+    _attr_device_class = MediaPlayerDeviceClass.SPEAKER
 
     def __init__(
         self,
@@ -275,6 +279,76 @@ class SnapcastBaseDevice(SnapcastCoordinatorEntity, MediaPlayerEntity):
         """Handle the unjoin service."""
         raise NotImplementedError
 
+    @property
+    def metadata(self) -> Mapping[str, Any]:
+        """Get metadata from the current stream."""
+        if metadata := self.coordinator.server.stream(
+            self._current_group.stream
+        ).metadata:
+            return metadata
+
+        # Fallback to an empty dict
+        return {}
+
+    @property
+    def media_title(self) -> str | None:
+        """Title of current playing media."""
+        return self.metadata.get("title")
+
+    @property
+    def media_image_url(self) -> str | None:
+        """Image url of current playing media."""
+        return self.metadata.get("artUrl")
+
+    @property
+    def media_artist(self) -> str | None:
+        """Artist of current playing media, music track only."""
+        if (value := self.metadata.get("artist")) is not None:
+            return ", ".join(value)
+
+        return None
+
+    @property
+    def media_album_name(self) -> str | None:
+        """Album name of current playing media, music track only."""
+        return self.metadata.get("album")
+
+    @property
+    def media_album_artist(self) -> str | None:
+        """Album artist of current playing media, music track only."""
+        if (value := self.metadata.get("albumArtist")) is not None:
+            return ", ".join(value)
+
+        return None
+
+    @property
+    def media_track(self) -> int | None:
+        """Track number of current playing media, music track only."""
+        if (value := self.metadata.get("trackNumber")) is not None:
+            return int(value)
+
+        return None
+
+    @property
+    def media_duration(self) -> int | None:
+        """Duration of current playing media in seconds."""
+        if (value := self.metadata.get("duration")) is not None:
+            return int(value)
+
+        return None
+
+    @property
+    def media_position(self) -> int | None:
+        """Position of current playing media in seconds."""
+        # Position is part of properties object, not metadata object
+        if properties := self.coordinator.server.stream(
+            self._current_group.stream
+        ).properties:
+            if (value := properties.get("position")) is not None:
+                return int(value)
+
+        return None
+
 
 class SnapcastGroupDevice(SnapcastBaseDevice):
     """Representation of a Snapcast group device."""

tests/components/snapcast/__init__.py

@@ -1 +1,13 @@
 """Tests for the Snapcast integration."""
+
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+
+
+async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
+    """Set up the Snapcast integration in Home Assistant."""
+    config_entry.add_to_hass(hass)
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()

tests/components/snapcast/conftest.py

@@ -1,9 +1,19 @@
 """Test the snapcast config flow."""
 
 from collections.abc import Generator
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, patch
 
 import pytest
+from snapcast.control.client import Snapclient
+from snapcast.control.group import Snapgroup
+from snapcast.control.server import CONTROL_PORT
+from snapcast.control.stream import Snapstream
+
+from homeassistant.components.snapcast.const import DOMAIN
+from homeassistant.components.snapcast.coordinator import Snapserver
+from homeassistant.const import CONF_HOST, CONF_PORT
+
+from tests.common import MockConfigEntry
 
 
 @pytest.fixture
@@ -16,10 +26,144 @@ def mock_setup_entry() -> Generator[AsyncMock]:
 
 
 @pytest.fixture
-def mock_create_server() -> Generator[AsyncMock]:
+def mock_create_server(
+    mock_group: AsyncMock,
+    mock_client: AsyncMock,
+    mock_stream_1: AsyncMock,
+    mock_stream_2: AsyncMock,
+) -> Generator[AsyncMock]:
     """Create mock snapcast connection."""
-    mock_connection = AsyncMock()
+    with patch(
-    mock_connection.start = AsyncMock(return_value=None)
+        "homeassistant.components.snapcast.coordinator.Snapserver", autospec=True
-    mock_connection.stop = MagicMock()
+    ) as mock_snapserver:
-    with patch("snapcast.control.create_server", return_value=mock_connection):
+        mock_server = mock_snapserver.return_value
-        yield mock_connection
+        mock_server.groups = [mock_group]
+        mock_server.clients = [mock_client]
+        mock_server.streams = [mock_stream_1, mock_stream_2]
+        mock_server.group.return_value = mock_group
+        mock_server.client.return_value = mock_client
+
+        def get_stream(identifier: str) -> AsyncMock:
+            return {s.identifier: s for s in mock_server.streams}[identifier]
+
+        mock_server.stream = get_stream
+        yield mock_server
+
+
+@pytest.fixture
+async def mock_config_entry() -> MockConfigEntry:
+    """Return a mock config entry."""
+
+    # Create a mock config entry
+    return MockConfigEntry(
+        domain=DOMAIN,
+        data={
+            CONF_HOST: "127.0.0.1",
+            CONF_PORT: CONTROL_PORT,
+        },
+    )
+
+
+@pytest.fixture
+def mock_server_connection() -> Generator[Snapserver]:
+    """Create a mock server connection."""
+
+    # Patch the start method of the Snapserver class to avoid network connections
+    with patch.object(Snapserver, "start", new_callable=AsyncMock) as mock_start:
+        yield mock_start
+
+
+@pytest.fixture
+def mock_group(stream: str, streams: dict[str, AsyncMock]) -> AsyncMock:
+    """Create a mock Snapgroup."""
+    group = AsyncMock(spec=Snapgroup)
+    group.identifier = "4dcc4e3b-c699-a04b-7f0c-8260d23c43e1"
+    group.name = "test_group"
+    group.friendly_name = "test_group"
+    group.stream = stream
+    group.muted = False
+    group.stream_status = streams[stream].status
+    group.volume = 48
+    group.streams_by_name.return_value = {s.friendly_name: s for s in streams.values()}
+    return group
+
+
+@pytest.fixture
+def mock_client(mock_group: AsyncMock) -> AsyncMock:
+    """Create a mock Snapclient."""
+    client = AsyncMock(spec=Snapclient)
+    client.identifier = "00:21:6a:7d:74:fc#2"
+    client.friendly_name = "test_client"
+    client.version = "0.10.0"
+    client.connected = True
+    client.name = "Snapclient"
+    client.latency = 6
+    client.muted = False
+    client.volume = 48
+    client.group = mock_group
+    mock_group.clients = [client.identifier]
+    return client
+
+
+@pytest.fixture
+def mock_stream_1() -> AsyncMock:
+    """Create a mock stream."""
+    stream = AsyncMock(spec=Snapstream)
+    stream.identifier = "test_stream_1"
+    stream.status = "playing"
+    stream.name = "Test Stream 1"
+    stream.friendly_name = "Test Stream 1"
+    stream.metadata = {
+        "album": "Test Album",
+        "artist": ["Test Artist 1", "Test Artist 2"],
+        "title": "Test Title",
+        "artUrl": "http://localhost/test_art.jpg",
+        "albumArtist": [
+            "Test Album Artist 1",
+            "Test Album Artist 2",
+        ],
+        "trackNumber": 10,
+        "duration": 60.0,
+    }
+    stream.meta = stream.metadata
+    stream.properties = {
+        "position": 30.0,
+        **stream.metadata,
+    }
+    stream.path = None
+    return stream
+
+
+@pytest.fixture
+def mock_stream_2() -> AsyncMock:
+    """Create a mock stream."""
+    stream = AsyncMock(spec=Snapstream)
+    stream.identifier = "test_stream_2"
+    stream.status = "idle"
+    stream.name = "Test Stream 2"
+    stream.friendly_name = "Test Stream 2"
+    stream.metadata = None
+    stream.meta = None
+    stream.properties = None
+    stream.path = None
+    return stream
+
+
+@pytest.fixture(
+    params=[
+        "test_stream_1",
+        "test_stream_2",
+    ]
+)
+def stream(request: pytest.FixtureRequest) -> Generator[str]:
+    """Return every device."""
+    return request.param
+
+
+@pytest.fixture
+def streams(mock_stream_1: AsyncMock, mock_stream_2: AsyncMock) -> dict[str, AsyncMock]:
+    """Return a dictionary of mock streams."""
+    return {
+        mock_stream_1.identifier: mock_stream_1,
+        mock_stream_2.identifier: mock_stream_2,
+    }

tests/components/snapcast/const.py

@@ -0,0 +1,4 @@
+"""Constants for Snapcast tests."""
+
+TEST_CLIENT_ENTITY_ID = "media_player.test_client_snapcast_client"
+TEST_GROUP_ENTITY_ID = "media_player.test_group_snapcast_group"

tests/components/snapcast/snapshots/test_media_player.ambr

@@ -0,0 +1,271 @@
+# serializer version: 1
+# name: test_state[test_stream_1][media_player.test_client_snapcast_client-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    'area_id': None,
+    'capabilities': dict({
+      'source_list': list([
+        'Test Stream 1',
+        'Test Stream 2',
+      ]),
+    }),
+    'config_entry_id': <ANY>,
+    'config_subentry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'media_player',
+    'entity_category': None,
+    'entity_id': 'media_player.test_client_snapcast_client',
+    'has_entity_name': False,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'options': dict({
+    }),
+    'original_device_class': <MediaPlayerDeviceClass.SPEAKER: 'speaker'>,
+    'original_icon': None,
+    'original_name': 'test_client Snapcast Client',
+    'platform': 'snapcast',
+    'previous_unique_id': None,
+    'suggested_object_id': None,
+    'supported_features': <MediaPlayerEntityFeature: 2060>,
+    'translation_key': None,
+    'unique_id': 'snapcast_client_127.0.0.1:1705_00:21:6a:7d:74:fc#2',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_state[test_stream_1][media_player.test_client_snapcast_client-state]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'device_class': 'speaker',
+      'entity_picture': '/api/media_player_proxy/media_player.test_client_snapcast_client?token=mock_token&cache=6e2dee674d9d1dc7',
+      'friendly_name': 'test_client Snapcast Client',
+      'is_volume_muted': False,
+      'latency': 6,
+      'media_album_artist': 'Test Album Artist 1, Test Album Artist 2',
+      'media_album_name': 'Test Album',
+      'media_artist': 'Test Artist 1, Test Artist 2',
+      'media_content_type': <MediaType.MUSIC: 'music'>,
+      'media_duration': 60,
+      'media_position': 30,
+      'media_title': 'Test Title',
+      'media_track': 10,
+      'source': 'test_stream_1',
+      'source_list': list([
+        'Test Stream 1',
+        'Test Stream 2',
+      ]),
+      'supported_features': <MediaPlayerEntityFeature: 2060>,
+      'volume_level': 0.48,
+    }),
+    'context': <ANY>,
+    'entity_id': 'media_player.test_client_snapcast_client',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'playing',
+  })
+# ---
+# name: test_state[test_stream_1][media_player.test_group_snapcast_group-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    'area_id': None,
+    'capabilities': dict({
+      'source_list': list([
+        'Test Stream 1',
+        'Test Stream 2',
+      ]),
+    }),
+    'config_entry_id': <ANY>,
+    'config_subentry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'media_player',
+    'entity_category': None,
+    'entity_id': 'media_player.test_group_snapcast_group',
+    'has_entity_name': False,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'options': dict({
+    }),
+    'original_device_class': <MediaPlayerDeviceClass.SPEAKER: 'speaker'>,
+    'original_icon': None,
+    'original_name': 'test_group Snapcast Group',
+    'platform': 'snapcast',
+    'previous_unique_id': None,
+    'suggested_object_id': None,
+    'supported_features': <MediaPlayerEntityFeature: 2060>,
+    'translation_key': None,
+    'unique_id': 'snapcast_group_127.0.0.1:1705_4dcc4e3b-c699-a04b-7f0c-8260d23c43e1',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_state[test_stream_1][media_player.test_group_snapcast_group-state]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'device_class': 'speaker',
+      'entity_picture': '/api/media_player_proxy/media_player.test_group_snapcast_group?token=mock_token&cache=6e2dee674d9d1dc7',
+      'friendly_name': 'test_group Snapcast Group',
+      'is_volume_muted': False,
+      'media_album_artist': 'Test Album Artist 1, Test Album Artist 2',
+      'media_album_name': 'Test Album',
+      'media_artist': 'Test Artist 1, Test Artist 2',
+      'media_content_type': <MediaType.MUSIC: 'music'>,
+      'media_duration': 60,
+      'media_position': 30,
+      'media_title': 'Test Title',
+      'media_track': 10,
+      'source': 'test_stream_1',
+      'source_list': list([
+        'Test Stream 1',
+        'Test Stream 2',
+      ]),
+      'supported_features': <MediaPlayerEntityFeature: 2060>,
+      'volume_level': 0.48,
+    }),
+    'context': <ANY>,
+    'entity_id': 'media_player.test_group_snapcast_group',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'playing',
+  })
+# ---
+# name: test_state[test_stream_2][media_player.test_client_snapcast_client-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    'area_id': None,
+    'capabilities': dict({
+      'source_list': list([
+        'Test Stream 1',
+        'Test Stream 2',
+      ]),
+    }),
+    'config_entry_id': <ANY>,
+    'config_subentry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'media_player',
+    'entity_category': None,
+    'entity_id': 'media_player.test_client_snapcast_client',
+    'has_entity_name': False,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'options': dict({
+    }),
+    'original_device_class': <MediaPlayerDeviceClass.SPEAKER: 'speaker'>,
+    'original_icon': None,
+    'original_name': 'test_client Snapcast Client',
+    'platform': 'snapcast',
+    'previous_unique_id': None,
+    'suggested_object_id': None,
+    'supported_features': <MediaPlayerEntityFeature: 2060>,
+    'translation_key': None,
+    'unique_id': 'snapcast_client_127.0.0.1:1705_00:21:6a:7d:74:fc#2',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_state[test_stream_2][media_player.test_client_snapcast_client-state]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'device_class': 'speaker',
+      'friendly_name': 'test_client Snapcast Client',
+      'is_volume_muted': False,
+      'latency': 6,
+      'media_content_type': <MediaType.MUSIC: 'music'>,
+      'source': 'test_stream_2',
+      'source_list': list([
+        'Test Stream 1',
+        'Test Stream 2',
+      ]),
+      'supported_features': <MediaPlayerEntityFeature: 2060>,
+      'volume_level': 0.48,
+    }),
+    'context': <ANY>,
+    'entity_id': 'media_player.test_client_snapcast_client',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'idle',
+  })
+# ---
+# name: test_state[test_stream_2][media_player.test_group_snapcast_group-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    'area_id': None,
+    'capabilities': dict({
+      'source_list': list([
+        'Test Stream 1',
+        'Test Stream 2',
+      ]),
+    }),
+    'config_entry_id': <ANY>,
+    'config_subentry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'media_player',
+    'entity_category': None,
+    'entity_id': 'media_player.test_group_snapcast_group',
+    'has_entity_name': False,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'options': dict({
+    }),
+    'original_device_class': <MediaPlayerDeviceClass.SPEAKER: 'speaker'>,
+    'original_icon': None,
+    'original_name': 'test_group Snapcast Group',
+    'platform': 'snapcast',
+    'previous_unique_id': None,
+    'suggested_object_id': None,
+    'supported_features': <MediaPlayerEntityFeature: 2060>,
+    'translation_key': None,
+    'unique_id': 'snapcast_group_127.0.0.1:1705_4dcc4e3b-c699-a04b-7f0c-8260d23c43e1',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_state[test_stream_2][media_player.test_group_snapcast_group-state]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'device_class': 'speaker',
+      'friendly_name': 'test_group Snapcast Group',
+      'is_volume_muted': False,
+      'media_content_type': <MediaType.MUSIC: 'music'>,
+      'source': 'test_stream_2',
+      'source_list': list([
+        'Test Stream 1',
+        'Test Stream 2',
+      ]),
+      'supported_features': <MediaPlayerEntityFeature: 2060>,
+      'volume_level': 0.48,
+    }),
+    'context': <ANY>,
+    'entity_id': 'media_player.test_group_snapcast_group',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'idle',
+  })
+# ---

tests/components/snapcast/test_config_flow.py

@@ -15,12 +15,10 @@ from tests.common import MockConfigEntry
 
 TEST_CONNECTION = {CONF_HOST: "snapserver.test", CONF_PORT: 1705}
 
-pytestmark = pytest.mark.usefixtures("mock_setup_entry", "mock_create_server")
+pytestmark = pytest.mark.usefixtures("mock_setup_entry")
 
 
-async def test_form(
+async def test_form(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> None:
-    hass: HomeAssistant, mock_setup_entry: AsyncMock, mock_create_server: AsyncMock
-) -> None:
     """Test we get the form and handle errors and successful connection."""
     await setup.async_setup_component(hass, "persistent_notification", {})
     result = await hass.config_entries.flow.async_init(
@@ -55,21 +53,19 @@ async def test_form(
     assert result["errors"] == {"base": "cannot_connect"}
 
     # test success
-    result = await hass.config_entries.flow.async_configure(
+    with patch("snapcast.control.create_server"):
-        result["flow_id"], TEST_CONNECTION
+        result = await hass.config_entries.flow.async_configure(
-    )
+            result["flow_id"],
-    await hass.async_block_till_done()
+            TEST_CONNECTION,
+        )
 
     assert result["type"] is FlowResultType.CREATE_ENTRY
     assert result["title"] == "Snapcast"
     assert result["data"] == {CONF_HOST: "snapserver.test", CONF_PORT: 1705}
-    assert len(mock_create_server.mock_calls) == 1
     assert len(mock_setup_entry.mock_calls) == 1
 
 
-async def test_abort(
+async def test_abort(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> None:
-    hass: HomeAssistant, mock_setup_entry: AsyncMock, mock_create_server: AsyncMock
-) -> None:
     """Test config flow abort if device is already configured."""
     entry = MockConfigEntry(
         domain=DOMAIN,

tests/components/snapcast/test_media_player.py

@@ -0,0 +1,30 @@
+"""Test the snapcast media player implementation."""
+
+from unittest.mock import AsyncMock, patch
+
+from syrupy.assertion import SnapshotAssertion
+
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_registry as er
+
+from . import setup_integration
+
+from tests.common import MockConfigEntry, snapshot_platform
+
+
+async def test_state(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_create_server: AsyncMock,
+    entity_registry: er.EntityRegistry,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test basic state information."""
+
+    # Setup and verify the integration is loaded
+    with patch("secrets.token_hex", return_value="mock_token"):
+        await setup_integration(hass, mock_config_entry)
+        assert mock_config_entry.state is ConfigEntryState.LOADED
+
+    await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)