github/ha-core
1
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-08-02 23:40:32 +02:00
Code Releases Activity

Add more dlna_dmr media_player services and attributes (#57827)

Browse Source
This commit is contained in:
Michael Chisholm 2021-10-29 08:44:41 +11:00 committed by GitHub
parent 3f50e444ca
commit 6cdc372dcb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 640 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

113
homeassistant/components/dlna_dmr/const.py
View File

@ -5,6 +5,8 @@ from collections.abc import Mapping
import logging
from typing import Final
from async_upnp_client.profiles.dlna import PlayMode as _PlayMode
from homeassistant.components.media_player import const as _mp_const
LOGGER = logging.getLogger(__package__)
@ -58,3 +60,114 @@ MEDIA_TYPE_MAP: Mapping[str, str] = {
"object.container.storageFolder": _mp_const.MEDIA_TYPE_PLAYLIST,
"object.container.bookmarkFolder": _mp_const.MEDIA_TYPE_PLAYLIST,
}
# Map media_player media_content_type to UPnP class. Not everything will map
# directly, in which case it's not specified and other defaults will be used.
MEDIA_UPNP_CLASS_MAP: Mapping[str, str] = {
_mp_const.MEDIA_TYPE_ALBUM: "object.container.album.musicAlbum",
_mp_const.MEDIA_TYPE_ARTIST: "object.container.person.musicArtist",
_mp_const.MEDIA_TYPE_CHANNEL: "object.item.videoItem.videoBroadcast",
_mp_const.MEDIA_TYPE_CHANNELS: "object.container.channelGroup",
_mp_const.MEDIA_TYPE_COMPOSER: "object.container.person.musicArtist",
_mp_const.MEDIA_TYPE_CONTRIBUTING_ARTIST: "object.container.person.musicArtist",
_mp_const.MEDIA_TYPE_EPISODE: "object.item.epgItem.videoProgram",
_mp_const.MEDIA_TYPE_GENRE: "object.container.genre",
_mp_const.MEDIA_TYPE_IMAGE: "object.item.imageItem",
_mp_const.MEDIA_TYPE_MOVIE: "object.item.videoItem.movie",
_mp_const.MEDIA_TYPE_MUSIC: "object.item.audioItem.musicTrack",
_mp_const.MEDIA_TYPE_PLAYLIST: "object.item.playlistItem",
_mp_const.MEDIA_TYPE_PODCAST: "object.item.audioItem.audioBook",
_mp_const.MEDIA_TYPE_SEASON: "object.item.epgItem.videoProgram",
_mp_const.MEDIA_TYPE_TRACK: "object.item.audioItem.musicTrack",
_mp_const.MEDIA_TYPE_TVSHOW: "object.item.videoItem.videoBroadcast",
_mp_const.MEDIA_TYPE_URL: "object.item.bookmarkItem",
_mp_const.MEDIA_TYPE_VIDEO: "object.item.videoItem",
}
# Translation of MediaMetadata keys to DIDL-Lite keys.
# See https://developers.google.com/cast/docs/reference/messages#MediaData via
# https://www.home-assistant.io/integrations/media_player/ for HA keys.
# See http://www.upnp.org/specs/av/UPnP-av-ContentDirectory-v4-Service.pdf for
# DIDL-Lite keys.
MEDIA_METADATA_DIDL: Mapping[str, str] = {
"subtitle": "longDescription",
"releaseDate": "date",
"studio": "publisher",
"season": "episodeSeason",
"episode": "episodeNumber",
"albumName": "album",
"trackNumber": "originalTrackNumber",
}
# For (un)setting repeat mode, map a combination of shuffle & repeat to a list
# of play modes in order of suitability. Fall back to _PlayMode.NORMAL in any
# case. NOTE: This list is slightly different to that in SHUFFLE_PLAY_MODES,
# due to fallback behaviour when turning on repeat modes.
REPEAT_PLAY_MODES: Mapping[tuple[bool, str], list[_PlayMode]] = {
(False, _mp_const.REPEAT_MODE_OFF): [
_PlayMode.NORMAL,
],
(False, _mp_const.REPEAT_MODE_ONE): [
_PlayMode.REPEAT_ONE,
_PlayMode.REPEAT_ALL,
_PlayMode.NORMAL,
],
(False, _mp_const.REPEAT_MODE_ALL): [
_PlayMode.REPEAT_ALL,
_PlayMode.REPEAT_ONE,
_PlayMode.NORMAL,
],
(True, _mp_const.REPEAT_MODE_OFF): [
_PlayMode.SHUFFLE,
_PlayMode.RANDOM,
_PlayMode.NORMAL,
],
(True, _mp_const.REPEAT_MODE_ONE): [
_PlayMode.REPEAT_ONE,
_PlayMode.RANDOM,
_PlayMode.SHUFFLE,
_PlayMode.NORMAL,
],
(True, _mp_const.REPEAT_MODE_ALL): [
_PlayMode.RANDOM,
_PlayMode.REPEAT_ALL,
_PlayMode.SHUFFLE,
_PlayMode.NORMAL,
],
}
# For (un)setting shuffle mode, map a combination of shuffle & repeat to a list
# of play modes in order of suitability. Fall back to _PlayMode.NORMAL in any
# case.
SHUFFLE_PLAY_MODES: Mapping[tuple[bool, str], list[_PlayMode]] = {
(False, _mp_const.REPEAT_MODE_OFF): [
_PlayMode.NORMAL,
],
(False, _mp_const.REPEAT_MODE_ONE): [
_PlayMode.REPEAT_ONE,
_PlayMode.REPEAT_ALL,
_PlayMode.NORMAL,
],
(False, _mp_const.REPEAT_MODE_ALL): [
_PlayMode.REPEAT_ALL,
_PlayMode.REPEAT_ONE,
_PlayMode.NORMAL,
],
(True, _mp_const.REPEAT_MODE_OFF): [
_PlayMode.SHUFFLE,
_PlayMode.RANDOM,
_PlayMode.NORMAL,
],
(True, _mp_const.REPEAT_MODE_ONE): [
_PlayMode.RANDOM,
_PlayMode.SHUFFLE,
_PlayMode.REPEAT_ONE,
_PlayMode.NORMAL,
],
(True, _mp_const.REPEAT_MODE_ALL): [
_PlayMode.RANDOM,
_PlayMode.SHUFFLE,
_PlayMode.REPEAT_ALL,
_PlayMode.NORMAL,
],
}

183
homeassistant/components/dlna_dmr/media_player.py
View File

@ -3,6 +3,7 @@ from __future__ import annotations
import asyncio
from collections.abc import Mapping, Sequence
import contextlib
from datetime import datetime, timedelta
import functools
from typing import Any, Callable, TypeVar, cast
@ -10,7 +11,7 @@ from typing import Any, Callable, TypeVar, cast
from async_upnp_client import UpnpService, UpnpStateVariable
from async_upnp_client.const import NotificationSubType
from async_upnp_client.exceptions import UpnpError, UpnpResponseError
from async_upnp_client.profiles.dlna import DmrDevice, TransportState
from async_upnp_client.profiles.dlna import DmrDevice, PlayMode, TransportState
from async_upnp_client.utils import async_get_local_ip
import voluptuous as vol
@ -18,12 +19,19 @@ from homeassistant import config_entries
from homeassistant.components import ssdp
from homeassistant.components.media_player import PLATFORM_SCHEMA, MediaPlayerEntity
from homeassistant.components.media_player.const import (
ATTR_MEDIA_EXTRA,
REPEAT_MODE_ALL,
REPEAT_MODE_OFF,
REPEAT_MODE_ONE,
SUPPORT_NEXT_TRACK,
SUPPORT_PAUSE,
SUPPORT_PLAY,
SUPPORT_PLAY_MEDIA,
SUPPORT_PREVIOUS_TRACK,
SUPPORT_REPEAT_SET,
SUPPORT_SEEK,
SUPPORT_SELECT_SOUND_MODE,
SUPPORT_SHUFFLE_SET,
SUPPORT_STOP,
SUPPORT_VOLUME_MUTE,
SUPPORT_VOLUME_SET,
@ -51,7 +59,11 @@ from .const import (
CONF_POLL_AVAILABILITY,
DOMAIN,
LOGGER as _LOGGER,
MEDIA_METADATA_DIDL,
MEDIA_TYPE_MAP,
MEDIA_UPNP_CLASS_MAP,
REPEAT_PLAY_MODES,
SHUFFLE_PLAY_MODES,
)
from .data import EventListenAddr, get_domain_data
@ -250,11 +262,9 @@ class DlnaDmrEntity(MediaPlayerEntity):
if self._bootid is not None and self._bootid == bootid:
# Store the new value (because our old value matches) so that we
# can ignore subsequent ssdp:alive messages
try:
with contextlib.suppress(KeyError, ValueError):
next_bootid_str = info[ssdp.ATTR_SSDP_NEXTBOOTID]
self._bootid = int(next_bootid_str, 10)
except (KeyError, ValueError):
pass
# Nothing left to do until ssdp:alive comes through
return
@ -445,7 +455,21 @@ class DlnaDmrEntity(MediaPlayerEntity):
if not state_variables:
# Indicates a failure to resubscribe, check if device is still available
self.check_available = True
self.async_write_ha_state()
force_refresh = False
if service.service_id == "urn:upnp-org:serviceId:AVTransport":
for state_variable in state_variables:
# Force a state refresh when player begins or pauses playback
# to update the position info.
if (
state_variable.name == "TransportState"
and state_variable.value
in (TransportState.PLAYING, TransportState.PAUSED_PLAYBACK)
):
force_refresh = True
self.async_schedule_update_ha_state(force_refresh)
@property
def available(self) -> bool:
@ -515,6 +539,15 @@ class DlnaDmrEntity(MediaPlayerEntity):
if self._device.can_seek_rel_time:
supported_features |= SUPPORT_SEEK
play_modes = self._device.valid_play_modes
if play_modes & {PlayMode.RANDOM, PlayMode.SHUFFLE}:
supported_features |= SUPPORT_SHUFFLE_SET
if play_modes & {PlayMode.REPEAT_ONE, PlayMode.REPEAT_ALL}:
supported_features |= SUPPORT_REPEAT_SET
if self._device.has_presets:
supported_features |= SUPPORT_SELECT_SOUND_MODE
return supported_features
@property
@ -575,23 +608,44 @@ class DlnaDmrEntity(MediaPlayerEntity):
) -> None:
"""Play a piece of media."""
_LOGGER.debug("Playing media: %s, %s, %s", media_type, media_id, kwargs)
title = "Home Assistant"
assert self._device is not None
extra: dict[str, Any] = kwargs.get(ATTR_MEDIA_EXTRA) or {}
metadata: dict[str, Any] = extra.get("metadata") or {}
title = extra.get("title") or metadata.get("title") or "Home Assistant"
thumb = extra.get("thumb")
if thumb:
metadata["album_art_uri"] = thumb
# Translate metadata keys from HA names to DIDL-Lite names
for hass_key, didl_key in MEDIA_METADATA_DIDL.items():
if hass_key in metadata:
metadata[didl_key] = metadata.pop(hass_key)
# Create metadata specific to the given media type; different fields are
# available depending on what the upnp_class is.
upnp_class = MEDIA_UPNP_CLASS_MAP.get(media_type)
didl_metadata = await self._device.construct_play_media_metadata(
media_url=media_id,
media_title=title,
override_upnp_class=upnp_class,
meta_data=metadata,
)
# Stop current playing media
if self._device.can_stop:
await self.async_media_stop()
# Queue media
await self._device.async_set_transport_uri(media_id, title)
await self._device.async_wait_for_can_play()
await self._device.async_set_transport_uri(media_id, title, didl_metadata)
# If already playing, no need to call Play
if self._device.transport_state == TransportState.PLAYING:
# If already playing, or don't want to autoplay, no need to call Play
autoplay = extra.get("autoplay", True)
if self._device.transport_state == TransportState.PLAYING or not autoplay:
return
# Play it
await self._device.async_wait_for_can_play()
await self.async_media_play()
@catch_request_errors
@ -606,6 +660,98 @@ class DlnaDmrEntity(MediaPlayerEntity):
assert self._device is not None
await self._device.async_next()
@property
def shuffle(self) -> bool | None:
"""Boolean if shuffle is enabled."""
if not self._device:
return None
play_mode = self._device.play_mode
if not play_mode:
return None
if play_mode == PlayMode.VENDOR_DEFINED:
return None
return play_mode in (PlayMode.SHUFFLE, PlayMode.RANDOM)
@catch_request_errors
async def async_set_shuffle(self, shuffle: bool) -> None:
"""Enable/disable shuffle mode."""
assert self._device is not None
repeat = self.repeat or REPEAT_MODE_OFF
potential_play_modes = SHUFFLE_PLAY_MODES[(shuffle, repeat)]
valid_play_modes = self._device.valid_play_modes
for mode in potential_play_modes:
if mode in valid_play_modes:
await self._device.async_set_play_mode(mode)
return
_LOGGER.debug(
"Couldn't find a suitable mode for shuffle=%s, repeat=%s", shuffle, repeat
)
@property
def repeat(self) -> str | None:
"""Return current repeat mode."""
if not self._device:
return None
play_mode = self._device.play_mode
if not play_mode:
return None
if play_mode == PlayMode.VENDOR_DEFINED:
return None
if play_mode == PlayMode.REPEAT_ONE:
return REPEAT_MODE_ONE
if play_mode in (PlayMode.REPEAT_ALL, PlayMode.RANDOM):
return REPEAT_MODE_ALL
return REPEAT_MODE_OFF
@catch_request_errors
async def async_set_repeat(self, repeat: str) -> None:
"""Set repeat mode."""
assert self._device is not None
shuffle = self.shuffle or False
potential_play_modes = REPEAT_PLAY_MODES[(shuffle, repeat)]
valid_play_modes = self._device.valid_play_modes
for mode in potential_play_modes:
if mode in valid_play_modes:
await self._device.async_set_play_mode(mode)
return
_LOGGER.debug(
"Couldn't find a suitable mode for shuffle=%s, repeat=%s", shuffle, repeat
)
@property
def sound_mode(self) -> str | None:
"""Name of the current sound mode, not supported by DLNA."""
return None
@property
def sound_mode_list(self) -> list[str] | None:
"""List of available sound modes."""
if not self._device:
return None
return self._device.preset_names
@catch_request_errors
async def async_select_sound_mode(self, sound_mode: str) -> None:
"""Select sound mode."""
assert self._device is not None
await self._device.async_select_preset(sound_mode)
@property
def media_title(self) -> str | None:
"""Title of current playing media."""
@ -705,12 +851,10 @@ class DlnaDmrEntity(MediaPlayerEntity):
not self._device.media_season_number
or self._device.media_season_number == "0"
) and self._device.media_episode_number:
try:
with contextlib.suppress(ValueError):
episode = int(self._device.media_episode_number, 10)
if episode > 100:
return str(episode // 100)
except ValueError:
pass
return self._device.media_season_number
@property
@ -723,12 +867,10 @@ class DlnaDmrEntity(MediaPlayerEntity):
not self._device.media_season_number
or self._device.media_season_number == "0"
) and self._device.media_episode_number:
try:
with contextlib.suppress(ValueError):
episode = int(self._device.media_episode_number, 10)
if episode > 100:
return str(episode % 100)
except ValueError:
pass
return self._device.media_episode_number
@property
@ -737,3 +879,10 @@ class DlnaDmrEntity(MediaPlayerEntity):
if not self._device:
return None
return self._device.media_channel_name
@property
def media_playlist(self) -> str | None:
"""Title of Playlist currently playing."""
if not self._device:
return None
return self._device.media_playlist_title

431
tests/components/dlna_dmr/test_media_player.py
View File

@ -8,12 +8,13 @@ from types import MappingProxyType
from typing import Any
from unittest.mock import ANY, DEFAULT, Mock, patch
from async_upnp_client import UpnpService, UpnpStateVariable
from async_upnp_client.exceptions import (
UpnpConnectionError,
UpnpError,
UpnpResponseError,
)
from async_upnp_client.profiles.dlna import TransportState
from async_upnp_client.profiles.dlna import PlayMode, TransportState
import pytest
from homeassistant import const as ha_const
@ -67,6 +68,16 @@ async def setup_mock_component(hass: HomeAssistant, mock_entry: MockConfigEntry)
return entity_id
async def get_attrs(hass: HomeAssistant, entity_id: str) -> Mapping[str, Any]:
"""Get updated device attributes."""
await async_update_entity(hass, entity_id)
entity_state = hass.states.get(entity_id)
assert entity_state is not None
attrs = entity_state.attributes
assert attrs is not None
return attrs
@pytest.fixture
async def mock_entity_id(
hass: HomeAssistant,
@ -335,9 +346,7 @@ async def test_setup_entry_with_options(
async def test_event_subscribe_failure(
hass: HomeAssistant,
config_entry_mock: MockConfigEntry,
dmr_device_mock: Mock,
hass: HomeAssistant, config_entry_mock: MockConfigEntry, dmr_device_mock: Mock
) -> None:
"""Test _device_connect aborts when async_subscribe_services fails."""
dmr_device_mock.async_subscribe_services.side_effect = UpnpError
@ -389,9 +398,7 @@ async def test_event_subscribe_rejected(
async def test_available_device(
hass: HomeAssistant,
dmr_device_mock: Mock,
mock_entity_id: str,
hass: HomeAssistant, dmr_device_mock: Mock, mock_entity_id: str
) -> None:
"""Test a DlnaDmrEntity with a connected DmrDevice."""
# Check hass device information is filled in
@ -429,19 +436,63 @@ async def test_available_device(
assert entity_state is not None
assert entity_state.state == ha_const.STATE_UNAVAILABLE
dmr_device_mock.profile_device.available = True
await async_update_entity(hass, mock_entity_id)
async def test_feature_flags(
hass: HomeAssistant, dmr_device_mock: Mock, mock_entity_id: str
) -> None:
"""Test feature flags of a connected DlnaDmrEntity."""
# Check supported feature flags, one at a time.
FEATURE_FLAGS: list[tuple[str, int]] = [
("has_volume_level", mp_const.SUPPORT_VOLUME_SET),
("has_volume_mute", mp_const.SUPPORT_VOLUME_MUTE),
("can_play", mp_const.SUPPORT_PLAY),
("can_pause", mp_const.SUPPORT_PAUSE),
("can_stop", mp_const.SUPPORT_STOP),
("can_previous", mp_const.SUPPORT_PREVIOUS_TRACK),
("can_next", mp_const.SUPPORT_NEXT_TRACK),
("has_play_media", mp_const.SUPPORT_PLAY_MEDIA),
("can_seek_rel_time", mp_const.SUPPORT_SEEK),
("has_presets", mp_const.SUPPORT_SELECT_SOUND_MODE),
]
# Clear all feature properties
dmr_device_mock.valid_play_modes = set()
for feat_prop, _ in FEATURE_FLAGS:
setattr(dmr_device_mock, feat_prop, False)
attrs = await get_attrs(hass, mock_entity_id)
assert attrs[ha_const.ATTR_SUPPORTED_FEATURES] == 0
# Test the properties cumulatively
expected_features = 0
for feat_prop, flag in FEATURE_FLAGS:
setattr(dmr_device_mock, feat_prop, True)
expected_features |= flag
attrs = await get_attrs(hass, mock_entity_id)
assert attrs[ha_const.ATTR_SUPPORTED_FEATURES] == expected_features
# shuffle and repeat features depend on the available play modes
PLAY_MODE_FEATURE_FLAGS: list[tuple[PlayMode, int]] = [
(PlayMode.NORMAL, 0),
(PlayMode.SHUFFLE, mp_const.SUPPORT_SHUFFLE_SET),
(PlayMode.REPEAT_ONE, mp_const.SUPPORT_REPEAT_SET),
(PlayMode.REPEAT_ALL, mp_const.SUPPORT_REPEAT_SET),
(PlayMode.RANDOM, mp_const.SUPPORT_SHUFFLE_SET),
(PlayMode.DIRECT_1, 0),
(PlayMode.INTRO, 0),
(PlayMode.VENDOR_DEFINED, 0),
]
for play_modes, flag in PLAY_MODE_FEATURE_FLAGS:
dmr_device_mock.valid_play_modes = {play_modes}
attrs = await get_attrs(hass, mock_entity_id)
assert attrs[ha_const.ATTR_SUPPORTED_FEATURES] == expected_features | flag
async def test_attributes(
hass: HomeAssistant, dmr_device_mock: Mock, mock_entity_id: str
) -> None:
"""Test attributes of a connected DlnaDmrEntity."""
# Check attributes come directly from the device
async def get_attrs() -> Mapping[str, Any]:
await async_update_entity(hass, mock_entity_id)
entity_state = hass.states.get(mock_entity_id)
assert entity_state is not None
attrs = entity_state.attributes
assert attrs is not None
return attrs
attrs = await get_attrs()
attrs = await get_attrs(hass, mock_entity_id)
assert attrs[mp_const.ATTR_MEDIA_VOLUME_LEVEL] is dmr_device_mock.volume_level
assert attrs[mp_const.ATTR_MEDIA_VOLUME_MUTED] is dmr_device_mock.is_volume_muted
assert attrs[mp_const.ATTR_MEDIA_DURATION] is dmr_device_mock.media_duration
@ -459,68 +510,65 @@ async def test_available_device(
assert attrs[mp_const.ATTR_MEDIA_SEASON] is dmr_device_mock.media_season_number
assert attrs[mp_const.ATTR_MEDIA_EPISODE] is dmr_device_mock.media_episode_number
assert attrs[mp_const.ATTR_MEDIA_CHANNEL] is dmr_device_mock.media_channel_name
assert attrs[mp_const.ATTR_SOUND_MODE_LIST] is dmr_device_mock.preset_names
# Entity picture is cached, won't correspond to remote image
assert isinstance(attrs[ha_const.ATTR_ENTITY_PICTURE], str)
# media_title depends on what is available
assert attrs[mp_const.ATTR_MEDIA_TITLE] is dmr_device_mock.media_program_title
dmr_device_mock.media_program_title = None
attrs = await get_attrs()
attrs = await get_attrs(hass, mock_entity_id)
assert attrs[mp_const.ATTR_MEDIA_TITLE] is dmr_device_mock.media_title
# media_content_type is mapped from UPnP class to MediaPlayer type
dmr_device_mock.media_class = "object.item.audioItem.musicTrack"
attrs = await get_attrs()
attrs = await get_attrs(hass, mock_entity_id)
assert attrs[mp_const.ATTR_MEDIA_CONTENT_TYPE] == mp_const.MEDIA_TYPE_MUSIC
dmr_device_mock.media_class = "object.item.videoItem.movie"
attrs = await get_attrs()
attrs = await get_attrs(hass, mock_entity_id)
assert attrs[mp_const.ATTR_MEDIA_CONTENT_TYPE] == mp_const.MEDIA_TYPE_MOVIE
dmr_device_mock.media_class = "object.item.videoItem.videoBroadcast"
attrs = await get_attrs()
attrs = await get_attrs(hass, mock_entity_id)
assert attrs[mp_const.ATTR_MEDIA_CONTENT_TYPE] == mp_const.MEDIA_TYPE_TVSHOW
# media_season & media_episode have a special case
dmr_device_mock.media_season_number = "0"
dmr_device_mock.media_episode_number = "123"
attrs = await get_attrs()
attrs = await get_attrs(hass, mock_entity_id)
assert attrs[mp_const.ATTR_MEDIA_SEASON] == "1"
assert attrs[mp_const.ATTR_MEDIA_EPISODE] == "23"
dmr_device_mock.media_season_number = "0"
dmr_device_mock.media_episode_number = "S1E23" # Unexpected and not parsed
attrs = await get_attrs()
attrs = await get_attrs(hass, mock_entity_id)
assert attrs[mp_const.ATTR_MEDIA_SEASON] == "0"
assert attrs[mp_const.ATTR_MEDIA_EPISODE] == "S1E23"
# Check supported feature flags, one at a time.
# tuple(async_upnp_client feature check property, HA feature flag)
FEATURE_FLAGS: list[tuple[str, int]] = [
("has_volume_level", mp_const.SUPPORT_VOLUME_SET),
("has_volume_mute", mp_const.SUPPORT_VOLUME_MUTE),
("can_play", mp_const.SUPPORT_PLAY),
("can_pause", mp_const.SUPPORT_PAUSE),
("can_stop", mp_const.SUPPORT_STOP),
("can_previous", mp_const.SUPPORT_PREVIOUS_TRACK),
("can_next", mp_const.SUPPORT_NEXT_TRACK),
("has_play_media", mp_const.SUPPORT_PLAY_MEDIA),
("can_seek_rel_time", mp_const.SUPPORT_SEEK),
]
# Clear all feature properties
for feat_prop, _ in FEATURE_FLAGS:
setattr(dmr_device_mock, feat_prop, False)
await async_update_entity(hass, mock_entity_id)
entity_state = hass.states.get(mock_entity_id)
assert entity_state is not None
assert entity_state.attributes[ha_const.ATTR_SUPPORTED_FEATURES] == 0
# Test the properties cumulatively
expected_features = 0
for feat_prop, flag in FEATURE_FLAGS:
setattr(dmr_device_mock, feat_prop, True)
expected_features |= flag
await async_update_entity(hass, mock_entity_id)
entity_state = hass.states.get(mock_entity_id)
assert entity_state is not None
assert (
entity_state.attributes[ha_const.ATTR_SUPPORTED_FEATURES]
== expected_features
)
# shuffle and repeat is based on device's play mode
for play_mode, shuffle, repeat in [
(PlayMode.NORMAL, False, mp_const.REPEAT_MODE_OFF),
(PlayMode.SHUFFLE, True, mp_const.REPEAT_MODE_OFF),
(PlayMode.REPEAT_ONE, False, mp_const.REPEAT_MODE_ONE),
(PlayMode.REPEAT_ALL, False, mp_const.REPEAT_MODE_ALL),
(PlayMode.RANDOM, True, mp_const.REPEAT_MODE_ALL),
(PlayMode.DIRECT_1, False, mp_const.REPEAT_MODE_OFF),
(PlayMode.INTRO, False, mp_const.REPEAT_MODE_OFF),
]:
dmr_device_mock.play_mode = play_mode
attrs = await get_attrs(hass, mock_entity_id)
assert attrs[mp_const.ATTR_MEDIA_SHUFFLE] is shuffle
assert attrs[mp_const.ATTR_MEDIA_REPEAT] == repeat
for bad_play_mode in [None, PlayMode.VENDOR_DEFINED]:
dmr_device_mock.play_mode = bad_play_mode
attrs = await get_attrs(hass, mock_entity_id)
assert mp_const.ATTR_MEDIA_SHUFFLE not in attrs
assert mp_const.ATTR_MEDIA_REPEAT not in attrs
async def test_services(
hass: HomeAssistant, dmr_device_mock: Mock, mock_entity_id: str
) -> None:
"""Test service calls of a connected DlnaDmrEntity."""
# Check interface methods interact directly with the device
await hass.services.async_call(
MP_DOMAIN,
@ -578,15 +626,22 @@ async def test_available_device(
blocking=True,
)
dmr_device_mock.async_seek_rel_time.assert_awaited_once_with(timedelta(seconds=33))
await hass.services.async_call(
MP_DOMAIN,
mp_const.SERVICE_SELECT_SOUND_MODE,
{ATTR_ENTITY_ID: mock_entity_id, mp_const.ATTR_SOUND_MODE: "Default"},
blocking=True,
)
dmr_device_mock.async_select_preset.assert_awaited_once_with("Default")
async def test_play_media_stopped(
hass: HomeAssistant, dmr_device_mock: Mock, mock_entity_id: str
) -> None:
"""Test play_media, starting from stopped and the device can stop."""
# play_media performs a few calls to the device for setup and play
# Start from stopped, and device can stop too
dmr_device_mock.can_stop = True
dmr_device_mock.transport_state = TransportState.STOPPED
dmr_device_mock.async_stop.reset_mock()
dmr_device_mock.async_set_transport_uri.reset_mock()
dmr_device_mock.async_wait_for_can_play.reset_mock()
dmr_device_mock.async_play.reset_mock()
await hass.services.async_call(
MP_DOMAIN,
mp_const.SERVICE_PLAY_MEDIA,
@ -598,20 +653,27 @@ async def test_available_device(
},
blocking=True,
)
dmr_device_mock.construct_play_media_metadata.assert_awaited_once_with(
media_url="http://192.88.99.20:8200/MediaItems/17621.mp3",
media_title="Home Assistant",
override_upnp_class="object.item.audioItem.musicTrack",
meta_data={},
)
dmr_device_mock.async_stop.assert_awaited_once_with()
dmr_device_mock.async_set_transport_uri.assert_awaited_once_with(
"http://192.88.99.20:8200/MediaItems/17621.mp3", "Home Assistant"
"http://192.88.99.20:8200/MediaItems/17621.mp3", "Home Assistant", ANY
)
dmr_device_mock.async_wait_for_can_play.assert_awaited_once_with()
dmr_device_mock.async_play.assert_awaited_once_with()
# play_media again, while the device is already playing and can't stop
async def test_play_media_playing(
hass: HomeAssistant, dmr_device_mock: Mock, mock_entity_id: str
) -> None:
"""Test play_media, device is already playing and can't stop."""
dmr_device_mock.can_stop = False
dmr_device_mock.transport_state = TransportState.PLAYING
dmr_device_mock.async_stop.reset_mock()
dmr_device_mock.async_set_transport_uri.reset_mock()
dmr_device_mock.async_wait_for_can_play.reset_mock()
dmr_device_mock.async_play.reset_mock()
await hass.services.async_call(
MP_DOMAIN,
mp_const.SERVICE_PLAY_MEDIA,
@ -623,14 +685,232 @@ async def test_available_device(
},
blocking=True,
)
dmr_device_mock.construct_play_media_metadata.assert_awaited_once_with(
media_url="http://192.88.99.20:8200/MediaItems/17621.mp3",
media_title="Home Assistant",
override_upnp_class="object.item.audioItem.musicTrack",
meta_data={},
)
dmr_device_mock.async_stop.assert_not_awaited()
dmr_device_mock.async_set_transport_uri.assert_awaited_once_with(
"http://192.88.99.20:8200/MediaItems/17621.mp3", "Home Assistant"
"http://192.88.99.20:8200/MediaItems/17621.mp3", "Home Assistant", ANY
)
dmr_device_mock.async_wait_for_can_play.assert_awaited_once_with()
dmr_device_mock.async_wait_for_can_play.assert_not_awaited()
dmr_device_mock.async_play.assert_not_awaited()
async def test_play_media_no_autoplay(
hass: HomeAssistant, dmr_device_mock: Mock, mock_entity_id: str
) -> None:
"""Test play_media with autoplay=False."""
# play_media performs a few calls to the device for setup and play
dmr_device_mock.can_stop = True
dmr_device_mock.transport_state = TransportState.STOPPED
await hass.services.async_call(
MP_DOMAIN,
mp_const.SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: mock_entity_id,
mp_const.ATTR_MEDIA_CONTENT_TYPE: mp_const.MEDIA_TYPE_MUSIC,
mp_const.ATTR_MEDIA_CONTENT_ID: "http://192.88.99.20:8200/MediaItems/17621.mp3",
mp_const.ATTR_MEDIA_ENQUEUE: False,
mp_const.ATTR_MEDIA_EXTRA: {"autoplay": False},
},
blocking=True,
)
dmr_device_mock.construct_play_media_metadata.assert_awaited_once_with(
media_url="http://192.88.99.20:8200/MediaItems/17621.mp3",
media_title="Home Assistant",
override_upnp_class="object.item.audioItem.musicTrack",
meta_data={},
)
dmr_device_mock.async_stop.assert_awaited_once_with()
dmr_device_mock.async_set_transport_uri.assert_awaited_once_with(
"http://192.88.99.20:8200/MediaItems/17621.mp3", "Home Assistant", ANY
)
dmr_device_mock.async_wait_for_can_play.assert_not_awaited()
dmr_device_mock.async_play.assert_not_awaited()
async def test_play_media_metadata(
hass: HomeAssistant, dmr_device_mock: Mock, mock_entity_id: str
) -> None:
"""Test play_media constructs useful metadata from user params."""
await hass.services.async_call(
MP_DOMAIN,
mp_const.SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: mock_entity_id,
mp_const.ATTR_MEDIA_CONTENT_TYPE: mp_const.MEDIA_TYPE_MUSIC,
mp_const.ATTR_MEDIA_CONTENT_ID: "http://192.88.99.20:8200/MediaItems/17621.mp3",
mp_const.ATTR_MEDIA_ENQUEUE: False,
mp_const.ATTR_MEDIA_EXTRA: {
"title": "Mock song",
"thumb": "http://192.88.99.20:8200/MediaItems/17621.jpg",
"metadata": {"artist": "Mock artist", "album": "Mock album"},
},
},
blocking=True,
)
dmr_device_mock.construct_play_media_metadata.assert_awaited_once_with(
media_url="http://192.88.99.20:8200/MediaItems/17621.mp3",
media_title="Mock song",
override_upnp_class="object.item.audioItem.musicTrack",
meta_data={
"artist": "Mock artist",
"album": "Mock album",
"album_art_uri": "http://192.88.99.20:8200/MediaItems/17621.jpg",
},
)
# Check again for a different media type
dmr_device_mock.construct_play_media_metadata.reset_mock()
await hass.services.async_call(
MP_DOMAIN,
mp_const.SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: mock_entity_id,
mp_const.ATTR_MEDIA_CONTENT_TYPE: mp_const.MEDIA_TYPE_TVSHOW,
mp_const.ATTR_MEDIA_CONTENT_ID: "http://192.88.99.20:8200/MediaItems/123.mkv",
mp_const.ATTR_MEDIA_ENQUEUE: False,
mp_const.ATTR_MEDIA_EXTRA: {
"title": "Mock show",
"metadata": {"season": 1, "episode": 12},
},
},
blocking=True,
)
dmr_device_mock.construct_play_media_metadata.assert_awaited_once_with(
media_url="http://192.88.99.20:8200/MediaItems/123.mkv",
media_title="Mock show",
override_upnp_class="object.item.videoItem.videoBroadcast",
meta_data={"episodeSeason": 1, "episodeNumber": 12},
)
async def test_shuffle_repeat_modes(
hass: HomeAssistant, dmr_device_mock: Mock, mock_entity_id: str
) -> None:
"""Test setting repeat and shuffle modes."""
# Test shuffle with all variations of existing play mode
dmr_device_mock.valid_play_modes = {mode.value for mode in PlayMode}
for init_mode, shuffle_set, expect_mode in [
(PlayMode.NORMAL, False, PlayMode.NORMAL),
(PlayMode.SHUFFLE, False, PlayMode.NORMAL),
(PlayMode.REPEAT_ONE, False, PlayMode.REPEAT_ONE),
(PlayMode.REPEAT_ALL, False, PlayMode.REPEAT_ALL),
(PlayMode.RANDOM, False, PlayMode.REPEAT_ALL),
(PlayMode.NORMAL, True, PlayMode.SHUFFLE),
(PlayMode.SHUFFLE, True, PlayMode.SHUFFLE),
(PlayMode.REPEAT_ONE, True, PlayMode.RANDOM),
(PlayMode.REPEAT_ALL, True, PlayMode.RANDOM),
(PlayMode.RANDOM, True, PlayMode.RANDOM),
]:
dmr_device_mock.play_mode = init_mode
await hass.services.async_call(
MP_DOMAIN,
ha_const.SERVICE_SHUFFLE_SET,
{ATTR_ENTITY_ID: mock_entity_id, mp_const.ATTR_MEDIA_SHUFFLE: shuffle_set},
blocking=True,
)
dmr_device_mock.async_set_play_mode.assert_awaited_with(expect_mode)
# Test repeat with all variations of existing play mode
for init_mode, repeat_set, expect_mode in [
(PlayMode.NORMAL, mp_const.REPEAT_MODE_OFF, PlayMode.NORMAL),
(PlayMode.SHUFFLE, mp_const.REPEAT_MODE_OFF, PlayMode.SHUFFLE),
(PlayMode.REPEAT_ONE, mp_const.REPEAT_MODE_OFF, PlayMode.NORMAL),
(PlayMode.REPEAT_ALL, mp_const.REPEAT_MODE_OFF, PlayMode.NORMAL),
(PlayMode.RANDOM, mp_const.REPEAT_MODE_OFF, PlayMode.SHUFFLE),
(PlayMode.NORMAL, mp_const.REPEAT_MODE_ONE, PlayMode.REPEAT_ONE),
(PlayMode.SHUFFLE, mp_const.REPEAT_MODE_ONE, PlayMode.REPEAT_ONE),
(PlayMode.REPEAT_ONE, mp_const.REPEAT_MODE_ONE, PlayMode.REPEAT_ONE),
(PlayMode.REPEAT_ALL, mp_const.REPEAT_MODE_ONE, PlayMode.REPEAT_ONE),
(PlayMode.RANDOM, mp_const.REPEAT_MODE_ONE, PlayMode.REPEAT_ONE),
(PlayMode.NORMAL, mp_const.REPEAT_MODE_ALL, PlayMode.REPEAT_ALL),
(PlayMode.SHUFFLE, mp_const.REPEAT_MODE_ALL, PlayMode.RANDOM),
(PlayMode.REPEAT_ONE, mp_const.REPEAT_MODE_ALL, PlayMode.REPEAT_ALL),
(PlayMode.REPEAT_ALL, mp_const.REPEAT_MODE_ALL, PlayMode.REPEAT_ALL),
(PlayMode.RANDOM, mp_const.REPEAT_MODE_ALL, PlayMode.RANDOM),
]:
dmr_device_mock.play_mode = init_mode
await hass.services.async_call(
MP_DOMAIN,
ha_const.SERVICE_REPEAT_SET,
{ATTR_ENTITY_ID: mock_entity_id, mp_const.ATTR_MEDIA_REPEAT: repeat_set},
blocking=True,
)
dmr_device_mock.async_set_play_mode.assert_awaited_with(expect_mode)
# Test shuffle when the device doesn't support the desired play mode.
# Trying to go from RANDOM -> REPEAT_MODE_ALL, but nothing in the list is supported.
dmr_device_mock.async_set_play_mode.reset_mock()
dmr_device_mock.play_mode = PlayMode.RANDOM
dmr_device_mock.valid_play_modes = {PlayMode.SHUFFLE, PlayMode.RANDOM}
await hass.services.async_call(
MP_DOMAIN,
ha_const.SERVICE_SHUFFLE_SET,
{ATTR_ENTITY_ID: mock_entity_id, mp_const.ATTR_MEDIA_SHUFFLE: False},
blocking=True,
)
dmr_device_mock.async_set_play_mode.assert_not_awaited()
# Test repeat when the device doesn't support the desired play mode.
# Trying to go from RANDOM -> SHUFFLE, but nothing in the list is supported.
dmr_device_mock.async_set_play_mode.reset_mock()
dmr_device_mock.play_mode = PlayMode.RANDOM
dmr_device_mock.valid_play_modes = {PlayMode.REPEAT_ONE, PlayMode.REPEAT_ALL}
await hass.services.async_call(
MP_DOMAIN,
ha_const.SERVICE_REPEAT_SET,
{
ATTR_ENTITY_ID: mock_entity_id,
mp_const.ATTR_MEDIA_REPEAT: mp_const.REPEAT_MODE_OFF,
},
blocking=True,
)
dmr_device_mock.async_set_play_mode.assert_not_awaited()
async def test_playback_update_state(
hass: HomeAssistant, dmr_device_mock: Mock, mock_entity_id: str
) -> None:
"""Test starting or pausing playback causes the state to be refreshed.
This is necessary for responsive updates of the current track position and
total track time.
"""
on_event = dmr_device_mock.on_event
mock_service = Mock(UpnpService)
mock_service.service_id = "urn:upnp-org:serviceId:AVTransport"
mock_state_variable = Mock(UpnpStateVariable)
mock_state_variable.name = "TransportState"
# Event update that device has started playing, device should get polled
mock_state_variable.value = TransportState.PLAYING
on_event(mock_service, [mock_state_variable])
await hass.async_block_till_done()
dmr_device_mock.async_update.assert_awaited_once_with(do_ping=False)
# Event update that device has paused playing, device should get polled
dmr_device_mock.async_update.reset_mock()
mock_state_variable.value = TransportState.PAUSED_PLAYBACK
on_event(mock_service, [mock_state_variable])
await hass.async_block_till_done()
dmr_device_mock.async_update.assert_awaited_once_with(do_ping=False)
# Different service shouldn't do anything
dmr_device_mock.async_update.reset_mock()
mock_service.service_id = "urn:upnp-org:serviceId:RenderingControl"
on_event(mock_service, [mock_state_variable])
await hass.async_block_till_done()
dmr_device_mock.async_update.assert_not_awaited()
async def test_unavailable_device(
hass: HomeAssistant,
domain_data_mock: Mock,
@ -691,6 +971,7 @@ async def test_unavailable_device(
assert attrs[ha_const.ATTR_FRIENDLY_NAME] == MOCK_DEVICE_NAME
assert attrs[ha_const.ATTR_SUPPORTED_FEATURES] == 0
assert mp_const.ATTR_SOUND_MODE_LIST not in attrs
# Check service calls do nothing
SERVICES: list[tuple[str, dict]] = [
@ -710,6 +991,9 @@ async def test_unavailable_device(
mp_const.ATTR_MEDIA_ENQUEUE: False,
},
),
(mp_const.SERVICE_SELECT_SOUND_MODE, {mp_const.ATTR_SOUND_MODE: "Default"}),
(ha_const.SERVICE_SHUFFLE_SET, {mp_const.ATTR_MEDIA_SHUFFLE: True}),
(ha_const.SERVICE_REPEAT_SET, {mp_const.ATTR_MEDIA_REPEAT: "all"}),
]
for service, data in SERVICES:
await hass.services.async_call(
@ -1312,6 +1596,9 @@ async def test_disappearing_device(
# media_image_url is normally hidden by entity_picture, but we want a direct check
assert entity.media_image_url is None
# Check attributes that are normally pre-checked
assert entity.sound_mode_list is None
# Test service calls
await entity.async_set_volume_level(0.1)
await entity.async_mute_volume(True)
@ -1322,6 +1609,9 @@ async def test_disappearing_device(
await entity.async_play_media("", "")
await entity.async_media_previous_track()
await entity.async_media_next_track()
await entity.async_set_shuffle(True)
await entity.async_set_repeat(mp_const.REPEAT_MODE_ALL)
await entity.async_select_sound_mode("Default")
async def test_resubscribe_failure(
@ -1335,7 +1625,8 @@ async def test_resubscribe_failure(
dmr_device_mock.async_update.reset_mock()
on_event = dmr_device_mock.on_event
on_event(None, [])
mock_service = Mock(UpnpService)
on_event(mock_service, [])
await hass.async_block_till_done()
await async_update_entity(hass, mock_entity_id)