Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add source selection to Heos component (#22592) 

* Add select source support

* Review feedback changes

* Removed unused import

* Ignore 'umused' import used in typing

* Only include trace back on useful errors

* Remove return from play_source
pull/22630/head
Andrew Sayre 2019-04-01 11:58:52 -05:00 committed by GitHub
parent a5b03541e9
commit 9f2c5b7231
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 365 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

137
homeassistant/components/heos/__init__.py
View File

@ -1,5 +1,6 @@
"""Denon HEOS Media Player."""
import asyncio
from datetime import timedelta
import logging
import voluptuous as vol
@ -8,13 +9,17 @@ from homeassistant.components.media_player.const import (
DOMAIN as MEDIA_PLAYER_DOMAIN)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, EVENT_HOMEASSISTANT_STOP
from homeassistant.exceptions import ConfigEntryNotReady
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from homeassistant.util import Throttle
from .config_flow import format_title
from .const import DATA_CONTROLLER, DOMAIN
from .const import (
COMMAND_RETRY_ATTEMPTS, COMMAND_RETRY_DELAY, DATA_CONTROLLER,
DATA_SOURCE_MANAGER, DOMAIN, SIGNAL_HEOS_SOURCES_UPDATED)
REQUIREMENTS = ['pyheos==0.2.0']
REQUIREMENTS = ['pyheos==0.3.0']
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
@ -22,6 +27,8 @@ CONFIG_SCHEMA = vol.Schema({
})
}, extra=vol.ALLOW_EXTRA)
MIN_UPDATE_SOURCES = timedelta(seconds=1)
_LOGGER = logging.getLogger(__name__)
@ -50,7 +57,7 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType):
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry):
"""Initialize config entry which represents the HEOS controller."""
from pyheos import Heos
from pyheos import Heos, CommandError
host = entry.data[CONF_HOST]
# Setting all_progress_events=False ensures that we only receive a
# media position update upon start of playback or when media changes
@ -58,26 +65,34 @@ async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry):
try:
await controller.connect(auto_reconnect=True)
# Auto reconnect only operates if initial connection was successful.
except (asyncio.TimeoutError, ConnectionError) as error:
except (asyncio.TimeoutError, ConnectionError, CommandError) as error:
await controller.disconnect()
_LOGGER.exception("Unable to connect to controller %s: %s",
host, type(error).__name__)
return False
_LOGGER.debug("Unable to connect to controller %s: %s", host, error)
raise ConfigEntryNotReady
# Disconnect when shutting down
async def disconnect_controller(event):
await controller.disconnect()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, disconnect_controller)
try:
players = await controller.get_players()
except (asyncio.TimeoutError, ConnectionError) as error:
players, favorites, inputs = await asyncio.gather(
controller.get_players(),
controller.get_favorites(),
controller.get_input_sources()
)
except (asyncio.TimeoutError, ConnectionError, CommandError) as error:
await controller.disconnect()
_LOGGER.exception("Unable to retrieve players: %s",
type(error).__name__)
return False
_LOGGER.debug("Unable to retrieve players and sources: %s", error,
exc_info=isinstance(error, CommandError))
raise ConfigEntryNotReady
source_manager = SourceManager(favorites, inputs)
source_manager.connect_update(hass, controller)
hass.data[DOMAIN] = {
DATA_CONTROLLER: controller,
DATA_SOURCE_MANAGER: source_manager,
MEDIA_PLAYER_DOMAIN: players
}
hass.async_create_task(hass.config_entries.async_forward_entry_setup(
@ -88,7 +103,105 @@ async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry):
async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry):
"""Unload a config entry."""
controller = hass.data[DOMAIN][DATA_CONTROLLER]
controller.dispatcher.disconnect_all()
await controller.disconnect()
hass.data.pop(DOMAIN)
return await hass.config_entries.async_forward_entry_unload(
entry, MEDIA_PLAYER_DOMAIN)
class SourceManager:
"""Class that manages sources for players."""
def __init__(self, favorites, inputs, *,
retry_delay: int = COMMAND_RETRY_DELAY,
max_retry_attempts: int = COMMAND_RETRY_ATTEMPTS):
"""Init input manager."""
self.retry_delay = retry_delay
self.max_retry_attempts = max_retry_attempts
self.favorites = favorites
self.inputs = inputs
self.source_list = self._build_source_list()
def _build_source_list(self):
"""Build a single list of inputs from various types."""
source_list = []
source_list.extend([favorite.name for favorite
in self.favorites.values()])
source_list.extend([source.name for source in self.inputs])
return source_list
async def play_source(self, source: str, player):
"""Determine type of source and play it."""
index = next((index for index, favorite in self.favorites.items()
if favorite.name == source), None)
if index is not None:
await player.play_favorite(index)
return
input_source = next((input_source for input_source in self.inputs
if input_source.name == source), None)
if input_source is not None:
await player.play_input_source(input_source)
return
_LOGGER.error("Unknown source: %s", source)
def get_current_source(self, now_playing_media):
"""Determine current source from now playing media."""
from pyheos import const
# Match input by input_name:media_id
if now_playing_media.source_id == const.MUSIC_SOURCE_AUX_INPUT:
return next((input_source.name for input_source in self.inputs
if input_source.input_name ==
now_playing_media.media_id), None)
# Try matching favorite by name:station or media_id:album_id
return next((source.name for source in self.favorites.values()
if source.name == now_playing_media.station
or source.media_id == now_playing_media.album_id), None)
def connect_update(self, hass, controller):
"""
Connect listener for when sources change and signal player update.
EVENT_SOURCES_CHANGED is often raised multiple times in response to a
physical event therefore throttle it. Retrieving sources immediately
after the event may fail so retry.
"""
from pyheos import CommandError, const
@Throttle(MIN_UPDATE_SOURCES)
async def get_sources():
retry_attempts = 0
while True:
try:
return await asyncio.gather(
controller.get_favorites(),
controller.get_input_sources())
except (asyncio.TimeoutError, ConnectionError, CommandError) \
as error:
if retry_attempts < self.max_retry_attempts:
retry_attempts += 1
_LOGGER.debug("Error retrieving sources and will "
"retry: %s", error,
exc_info=isinstance(error, CommandError))
await asyncio.sleep(self.retry_delay)
else:
_LOGGER.error("Unable to update sources: %s", error,
exc_info=isinstance(error, CommandError))
return
async def update_sources(event):
if event in const.EVENT_SOURCES_CHANGED:
sources = await get_sources()
# If throttled, it will return None
if sources:
self.favorites, self.inputs = sources
self.source_list = self._build_source_list()
_LOGGER.debug("Sources updated due to changed event")
# Let players know to update
hass.helpers.dispatcher.async_dispatcher_send(
SIGNAL_HEOS_SOURCES_UPDATED)
controller.dispatcher.connect(
const.SIGNAL_CONTROLLER_EVENT, update_sources)

4
homeassistant/components/heos/const.py
View File

@ -1,4 +1,8 @@
"""Const for the HEOS integration."""
COMMAND_RETRY_ATTEMPTS = 2
COMMAND_RETRY_DELAY = 1
DATA_CONTROLLER = "controller"
DATA_SOURCE_MANAGER = "source_manager"
DOMAIN = 'heos'
SIGNAL_HEOS_SOURCES_UPDATED = "heos_sources_updated"

47
homeassistant/components/heos/media_player.py
View File

@ -1,22 +1,27 @@
"""Denon HEOS Media Player."""
from functools import reduce
from operator import ior
from typing import Sequence
from homeassistant.components.media_player import MediaPlayerDevice
from homeassistant.components.media_player.const import (
DOMAIN, MEDIA_TYPE_MUSIC, SUPPORT_CLEAR_PLAYLIST, SUPPORT_NEXT_TRACK,
SUPPORT_PAUSE, SUPPORT_PLAY, SUPPORT_PREVIOUS_TRACK, SUPPORT_SHUFFLE_SET,
SUPPORT_STOP, SUPPORT_VOLUME_MUTE, SUPPORT_VOLUME_SET, SUPPORT_VOLUME_STEP)
SUPPORT_PAUSE, SUPPORT_PLAY, SUPPORT_PREVIOUS_TRACK, SUPPORT_SELECT_SOURCE,
SUPPORT_SHUFFLE_SET, SUPPORT_STOP, SUPPORT_VOLUME_MUTE, SUPPORT_VOLUME_SET,
SUPPORT_VOLUME_STEP)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import STATE_IDLE, STATE_PAUSED, STATE_PLAYING
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.util.dt import utcnow
from .const import DOMAIN as HEOS_DOMAIN
from .const import (
DATA_SOURCE_MANAGER, DOMAIN as HEOS_DOMAIN, SIGNAL_HEOS_SOURCES_UPDATED)
DEPENDENCIES = ['heos']
BASE_SUPPORTED_FEATURES = SUPPORT_VOLUME_MUTE | SUPPORT_VOLUME_SET | \
SUPPORT_VOLUME_STEP | SUPPORT_CLEAR_PLAYLIST | \
SUPPORT_SHUFFLE_SET
SUPPORT_SHUFFLE_SET | SUPPORT_SELECT_SOURCE
async def async_setup_platform(
@ -25,8 +30,9 @@ async def async_setup_platform(
pass
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Add binary sensors for a config entry."""
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry,
async_add_entities):
"""Add media players for a config entry."""
players = hass.data[HEOS_DOMAIN][DOMAIN]
devices = [HeosMediaPlayer(player) for player in players.values()]
async_add_entities(devices, True)
@ -42,6 +48,7 @@ class HeosMediaPlayer(MediaPlayerDevice):
self._player = player
self._signals = []
self._supported_features = BASE_SUPPORTED_FEATURES
self._source_manager = None
self._play_state_to_state = {
const.PLAY_STATE_PLAY: STATE_PLAYING,
const.PLAY_STATE_STOP: STATE_IDLE,
@ -74,9 +81,14 @@ class HeosMediaPlayer(MediaPlayerDevice):
self._media_position_updated_at = utcnow()
await self.async_update_ha_state(True)
async def _sources_updated(self):
"""Handle sources changed."""
await self.async_update_ha_state(True)
async def async_added_to_hass(self):
"""Device added to hass."""
from pyheos import const
self._source_manager = self.hass.data[HEOS_DOMAIN][DATA_SOURCE_MANAGER]
# Update state when attributes of the player change
self._signals.append(self._player.heos.dispatcher.connect(
const.SIGNAL_PLAYER_EVENT, self._player_update))
@ -86,6 +98,10 @@ class HeosMediaPlayer(MediaPlayerDevice):
# Update state upon connect/disconnects
self._signals.append(self._player.heos.dispatcher.connect(
const.SIGNAL_HEOS_EVENT, self._heos_event))
# Update state when sources change
self._signals.append(
self.hass.helpers.dispatcher.async_dispatcher_connect(
SIGNAL_HEOS_SOURCES_UPDATED, self._sources_updated))
async def async_clear_playlist(self):
"""Clear players playlist."""
@ -115,6 +131,10 @@ class HeosMediaPlayer(MediaPlayerDevice):
"""Mute the volume."""
await self._player.set_mute(mute)
async def async_select_source(self, source):
"""Select input source."""
await self._source_manager.play_source(source, self._player)
async def async_set_shuffle(self, shuffle):
"""Enable/disable shuffle mode."""
await self._player.set_play_mode(self._player.repeat, shuffle)
@ -218,7 +238,9 @@ class HeosMediaPlayer(MediaPlayerDevice):
@property
def media_image_url(self) -> str:
"""Image url of current playing media."""
return self._player.now_playing_media.image_url
# May be an empty string, if so, return None
image_url = self._player.now_playing_media.image_url
return image_url if image_url else None
@property
def media_title(self) -> str:
@ -240,6 +262,17 @@ class HeosMediaPlayer(MediaPlayerDevice):
"""Boolean if shuffle is enabled."""
return self._player.shuffle
@property
def source(self) -> str:
"""Name of the current input source."""
return self._source_manager.get_current_source(
self._player.now_playing_media)
@property
def source_list(self) -> Sequence[str]:
"""List of available input sources."""
return self._source_manager.source_list
@property
def state(self) -> str:
"""State of the player."""

2
requirements_all.txt
View File

@ -1077,7 +1077,7 @@ pygtt==1.1.2
pyhaversion==2.0.3
# homeassistant.components.heos
pyheos==0.2.0
pyheos==0.3.0
# homeassistant.components.hikvision.binary_sensor
pyhik==0.2.2

2
requirements_test_all.txt
View File

@ -206,7 +206,7 @@ pydeconz==54
pydispatcher==2.0.5
# homeassistant.components.heos
pyheos==0.2.0
pyheos==0.3.0
# homeassistant.components.homematic
pyhomematic==0.1.58

46
tests/components/heos/conftest.py
View File

@ -1,6 +1,8 @@
"""Configuration for HEOS tests."""
from typing import Dict, Sequence
from asynctest.mock import Mock, patch as patch
from pyheos import Dispatcher, HeosPlayer, const
from pyheos import Dispatcher, HeosPlayer, HeosSource, InputSource, const
import pytest
from homeassistant.components.heos import DOMAIN
@ -17,12 +19,15 @@ def config_entry_fixture():
@pytest.fixture(name="controller")
def controller_fixture(players):
def controller_fixture(players, favorites, input_sources, dispatcher):
"""Create a mock Heos controller fixture."""
with patch("pyheos.Heos", autospec=True) as mock:
mock_heos = mock.return_value
mock_heos.dispatcher = dispatcher
mock_heos.get_players.return_value = players
mock_heos.players = players
mock_heos.get_favorites.return_value = favorites
mock_heos.get_input_sources.return_value = input_sources
yield mock_heos
@ -35,10 +40,10 @@ def config_fixture():
@pytest.fixture(name="players")
def player_fixture():
def player_fixture(dispatcher):
"""Create a mock HeosPlayer."""
player = Mock(HeosPlayer, autospec=True)
player.heos.dispatcher = Dispatcher()
player.heos.dispatcher = dispatcher
player.player_id = 1
player.name = "Test Player"
player.model = "Test Model"
@ -65,3 +70,36 @@ def player_fixture():
player.now_playing_media.image_url = "http://"
player.now_playing_media.song = "Song"
return {player.player_id: player}
@pytest.fixture(name="favorites")
def favorites_fixture() -> Dict[int, HeosSource]:
"""Create favorites fixture."""
station = Mock(HeosSource, autospec=True)
station.type = const.TYPE_STATION
station.name = "Today's Hits Radio"
station.media_id = '123456789'
radio = Mock(HeosSource, autospec=True)
radio.type = const.TYPE_STATION
radio.name = "Classical MPR (Classical Music)"
radio.media_id = 's1234'
return {
1: station,
2: radio
}
@pytest.fixture(name="input_sources")
def input_sources_fixture() -> Sequence[InputSource]:
"""Create a set of input sources for testing."""
source = Mock(InputSource, autospec=True)
source.player_id = 1
source.input_name = const.INPUT_AUX_IN_1
source.name = "HEOS Drive - Line In 1"
return [source]
@pytest.fixture(name="dispatcher")
def dispatcher_fixture() -> Dispatcher:
"""Create a dispatcher for testing."""
return Dispatcher()

51
tests/components/heos/test_init.py
View File

@ -2,12 +2,17 @@
import asyncio
from asynctest import patch
from pyheos import CommandError, const
import pytest
from homeassistant.components.heos import async_setup_entry, async_unload_entry
from homeassistant.components.heos.const import DATA_CONTROLLER, DOMAIN
from homeassistant.components.heos import (
SourceManager, async_setup_entry, async_unload_entry)
from homeassistant.components.heos.const import (
DATA_CONTROLLER, DATA_SOURCE_MANAGER, DOMAIN)
from homeassistant.components.media_player.const import (
DOMAIN as MEDIA_PLAYER_DOMAIN)
from homeassistant.const import CONF_HOST
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.setup import async_setup_component
@ -36,7 +41,7 @@ async def test_async_setup_updates_entry(hass, config_entry, config):
async def test_async_setup_returns_true(hass, config_entry, config):
"""Test component setup updates entry from config."""
"""Test component setup from config."""
config_entry.add_to_hass(hass)
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
@ -46,7 +51,7 @@ async def test_async_setup_returns_true(hass, config_entry, config):
async def test_async_setup_no_config_returns_true(hass, config_entry):
"""Test component setup updates entry from entry only."""
"""Test component setup from entry only."""
config_entry.add_to_hass(hass)
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
@ -67,20 +72,20 @@ async def test_async_setup_entry_loads_platforms(
assert forward_mock.call_count == 1
assert controller.connect.call_count == 1
controller.disconnect.assert_not_called()
assert hass.data[DOMAIN] == {
DATA_CONTROLLER: controller,
MEDIA_PLAYER_DOMAIN: controller.players
}
assert hass.data[DOMAIN][DATA_CONTROLLER] == controller
assert hass.data[DOMAIN][MEDIA_PLAYER_DOMAIN] == controller.players
assert isinstance(hass.data[DOMAIN][DATA_SOURCE_MANAGER], SourceManager)
async def test_async_setup_entry_connect_failure(
hass, config_entry, controller):
"""Test failure to connect does not load entry."""
"""Connection failure raises ConfigEntryNotReady."""
config_entry.add_to_hass(hass)
errors = [ConnectionError, asyncio.TimeoutError]
for error in errors:
controller.connect.side_effect = error
assert not await async_setup_entry(hass, config_entry)
with pytest.raises(ConfigEntryNotReady):
await async_setup_entry(hass, config_entry)
await hass.async_block_till_done()
assert controller.connect.call_count == 1
assert controller.disconnect.call_count == 1
@ -90,12 +95,13 @@ async def test_async_setup_entry_connect_failure(
async def test_async_setup_entry_player_failure(
hass, config_entry, controller):
"""Test failure to retrieve players does not load entry."""
"""Failure to retrieve players/sources raises ConfigEntryNotReady."""
config_entry.add_to_hass(hass)
errors = [ConnectionError, asyncio.TimeoutError]
for error in errors:
controller.get_players.side_effect = error
assert not await async_setup_entry(hass, config_entry)
with pytest.raises(ConfigEntryNotReady):
await async_setup_entry(hass, config_entry)
await hass.async_block_till_done()
assert controller.connect.call_count == 1
assert controller.disconnect.call_count == 1
@ -112,3 +118,24 @@ async def test_unload_entry(hass, config_entry, controller):
await hass.async_block_till_done()
assert controller.disconnect.call_count == 1
assert unload.call_count == 1
assert DOMAIN not in hass.data
async def test_update_sources_retry(hass, config_entry, config, controller,
caplog):
"""Test update sources retries on failures to max attempts."""
config_entry.add_to_hass(hass)
assert await async_setup_component(hass, DOMAIN, config)
controller.get_favorites.reset_mock()
controller.get_input_sources.reset_mock()
source_manager = hass.data[DOMAIN][DATA_SOURCE_MANAGER]
source_manager.retry_delay = 0
source_manager.max_retry_attempts = 1
controller.get_favorites.side_effect = CommandError("Test", "test", 0)
controller.dispatcher.send(
const.SIGNAL_CONTROLLER_EVENT, const.EVENT_SOURCES_CHANGED)
# Wait until it's finished
while "Unable to update sources" not in caplog.text:
await asyncio.sleep(0.1)
assert controller.get_favorites.call_count == 2
assert controller.get_input_sources.call_count == 2

120
tests/components/heos/test_media_player.py
View File

@ -1,16 +1,19 @@
"""Tests for the Heos Media Player platform."""
import asyncio
from pyheos import const
from homeassistant.components.heos import media_player
from homeassistant.components.heos.const import DOMAIN
from homeassistant.components.heos.const import (
DATA_SOURCE_MANAGER, DOMAIN, SIGNAL_HEOS_SOURCES_UPDATED)
from homeassistant.components.media_player.const import (
ATTR_MEDIA_ALBUM_NAME, ATTR_MEDIA_ARTIST, ATTR_MEDIA_CONTENT_ID,
ATTR_MEDIA_CONTENT_TYPE, ATTR_MEDIA_DURATION, ATTR_MEDIA_POSITION,
ATTR_MEDIA_POSITION_UPDATED_AT, ATTR_MEDIA_SHUFFLE, ATTR_MEDIA_TITLE,
ATTR_MEDIA_VOLUME_LEVEL, ATTR_MEDIA_VOLUME_MUTED,
DOMAIN as MEDIA_PLAYER_DOMAIN, MEDIA_TYPE_MUSIC, SERVICE_CLEAR_PLAYLIST,
SUPPORT_NEXT_TRACK, SUPPORT_PAUSE, SUPPORT_PLAY, SUPPORT_PREVIOUS_TRACK,
SUPPORT_STOP)
ATTR_INPUT_SOURCE, ATTR_INPUT_SOURCE_LIST, ATTR_MEDIA_ALBUM_NAME,
ATTR_MEDIA_ARTIST, ATTR_MEDIA_CONTENT_ID, ATTR_MEDIA_CONTENT_TYPE,
ATTR_MEDIA_DURATION, ATTR_MEDIA_POSITION, ATTR_MEDIA_POSITION_UPDATED_AT,
ATTR_MEDIA_SHUFFLE, ATTR_MEDIA_TITLE, ATTR_MEDIA_VOLUME_LEVEL,
ATTR_MEDIA_VOLUME_MUTED, DOMAIN as MEDIA_PLAYER_DOMAIN, MEDIA_TYPE_MUSIC,
SERVICE_CLEAR_PLAYLIST, SERVICE_SELECT_SOURCE, SUPPORT_NEXT_TRACK,
SUPPORT_PAUSE, SUPPORT_PLAY, SUPPORT_PREVIOUS_TRACK, SUPPORT_STOP)
from homeassistant.const import (
ATTR_ENTITY_ID, ATTR_FRIENDLY_NAME, ATTR_SUPPORTED_FEATURES,
SERVICE_MEDIA_NEXT_TRACK, SERVICE_MEDIA_PAUSE, SERVICE_MEDIA_PLAY,
@ -56,10 +59,13 @@ async def test_state_attributes(hass, config_entry, config, controller):
assert state.attributes[ATTR_SUPPORTED_FEATURES] == \
SUPPORT_PLAY | SUPPORT_PAUSE | SUPPORT_STOP | SUPPORT_NEXT_TRACK | \
SUPPORT_PREVIOUS_TRACK | media_player.BASE_SUPPORTED_FEATURES
assert ATTR_INPUT_SOURCE not in state.attributes
assert state.attributes[ATTR_INPUT_SOURCE_LIST] == \
hass.data[DOMAIN][DATA_SOURCE_MANAGER].source_list
async def test_updates_start_from_signals(
hass, config_entry, config, controller):
hass, config_entry, config, controller, favorites):
"""Tests dispatched signals update player."""
await setup_platform(hass, config_entry, config)
player = controller.players[1]
@ -110,6 +116,23 @@ async def test_updates_start_from_signals(
state = hass.states.get('media_player.test_player')
assert state.state == STATE_PLAYING
# Test sources event update
event = asyncio.Event()
async def set_signal():
event.set()
hass.helpers.dispatcher.async_dispatcher_connect(
SIGNAL_HEOS_SOURCES_UPDATED, set_signal)
favorites.clear()
player.heos.dispatcher.send(
const.SIGNAL_CONTROLLER_EVENT, const.EVENT_SOURCES_CHANGED)
await event.wait()
source_list = hass.data[DOMAIN][DATA_SOURCE_MANAGER].source_list
assert len(source_list) == 1
state = hass.states.get('media_player.test_player')
assert state.attributes[ATTR_INPUT_SOURCE_LIST] == source_list
async def test_services(hass, config_entry, config, controller):
"""Tests player commands."""
@ -173,6 +196,85 @@ async def test_services(hass, config_entry, config, controller):
player.set_volume.assert_called_once_with(100)
async def test_select_favorite(
hass, config_entry, config, controller, favorites):
"""Tests selecting a music service favorite and state."""
await setup_platform(hass, config_entry, config)
player = controller.players[1]
# Test set music service preset
favorite = favorites[1]
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN, SERVICE_SELECT_SOURCE,
{ATTR_ENTITY_ID: 'media_player.test_player',
ATTR_INPUT_SOURCE: favorite.name}, blocking=True)
player.play_favorite.assert_called_once_with(1)
# Test state is matched by station name
player.now_playing_media.station = favorite.name
player.heos.dispatcher.send(
const.SIGNAL_PLAYER_EVENT, player.player_id,
const.EVENT_PLAYER_STATE_CHANGED)
await hass.async_block_till_done()
state = hass.states.get('media_player.test_player')
assert state.attributes[ATTR_INPUT_SOURCE] == favorite.name
async def test_select_radio_favorite(
hass, config_entry, config, controller, favorites):
"""Tests selecting a radio favorite and state."""
await setup_platform(hass, config_entry, config)
player = controller.players[1]
# Test set radio preset
favorite = favorites[2]
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN, SERVICE_SELECT_SOURCE,
{ATTR_ENTITY_ID: 'media_player.test_player',
ATTR_INPUT_SOURCE: favorite.name}, blocking=True)
player.play_favorite.assert_called_once_with(2)
# Test state is matched by album id
player.now_playing_media.station = "Classical"
player.now_playing_media.album_id = favorite.media_id
player.heos.dispatcher.send(
const.SIGNAL_PLAYER_EVENT, player.player_id,
const.EVENT_PLAYER_STATE_CHANGED)
await hass.async_block_till_done()
state = hass.states.get('media_player.test_player')
assert state.attributes[ATTR_INPUT_SOURCE] == favorite.name
async def test_select_input_source(
hass, config_entry, config, controller, input_sources):
"""Tests selecting input source and state."""
await setup_platform(hass, config_entry, config)
player = controller.players[1]
# Test proper service called
input_source = input_sources[0]
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN, SERVICE_SELECT_SOURCE,
{ATTR_ENTITY_ID: 'media_player.test_player',
ATTR_INPUT_SOURCE: input_source.name}, blocking=True)
player.play_input_source.assert_called_once_with(input_source)
# Test state is matched by media id
player.now_playing_media.source_id = const.MUSIC_SOURCE_AUX_INPUT
player.now_playing_media.media_id = const.INPUT_AUX_IN_1
player.heos.dispatcher.send(
const.SIGNAL_PLAYER_EVENT, player.player_id,
const.EVENT_PLAYER_STATE_CHANGED)
await hass.async_block_till_done()
state = hass.states.get('media_player.test_player')
assert state.attributes[ATTR_INPUT_SOURCE] == input_source.name
async def test_select_input_unknown(
hass, config_entry, config, controller, caplog):
"""Tests selecting an unknown input."""
await setup_platform(hass, config_entry, config)
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN, SERVICE_SELECT_SOURCE,
{ATTR_ENTITY_ID: 'media_player.test_player',
ATTR_INPUT_SOURCE: "Unknown"}, blocking=True)
assert "Unknown source: Unknown" in caplog.text
async def test_unload_config_entry(hass, config_entry, config, controller):
"""Test the player is removed when the config entry is unloaded."""
await setup_platform(hass, config_entry, config)