Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-04-02 00:20:30 +01:00
Code Activity

Add WiiM media player integration (#148948)

Browse Source 
Co-authored-by: Tao Jiang <tao.jiang@linkplay.com>
Co-authored-by: Erik Montnemery <erik@montnemery.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Paulus Schoutsen <balloob@gmail.com>
Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
Linkplay2020
2026-03-20 01:33:53 +08:00
committed by GitHub
parent 29309d1315
commit fa57f72f37
20 changed files with 2505 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
CODEOWNERS generated
View File

@@ -1915,6 +1915,8 @@ build.json @home-assistant/supervisor
/tests/components/whois/ @frenck
/homeassistant/components/wiffi/ @mampfes
/tests/components/wiffi/ @mampfes
/homeassistant/components/wiim/ @Linkplay2020
/tests/components/wiim/ @Linkplay2020
/homeassistant/components/wilight/ @leofig-rj
/tests/components/wilight/ @leofig-rj
/homeassistant/components/window/ @home-assistant/core

128
homeassistant/components/wiim/__init__.py Normal file
View File

@@ -0,0 +1,128 @@
"""The WiiM integration."""
from __future__ import annotations
from typing import TYPE_CHECKING
from urllib.parse import urlparse
from wiim.controller import WiimController
from wiim.discovery import async_create_wiim_device
from wiim.exceptions import WiimDeviceException, WiimRequestException
from homeassistant.const import CONF_HOST, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import Event, HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.network import NoURLAvailableError, get_url
from .const import DATA_WIIM, DOMAIN, LOGGER, PLATFORMS, UPNP_PORT, WiimConfigEntry
from .models import WiimData
DEFAULT_AVAILABILITY_POLLING_INTERVAL = 60
async def async_setup_entry(hass: HomeAssistant, entry: WiimConfigEntry) -> bool:
"""Set up WiiM from a config entry.
This method owns the device connect/disconnect lifecycle.
"""
LOGGER.debug(
"Setting up WiiM entry: %s (UDN: %s, Source: %s)",
entry.title,
entry.unique_id,
entry.source,
)
# This integration maintains shared domain-level state because:
# - Multiple config entries can be loaded simultaneously.
# - All WiiM devices share a single WiimController instance
# to coordinate network communication and event handling.
# - We also maintain a global entity_id -> UDN mapping
# used for cross-entity event routing.
#
# The domain data must therefore be initialized once and reused
# across all config entries.
session = async_get_clientsession(hass)
if DATA_WIIM not in hass.data:
hass.data[DATA_WIIM] = WiimData(controller=WiimController(session))
wiim_domain_data = hass.data[DATA_WIIM]
controller = wiim_domain_data.controller
host = entry.data[CONF_HOST]
upnp_location = f"http://{host}:{UPNP_PORT}/description.xml"
try:
base_url = get_url(hass, prefer_external=False)
except NoURLAvailableError as err:
raise ConfigEntryNotReady("Failed to determine Home Assistant URL") from err
local_host = urlparse(base_url).hostname
if TYPE_CHECKING:
assert local_host is not None
try:
wiim_device = await async_create_wiim_device(
upnp_location,
session,
host=host,
local_host=local_host,
polling_interval=DEFAULT_AVAILABILITY_POLLING_INTERVAL,
)
except WiimRequestException as err:
raise ConfigEntryNotReady(f"HTTP API request failed for {host}: {err}") from err
except WiimDeviceException as err:
raise ConfigEntryNotReady(f"Device setup failed for {host}: {err}") from err
await controller.add_device(wiim_device)
entry.runtime_data = wiim_device
LOGGER.info(
"WiiM device %s (UDN: %s) linked to HASS. Name: '%s', HTTP: %s, UPnP Location: %s",
entry.entry_id,
wiim_device.udn,
wiim_device.name,
host,
upnp_location or "N/A",
)
async def _async_shutdown_event_handler(event: Event) -> None:
LOGGER.info(
"Home Assistant stopping, disconnecting WiiM device: %s",
wiim_device.name,
)
await wiim_device.disconnect()
entry.async_on_unload(
hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_STOP, _async_shutdown_event_handler
)
)
async def _unload_entry_cleanup():
"""Cleanup when unloading the config entry.
Removes the device from the controller and disconnects it.
"""
LOGGER.debug("Running unload cleanup for %s", wiim_device.name)
await controller.remove_device(wiim_device.udn)
await wiim_device.disconnect()
entry.async_on_unload(_unload_entry_cleanup)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: WiimConfigEntry) -> bool:
"""Unload a config entry."""
LOGGER.info("Unloading WiiM entry: %s (UDN: %s)", entry.title, entry.unique_id)
if not await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
return False
if not hass.config_entries.async_loaded_entries(DOMAIN):
hass.data.pop(DATA_WIIM)
LOGGER.info("Last WiiM entry unloaded, cleaning up domain data")
return True

132
homeassistant/components/wiim/config_flow.py Normal file
View File

@@ -0,0 +1,132 @@
"""Config flow for WiiM integration."""
from __future__ import annotations
from typing import Any
import voluptuous as vol
from wiim.discovery import async_probe_wiim_device
from wiim.models import WiimProbeResult
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
from .const import DOMAIN, LOGGER, UPNP_PORT
STEP_USER_DATA_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str})
async def _async_probe_wiim_host(hass: HomeAssistant, host: str) -> WiimProbeResult:
"""Probe the given host and return WiiM device information."""
session = async_get_clientsession(hass)
location = f"http://{host}:{UPNP_PORT}/description.xml"
LOGGER.debug("Validating UPnP device at location: %s", location)
try:
probe_result = await async_probe_wiim_device(
location,
session,
host=host,
)
except TimeoutError as err:
raise CannotConnect from err
if probe_result is None:
raise CannotConnect
return probe_result
class WiimConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for WiiM."""
_discovered_info: WiimProbeResult | None = None
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step when user adds integration manually."""
errors: dict[str, str] = {}
if user_input is not None:
host = user_input[CONF_HOST]
try:
device_info = await _async_probe_wiim_host(self.hass, host)
except CannotConnect:
errors["base"] = "cannot_connect"
else:
await self.async_set_unique_id(device_info.udn)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=device_info.name,
data={
CONF_HOST: device_info.host,
},
)
return self.async_show_form(
step_id="user",
data_schema=self.add_suggested_values_to_schema(
STEP_USER_DATA_SCHEMA, user_input
),
errors=errors,
)
async def async_step_zeroconf(
self, discovery_info: ZeroconfServiceInfo
) -> ConfigFlowResult:
"""Handle Zeroconf discovery."""
LOGGER.debug(
"Zeroconf discovery received: Name: %s, Host: %s, Port: %s, Properties: %s",
discovery_info.name,
discovery_info.host,
discovery_info.port,
discovery_info.properties,
)
host = discovery_info.host
udn_from_txt = discovery_info.properties.get("uuid")
if udn_from_txt:
await self.async_set_unique_id(udn_from_txt)
self._abort_if_unique_id_configured(updates={CONF_HOST: host})
try:
device_info = await _async_probe_wiim_host(self.hass, host)
except CannotConnect:
return self.async_abort(reason="cannot_connect")
await self.async_set_unique_id(device_info.udn)
self._abort_if_unique_id_configured(updates={CONF_HOST: device_info.host})
self._discovered_info = device_info
self.context["title_placeholders"] = {"name": device_info.name}
return await self.async_step_discovery_confirm()
async def async_step_discovery_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle user confirmation of discovered device."""
discovered_info = self._discovered_info
if user_input is not None and discovered_info is not None:
return self.async_create_entry(
title=discovered_info.name,
data={
CONF_HOST: discovered_info.host,
},
)
return self.async_show_form(
step_id="discovery_confirm",
description_placeholders={
"name": (
discovered_info.name
if discovered_info is not None
else "Discovered WiiM Device"
)
},
)
class CannotConnect(HomeAssistantError):
"""Error to indicate we cannot connect."""

27
homeassistant/components/wiim/const.py Normal file
View File

@@ -0,0 +1,27 @@
"""Constants for the WiiM integration."""
import logging
from typing import TYPE_CHECKING, Final
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.util.hass_dict import HassKey
if TYPE_CHECKING:
from wiim import WiimDevice
from .models import WiimData
type WiimConfigEntry = ConfigEntry[WiimDevice]
DOMAIN: Final = "wiim"
LOGGER = logging.getLogger(__package__)
DATA_WIIM: HassKey[WiimData] = HassKey(DOMAIN)
PLATFORMS: Final[list[Platform]] = [
Platform.MEDIA_PLAYER,
]
UPNP_PORT = 49152
ZEROCONF_TYPE_LINKPLAY: Final = "_linkplay._tcp.local."

36
homeassistant/components/wiim/entity.py Normal file
View File

@@ -0,0 +1,36 @@
"""Base entity for the WiiM integration."""
from __future__ import annotations
from wiim.wiim_device import WiimDevice
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.entity import Entity
from .const import DOMAIN
class WiimBaseEntity(Entity):
"""Base representation of a WiiM entity."""
_attr_has_entity_name = True
def __init__(self, wiim_device: WiimDevice) -> None:
"""Initialize the WiiM base entity."""
self._device = wiim_device
self._attr_device_info = dr.DeviceInfo(
identifiers={(DOMAIN, self._device.udn)},
name=self._device.name,
manufacturer=self._device.manufacturer,
model=self._device.model_name,
sw_version=self._device.firmware_version,
)
if self._device.presentation_url:
self._attr_device_info["configuration_url"] = self._device.presentation_url
elif self._device.http_api_url:
self._attr_device_info["configuration_url"] = self._device.http_api_url
@property
def available(self) -> bool:
"""Return True if entity is available."""
return self._device.available

13
homeassistant/components/wiim/manifest.json Normal file
View File

@@ -0,0 +1,13 @@
{
"domain": "wiim",
"name": "WiiM",
"codeowners": ["@Linkplay2020"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/wiim",
"integration_type": "hub",
"iot_class": "local_push",
"loggers": ["wiim.sdk", "async_upnp_client"],
"quality_scale": "bronze",
"requirements": ["wiim==0.1.0"],
"zeroconf": ["_linkplay._tcp.local."]
}

794
homeassistant/components/wiim/media_player.py Normal file
View File

@@ -0,0 +1,794 @@
"""Support for WiiM Media Players."""
from __future__ import annotations
from collections.abc import Awaitable, Callable, Coroutine
from functools import wraps
from typing import Any, Concatenate
from async_upnp_client.client import UpnpService, UpnpStateVariable
from wiim.consts import PlayingStatus as SDKPlayingStatus
from wiim.exceptions import WiimDeviceException, WiimException, WiimRequestException
from wiim.models import (
WiimGroupRole,
WiimGroupSnapshot,
WiimRepeatMode,
WiimTransportCapabilities,
)
from wiim.wiim_device import WiimDevice
from homeassistant.components import media_source
from homeassistant.components.media_player import (
BrowseError,
BrowseMedia,
MediaClass,
MediaPlayerDeviceClass,
MediaPlayerEntity,
MediaPlayerEntityFeature,
MediaPlayerState,
MediaType,
RepeatMode,
async_process_play_media_url,
)
from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util.dt import utcnow
from .const import DATA_WIIM, LOGGER, WiimConfigEntry
from .entity import WiimBaseEntity
from .models import WiimData
MEDIA_TYPE_WIIM_LIBRARY = "wiim_library"
MEDIA_CONTENT_ID_ROOT = "library_root"
MEDIA_CONTENT_ID_FAVORITES = (
f"{MEDIA_TYPE_WIIM_LIBRARY}/{MEDIA_CONTENT_ID_ROOT}/favorites"
)
MEDIA_CONTENT_ID_PLAYLISTS = (
f"{MEDIA_TYPE_WIIM_LIBRARY}/{MEDIA_CONTENT_ID_ROOT}/playlists"
)
SDK_TO_HA_STATE: dict[SDKPlayingStatus, MediaPlayerState] = {
SDKPlayingStatus.PLAYING: MediaPlayerState.PLAYING,
SDKPlayingStatus.PAUSED: MediaPlayerState.PAUSED,
SDKPlayingStatus.STOPPED: MediaPlayerState.IDLE,
SDKPlayingStatus.LOADING: MediaPlayerState.BUFFERING,
}
# Define supported features
SUPPORT_WIIM_BASE = (
MediaPlayerEntityFeature.PLAY
| MediaPlayerEntityFeature.PAUSE
| MediaPlayerEntityFeature.STOP
| MediaPlayerEntityFeature.VOLUME_SET
| MediaPlayerEntityFeature.VOLUME_MUTE
| MediaPlayerEntityFeature.BROWSE_MEDIA
| MediaPlayerEntityFeature.PLAY_MEDIA
| MediaPlayerEntityFeature.SELECT_SOURCE
| MediaPlayerEntityFeature.GROUPING
| MediaPlayerEntityFeature.SEEK
)
def media_player_exception_wrap[
_WiimMediaPlayerEntityT: "WiimMediaPlayerEntity",
**_P,
_R,
](
func: Callable[Concatenate[_WiimMediaPlayerEntityT, _P], Awaitable[_R]],
) -> Callable[Concatenate[_WiimMediaPlayerEntityT, _P], Coroutine[Any, Any, _R]]:
"""Wrap media player commands to handle SDK exceptions consistently."""
@wraps(func)
async def _wrap(
self: _WiimMediaPlayerEntityT, *args: _P.args, **kwargs: _P.kwargs
) -> _R:
try:
result = await func(self, *args, **kwargs)
except (WiimDeviceException, WiimRequestException, WiimException) as err:
await self._async_handle_critical_error(err)
raise HomeAssistantError(
f"{func.__name__} failed for {self.entity_id}"
) from err
except RuntimeError as err:
raise HomeAssistantError(
f"{func.__name__} failed for {self.entity_id}"
) from err
self._update_ha_state_from_sdk_cache()
return result
return _wrap
async def async_setup_entry(
hass: HomeAssistant,
entry: WiimConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up WiiM media player from a config entry."""
async_add_entities([WiimMediaPlayerEntity(entry.runtime_data, entry)])
class WiimMediaPlayerEntity(WiimBaseEntity, MediaPlayerEntity):
"""Representation of a WiiM media player."""
_attr_device_class = MediaPlayerDeviceClass.SPEAKER
_attr_media_image_remotely_accessible = True
_attr_name = None
_attr_should_poll = False
def __init__(self, device: WiimDevice, entry: WiimConfigEntry) -> None:
"""Initialize the WiiM entity."""
super().__init__(device)
self._entry = entry
self._attr_unique_id = device.udn
self._attr_source_list = list(device.supported_input_modes) or None
self._attr_shuffle: bool = False
self._attr_repeat = RepeatMode.OFF
self._transport_capabilities: WiimTransportCapabilities | None = None
self._supported_features_update_in_flight = False
@property
def _wiim_data(self) -> WiimData:
"""Return shared WiiM domain data."""
return self.hass.data[DATA_WIIM]
@property
def supported_features(self) -> MediaPlayerEntityFeature:
"""Return the features supported by the current device state."""
features = SUPPORT_WIIM_BASE
if self._transport_capabilities is None:
return features
if self._transport_capabilities.can_next:
features |= MediaPlayerEntityFeature.NEXT_TRACK
if self._transport_capabilities.can_previous:
features |= MediaPlayerEntityFeature.PREVIOUS_TRACK
if self._transport_capabilities.can_repeat:
features |= MediaPlayerEntityFeature.REPEAT_SET
if self._transport_capabilities.can_shuffle:
features |= MediaPlayerEntityFeature.SHUFFLE_SET
return features
@callback
def _get_entity_id_for_udn(self, udn: str) -> str | None:
"""Helper to get a WiimMediaPlayerEntity ID by UDN from shared data."""
for entity_id, stored_udn in self._wiim_data.entity_id_to_udn_map.items():
if stored_udn == udn:
return entity_id
LOGGER.debug("No entity ID found for UDN: %s", udn)
return None
def _get_group_snapshot(self) -> WiimGroupSnapshot:
"""Return the typed group snapshot for the current device."""
return self._wiim_data.controller.get_group_snapshot(self._device.udn)
@property
def _metadata_device(self) -> WiimDevice:
"""Return the device whose metadata should back this entity."""
group_snapshot = self._get_group_snapshot()
if group_snapshot.role != WiimGroupRole.FOLLOWER:
return self._device
return self._wiim_data.controller.get_device(group_snapshot.leader_udn)
@callback
def _clear_media_metadata(self) -> None:
"""Clear media metadata attributes."""
self._attr_media_title = None
self._attr_media_artist = None
self._attr_media_album_name = None
self._attr_media_image_url = None
self._attr_media_content_id = None
self._attr_media_content_type = None
self._attr_media_duration = None
self._attr_media_position = None
self._attr_media_position_updated_at = None
@callback
def _get_command_target_device(self, action_name: str) -> WiimDevice:
"""Return the device that should receive a grouped playback command."""
group_snapshot = self._get_group_snapshot()
if group_snapshot.role != WiimGroupRole.FOLLOWER:
return self._device
target_device = self._wiim_data.controller.get_device(
group_snapshot.command_target_udn
)
LOGGER.info(
"Routing %s command from follower %s to leader %s",
action_name,
self.entity_id,
target_device.udn,
)
return target_device
@callback
def _update_ha_state_from_sdk_cache(
self,
*,
write_state: bool = True,
update_supported_features: bool = True,
) -> None:
"""Update HA state from SDK's cache/HTTP poll attributes.
This is the main method for updating this entity's HA attributes.
Crucially, it also handles propagating metadata to followers if this is a leader.
"""
LOGGER.debug(
"Device %s: Updating HA state from SDK cache/HTTP poll",
self.name or self.unique_id,
)
self._attr_available = self._device.available
if not self._attr_available:
self._attr_state = None
self._clear_media_metadata()
self._attr_source = None
self._transport_capabilities = None
if write_state:
self.async_write_ha_state()
return
# Update common attributes first
self._attr_volume_level = self._device.volume / 100
self._attr_is_volume_muted = self._device.is_muted
self._attr_source_list = list(self._device.supported_input_modes) or None
# Determine current group role (leader/follower/standalone)
group_snapshot = self._get_group_snapshot()
metadata_device = self._metadata_device
if group_snapshot.role == WiimGroupRole.FOLLOWER:
LOGGER.debug(
"Follower %s: Actively pulling metadata from leader %s",
self.entity_id,
metadata_device.udn,
)
if metadata_device.playing_status is not None:
self._attr_state = SDK_TO_HA_STATE.get(
metadata_device.playing_status, MediaPlayerState.IDLE
)
if metadata_device.play_mode is not None:
self._attr_source = metadata_device.play_mode
loop_state = metadata_device.loop_state
self._attr_repeat = RepeatMode(loop_state.repeat)
self._attr_shuffle = loop_state.shuffle
if media := metadata_device.current_media:
self._attr_media_title = media.title
self._attr_media_artist = media.artist
self._attr_media_album_name = media.album
self._attr_media_image_url = media.image_url
self._attr_media_content_id = media.uri
self._attr_media_content_type = MediaType.MUSIC
self._attr_media_duration = media.duration
if self._attr_media_position != media.position:
self._attr_media_position = media.position
self._attr_media_position_updated_at = utcnow()
else:
self._clear_media_metadata()
group_members = [
entity_id
for udn in group_snapshot.member_udns
if (entity_id := self._get_entity_id_for_udn(udn)) is not None
]
self._attr_group_members = group_members or ([self.entity_id])
if update_supported_features:
self._async_schedule_update_supported_features()
if write_state:
self.async_write_ha_state()
@callback
def _handle_sdk_general_device_update(self, device: WiimDevice) -> None:
"""Handle general updates from the SDK (e.g., availability, polled data)."""
LOGGER.debug(
"Device %s: Received general SDK update from %s",
self.entity_id,
device.name,
)
if not self._device.available:
self._update_ha_state_from_sdk_cache()
self._entry.async_create_background_task(
self.hass,
self._async_handle_critical_error(WiimException("Device offline.")),
name=f"wiim_{self.entity_id}_critical_error",
)
return
async def _wrapped() -> None:
await self._device.ensure_subscriptions()
self._update_ha_state_from_sdk_cache()
if self._device.supports_http_api:
self._entry.async_create_background_task(
self.hass,
_wrapped(),
name=f"wiim_{self.entity_id}_general_update",
)
else:
self._update_ha_state_from_sdk_cache()
@callback
def _handle_sdk_av_transport_event(
self, service: UpnpService, state_variables: list[UpnpStateVariable]
) -> None:
"""Handle AVTransport events from the SDK.
This method updates the internal SDK device state based on events,
then triggers a full HA state refresh from the device's cache.
"""
LOGGER.debug(
"Device %s: Received AVTransport event: %s",
self.entity_id,
self._device.event_data,
)
event_data = self._device.event_data
if "TransportState" in event_data:
sdk_status_str = event_data["TransportState"]
try:
sdk_status = SDKPlayingStatus(sdk_status_str)
except ValueError:
LOGGER.warning(
"Device %s: Unknown TransportState from event: %s",
self.entity_id,
sdk_status_str,
)
else:
self._device.playing_status = sdk_status
if sdk_status == SDKPlayingStatus.STOPPED:
LOGGER.debug(
"Device %s: TransportState is STOPPED. Resetting media position and metadata",
self.entity_id,
)
self._device.current_position = 0
self._device.current_track_duration = 0
self._attr_media_position_updated_at = None
self._attr_media_duration = None
self._attr_media_position = None
elif sdk_status in {SDKPlayingStatus.PAUSED, SDKPlayingStatus.PLAYING}:
self._entry.async_create_background_task(
self.hass,
self._device.sync_device_duration_and_position(),
name=f"wiim_{self.entity_id}_sync_position",
)
self._update_ha_state_from_sdk_cache()
@callback
def _handle_sdk_refresh_event(
self, _service: UpnpService, state_variables: list[UpnpStateVariable]
) -> None:
"""Handle SDK events that only require a state refresh."""
LOGGER.debug(
"Device %s: Received SDK refresh event: %s", self.entity_id, state_variables
)
self._update_ha_state_from_sdk_cache()
async def _async_get_transport_capabilities_for_device(
self, device: WiimDevice
) -> WiimTransportCapabilities | None:
"""Return transport capabilities for a device."""
try:
return await device.async_get_transport_capabilities()
except WiimRequestException as err:
LOGGER.warning(
"Device %s: Failed to fetch transport capabilities: %s",
device.udn,
err,
)
return None
except RuntimeError as err:
LOGGER.error(
"Device %s: Unexpected error in transport capability detection: %s",
device.udn,
err,
)
return None
async def _from_device_update_supported_features(
self, *, write_state: bool = True
) -> None:
"""Fetches media info from the device to dynamically update supported features.
This method is asynchronous and makes a network call.
"""
metadata_device = self._metadata_device
previous_capabilities = self._transport_capabilities
if (
transport_capabilities
:= await self._async_get_transport_capabilities_for_device(metadata_device)
) is not None:
if self._transport_capabilities != transport_capabilities:
self._transport_capabilities = transport_capabilities
LOGGER.debug(
"Device %s: Updated transport capabilities to %s",
self.entity_id,
transport_capabilities,
)
elif (
metadata_device is not self._device
and self._transport_capabilities is not None
):
self._transport_capabilities = None
LOGGER.debug(
"Device %s: Follower transport capabilities unavailable, using base features",
self.entity_id,
)
if write_state and self._transport_capabilities != previous_capabilities:
self.async_write_ha_state()
@callback
def _async_schedule_update_supported_features(self) -> None:
"""Update supported features based on current state."""
# Avoid parallel MEDIA_INFO request.
if self._supported_features_update_in_flight:
return
self._supported_features_update_in_flight = True
async def _refresh_supported_features() -> None:
try:
await self._from_device_update_supported_features()
finally:
self._supported_features_update_in_flight = False
self._entry.async_create_background_task(
self.hass,
_refresh_supported_features(),
name=f"wiim_{self.entity_id}_refresh_supported_features",
)
@callback
def _async_registry_updated(
self, event: Event[er.EventEntityRegistryUpdatedData]
) -> None:
"""Keep the entity-to-UDN map in sync with entity registry updates."""
if (
event.data["action"] == "update"
and (old_entity_id := event.data.get("old_entity_id"))
and old_entity_id != (entity_id := event.data["entity_id"])
):
self._wiim_data.entity_id_to_udn_map.pop(old_entity_id, None)
self._wiim_data.entity_id_to_udn_map[entity_id] = self._device.udn
super()._async_registry_updated(event)
async def async_added_to_hass(self) -> None:
"""Run when entity is added to Home Assistant."""
await super().async_added_to_hass()
self._wiim_data.entity_id_to_udn_map[self.entity_id] = self._device.udn
LOGGER.debug(
"Added %s (UDN: %s) to entity maps in hass.data",
self.entity_id,
self._device.udn,
)
await self._from_device_update_supported_features(write_state=False)
self._update_ha_state_from_sdk_cache(
write_state=False, update_supported_features=False
)
self._device.general_event_callback = self._handle_sdk_general_device_update
self._device.av_transport_event_callback = self._handle_sdk_av_transport_event
self._device.rendering_control_event_callback = self._handle_sdk_refresh_event
self._device.play_queue_event_callback = self._handle_sdk_refresh_event
LOGGER.debug(
"Entity %s registered callbacks with WiimDevice %s",
self.entity_id,
self._device.name,
)
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from Home Assistant."""
# Unregister SDK callbacks
self._device.general_event_callback = None
self._device.av_transport_event_callback = None
self._device.rendering_control_event_callback = None
self._device.play_queue_event_callback = None
LOGGER.debug(
"Entity %s unregistered callbacks from WiimDevice %s",
self.entity_id,
self._device.name,
)
self._wiim_data.entity_id_to_udn_map.pop(self.entity_id, None)
LOGGER.debug("Removed %s from entity_id_to_udn_map", self.entity_id)
await super().async_will_remove_from_hass()
async def _async_handle_critical_error(self, error: WiimException) -> None:
"""Handle communication failures by marking the device unavailable."""
if self._device.available:
LOGGER.info(
"Lost connection to WiiM device %s: %s",
self.entity_id,
error,
)
self._device.set_available(False)
self._update_ha_state_from_sdk_cache()
await self._wiim_data.controller.async_update_all_multiroom_status()
@media_player_exception_wrap
async def async_set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0-1."""
await self._device.async_set_volume(round(volume * 100))
@media_player_exception_wrap
async def async_mute_volume(self, mute: bool) -> None:
"""Mute (true) or unmute (false) media player."""
await self._device.async_set_mute(mute)
@media_player_exception_wrap
async def async_media_play(self) -> None:
"""Send play command."""
await self._get_command_target_device("media_play").async_play()
@media_player_exception_wrap
async def async_media_pause(self) -> None:
"""Send pause command."""
target_device = self._get_command_target_device("media_pause")
await target_device.async_pause()
await target_device.sync_device_duration_and_position()
@media_player_exception_wrap
async def async_media_stop(self) -> None:
"""Send stop command."""
await self._get_command_target_device("media_stop").async_stop()
@media_player_exception_wrap
async def async_media_next_track(self) -> None:
"""Send next track command."""
await self._get_command_target_device("media_next_track").async_next()
@media_player_exception_wrap
async def async_media_previous_track(self) -> None:
"""Send previous track command."""
await self._get_command_target_device("media_previous_track").async_previous()
@media_player_exception_wrap
async def async_media_seek(self, position: float) -> None:
"""Seek to a specific position in the track."""
await self._get_command_target_device("media_seek").async_seek(int(position))
@media_player_exception_wrap
async def async_play_media(
self, media_type: MediaType | str, media_id: str, **kwargs: Any
) -> None:
"""Play a piece of media."""
LOGGER.debug(
"async_play_media: type=%s, id=%s, kwargs=%s", media_type, media_id, kwargs
)
target_device = self._get_command_target_device("play_media")
if media_source.is_media_source_id(media_id):
play_item = await media_source.async_resolve_media(
self.hass, media_id, self.entity_id
)
await self._async_play_url(target_device, play_item.url)
elif media_type == MEDIA_TYPE_WIIM_LIBRARY:
if not media_id.isdigit():
raise ServiceValidationError(f"Invalid preset ID: {media_id}")
preset_number = int(media_id)
await target_device.play_preset(preset_number)
self._attr_media_content_id = f"wiim_preset_{preset_number}"
self._attr_media_content_type = MediaType.PLAYLIST
self._attr_state = MediaPlayerState.PLAYING
elif media_type == MediaType.MUSIC:
if media_id.isdigit():
preset_number = int(media_id)
await target_device.play_preset(preset_number)
self._attr_media_content_id = f"wiim_preset_{preset_number}"
self._attr_media_content_type = MediaType.PLAYLIST
self._attr_state = MediaPlayerState.PLAYING
else:
await self._async_play_url(target_device, media_id)
elif media_type == MediaType.URL:
await self._async_play_url(target_device, media_id)
elif media_type == MediaType.TRACK:
if not media_id.isdigit():
raise ServiceValidationError(
f"Invalid media_id: {media_id}. Expected a valid track index."
)
track_index = int(media_id)
await target_device.async_play_queue_with_index(track_index)
self._attr_media_content_id = f"wiim_track_{track_index}"
self._attr_media_content_type = MediaType.TRACK
self._attr_state = MediaPlayerState.PLAYING
else:
raise ServiceValidationError(f"Unsupported media type: {media_type}")
async def _async_play_url(self, target_device: WiimDevice, media_id: str) -> None:
"""Play a direct media URL on the target device."""
if not target_device.supports_http_api:
raise ServiceValidationError(
"Direct URL playback is not supported on this device"
)
url = async_process_play_media_url(self.hass, media_id)
LOGGER.debug("HTTP media_type for play_media: %s", url)
await target_device.play_url(url)
self._attr_state = MediaPlayerState.PLAYING
@media_player_exception_wrap
async def async_set_repeat(self, repeat: RepeatMode) -> None:
"""Set repeat mode."""
target_device = self._get_command_target_device("repeat_set")
await target_device.async_set_loop_mode(
target_device.build_loop_mode(WiimRepeatMode(repeat), self._attr_shuffle)
)
@media_player_exception_wrap
async def async_set_shuffle(self, shuffle: bool) -> None:
"""Enable/disable shuffle mode."""
repeat = self._attr_repeat or WiimRepeatMode.OFF
target_device = self._get_command_target_device("shuffle_set")
await target_device.async_set_loop_mode(
target_device.build_loop_mode(WiimRepeatMode(repeat), shuffle)
)
@media_player_exception_wrap
async def async_select_source(self, source: str) -> None:
"""Select input mode."""
await self._get_command_target_device("select_source").async_set_play_mode(
source
)
async def async_browse_media(
self,
media_content_type: MediaType | str | None = None,
media_content_id: str | None = None,
) -> BrowseMedia:
"""Implement media Browse helper."""
LOGGER.debug(
"Browsing media: content_type=%s, content_id=%s",
media_content_type,
media_content_id,
)
if media_content_id is not None and media_source.is_media_source_id(
media_content_id
):
if not self._device.supports_http_api:
raise BrowseError("Media sources are not supported on this device")
return await media_source.async_browse_media(
self.hass,
media_content_id,
content_filter=lambda item: item.media_content_type.startswith(
"audio/"
),
)
# Root browse
if media_content_id is None or media_content_id == MEDIA_CONTENT_ID_ROOT:
children: list[BrowseMedia] = []
children.append(
BrowseMedia(
media_class=MediaClass.DIRECTORY,
media_content_id=MEDIA_CONTENT_ID_FAVORITES,
media_content_type=MediaType.PLAYLIST,
title="Presets",
can_play=False,
can_expand=True,
thumbnail=None,
),
)
children.append(
BrowseMedia(
media_class=MediaClass.DIRECTORY,
media_content_id=MEDIA_CONTENT_ID_PLAYLISTS,
media_content_type=MediaType.PLAYLIST,
title="Queue",
can_play=False,
can_expand=True,
thumbnail=None,
),
)
if self._device.supports_http_api:
media_sources_item = await media_source.async_browse_media(
self.hass,
None,
content_filter=lambda item: item.media_content_type.startswith(
"audio/"
),
)
if media_sources_item.children:
children.extend(media_sources_item.children)
return BrowseMedia(
media_class=MediaClass.DIRECTORY,
media_content_id=MEDIA_CONTENT_ID_ROOT,
media_content_type=MEDIA_TYPE_WIIM_LIBRARY,
title=self._device.name,
can_play=False,
can_expand=True,
children=children,
)
if media_content_id == MEDIA_CONTENT_ID_FAVORITES:
sdk_favorites = await self._device.async_get_presets()
favorites_items = [
BrowseMedia(
media_class=MediaClass.PLAYLIST,
media_content_id=str(item.preset_id),
media_content_type=MediaType.MUSIC,
title=item.title,
can_play=True,
can_expand=False,
thumbnail=item.image_url,
)
for item in sdk_favorites
]
return BrowseMedia(
media_class=MediaClass.PLAYLIST,
media_content_id=MEDIA_CONTENT_ID_FAVORITES,
media_content_type=MediaType.PLAYLIST,
title="Presets",
can_play=False,
can_expand=True,
children=favorites_items,
)
if media_content_id == MEDIA_CONTENT_ID_PLAYLISTS:
queue_snapshot = await self._device.async_get_queue_snapshot()
if not queue_snapshot.is_active:
return BrowseMedia(
media_class=MediaClass.PLAYLIST,
media_content_id=MEDIA_CONTENT_ID_PLAYLISTS,
media_content_type=MediaType.PLAYLIST,
title="Queue",
can_play=False,
can_expand=True,
children=[],
)
playlist_track_items = [
BrowseMedia(
media_class=MediaClass.TRACK,
media_content_id=str(item.queue_index),
media_content_type=MediaType.TRACK,
title=item.title,
can_play=True,
can_expand=False,
thumbnail=item.image_url,
)
for item in queue_snapshot.items
]
return BrowseMedia(
media_class=MediaClass.PLAYLIST,
media_content_id=MEDIA_CONTENT_ID_PLAYLISTS,
media_content_type=MediaType.PLAYLIST,
title="Queue",
can_play=False,
can_expand=True,
children=playlist_track_items,
)
LOGGER.warning(
"Unhandled browse_media request: content_type=%s, content_id=%s",
media_content_type,
media_content_id,
)
raise BrowseError(f"Invalid browse path: {media_content_id}")

13
homeassistant/components/wiim/models.py Normal file
View File

@@ -0,0 +1,13 @@
"""Runtime models for the WiiM integration."""
from dataclasses import dataclass, field
from wiim.controller import WiimController
@dataclass
class WiimData:
"""Runtime data for the WiiM integration shared across platforms."""
controller: WiimController
entity_id_to_udn_map: dict[str, str] = field(default_factory=dict)

78
homeassistant/components/wiim/quality_scale.yaml Normal file
View File

@@ -0,0 +1,78 @@
rules:
# Bronze
action-setup:
status: exempt
comment: |
The integration does not provide any additional actions.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: |
This integration does not provide additional actions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup: done
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions:
status: todo
comment: |
- The calls to the api can be changed to return bool, and services can then raise HomeAssistantError
config-entry-unloading: done
docs-configuration-parameters:
status: exempt
comment: Integration has no configuration parameters
docs-installation-parameters: todo
entity-unavailable: done
integration-owner: done
log-when-unavailable: done
parallel-updates: todo
reauthentication-flow: todo
test-coverage:
status: todo
comment: |
- Increase test coverage for the media_player platform
# Gold
devices: done
diagnostics: todo
discovery-update-info: done
discovery: done
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: done
docs-supported-functions: done
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices: todo
entity-category: done
entity-device-class:
status: todo
comment: |
Set appropriate device classes for all entities where applicable.
entity-disabled-by-default: done
entity-translations: done
exception-translations: todo
icon-translations: done
reconfiguration-flow: todo
repair-issues:
status: exempt
comment: No known use cases for repair issues or flows, yet
stale-devices: todo
# Platinum
async-dependency: done
inject-websession: done
strict-typing: todo

35
homeassistant/components/wiim/strings.json Normal file
View File

@@ -0,0 +1,35 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"flow_title": "{name}",
"step": {
"discovery_confirm": {
"description": "Do you want to set up {name}?"
},
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]"
},
"data_description": {
"host": "The hostname or IP address of the WiiM device."
}
}
}
},
"exceptions": {
"invalid_grouping_entity": {
"message": "Entity with ID {entity_id} can't be added to the WiiM multiroom. Is the entity a WiiM media player?"
},
"missing_homeassistant_url": {
"message": "Failed to determine Home Assistant URL"
}
}
}

1
homeassistant/generated/config_flows.py generated
View File

@@ -803,6 +803,7 @@ FLOWS = {
"whirlpool",
"whois",
"wiffi",
"wiim",
"wilight",
"withings",
"wiz",

6
homeassistant/generated/integrations.json
View File

@@ -7777,6 +7777,12 @@
"config_flow": true,
"iot_class": "local_push"
},
"wiim": {
"name": "WiiM",
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_push"
},
"wilight": {
"name": "WiLight",
"integration_type": "hub",

3
homeassistant/generated/zeroconf.py generated
View File

@@ -711,6 +711,9 @@ ZEROCONF = {
{
"domain": "linkplay",
},
{
"domain": "wiim",
},
],
"_lookin._tcp.local.": [
{

3
requirements_all.txt generated
View File

@@ -3294,6 +3294,9 @@ whois==0.9.27
# homeassistant.components.wiffi
wiffi==1.1.2
# homeassistant.components.wiim
wiim==0.1.0
# homeassistant.components.wirelesstag
wirelesstagpy==0.8.1

3
requirements_test_all.txt generated
View File

@@ -2779,6 +2779,9 @@ whois==0.9.27
# homeassistant.components.wiffi
wiffi==1.1.2
# homeassistant.components.wiim
wiim==0.1.0
# homeassistant.components.wled
wled==0.21.0

42
tests/components/wiim/__init__.py Normal file
View File

@@ -0,0 +1,42 @@
"""Tests for the wiim integration."""
from unittest.mock import MagicMock
from wiim.consts import PlayingStatus
from homeassistant.core import HomeAssistant
from homeassistant.core_config import async_process_ha_core_config
from tests.common import MockConfigEntry
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Set up the component."""
await async_process_ha_core_config(
hass,
{"internal_url": "http://192.168.1.10:8123"},
)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
async def fire_general_update(hass: HomeAssistant, mock_device: MagicMock) -> None:
"""Trigger the registered general update callback on the mock device."""
assert mock_device.general_event_callback is not None
mock_device.general_event_callback(mock_device)
await hass.async_block_till_done()
async def fire_transport_update(
hass: HomeAssistant,
mock_device: MagicMock,
transport_state: PlayingStatus,
) -> None:
"""Trigger the registered AVTransport callback on the mock device."""
assert mock_device.av_transport_event_callback is not None
mock_device.event_data = {"TransportState": transport_state.value}
mock_device.av_transport_event_callback(MagicMock(), [])
await hass.async_block_till_done()

171
tests/components/wiim/conftest.py Normal file
View File

@@ -0,0 +1,171 @@
"""Pytest fixtures and shared setup for the WiiM integration tests."""
from __future__ import annotations
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from wiim.consts import AudioOutputHwMode, InputMode, LoopMode, PlayingStatus
from wiim.controller import WiimController
from wiim.models import (
WiimGroupRole,
WiimGroupSnapshot,
WiimLoopState,
WiimProbeResult,
WiimQueueSnapshot,
WiimRepeatMode,
WiimTransportCapabilities,
)
from homeassistant.components.wiim import DOMAIN
from homeassistant.const import CONF_HOST
from tests.common import MockConfigEntry
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Mock setting up a config entry."""
with patch(
"homeassistant.components.wiim.async_setup_entry", return_value=True
) as mock_setup:
yield mock_setup
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Mock Home Assistant ConfigEntry."""
return MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: "192.168.1.100"},
title="Test WiiM Device",
unique_id="uuid:test-udn-1234",
)
@pytest.fixture
def mock_wiim_device() -> Generator[AsyncMock]:
"""Patch async_create_wiim_device and return the mocked WiimDevice instance."""
with patch(
"homeassistant.components.wiim.async_create_wiim_device",
autospec=True,
) as mock_create:
mock = mock_create.return_value
mock.udn = "uuid:test-udn-1234"
mock.name = "Test WiiM Device"
mock.model_name = "WiiM Pro"
mock.manufacturer = "Linkplay Tech"
mock.firmware_version = "4.8.523456"
mock.ip_address = "192.168.1.100"
mock.http_api_url = "http://192.168.1.100:8080"
mock.presentation_url = "http://192.168.1.100:8080/web_interface"
mock.available = True
mock.model = "WiiM Pro"
mock.volume = 50
mock.is_muted = False
mock.supports_http_api = False
mock.playing_status = PlayingStatus.STOPPED
mock.loop_mode = LoopMode.SHUFFLE_DISABLE_REPEAT_NONE
mock.loop_state = WiimLoopState(
repeat=WiimRepeatMode.OFF,
shuffle=False,
)
mock.input_mode = InputMode.LINE_IN
mock.audio_output_hw_mode = AudioOutputHwMode.SPEAKER_OUT.display_name # type: ignore[attr-defined]
mock.mac_address = "AA:BB:CC:DD:EE:FF"
mock.current_track_info = {}
mock.current_media = None
mock.current_track_duration = 0
mock.play_mode = "Network"
mock.equalizer_mode = ""
mock.current_position = 0
mock.next_track_uri = ""
mock.event_data = {}
mock.general_event_callback = None
mock.av_transport_event_callback = None
mock.rendering_control_event_callback = None
mock.play_queue_event_callback = None
mock.output_mode = "speaker"
mock.supported_input_modes = (InputMode.LINE_IN.display_name,) # type: ignore[attr-defined]
mock.supported_output_modes = (
AudioOutputHwMode.SPEAKER_OUT.display_name, # type: ignore[attr-defined]
)
upnp_device = MagicMock()
upnp_device.udn = mock.udn
upnp_device.friendly_name = mock.name
upnp_device.model_name = mock.model_name
upnp_device.manufacturer = mock.manufacturer
upnp_device.serial_number = "TESTSERIAL123"
upnp_device.presentation_url = mock.presentation_url
upnp_device.get_device_info = MagicMock(
return_value={
"udn": mock.udn,
"friendly_name": mock.name,
"model_name": mock.model_name,
"manufacturer": mock.manufacturer,
"serial_number": upnp_device.serial_number,
}
)
mock.upnp_device = upnp_device
mock.get_audio_output_hw_mode = AsyncMock(return_value="speaker")
mock.async_get_play_queue = AsyncMock(return_value=[])
mock.async_get_audio_output_modes = AsyncMock(return_value=[])
mock.async_get_input_modes = AsyncMock(return_value=[])
mock.async_get_play_mediums = AsyncMock(return_value=[])
mock.async_get_transport_capabilities = AsyncMock(
return_value=WiimTransportCapabilities(
can_next=False,
can_previous=False,
can_repeat=False,
can_shuffle=False,
)
)
mock.async_get_presets = AsyncMock(return_value=())
mock.async_get_queue_snapshot = AsyncMock(
return_value=WiimQueueSnapshot(items=())
)
mock.build_loop_mode = MagicMock(
return_value=LoopMode.SHUFFLE_DISABLE_REPEAT_NONE
)
yield mock
@pytest.fixture
def mock_wiim_controller(mock_wiim_device: AsyncMock) -> Generator[MagicMock]:
"""Mock a WiimController instance."""
mock = MagicMock(spec=WiimController)
mock.add_device = AsyncMock()
mock.disconnect = AsyncMock()
mock.remove_device = AsyncMock()
mock.async_update_all_multiroom_status = AsyncMock()
mock.get_group_snapshot.return_value = WiimGroupSnapshot(
role=WiimGroupRole.STANDALONE,
leader_udn=mock_wiim_device.udn,
member_udns=(mock_wiim_device.udn,),
)
mock.get_device.side_effect = lambda _udn: mock_wiim_device
with patch(
"homeassistant.components.wiim.WiimController",
return_value=mock,
):
yield mock
@pytest.fixture
def mock_probe_player() -> Generator[AsyncMock]:
"""Mock a WiimProbePlayer instance."""
with patch(
"homeassistant.components.wiim.config_flow.async_probe_wiim_device"
) as mock_probe:
mock_probe.return_value = WiimProbeResult(
host="192.168.1.100",
udn="uuid:test-udn-1234",
name="WiiM Pro",
location="http://192.168.1.100:49152/description.xml",
model="WiiM Pro",
)
yield mock_probe

208
tests/components/wiim/test_config_flow.py Normal file
View File

@@ -0,0 +1,208 @@
"""Tests for the WiiM config flow."""
from ipaddress import ip_address
from unittest.mock import AsyncMock
import pytest
from homeassistant.components.wiim.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER, SOURCE_ZEROCONF
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
from tests.common import MockConfigEntry
pytestmark = pytest.mark.usefixtures("mock_setup_entry")
DISCOVERY_INFO = ZeroconfServiceInfo(
ip_address=ip_address("192.168.1.100"),
ip_addresses=[ip_address("192.168.1.100")],
hostname="wiim-pro.local.",
name="WiiM Pro._linkplay._tcp.local.",
port=49152,
properties={"uuid": "uuid:test-udn-1234"},
type="_linkplay._tcp.local.",
)
@pytest.mark.usefixtures("mock_probe_player", "mock_setup_entry")
async def test_user_flow_create_entry(hass: HomeAssistant) -> None:
"""Test the user flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "WiiM Pro"
assert result["data"] == {CONF_HOST: "192.168.1.100"}
assert result["result"].unique_id == "uuid:test-udn-1234"
@pytest.mark.usefixtures("mock_setup_entry")
async def test_user_flow_cannot_connect(
hass: HomeAssistant, mock_probe_player: AsyncMock
) -> None:
"""Test the user flow handles connection failures."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
mock_probe_player.side_effect = TimeoutError
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "cannot_connect"}
mock_probe_player.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
@pytest.mark.usefixtures("mock_setup_entry")
async def test_user_flow_no_probe(
hass: HomeAssistant, mock_probe_player: AsyncMock
) -> None:
"""Test the user flow handles connection failures."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
old_return_value = mock_probe_player.return_value
mock_probe_player.return_value = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "cannot_connect"}
mock_probe_player.return_value = old_return_value
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
@pytest.mark.usefixtures("mock_probe_player")
async def test_user_flow_already_configured(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test the user flow aborts for an already configured device."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.usefixtures("mock_probe_player", "mock_setup_entry")
async def test_zeroconf_flow(hass: HomeAssistant) -> None:
"""Test the zeroconf discovery flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=DISCOVERY_INFO,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "discovery_confirm"
assert result["description_placeholders"] == {"name": "WiiM Pro"}
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "WiiM Pro"
assert result["data"] == {CONF_HOST: "192.168.1.100"}
assert result["result"].unique_id == "uuid:test-udn-1234"
async def test_zeroconf_flow_cannot_connect(
hass: HomeAssistant, mock_probe_player: AsyncMock
) -> None:
"""Test the zeroconf flow aborts on connection errors."""
mock_probe_player.side_effect = TimeoutError
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=DISCOVERY_INFO,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
async def test_zeroconf_flow_no_probe(
hass: HomeAssistant, mock_probe_player: AsyncMock
) -> None:
"""Test the zeroconf flow aborts when probing failed."""
mock_probe_player.return_value = None
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=DISCOVERY_INFO,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
@pytest.mark.usefixtures("mock_setup_entry")
async def test_zeroconf_flow_already_configured(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test the zeroconf flow aborts for an already configured device."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=ZeroconfServiceInfo(
ip_address=ip_address("192.168.1.101"),
ip_addresses=[ip_address("192.168.1.101")],
hostname="wiim-pro.local.",
name="WiiM Pro._linkplay._tcp.local.",
port=49152,
properties={"uuid": "uuid:test-udn-1234"},
type="_linkplay._tcp.local.",
),
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert mock_config_entry.data[CONF_HOST] == "192.168.1.101"

87
tests/components/wiim/test_init.py Normal file
View File

@@ -0,0 +1,87 @@
"""Tests for the WiiM integration initialization."""
from unittest.mock import AsyncMock, patch
import pytest
from wiim.exceptions import WiimDeviceException, WiimRequestException
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.core_config import async_process_ha_core_config
from . import setup_integration
from tests.common import MockConfigEntry
async def test_load_unload_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_wiim_device: AsyncMock,
mock_wiim_controller: AsyncMock,
) -> None:
"""Test loading and unloading a config entry."""
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
await hass.config_entries.async_remove(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
@pytest.mark.parametrize(
("exc", "state"),
[
(WiimDeviceException("device init failed"), ConfigEntryState.SETUP_RETRY),
(WiimRequestException("http failure"), ConfigEntryState.SETUP_RETRY),
],
)
async def test_setup_raises_config_entry_not_ready(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_wiim_controller: AsyncMock,
exc: Exception,
state: ConfigEntryState,
) -> None:
"""Test that setup raises ConfigEntryNotReady on device/request exceptions."""
with patch(
"homeassistant.components.wiim.async_create_wiim_device",
side_effect=exc,
):
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is state
async def test_setup_raises_config_entry_not_ready_when_no_url(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_wiim_device: AsyncMock,
mock_wiim_controller: AsyncMock,
) -> None:
"""Test setup raises ConfigEntryNotReady when no internal URL is configured."""
# Do NOT call async_process_ha_core_config so get_url raises NoURLAvailableError.
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_setup_no_url_after_core_config(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_wiim_device: AsyncMock,
mock_wiim_controller: AsyncMock,
) -> None:
"""Test that setup succeeds once internal_url is configured."""
await async_process_ha_core_config(
hass, {"internal_url": "http://192.168.1.10:8123"}
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED

723
tests/components/wiim/test_media_player.py Normal file
View File

@@ -0,0 +1,723 @@
"""Tests for the WiiM media player via services and the state machine."""
from __future__ import annotations
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from wiim.consts import PlayingStatus
from wiim.models import (
WiimGroupRole,
WiimGroupSnapshot,
WiimLoopState,
WiimMediaMetadata,
WiimPreset,
WiimQueueItem,
WiimQueueSnapshot,
WiimRepeatMode,
WiimTransportCapabilities,
)
from wiim.wiim_device import WiimDevice
from homeassistant.components.media_player import (
ATTR_INPUT_SOURCE,
ATTR_MEDIA_ALBUM_NAME,
ATTR_MEDIA_CONTENT_ID,
ATTR_MEDIA_CONTENT_TYPE,
ATTR_MEDIA_DURATION,
ATTR_MEDIA_POSITION,
ATTR_MEDIA_REPEAT,
ATTR_MEDIA_SHUFFLE,
ATTR_MEDIA_TITLE,
ATTR_MEDIA_VOLUME_LEVEL,
ATTR_MEDIA_VOLUME_MUTED,
DOMAIN as MEDIA_PLAYER_DOMAIN,
SERVICE_BROWSE_MEDIA,
SERVICE_MEDIA_PAUSE,
SERVICE_MEDIA_PLAY,
SERVICE_MEDIA_SEEK,
SERVICE_PLAY_MEDIA,
SERVICE_REPEAT_SET,
SERVICE_SELECT_SOURCE,
SERVICE_SHUFFLE_SET,
SERVICE_VOLUME_MUTE,
SERVICE_VOLUME_SET,
BrowseMedia,
MediaClass,
MediaPlayerEntityFeature,
MediaPlayerState,
MediaType,
RepeatMode,
)
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from . import fire_general_update, fire_transport_update, setup_integration
from tests.common import MockConfigEntry
MEDIA_PLAYER_ENTITY_ID = "media_player.test_wiim_device"
async def test_state_machine_updates_from_device_callbacks(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_wiim_device: MagicMock,
mock_wiim_controller: MagicMock,
) -> None:
"""Test cached device state is reflected in Home Assistant."""
await setup_integration(hass, mock_config_entry)
state = hass.states.get(MEDIA_PLAYER_ENTITY_ID)
assert state.state == MediaPlayerState.IDLE
assert state.attributes[ATTR_MEDIA_VOLUME_LEVEL] == 0.5
assert state.attributes[ATTR_INPUT_SOURCE] == "Network"
assert state.attributes["supported_features"] == int(
MediaPlayerEntityFeature.PLAY
| MediaPlayerEntityFeature.PAUSE
| MediaPlayerEntityFeature.STOP
| MediaPlayerEntityFeature.VOLUME_SET
| MediaPlayerEntityFeature.VOLUME_MUTE
| MediaPlayerEntityFeature.BROWSE_MEDIA
| MediaPlayerEntityFeature.PLAY_MEDIA
| MediaPlayerEntityFeature.SELECT_SOURCE
| MediaPlayerEntityFeature.GROUPING
| MediaPlayerEntityFeature.SEEK
)
mock_wiim_device.volume = 60
mock_wiim_device.playing_status = PlayingStatus.PLAYING
mock_wiim_device.play_mode = "Bluetooth"
mock_wiim_device.output_mode = "optical"
mock_wiim_device.loop_state = WiimLoopState(
repeat=WiimRepeatMode.ALL,
shuffle=True,
)
mock_wiim_device.current_media = WiimMediaMetadata(
title="New Song",
artist="Test Artist",
album="Test Album",
uri="http://example.com/song.flac",
duration=180,
position=42,
)
mock_wiim_device.async_get_transport_capabilities.return_value = (
WiimTransportCapabilities(
can_next=True,
can_previous=False,
can_repeat=True,
can_shuffle=True,
)
)
await fire_general_update(hass, mock_wiim_device)
state = hass.states.get(MEDIA_PLAYER_ENTITY_ID)
assert state.state == MediaPlayerState.PLAYING
assert state.attributes[ATTR_MEDIA_TITLE] == "New Song"
assert state.attributes[ATTR_MEDIA_ALBUM_NAME] == "Test Album"
assert state.attributes[ATTR_MEDIA_DURATION] == 180
assert state.attributes[ATTR_MEDIA_POSITION] == 42
assert state.attributes[ATTR_MEDIA_VOLUME_LEVEL] == 0.6
assert state.attributes[ATTR_INPUT_SOURCE] == "Bluetooth"
assert state.attributes[ATTR_MEDIA_REPEAT] == RepeatMode.ALL
assert state.attributes[ATTR_MEDIA_SHUFFLE] is True
assert state.attributes["supported_features"] == int(
MediaPlayerEntityFeature.PLAY
| MediaPlayerEntityFeature.PAUSE
| MediaPlayerEntityFeature.STOP
| MediaPlayerEntityFeature.VOLUME_SET
| MediaPlayerEntityFeature.VOLUME_MUTE
| MediaPlayerEntityFeature.BROWSE_MEDIA
| MediaPlayerEntityFeature.PLAY_MEDIA
| MediaPlayerEntityFeature.SELECT_SOURCE
| MediaPlayerEntityFeature.GROUPING
| MediaPlayerEntityFeature.SEEK
| MediaPlayerEntityFeature.NEXT_TRACK
| MediaPlayerEntityFeature.REPEAT_SET
| MediaPlayerEntityFeature.SHUFFLE_SET
)
async def test_state_machine_updates_from_transport_events(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_wiim_device: MagicMock,
mock_wiim_controller: MagicMock,
) -> None:
"""Test transport events update the state machine."""
await setup_integration(hass, mock_config_entry)
mock_wiim_device.current_media = WiimMediaMetadata(
title="Queued Song",
duration=240,
position=30,
)
await fire_transport_update(hass, mock_wiim_device, PlayingStatus.PLAYING)
state = hass.states.get(MEDIA_PLAYER_ENTITY_ID)
assert state.state == MediaPlayerState.PLAYING
assert state.attributes[ATTR_MEDIA_TITLE] == "Queued Song"
await fire_transport_update(hass, mock_wiim_device, PlayingStatus.PAUSED)
state = hass.states.get(MEDIA_PLAYER_ENTITY_ID)
assert state.state == MediaPlayerState.PAUSED
mock_wiim_device.current_media = None
await fire_transport_update(hass, mock_wiim_device, PlayingStatus.STOPPED)
state = hass.states.get(MEDIA_PLAYER_ENTITY_ID)
assert state.state == MediaPlayerState.IDLE
assert state.attributes.get(ATTR_MEDIA_TITLE) is None
@pytest.mark.parametrize(
(
"service",
"service_data",
"device_method",
"expected_args",
"state_update",
"state_attr",
"state_value",
),
[
(
SERVICE_VOLUME_SET,
{ATTR_MEDIA_VOLUME_LEVEL: 0.75},
"async_set_volume",
(75,),
{"volume": 75},
ATTR_MEDIA_VOLUME_LEVEL,
0.75,
),
(
SERVICE_VOLUME_MUTE,
{ATTR_MEDIA_VOLUME_MUTED: True},
"async_set_mute",
(True,),
{"is_muted": True},
ATTR_MEDIA_VOLUME_MUTED,
True,
),
(
SERVICE_SELECT_SOURCE,
{ATTR_INPUT_SOURCE: "Bluetooth"},
"async_set_play_mode",
("Bluetooth",),
{"play_mode": "Bluetooth"},
ATTR_INPUT_SOURCE,
"Bluetooth",
),
],
)
async def test_control_services_update_state_machine(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_wiim_device: MagicMock,
mock_wiim_controller: MagicMock,
service: str,
service_data: dict[str, object],
device_method: str,
expected_args: tuple[object, ...],
state_update: dict[str, object],
state_attr: str,
state_value: object,
) -> None:
"""Test control services are exercised through Home Assistant."""
await setup_integration(hass, mock_config_entry)
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
service,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID, **service_data},
blocking=True,
)
getattr(mock_wiim_device, device_method).assert_awaited_once_with(*expected_args)
for attr_name, attr_value in state_update.items():
setattr(mock_wiim_device, attr_name, attr_value)
await fire_general_update(hass, mock_wiim_device)
state = hass.states.get(MEDIA_PLAYER_ENTITY_ID)
assert state.attributes[state_attr] == state_value
async def test_repeat_and_shuffle_services_update_state_machine(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_wiim_device: MagicMock,
mock_wiim_controller: MagicMock,
) -> None:
"""Test repeat and shuffle go through services and state updates."""
await setup_integration(hass, mock_config_entry)
mock_wiim_device.async_get_transport_capabilities.return_value = (
WiimTransportCapabilities(
can_next=True,
can_previous=True,
can_repeat=True,
can_shuffle=True,
)
)
await fire_general_update(hass, mock_wiim_device)
repeat_loop_mode = object()
mock_wiim_device.build_loop_mode.return_value = repeat_loop_mode
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_REPEAT_SET,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID, ATTR_MEDIA_REPEAT: RepeatMode.ALL},
blocking=True,
)
mock_wiim_device.build_loop_mode.assert_called_once_with(WiimRepeatMode.ALL, False)
mock_wiim_device.async_set_loop_mode.assert_awaited_once_with(repeat_loop_mode)
mock_wiim_device.loop_state = WiimLoopState(
repeat=WiimRepeatMode.ALL,
shuffle=False,
)
await fire_general_update(hass, mock_wiim_device)
state = hass.states.get(MEDIA_PLAYER_ENTITY_ID)
assert state.attributes[ATTR_MEDIA_REPEAT] == RepeatMode.ALL
mock_wiim_device.build_loop_mode.reset_mock()
mock_wiim_device.async_set_loop_mode.reset_mock()
shuffle_loop_mode = object()
mock_wiim_device.build_loop_mode.return_value = shuffle_loop_mode
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_SHUFFLE_SET,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID, ATTR_MEDIA_SHUFFLE: True},
blocking=True,
)
mock_wiim_device.build_loop_mode.assert_called_once_with(WiimRepeatMode.ALL, True)
mock_wiim_device.async_set_loop_mode.assert_awaited_once_with(shuffle_loop_mode)
mock_wiim_device.loop_state = WiimLoopState(
repeat=WiimRepeatMode.ALL,
shuffle=True,
)
await fire_general_update(hass, mock_wiim_device)
state = hass.states.get(MEDIA_PLAYER_ENTITY_ID)
assert state.attributes[ATTR_MEDIA_SHUFFLE] is True
async def test_play_pause_and_seek_services_update_state_machine(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_wiim_device: MagicMock,
mock_wiim_controller: MagicMock,
) -> None:
"""Test playback services drive the device and state machine."""
await setup_integration(hass, mock_config_entry)
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_PLAY,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID},
blocking=True,
)
mock_wiim_device.async_play.assert_awaited_once()
mock_wiim_device.current_media = WiimMediaMetadata(
title="Playing Song",
duration=200,
position=12,
)
mock_wiim_device.playing_status = PlayingStatus.PLAYING
await fire_general_update(hass, mock_wiim_device)
state = hass.states.get(MEDIA_PLAYER_ENTITY_ID)
assert state.state == MediaPlayerState.PLAYING
assert state.attributes[ATTR_MEDIA_TITLE] == "Playing Song"
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_PAUSE,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID},
blocking=True,
)
mock_wiim_device.async_pause.assert_awaited_once()
mock_wiim_device.sync_device_duration_and_position.assert_awaited_once()
await fire_transport_update(hass, mock_wiim_device, PlayingStatus.PAUSED)
state = hass.states.get(MEDIA_PLAYER_ENTITY_ID)
assert state.state == MediaPlayerState.PAUSED
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_SEEK,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID, "seek_position": 60},
blocking=True,
)
mock_wiim_device.async_seek.assert_awaited_once_with(60)
mock_wiim_device.current_media = WiimMediaMetadata(
title="Playing Song",
duration=200,
position=60,
)
await fire_general_update(hass, mock_wiim_device)
state = hass.states.get(MEDIA_PLAYER_ENTITY_ID)
assert state.attributes[ATTR_MEDIA_POSITION] == 60
async def test_follower_routes_commands_and_reads_leader_metadata(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_wiim_device: MagicMock,
mock_wiim_controller: MagicMock,
) -> None:
"""Test follower commands are routed to the leader device."""
await setup_integration(hass, mock_config_entry)
leader_device = AsyncMock(spec=WiimDevice)
leader_device.udn = "uuid:leader-1234"
leader_device.name = "Leader WiiM Device"
leader_device.playing_status = PlayingStatus.STOPPED
leader_device.play_mode = "Network"
leader_device.loop_state = WiimLoopState(
repeat=WiimRepeatMode.OFF,
shuffle=False,
)
leader_device.output_mode = "speaker"
leader_device.current_media = None
leader_device.async_get_transport_capabilities = AsyncMock(
return_value=WiimTransportCapabilities(
can_next=True,
can_previous=False,
can_repeat=True,
can_shuffle=False,
)
)
mock_wiim_controller.get_group_snapshot.return_value = WiimGroupSnapshot(
role=WiimGroupRole.FOLLOWER,
leader_udn=leader_device.udn,
member_udns=(leader_device.udn, mock_wiim_device.udn),
)
mock_wiim_controller.get_device.side_effect = lambda udn: (
leader_device if udn == leader_device.udn else mock_wiim_device
)
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_PLAY,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID},
blocking=True,
)
leader_device.async_play.assert_awaited_once()
mock_wiim_device.async_play.assert_not_awaited()
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_SEEK,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID, "seek_position": 90},
blocking=True,
)
leader_device.async_seek.assert_awaited_once_with(90)
mock_wiim_device.async_seek.assert_not_awaited()
leader_device.playing_status = PlayingStatus.PLAYING
leader_device.play_mode = "Spotify"
leader_device.current_media = WiimMediaMetadata(
title="Leader Song",
album="Leader Album",
duration=210,
position=90,
)
await fire_general_update(hass, mock_wiim_device)
state = hass.states.get(MEDIA_PLAYER_ENTITY_ID)
assert state.state == MediaPlayerState.PLAYING
assert state.attributes[ATTR_MEDIA_TITLE] == "Leader Song"
assert state.attributes[ATTR_MEDIA_ALBUM_NAME] == "Leader Album"
assert state.attributes[ATTR_INPUT_SOURCE] == "Spotify"
assert state.attributes[ATTR_MEDIA_POSITION] == 90
async def test_follower_routes_repeat_shuffle_and_source_commands_to_leader(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_wiim_device: MagicMock,
mock_wiim_controller: MagicMock,
) -> None:
"""Test follower repeat, shuffle, and source changes are sent to the leader."""
await setup_integration(hass, mock_config_entry)
leader_device = AsyncMock(spec=WiimDevice)
leader_device.udn = "uuid:leader-1234"
leader_device.playing_status = PlayingStatus.STOPPED
leader_device.current_media = None
leader_device.loop_state = WiimLoopState(
repeat=WiimRepeatMode.OFF,
shuffle=False,
)
leader_device.play_mode = "Network"
leader_device.async_get_transport_capabilities = AsyncMock(
return_value=WiimTransportCapabilities(
can_next=True,
can_previous=True,
can_repeat=True,
can_shuffle=True,
)
)
mock_wiim_controller.get_group_snapshot.return_value = WiimGroupSnapshot(
role=WiimGroupRole.FOLLOWER,
leader_udn=leader_device.udn,
member_udns=(leader_device.udn, mock_wiim_device.udn),
)
mock_wiim_controller.get_device.side_effect = lambda udn: (
leader_device if udn == leader_device.udn else mock_wiim_device
)
await fire_general_update(hass, mock_wiim_device)
repeat_loop_mode = object()
leader_device.build_loop_mode.return_value = repeat_loop_mode
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_REPEAT_SET,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID, ATTR_MEDIA_REPEAT: RepeatMode.ALL},
blocking=True,
)
leader_device.build_loop_mode.assert_called_once_with(WiimRepeatMode.ALL, False)
leader_device.async_set_loop_mode.assert_awaited_once_with(repeat_loop_mode)
mock_wiim_device.async_set_loop_mode.assert_not_awaited()
leader_device.build_loop_mode.reset_mock()
leader_device.async_set_loop_mode.reset_mock()
shuffle_loop_mode = object()
leader_device.build_loop_mode.return_value = shuffle_loop_mode
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_SHUFFLE_SET,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID, ATTR_MEDIA_SHUFFLE: True},
blocking=True,
)
leader_device.build_loop_mode.assert_called_once_with(WiimRepeatMode.OFF, True)
leader_device.async_set_loop_mode.assert_awaited_once_with(shuffle_loop_mode)
mock_wiim_device.async_set_loop_mode.assert_not_awaited()
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_SELECT_SOURCE,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID, ATTR_INPUT_SOURCE: "Bluetooth"},
blocking=True,
)
leader_device.async_set_play_mode.assert_awaited_once_with("Bluetooth")
mock_wiim_device.async_set_play_mode.assert_not_awaited()
async def test_play_media_services_call_device_commands(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_wiim_device: MagicMock,
mock_wiim_controller: MagicMock,
) -> None:
"""Test play_media services are driven through Home Assistant."""
await setup_integration(hass, mock_config_entry)
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID,
ATTR_MEDIA_CONTENT_TYPE: MediaType.MUSIC,
ATTR_MEDIA_CONTENT_ID: "1",
},
blocking=True,
)
mock_wiim_device.play_preset.assert_awaited_once_with(1)
mock_wiim_device.current_media = WiimMediaMetadata(title="Preset 1")
mock_wiim_device.playing_status = PlayingStatus.PLAYING
await fire_general_update(hass, mock_wiim_device)
state = hass.states.get(MEDIA_PLAYER_ENTITY_ID)
assert state.state == MediaPlayerState.PLAYING
assert state.attributes[ATTR_MEDIA_TITLE] == "Preset 1"
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID,
ATTR_MEDIA_CONTENT_TYPE: MediaType.TRACK,
ATTR_MEDIA_CONTENT_ID: "2",
},
blocking=True,
)
mock_wiim_device.async_play_queue_with_index.assert_awaited_once_with(2)
@pytest.mark.parametrize("media_type", [MediaType.MUSIC, MediaType.URL])
async def test_play_media_url_service_uses_processed_url(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_wiim_device: MagicMock,
mock_wiim_controller: MagicMock,
media_type: MediaType,
) -> None:
"""Test direct URL playback goes through the URL processor."""
await setup_integration(hass, mock_config_entry)
mock_wiim_device.supports_http_api = True
with patch(
"homeassistant.components.wiim.media_player.async_process_play_media_url",
return_value="http://processed/song.mp3",
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID,
ATTR_MEDIA_CONTENT_TYPE: media_type,
ATTR_MEDIA_CONTENT_ID: "http://example.com/song.mp3",
},
blocking=True,
)
mock_wiim_device.play_url.assert_awaited_once_with("http://processed/song.mp3")
mock_wiim_device.play_preset.assert_not_awaited()
async def test_play_media_source_service_uses_resolved_url(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_wiim_device: MagicMock,
mock_wiim_controller: MagicMock,
) -> None:
"""Test media_source playback goes through the resolver."""
await setup_integration(hass, mock_config_entry)
mock_wiim_device.supports_http_api = True
with (
patch(
"homeassistant.components.wiim.media_player.media_source.is_media_source_id",
return_value=True,
),
patch(
"homeassistant.components.wiim.media_player.media_source.async_resolve_media",
AsyncMock(return_value=MagicMock(url="http://resolved/song.mp3")),
),
patch(
"homeassistant.components.wiim.media_player.async_process_play_media_url",
return_value="http://processed/song.mp3",
),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID,
ATTR_MEDIA_CONTENT_TYPE: MediaType.MUSIC,
ATTR_MEDIA_CONTENT_ID: "media-source://media_source/local/song.mp3",
},
blocking=True,
)
mock_wiim_device.play_url.assert_awaited_once_with("http://processed/song.mp3")
async def test_browse_media_service_returns_wiim_library(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_wiim_device: MagicMock,
mock_wiim_controller: MagicMock,
) -> None:
"""Test browsing WiiM presets and queue via the media_player service."""
await setup_integration(hass, mock_config_entry)
root_result = await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_BROWSE_MEDIA,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID},
blocking=True,
return_response=True,
)
root_browse = root_result[MEDIA_PLAYER_ENTITY_ID]
assert root_browse.title == mock_wiim_device.name
assert [child.title for child in root_browse.children] == ["Presets", "Queue"]
mock_wiim_device.async_get_presets.return_value = (
WiimPreset(1, "Preset 1", "http://image1"),
WiimPreset(2, "Preset 2", "http://image2"),
)
preset_result = await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_BROWSE_MEDIA,
{
ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID,
ATTR_MEDIA_CONTENT_TYPE: MediaType.PLAYLIST,
ATTR_MEDIA_CONTENT_ID: "wiim_library/library_root/favorites",
},
blocking=True,
return_response=True,
)
preset_browse = preset_result[MEDIA_PLAYER_ENTITY_ID]
assert [child.title for child in preset_browse.children] == ["Preset 1", "Preset 2"]
mock_wiim_device.async_get_queue_snapshot.return_value = WiimQueueSnapshot(
items=(
WiimQueueItem(1, "Song A", "http://image-a"),
WiimQueueItem(2, "Song B", "http://image-b"),
),
is_active=True,
)
queue_result = await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_BROWSE_MEDIA,
{
ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID,
ATTR_MEDIA_CONTENT_TYPE: MediaType.PLAYLIST,
ATTR_MEDIA_CONTENT_ID: "wiim_library/library_root/playlists",
},
blocking=True,
return_response=True,
)
queue_browse = queue_result[MEDIA_PLAYER_ENTITY_ID]
assert [child.title for child in queue_browse.children] == ["Song A", "Song B"]
async def test_browse_media_service_includes_media_sources_when_supported(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_wiim_device: MagicMock,
mock_wiim_controller: MagicMock,
) -> None:
"""Test media sources are exposed through browse_media when HTTP API exists."""
await setup_integration(hass, mock_config_entry)
mock_wiim_device.supports_http_api = True
media_source_root = BrowseMedia(
media_class=MediaClass.DIRECTORY,
media_content_id="media-source://media_source",
media_content_type=MediaType.APPS,
title="Media Sources",
can_play=False,
can_expand=True,
children=[
BrowseMedia(
media_class=MediaClass.MUSIC,
media_content_id="media-source://media_source/local/song.mp3",
media_content_type="audio/mpeg",
title="song.mp3",
can_play=True,
can_expand=False,
)
],
)
with patch(
"homeassistant.components.wiim.media_player.media_source.async_browse_media",
AsyncMock(return_value=media_source_root),
):
result = await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_BROWSE_MEDIA,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID},
blocking=True,
return_response=True,
)
browse = result[MEDIA_PLAYER_ENTITY_ID]
assert [child.title for child in browse.children] == [
"Presets",
"Queue",
"song.mp3",
]