Add Bang & Olufsen media_player grouping (#123020)

* Add Beolink custom services
Add support for media player grouping via beolink
Give media player entity name

* Fix progress not being set to None as Beolink listener
Revert naming changes

* Update API
simplify Beolink attributes

* Improve beolink custom services

* Fix Beolink expandable source check
Add unexpand return value
Set entity name on initialization

* Handle entity naming as intended

* Fix "null" Beolink self friendly name

* Add regex service input validation
Add all_discovered to beolink_expand service
Improve beolink_expand response

* Add service icons

* Fix merge
Remove unnecessary assignment

* Remove invalid typing
Update response typing for updated API

* Revert to old typed response dict method
Remove mypy ignore line
Fix jid possibly used before assignment

* Re add debugging logging

* Fix coroutine
Fix formatting

* Remove unnecessary update control

* Make tests pass
Fix remote leader media position bug
Improve remote leader BangOlufsenSource comparison

* Fix naming and add callback decorators

* Move regex service check to variable
Suppress KeyError
Update tests

* Re-add hass running check

* Improve comments, naming and type hinting

* Remove old temporary fix

* Convert logged warning to raised exception for invalid media_player
Simplify code using walrus operator

* Fix test for invalid media_player grouping

* Improve method naming

* Improve _beolink_sources explanation

* Improve _beolink_sources explanation

* Add initial media_player grouping

* Convert custom service methods to media_player methods
Fix testing

* Remove beolink JID extra state attribute

* Modify custom services to only work as expected for media_player grouping
Fix tests

* Remove unused dispatch

* Remove wrong comment

* Remove commented out code

* Add config entry mock typing

* Fix beolink listener playback progress
Fix formatting
Add and use get_serial_number_from_jid function

* Fix testing

* Clarify beolink WebSocket notifications

* Further clarify beolink WebSocket notifications

* Convert notification value to enum value

* Improve comments for touch to join

* Fix None being cast to str if leader is not in HA

* Add error messages to devices in Beolink session and not Home Assistant
Rework _get_beolink_jid

* Replace redundant function call

* Show friendly name for unavailable remote leader instead of JID

* Update homeassistant/components/bang_olufsen/media_player.py

Co-authored-by: Erik Montnemery <erik@montnemery.com>

* Remove unneeded typing

* Rework _get_beolink_jid entity check
Clarify invalid entity error message

* Remove redundant "entity" from string

* Fix invalid typing
fix state assertions

* Fix raised error type

---------

Co-authored-by: Erik Montnemery <erik@montnemery.com>

homeassistant/components/bang_olufsen/config_flow.py

@@ -25,6 +25,7 @@ from .const import (
     DEFAULT_MODEL,
     DOMAIN,
 )
+from .util import get_serial_number_from_jid
 
 
 class EntryData(TypedDict, total=False):
@@ -107,7 +108,7 @@ class BangOlufsenConfigFlowHandler(ConfigFlow, domain=DOMAIN):
                     )
 
             self._beolink_jid = beolink_self.jid
-            self._serial_number = beolink_self.jid.split(".")[2].split("@")[0]
+            self._serial_number = get_serial_number_from_jid(beolink_self.jid)
 
             await self.async_set_unique_id(self._serial_number)
             self._abort_if_unique_id_configured()

homeassistant/components/bang_olufsen/const.py

@@ -78,6 +78,11 @@ class WebsocketNotification(StrEnum):
     VOLUME = "volume"
 
     # Sub-notifications
+    BEOLINK = "beolink"
+    BEOLINK_PEERS = "beolinkPeers"
+    BEOLINK_LISTENERS = "beolinkListeners"
+    BEOLINK_AVAILABLE_LISTENERS = "beolinkAvailableListeners"
+    CONFIGURATION = "configuration"
     NOTIFICATION = "notification"
     REMOTE_MENU_CHANGED = "remoteMenuChanged"
 

homeassistant/components/bang_olufsen/media_player.py

@@ -5,13 +5,14 @@ from __future__ import annotations
 from collections.abc import Callable
 import json
 import logging
-from typing import Any, cast
+from typing import TYPE_CHECKING, Any, cast
 
 from mozart_api import __version__ as MOZART_API_VERSION
 from mozart_api.exceptions import ApiException
 from mozart_api.models import (
     Action,
     Art,
+    BeolinkLeader,
     OverlayPlayRequest,
     OverlayPlayRequestTextToSpeechTextToSpeech,
     PlaybackContentMetadata,
@@ -44,9 +45,10 @@ from homeassistant.components.media_player import (
     async_process_play_media_url,
 )
 from homeassistant.config_entries import ConfigEntry
-from homeassistant.const import CONF_MODEL
+from homeassistant.const import CONF_MODEL, Platform
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
+from homeassistant.helpers import entity_registry as er
 from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.dispatcher import async_dispatcher_connect
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
@@ -66,12 +68,14 @@ from .const import (
     WebsocketNotification,
 )
 from .entity import BangOlufsenEntity
+from .util import get_serial_number_from_jid
 
 _LOGGER = logging.getLogger(__name__)
 
 BANG_OLUFSEN_FEATURES = (
     MediaPlayerEntityFeature.BROWSE_MEDIA
     | MediaPlayerEntityFeature.CLEAR_PLAYLIST
+    | MediaPlayerEntityFeature.GROUPING
     | MediaPlayerEntityFeature.MEDIA_ANNOUNCE
     | MediaPlayerEntityFeature.NEXT_TRACK
     | MediaPlayerEntityFeature.PAUSE
@@ -134,14 +138,19 @@ class BangOlufsenMediaPlayer(BangOlufsenEntity, MediaPlayerEntity):
         self._state: str = MediaPlayerState.IDLE
         self._video_sources: dict[str, str] = {}
 
+        # Beolink compatible sources
+        self._beolink_sources: dict[str, bool] = {}
+        self._remote_leader: BeolinkLeader | None = None
+
     async def async_added_to_hass(self) -> None:
         """Turn on the dispatchers."""
         await self._initialize()
 
         signal_handlers: dict[str, Callable] = {
             CONNECTION_STATUS: self._async_update_connection_state,
+            WebsocketNotification.BEOLINK: self._async_update_beolink,
             WebsocketNotification.PLAYBACK_ERROR: self._async_update_playback_error,
-            WebsocketNotification.PLAYBACK_METADATA: self._async_update_playback_metadata,
+            WebsocketNotification.PLAYBACK_METADATA: self._async_update_playback_metadata_and_beolink,
             WebsocketNotification.PLAYBACK_PROGRESS: self._async_update_playback_progress,
             WebsocketNotification.PLAYBACK_STATE: self._async_update_playback_state,
             WebsocketNotification.REMOTE_MENU_CHANGED: self._async_update_sources,
@@ -183,6 +192,7 @@ class BangOlufsenMediaPlayer(BangOlufsenEntity, MediaPlayerEntity):
         if product_state.playback:
             if product_state.playback.metadata:
                 self._playback_metadata = product_state.playback.metadata
+                self._remote_leader = product_state.playback.metadata.remote_leader
             if product_state.playback.progress:
                 self._playback_progress = product_state.playback.progress
             if product_state.playback.source:
@@ -201,9 +211,6 @@ class BangOlufsenMediaPlayer(BangOlufsenEntity, MediaPlayerEntity):
         # If the device has been updated with new sources, then the API will fail here.
         await self._async_update_sources()
 
-        # Set the static entity attributes that needed more information.
-        self._attr_source_list = list(self._sources.values())
-
     async def _async_update_sources(self) -> None:
         """Get sources for the specific product."""
 
@@ -237,6 +244,21 @@ class BangOlufsenMediaPlayer(BangOlufsenEntity, MediaPlayerEntity):
             and source.id not in HIDDEN_SOURCE_IDS
         }
 
+        # Some sources are not Beolink expandable, meaning that they can't be joined by
+        # or expand to other Bang & Olufsen devices for a multi-room experience.
+        # _source_change, which is used throughout the entity for current source
+        # information, lacks this information, so source ID's and their expandability is
+        # stored in the self._beolink_sources variable.
+        self._beolink_sources = {
+            source.id: (
+                source.is_multiroom_available
+                if source.is_multiroom_available is not None
+                else False
+            )
+            for source in cast(list[Source], sources.items)
+            if source.id
+        }
+
         # Video sources from remote menu
         menu_items = await self._client.get_remote_menu()
 
@@ -260,19 +282,22 @@ class BangOlufsenMediaPlayer(BangOlufsenEntity, MediaPlayerEntity):
         # Combine the source dicts
         self._sources = self._audio_sources | self._video_sources
 
+        self._attr_source_list = list(self._sources.values())
+
         # HASS won't necessarily be running the first time this method is run
         if self.hass.is_running:
             self.async_write_ha_state()
 
     @callback
-    def _async_update_playback_metadata(self, data: PlaybackContentMetadata) -> None:
+    async def _async_update_playback_metadata_and_beolink(
+        self, data: PlaybackContentMetadata
+    ) -> None:
         """Update _playback_metadata and related."""
         self._playback_metadata = data
 
-        # Update current artwork.
+        # Update current artwork and remote_leader.
         self._media_image = get_highest_resolution_artwork(self._playback_metadata)
-
+        await self._async_update_beolink()
-        self.async_write_ha_state()
 
     @callback
     def _async_update_playback_error(self, data: PlaybackError) -> None:
@@ -319,6 +344,96 @@ class BangOlufsenMediaPlayer(BangOlufsenEntity, MediaPlayerEntity):
 
         self.async_write_ha_state()
 
+    @callback
+    async def _async_update_beolink(self) -> None:
+        """Update the current Beolink leader, listeners, peers and self."""
+
+        # Add Beolink listeners / leader
+        self._remote_leader = self._playback_metadata.remote_leader
+
+        # Create group members list
+        group_members = []
+
+        # If the device is a listener.
+        if self._remote_leader is not None:
+            # Add leader if available in Home Assistant
+            leader = self._get_entity_id_from_jid(self._remote_leader.jid)
+            group_members.append(
+                leader
+                if leader is not None
+                else f"leader_not_in_hass-{self._remote_leader.friendly_name}"
+            )
+
+            # Add self
+            group_members.append(self.entity_id)
+
+        # If not listener, check if leader.
+        else:
+            beolink_listeners = await self._client.get_beolink_listeners()
+
+            # Check if the device is a leader.
+            if len(beolink_listeners) > 0:
+                # Add self
+                group_members.append(self.entity_id)
+
+                # Get the entity_ids of the listeners if available in Home Assistant
+                group_members.extend(
+                    [
+                        listener
+                        if (
+                            listener := self._get_entity_id_from_jid(
+                                beolink_listener.jid
+                            )
+                        )
+                        is not None
+                        else f"listener_not_in_hass-{beolink_listener.jid}"
+                        for beolink_listener in beolink_listeners
+                    ]
+                )
+
+        self._attr_group_members = group_members
+
+        self.async_write_ha_state()
+
+    def _get_entity_id_from_jid(self, jid: str) -> str | None:
+        """Get entity_id from Beolink JID (if available)."""
+
+        unique_id = get_serial_number_from_jid(jid)
+
+        entity_registry = er.async_get(self.hass)
+        return entity_registry.async_get_entity_id(
+            Platform.MEDIA_PLAYER, DOMAIN, unique_id
+        )
+
+    def _get_beolink_jid(self, entity_id: str) -> str:
+        """Get beolink JID from entity_id."""
+
+        entity_registry = er.async_get(self.hass)
+
+        # Check for valid bang_olufsen media_player entity
+        entity_entry = entity_registry.async_get(entity_id)
+
+        if (
+            entity_entry is None
+            or entity_entry.domain != Platform.MEDIA_PLAYER
+            or entity_entry.platform != DOMAIN
+            or entity_entry.config_entry_id is None
+        ):
+            raise ServiceValidationError(
+                translation_domain=DOMAIN,
+                translation_key="invalid_grouping_entity",
+                translation_placeholders={"entity_id": entity_id},
+            )
+
+        config_entry = self.hass.config_entries.async_get_entry(
+            entity_entry.config_entry_id
+        )
+        if TYPE_CHECKING:
+            assert config_entry
+
+        # Return JID
+        return cast(str, config_entry.data[CONF_BEOLINK_JID])
+
     @property
     def state(self) -> MediaPlayerState:
         """Return the current state of the media player."""
@@ -664,3 +779,47 @@ class BangOlufsenMediaPlayer(BangOlufsenEntity, MediaPlayerEntity):
             media_content_id,
             content_filter=lambda item: item.media_content_type.startswith("audio/"),
         )
+
+    async def async_join_players(self, group_members: list[str]) -> None:
+        """Create a Beolink session with defined group members."""
+
+        # Use the touch to join if no entities have been defined
+        # Touch to join will make the device connect to any other currently-playing
+        # Beolink compatible B&O device.
+        # Repeated presses / calls will cycle between compatible playing devices.
+        if len(group_members) == 0:
+            await self._async_beolink_join()
+            return
+
+        # Get JID for each group member
+        jids = [self._get_beolink_jid(group_member) for group_member in group_members]
+        await self._async_beolink_expand(jids)
+
+    async def async_unjoin_player(self) -> None:
+        """Unjoin Beolink session. End session if leader."""
+        await self._async_beolink_leave()
+
+    async def _async_beolink_join(self) -> None:
+        """Join a Beolink multi-room experience."""
+        await self._client.join_latest_beolink_experience()
+
+    async def _async_beolink_expand(self, beolink_jids: list[str]) -> None:
+        """Expand a Beolink multi-room experience with a device or devices."""
+        # Ensure that the current source is expandable
+        if not self._beolink_sources[cast(str, self._source_change.id)]:
+            raise ServiceValidationError(
+                translation_domain=DOMAIN,
+                translation_key="invalid_source",
+                translation_placeholders={
+                    "invalid_source": cast(str, self._source_change.id),
+                    "valid_sources": ", ".join(list(self._beolink_sources.keys())),
+                },
+            )
+
+        # Try to expand to all defined devices
+        for beolink_jid in beolink_jids:
+            await self._client.post_beolink_expand(jid=beolink_jid)
+
+    async def _async_beolink_leave(self) -> None:
+        """Leave the current Beolink experience."""
+        await self._client.post_beolink_leave()

homeassistant/components/bang_olufsen/strings.json

@@ -40,6 +40,9 @@
     },
     "play_media_error": {
       "message": "An error occurred while attempting to play {media_type}: {error_message}."
+    },
+    "invalid_grouping_entity": {
+      "message": "Entity with id: {entity_id} can't be added to the Beolink session. Is the entity a Bang & Olufsen media_player?"
     }
   }
 }

homeassistant/components/bang_olufsen/util.py

@@ -16,3 +16,8 @@ def get_device(hass: HomeAssistant, unique_id: str) -> DeviceEntry:
     assert device
 
     return device
+
+
+def get_serial_number_from_jid(jid: str) -> str:
+    """Get serial number from Beolink JID."""
+    return jid.split(".")[2].split("@")[0]

homeassistant/components/bang_olufsen/websocket.py

@@ -96,7 +96,16 @@ class BangOlufsenWebsocket(BangOlufsenBase):
         # Try to match the notification type with available WebsocketNotification members
         notification_type = try_parse_enum(WebsocketNotification, notification.value)
 
-        if notification_type is WebsocketNotification.REMOTE_MENU_CHANGED:
+        if notification_type in (
+            WebsocketNotification.BEOLINK_PEERS,
+            WebsocketNotification.BEOLINK_LISTENERS,
+            WebsocketNotification.BEOLINK_AVAILABLE_LISTENERS,
+        ):
+            async_dispatcher_send(
+                self.hass,
+                f"{self._unique_id}_{WebsocketNotification.BEOLINK}",
+            )
+        elif notification_type is WebsocketNotification.REMOTE_MENU_CHANGED:
             async_dispatcher_send(
                 self.hass,
                 f"{self._unique_id}_{WebsocketNotification.REMOTE_MENU_CHANGED}",

tests/components/bang_olufsen/conftest.py

@@ -27,10 +27,17 @@ from homeassistant.core import HomeAssistant
 
 from .const import (
     TEST_DATA_CREATE_ENTRY,
+    TEST_DATA_CREATE_ENTRY_2,
     TEST_FRIENDLY_NAME,
+    TEST_FRIENDLY_NAME_2,
+    TEST_FRIENDLY_NAME_3,
     TEST_JID_1,
+    TEST_JID_2,
+    TEST_JID_3,
     TEST_NAME,
+    TEST_NAME_2,
     TEST_SERIAL_NUMBER,
+    TEST_SERIAL_NUMBER_2,
 )
 
 from tests.common import MockConfigEntry
@@ -47,6 +54,17 @@ def mock_config_entry() -> MockConfigEntry:
     )
 
 
+@pytest.fixture
+def mock_config_entry_2() -> MockConfigEntry:
+    """Mock config entry."""
+    return MockConfigEntry(
+        domain=DOMAIN,
+        unique_id=TEST_SERIAL_NUMBER_2,
+        data=TEST_DATA_CREATE_ENTRY_2,
+        title=TEST_NAME_2,
+    )
+
+
 @pytest.fixture
 async def mock_media_player(
     hass: HomeAssistant,
@@ -102,13 +120,19 @@ def mock_mozart_client() -> Generator[AsyncMock]:
                     is_enabled=True,
                     is_multiroom_available=False,
                 ),
-                # The only available source
+                # The only available beolink source
                 Source(
                     name="Tidal",
                     id="tidal",
                     is_enabled=True,
                     is_multiroom_available=True,
                 ),
+                Source(
+                    name="Line-In",
+                    id="lineIn",
+                    is_enabled=True,
+                    is_multiroom_available=False,
+                ),
                 # Is disabled, so should not be user selectable
                 Source(
                     name="Powerlink",
@@ -228,6 +252,17 @@ def mock_mozart_client() -> Generator[AsyncMock]:
                 id="64c9da45-3682-44a4-8030-09ed3ef44160",
             ),
         }
+        client.get_beolink_peers = AsyncMock()
+        client.get_beolink_peers.return_value = [
+            BeolinkPeer(friendly_name=TEST_FRIENDLY_NAME_2, jid=TEST_JID_2),
+            BeolinkPeer(friendly_name=TEST_FRIENDLY_NAME_3, jid=TEST_JID_3),
+        ]
+        client.get_beolink_listeners = AsyncMock()
+        client.get_beolink_listeners.return_value = [
+            BeolinkPeer(friendly_name=TEST_FRIENDLY_NAME_2, jid=TEST_JID_2),
+            BeolinkPeer(friendly_name=TEST_FRIENDLY_NAME_3, jid=TEST_JID_3),
+        ]
+
         client.post_standby = AsyncMock()
         client.set_current_volume_level = AsyncMock()
         client.set_volume_mute = AsyncMock()
@@ -242,6 +277,12 @@ def mock_mozart_client() -> Generator[AsyncMock]:
         client.add_to_queue = AsyncMock()
         client.post_remote_trigger = AsyncMock()
         client.set_active_source = AsyncMock()
+        client.post_beolink_expand = AsyncMock()
+        client.join_beolink_peer = AsyncMock()
+        client.post_beolink_unexpand = AsyncMock()
+        client.post_beolink_leave = AsyncMock()
+        client.post_beolink_allstandby = AsyncMock()
+        client.join_latest_beolink_experience = AsyncMock()
 
         # Non-REST API client methods
         client.check_device_connection = AsyncMock()

tests/components/bang_olufsen/const.py

@@ -39,13 +39,27 @@ TEST_MODEL_BALANCE = "Beosound Balance"
 TEST_MODEL_THEATRE = "Beosound Theatre"
 TEST_MODEL_LEVEL = "Beosound Level"
 TEST_SERIAL_NUMBER = "11111111"
+TEST_SERIAL_NUMBER_2 = "22222222"
 TEST_NAME = f"{TEST_MODEL_BALANCE}-{TEST_SERIAL_NUMBER}"
+TEST_NAME_2 = f"{TEST_MODEL_BALANCE}-{TEST_SERIAL_NUMBER_2}"
 TEST_FRIENDLY_NAME = "Living room Balance"
 TEST_TYPE_NUMBER = "1111"
 TEST_ITEM_NUMBER = "1111111"
 TEST_JID_1 = f"{TEST_TYPE_NUMBER}.{TEST_ITEM_NUMBER}.{TEST_SERIAL_NUMBER}@products.bang-olufsen.com"
 TEST_MEDIA_PLAYER_ENTITY_ID = "media_player.beosound_balance_11111111"
 
+TEST_FRIENDLY_NAME_2 = "Laundry room Balance"
+TEST_JID_2 = f"{TEST_TYPE_NUMBER}.{TEST_ITEM_NUMBER}.22222222@products.bang-olufsen.com"
+TEST_MEDIA_PLAYER_ENTITY_ID_2 = "media_player.beosound_balance_22222222"
+
+TEST_FRIENDLY_NAME_3 = "Lego room Balance"
+TEST_JID_3 = f"{TEST_TYPE_NUMBER}.{TEST_ITEM_NUMBER}.33333333@products.bang-olufsen.com"
+TEST_MEDIA_PLAYER_ENTITY_ID_3 = "media_player.beosound_balance_33333333"
+
+TEST_FRIENDLY_NAME_4 = "Lounge room Balance"
+TEST_JID_4 = f"{TEST_TYPE_NUMBER}.{TEST_ITEM_NUMBER}.44444444@products.bang-olufsen.com"
+TEST_MEDIA_PLAYER_ENTITY_ID_4 = "media_player.beosound_balance_44444444"
+
 TEST_HOSTNAME_ZEROCONF = TEST_NAME.replace(" ", "-") + ".local."
 TEST_TYPE_ZEROCONF = "_bangolufsen._tcp.local."
 TEST_NAME_ZEROCONF = TEST_NAME.replace(" ", "-") + "." + TEST_TYPE_ZEROCONF
@@ -60,6 +74,12 @@ TEST_DATA_CREATE_ENTRY = {
     CONF_BEOLINK_JID: TEST_JID_1,
     CONF_NAME: TEST_NAME,
 }
+TEST_DATA_CREATE_ENTRY_2 = {
+    CONF_HOST: TEST_HOST,
+    CONF_MODEL: TEST_MODEL_BALANCE,
+    CONF_BEOLINK_JID: TEST_JID_2,
+    CONF_NAME: TEST_NAME_2,
+}
 
 TEST_DATA_ZEROCONF = ZeroconfServiceInfo(
     ip_address=IPv4Address(TEST_HOST),
@@ -101,7 +121,7 @@ TEST_DATA_ZEROCONF_IPV6 = ZeroconfServiceInfo(
     },
 )
 
-TEST_AUDIO_SOURCES = [BangOlufsenSource.TIDAL.name]
+TEST_AUDIO_SOURCES = [BangOlufsenSource.TIDAL.name, BangOlufsenSource.LINE_IN.name]
 TEST_VIDEO_SOURCES = ["HDMI A"]
 TEST_SOURCES = TEST_AUDIO_SOURCES + TEST_VIDEO_SOURCES
 TEST_FALLBACK_SOURCES = [

tests/components/bang_olufsen/test_media_player.py

@@ -5,6 +5,7 @@ import logging
 from unittest.mock import AsyncMock, patch
 
 from mozart_api.models import (
+    BeolinkLeader,
     PlaybackContentMetadata,
     RenderingState,
     Source,
@@ -18,6 +19,7 @@ from homeassistant.components.bang_olufsen.const import (
     BangOlufsenSource,
 )
 from homeassistant.components.media_player import (
+    ATTR_GROUP_MEMBERS,
     ATTR_INPUT_SOURCE,
     ATTR_INPUT_SOURCE_LIST,
     ATTR_MEDIA_ALBUM_ARTIST,
@@ -62,7 +64,11 @@ from .const import (
     TEST_DEEZER_PLAYLIST,
     TEST_DEEZER_TRACK,
     TEST_FALLBACK_SOURCES,
+    TEST_FRIENDLY_NAME_2,
+    TEST_JID_2,
     TEST_MEDIA_PLAYER_ENTITY_ID,
+    TEST_MEDIA_PLAYER_ENTITY_ID_2,
+    TEST_MEDIA_PLAYER_ENTITY_ID_3,
     TEST_OVERLAY_INVALID_OFFSET_VOLUME_TTS,
     TEST_OVERLAY_OFFSET_VOLUME_TTS,
     TEST_PLAYBACK_ERROR,
@@ -452,6 +458,70 @@ async def test_async_set_volume_level(
     )
 
 
+async def test_async_update_beolink_line_in(
+    hass: HomeAssistant,
+    mock_mozart_client: AsyncMock,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test _async_update_beolink with line-in and no active Beolink session."""
+    # Ensure no listeners
+    mock_mozart_client.get_beolink_listeners.return_value = []
+
+    mock_config_entry.add_to_hass(hass)
+    await hass.config_entries.async_setup(mock_config_entry.entry_id)
+
+    source_change_callback = (
+        mock_mozart_client.get_source_change_notifications.call_args[0][0]
+    )
+    beolink_callback = mock_mozart_client.get_notification_notifications.call_args[0][0]
+
+    # Set source
+    source_change_callback(BangOlufsenSource.LINE_IN)
+    beolink_callback(WebsocketNotificationTag(value="beolinkListeners"))
+
+    assert (states := hass.states.get(TEST_MEDIA_PLAYER_ENTITY_ID))
+    assert states.attributes["group_members"] == []
+
+    assert mock_mozart_client.get_beolink_listeners.call_count == 1
+
+
+async def test_async_update_beolink_listener(
+    hass: HomeAssistant,
+    mock_mozart_client: AsyncMock,
+    mock_config_entry: MockConfigEntry,
+    mock_config_entry_2: MockConfigEntry,
+) -> None:
+    """Test _async_update_beolink as a listener."""
+
+    mock_config_entry.add_to_hass(hass)
+    await hass.config_entries.async_setup(mock_config_entry.entry_id)
+
+    playback_metadata_callback = (
+        mock_mozart_client.get_playback_metadata_notifications.call_args[0][0]
+    )
+
+    # Add another entity
+    mock_config_entry_2.add_to_hass(hass)
+    await hass.config_entries.async_setup(mock_config_entry_2.entry_id)
+
+    # Runs _async_update_beolink
+    playback_metadata_callback(
+        PlaybackContentMetadata(
+            remote_leader=BeolinkLeader(
+                friendly_name=TEST_FRIENDLY_NAME_2, jid=TEST_JID_2
+            )
+        )
+    )
+
+    assert (states := hass.states.get(TEST_MEDIA_PLAYER_ENTITY_ID))
+    assert states.attributes["group_members"] == [
+        TEST_MEDIA_PLAYER_ENTITY_ID_2,
+        TEST_MEDIA_PLAYER_ENTITY_ID,
+    ]
+
+    assert mock_mozart_client.get_beolink_listeners.call_count == 0
+
+
 async def test_async_mute_volume(
     hass: HomeAssistant,
     mock_mozart_client: AsyncMock,
@@ -1147,3 +1217,133 @@ async def test_async_browse_media(
     assert response["success"]
 
     assert (child in response["result"]["children"]) is present
+
+
+@pytest.mark.parametrize(
+    ("group_members", "expand_count", "join_count"),
+    [
+        # Valid member
+        ([TEST_MEDIA_PLAYER_ENTITY_ID_2], 1, 0),
+        # Touch to join
+        ([], 0, 1),
+    ],
+)
+async def test_async_join_players(
+    hass: HomeAssistant,
+    mock_mozart_client: AsyncMock,
+    mock_config_entry: MockConfigEntry,
+    mock_config_entry_2: MockConfigEntry,
+    group_members: list[str],
+    expand_count: int,
+    join_count: int,
+) -> None:
+    """Test async_join_players."""
+
+    mock_config_entry.add_to_hass(hass)
+    await hass.config_entries.async_setup(mock_config_entry.entry_id)
+
+    source_change_callback = (
+        mock_mozart_client.get_source_change_notifications.call_args[0][0]
+    )
+
+    # Add another entity
+    mock_config_entry_2.add_to_hass(hass)
+    await hass.config_entries.async_setup(mock_config_entry_2.entry_id)
+
+    # Set the source to a beolink expandable source
+    source_change_callback(BangOlufsenSource.TIDAL)
+
+    await hass.services.async_call(
+        "media_player",
+        "join",
+        {
+            ATTR_ENTITY_ID: TEST_MEDIA_PLAYER_ENTITY_ID,
+            ATTR_GROUP_MEMBERS: group_members,
+        },
+        blocking=True,
+    )
+
+    assert mock_mozart_client.post_beolink_expand.call_count == expand_count
+    assert mock_mozart_client.join_latest_beolink_experience.call_count == join_count
+
+
+@pytest.mark.parametrize(
+    ("source", "group_members", "expected_result", "error_type"),
+    [
+        # Invalid source
+        (
+            BangOlufsenSource.LINE_IN,
+            [TEST_MEDIA_PLAYER_ENTITY_ID_2],
+            pytest.raises(ServiceValidationError),
+            "invalid_source",
+        ),
+        # Invalid media_player entity
+        (
+            BangOlufsenSource.TIDAL,
+            [TEST_MEDIA_PLAYER_ENTITY_ID_3],
+            pytest.raises(ServiceValidationError),
+            "invalid_grouping_entity",
+        ),
+    ],
+)
+async def test_async_join_players_invalid(
+    hass: HomeAssistant,
+    mock_mozart_client: AsyncMock,
+    mock_config_entry: MockConfigEntry,
+    mock_config_entry_2: MockConfigEntry,
+    source: Source,
+    group_members: list[str],
+    expected_result: AbstractContextManager,
+    error_type: str,
+) -> None:
+    """Test async_join_players with an invalid media_player entity."""
+
+    mock_config_entry.add_to_hass(hass)
+    await hass.config_entries.async_setup(mock_config_entry.entry_id)
+
+    source_change_callback = (
+        mock_mozart_client.get_source_change_notifications.call_args[0][0]
+    )
+
+    mock_config_entry_2.add_to_hass(hass)
+    await hass.config_entries.async_setup(mock_config_entry_2.entry_id)
+
+    source_change_callback(source)
+
+    with expected_result as exc_info:
+        await hass.services.async_call(
+            "media_player",
+            "join",
+            {
+                ATTR_ENTITY_ID: TEST_MEDIA_PLAYER_ENTITY_ID,
+                ATTR_GROUP_MEMBERS: group_members,
+            },
+            blocking=True,
+        )
+
+    assert exc_info.value.translation_domain == DOMAIN
+    assert exc_info.value.translation_key == error_type
+    assert exc_info.errisinstance(HomeAssistantError)
+
+    assert mock_mozart_client.post_beolink_expand.call_count == 0
+    assert mock_mozart_client.join_latest_beolink_experience.call_count == 0
+
+
+async def test_async_unjoin_player(
+    hass: HomeAssistant,
+    mock_mozart_client: AsyncMock,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test async_unjoin_player."""
+
+    mock_config_entry.add_to_hass(hass)
+    await hass.config_entries.async_setup(mock_config_entry.entry_id)
+
+    await hass.services.async_call(
+        "media_player",
+        "unjoin",
+        {ATTR_ENTITY_ID: TEST_MEDIA_PLAYER_ENTITY_ID},
+        blocking=True,
+    )
+
+    mock_mozart_client.post_beolink_leave.assert_called_once()