Add tests to the media_player platform of the Squeezebox integration (#125378)
* Squeezebox media_player platform tests * Fix play-pause test * Squeezebox remove stray reference to deprecated property * More tests for squeezebox * Update tests to fix merge conflict with binary_sensor * Refactor tests to use autospec * Use freeze and snapshot * Update media player entity before adding * Consolidate test fixtures for different platforms * Merge in sensor platform * Use deepcopy * Update tests with suggestions from code review
This commit is contained in:
Raj Laud
GitHub
parent
0af913cc9a
commit
a01036760e
7 changed files with 1125 additions and 135 deletions
|
|
@ -14,6 +14,7 @@ import voluptuous as vol
|
|||
from homeassistant.components import media_source
|
||||
from homeassistant.components.media_player import (
|
||||
ATTR_MEDIA_ENQUEUE,
|
||||
BrowseError,
|
||||
BrowseMedia,
|
||||
MediaPlayerEnqueue,
|
||||
MediaPlayerEntity,
|
||||
|
|
@ -26,6 +27,7 @@ from homeassistant.components.media_player import (
|
|||
from homeassistant.config_entries import SOURCE_INTEGRATION_DISCOVERY
|
||||
from homeassistant.const import ATTR_COMMAND, CONF_HOST, CONF_PORT
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.exceptions import ServiceValidationError
|
||||
from homeassistant.helpers import (
|
||||
config_validation as cv,
|
||||
discovery_flow,
|
||||
|
|
@ -138,7 +140,7 @@ async def async_setup_entry(
|
|||
_LOGGER.debug("Adding new entity: %s", player)
|
||||
entity = SqueezeBoxEntity(player, lms)
|
||||
known_players.append(entity)
|
||||
async_add_entities([entity])
|
||||
async_add_entities([entity], True)
|
||||
|
||||
if players := await lms.async_get_players():
|
||||
for player in players:
|
||||
|
|
@ -248,8 +250,11 @@ class SqueezeBoxEntity(MediaPlayerEntity):
|
|||
"""Return the state of the device."""
|
||||
if not self._player.power:
|
||||
return MediaPlayerState.OFF
|
||||
if self._player.mode:
|
||||
return SQUEEZEBOX_MODE.get(self._player.mode)
|
||||
if self._player.mode and self._player.mode in SQUEEZEBOX_MODE:
|
||||
return SQUEEZEBOX_MODE[self._player.mode]
|
||||
_LOGGER.error(
|
||||
"Received unknown mode %s from player %s", self._player.mode, self.name
|
||||
)
|
||||
return None
|
||||
|
||||
async def async_update(self) -> None:
|
||||
|
|
@ -278,6 +283,7 @@ class SqueezeBoxEntity(MediaPlayerEntity):
|
|||
"""Volume level of the media player (0..1)."""
|
||||
if self._player.volume:
|
||||
return int(float(self._player.volume)) / 100.0
|
||||
|
||||
return None
|
||||
|
||||
@property
|
||||
|
|
@ -322,7 +328,7 @@ class SqueezeBoxEntity(MediaPlayerEntity):
|
|||
@property
|
||||
def media_image_url(self) -> str | None:
|
||||
"""Image url of current playing media."""
|
||||
return str(self._player.image_url)
|
||||
return str(self._player.image_url) if self._player.image_url else None
|
||||
|
||||
@property
|
||||
def media_title(self) -> str | None:
|
||||
|
|
@ -371,11 +377,6 @@ class SqueezeBoxEntity(MediaPlayerEntity):
|
|||
if player in player_ids
|
||||
]
|
||||
|
||||
@property
|
||||
def sync_group(self) -> list[str]:
|
||||
"""List players we are synced with. Deprecated."""
|
||||
return self.group_members
|
||||
|
||||
@property
|
||||
def query_result(self) -> dict | bool:
|
||||
"""Return the result from the call_query service."""
|
||||
|
|
@ -474,7 +475,7 @@ class SqueezeBoxEntity(MediaPlayerEntity):
|
|||
"search_type": MediaType.PLAYLIST,
|
||||
}
|
||||
playlist = await generate_playlist(self._player, payload)
|
||||
except ValueError:
|
||||
except BrowseError:
|
||||
# a list of urls
|
||||
content = json.loads(media_id)
|
||||
playlist = content["urls"]
|
||||
|
|
@ -553,8 +554,8 @@ class SqueezeBoxEntity(MediaPlayerEntity):
|
|||
if other_player_id := player_ids.get(other_player):
|
||||
await self._player.async_sync(other_player_id)
|
||||
else:
|
||||
_LOGGER.debug(
|
||||
"Could not find player_id for %s. Not syncing", other_player
|
||||
raise ServiceValidationError(
|
||||
f"Could not join unknown player {other_player}"
|
||||
)
|
||||
|
||||
async def async_unjoin_player(self) -> None:
|
||||
|
|
|
|||
|
|
@ -1,85 +1 @@
|
|||
"""Tests for the Logitech Squeezebox integration."""
|
||||
|
||||
from homeassistant.components.squeezebox.const import (
|
||||
DOMAIN,
|
||||
STATUS_QUERY_LIBRARYNAME,
|
||||
STATUS_QUERY_MAC,
|
||||
STATUS_QUERY_UUID,
|
||||
STATUS_QUERY_VERSION,
|
||||
STATUS_SENSOR_INFO_TOTAL_ALBUMS,
|
||||
STATUS_SENSOR_INFO_TOTAL_ARTISTS,
|
||||
STATUS_SENSOR_INFO_TOTAL_DURATION,
|
||||
STATUS_SENSOR_INFO_TOTAL_GENRES,
|
||||
STATUS_SENSOR_INFO_TOTAL_SONGS,
|
||||
STATUS_SENSOR_LASTSCAN,
|
||||
STATUS_SENSOR_OTHER_PLAYER_COUNT,
|
||||
STATUS_SENSOR_PLAYER_COUNT,
|
||||
STATUS_SENSOR_RESCAN,
|
||||
)
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
# from homeassistant.setup import async_setup_component
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
FAKE_IP = "42.42.42.42"
|
||||
FAKE_MAC = "deadbeefdead"
|
||||
FAKE_UUID = "deadbeefdeadbeefbeefdeafbeef42"
|
||||
FAKE_PORT = 9000
|
||||
FAKE_VERSION = "42.0"
|
||||
|
||||
FAKE_QUERY_RESPONSE = {
|
||||
STATUS_QUERY_UUID: FAKE_UUID,
|
||||
STATUS_QUERY_MAC: FAKE_MAC,
|
||||
STATUS_QUERY_VERSION: FAKE_VERSION,
|
||||
STATUS_SENSOR_RESCAN: 1,
|
||||
STATUS_SENSOR_LASTSCAN: 0,
|
||||
STATUS_QUERY_LIBRARYNAME: "FakeLib",
|
||||
STATUS_SENSOR_INFO_TOTAL_ALBUMS: 4,
|
||||
STATUS_SENSOR_INFO_TOTAL_ARTISTS: 2,
|
||||
STATUS_SENSOR_INFO_TOTAL_DURATION: 500,
|
||||
STATUS_SENSOR_INFO_TOTAL_GENRES: 1,
|
||||
STATUS_SENSOR_INFO_TOTAL_SONGS: 42,
|
||||
STATUS_SENSOR_PLAYER_COUNT: 10,
|
||||
STATUS_SENSOR_OTHER_PLAYER_COUNT: 0,
|
||||
"players_loop": [
|
||||
{
|
||||
"isplaying": 0,
|
||||
"name": "SqueezeLite-HA-Addon",
|
||||
"seq_no": 0,
|
||||
"modelname": "SqueezeLite-HA-Addon",
|
||||
"playerindex": "status",
|
||||
"model": "squeezelite",
|
||||
"uuid": FAKE_UUID,
|
||||
"canpoweroff": 1,
|
||||
"ip": "192.168.78.86:57700",
|
||||
"displaytype": "none",
|
||||
"playerid": "f9:23:cd:37:c5:ff",
|
||||
"power": 0,
|
||||
"isplayer": 1,
|
||||
"connected": 1,
|
||||
"firmware": "v2.0.0-1488",
|
||||
}
|
||||
],
|
||||
"count": 1,
|
||||
}
|
||||
|
||||
|
||||
async def setup_mocked_integration(hass: HomeAssistant) -> MockConfigEntry:
|
||||
"""Mock ConfigEntry in Home Assistant."""
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
unique_id=FAKE_UUID,
|
||||
data={
|
||||
CONF_HOST: FAKE_IP,
|
||||
CONF_PORT: FAKE_PORT,
|
||||
},
|
||||
)
|
||||
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
return entry
|
||||
|
|
|
|||
|
|
@ -11,17 +11,82 @@ from homeassistant.components.squeezebox.browse_media import (
|
|||
MEDIA_TYPE_TO_SQUEEZEBOX,
|
||||
SQUEEZEBOX_ID_BY_TYPE,
|
||||
)
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT
|
||||
from homeassistant.components.squeezebox.const import (
|
||||
STATUS_QUERY_LIBRARYNAME,
|
||||
STATUS_QUERY_MAC,
|
||||
STATUS_QUERY_UUID,
|
||||
STATUS_QUERY_VERSION,
|
||||
STATUS_SENSOR_INFO_TOTAL_ALBUMS,
|
||||
STATUS_SENSOR_INFO_TOTAL_ARTISTS,
|
||||
STATUS_SENSOR_INFO_TOTAL_DURATION,
|
||||
STATUS_SENSOR_INFO_TOTAL_GENRES,
|
||||
STATUS_SENSOR_INFO_TOTAL_SONGS,
|
||||
STATUS_SENSOR_LASTSCAN,
|
||||
STATUS_SENSOR_OTHER_PLAYER_COUNT,
|
||||
STATUS_SENSOR_PLAYER_COUNT,
|
||||
STATUS_SENSOR_RESCAN,
|
||||
)
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT, Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.device_registry import format_mac
|
||||
|
||||
# from homeassistant.setup import async_setup_component
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
TEST_HOST = "1.2.3.4"
|
||||
TEST_PORT = "9000"
|
||||
TEST_USE_HTTPS = False
|
||||
SERVER_UUID = "12345678-1234-1234-1234-123456789012"
|
||||
TEST_MAC = "aa:bb:cc:dd:ee:ff"
|
||||
SERVER_UUIDS = [
|
||||
"12345678-1234-1234-1234-123456789012",
|
||||
"87654321-4321-4321-4321-210987654321",
|
||||
]
|
||||
TEST_MAC = ["aa:bb:cc:dd:ee:ff", "ff:ee:dd:cc:bb:aa"]
|
||||
TEST_PLAYER_NAME = "Test Player"
|
||||
TEST_SERVER_NAME = "Test Server"
|
||||
FAKE_VALID_ITEM_ID = "1234"
|
||||
FAKE_INVALID_ITEM_ID = "4321"
|
||||
|
||||
FAKE_IP = "42.42.42.42"
|
||||
FAKE_MAC = "deadbeefdead"
|
||||
FAKE_UUID = "deadbeefdeadbeefbeefdeafbeef42"
|
||||
FAKE_PORT = 9000
|
||||
FAKE_VERSION = "42.0"
|
||||
|
||||
FAKE_QUERY_RESPONSE = {
|
||||
STATUS_QUERY_UUID: FAKE_UUID,
|
||||
STATUS_QUERY_MAC: FAKE_MAC,
|
||||
STATUS_QUERY_VERSION: FAKE_VERSION,
|
||||
STATUS_SENSOR_RESCAN: 1,
|
||||
STATUS_SENSOR_LASTSCAN: 0,
|
||||
STATUS_QUERY_LIBRARYNAME: "FakeLib",
|
||||
STATUS_SENSOR_INFO_TOTAL_ALBUMS: 4,
|
||||
STATUS_SENSOR_INFO_TOTAL_ARTISTS: 2,
|
||||
STATUS_SENSOR_INFO_TOTAL_DURATION: 500,
|
||||
STATUS_SENSOR_INFO_TOTAL_GENRES: 1,
|
||||
STATUS_SENSOR_INFO_TOTAL_SONGS: 42,
|
||||
STATUS_SENSOR_PLAYER_COUNT: 10,
|
||||
STATUS_SENSOR_OTHER_PLAYER_COUNT: 0,
|
||||
"players_loop": [
|
||||
{
|
||||
"isplaying": 0,
|
||||
"name": "SqueezeLite-HA-Addon",
|
||||
"seq_no": 0,
|
||||
"modelname": "SqueezeLite-HA-Addon",
|
||||
"playerindex": "status",
|
||||
"model": "squeezelite",
|
||||
"uuid": FAKE_UUID,
|
||||
"canpoweroff": 1,
|
||||
"ip": "192.168.78.86:57700",
|
||||
"displaytype": "none",
|
||||
"playerid": "f9:23:cd:37:c5:ff",
|
||||
"power": 0,
|
||||
"isplayer": 1,
|
||||
"connected": 1,
|
||||
"firmware": "v2.0.0-1488",
|
||||
}
|
||||
],
|
||||
"count": 1,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -38,7 +103,7 @@ def config_entry(hass: HomeAssistant) -> MockConfigEntry:
|
|||
"""Add the squeezebox mock config entry to hass."""
|
||||
config_entry = MockConfigEntry(
|
||||
domain=const.DOMAIN,
|
||||
unique_id=SERVER_UUID,
|
||||
unique_id=SERVER_UUIDS[0],
|
||||
data={
|
||||
CONF_HOST: TEST_HOST,
|
||||
CONF_PORT: TEST_PORT,
|
||||
|
|
@ -69,29 +134,41 @@ async def mock_async_browse(
|
|||
fake_items = [
|
||||
{
|
||||
"title": "Fake Item 1",
|
||||
"id": "1234",
|
||||
"id": FAKE_VALID_ITEM_ID,
|
||||
"hasitems": False,
|
||||
"item_type": child_types[media_type],
|
||||
"artwork_track_id": "b35bb9e9",
|
||||
"url": "file:///var/lib/squeezeboxserver/music/track_1.mp3",
|
||||
},
|
||||
{
|
||||
"title": "Fake Item 2",
|
||||
"id": "12345",
|
||||
"id": FAKE_VALID_ITEM_ID + "_2",
|
||||
"hasitems": media_type == "favorites",
|
||||
"item_type": child_types[media_type],
|
||||
"image_url": "http://lms.internal:9000/html/images/favorites.png",
|
||||
"url": "file:///var/lib/squeezeboxserver/music/track_2.mp3",
|
||||
},
|
||||
{
|
||||
"title": "Fake Item 3",
|
||||
"id": "123456",
|
||||
"id": FAKE_VALID_ITEM_ID + "_3",
|
||||
"hasitems": media_type == "favorites",
|
||||
"album_id": "123456" if media_type == "favorites" else None,
|
||||
"album_id": FAKE_VALID_ITEM_ID if media_type == "favorites" else None,
|
||||
"url": "file:///var/lib/squeezeboxserver/music/track_3.mp3",
|
||||
},
|
||||
]
|
||||
|
||||
if browse_id:
|
||||
search_type, search_id = browse_id
|
||||
if search_id:
|
||||
if search_type == "playlist_id":
|
||||
return (
|
||||
{
|
||||
"title": "Fake Item 1",
|
||||
"items": fake_items,
|
||||
}
|
||||
if search_id == FAKE_VALID_ITEM_ID
|
||||
else None
|
||||
)
|
||||
if search_type in SQUEEZEBOX_ID_BY_TYPE.values():
|
||||
for item in fake_items:
|
||||
if item["id"] == search_id:
|
||||
|
|
@ -115,20 +192,96 @@ async def mock_async_browse(
|
|||
|
||||
|
||||
@pytest.fixture
|
||||
def lms() -> MagicMock:
|
||||
"""Mock a Lyrion Media Server with one mock player attached."""
|
||||
lms = MagicMock()
|
||||
player = MagicMock()
|
||||
player.player_id = TEST_MAC
|
||||
player.name = "Test Player"
|
||||
player.power = False
|
||||
player.async_browse = AsyncMock(side_effect=mock_async_browse)
|
||||
player.async_load_playlist = AsyncMock()
|
||||
player.async_update = AsyncMock()
|
||||
player.generate_image_url_from_track_id = MagicMock(
|
||||
return_value="http://lms.internal:9000/html/images/favorites.png"
|
||||
def player() -> MagicMock:
|
||||
"""Return a mock player."""
|
||||
return mock_pysqueezebox_player()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def player_factory() -> MagicMock:
|
||||
"""Return a factory for creating mock players."""
|
||||
return mock_pysqueezebox_player
|
||||
|
||||
|
||||
def mock_pysqueezebox_player(uuid: str) -> MagicMock:
|
||||
"""Mock a Lyrion Media Server player."""
|
||||
with patch(
|
||||
"homeassistant.components.squeezebox.media_player.Player", autospec=True
|
||||
) as mock_player:
|
||||
mock_player.async_browse = AsyncMock(side_effect=mock_async_browse)
|
||||
mock_player.generate_image_url_from_track_id = MagicMock(
|
||||
return_value="http://lms.internal:9000/html/images/favorites.png"
|
||||
)
|
||||
mock_player.name = TEST_PLAYER_NAME
|
||||
mock_player.player_id = uuid
|
||||
mock_player.mode = "stop"
|
||||
mock_player.playlist = None
|
||||
mock_player.album = None
|
||||
mock_player.artist = None
|
||||
mock_player.remote_title = None
|
||||
mock_player.title = None
|
||||
mock_player.image_url = None
|
||||
|
||||
return mock_player
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def lms_factory(player_factory: MagicMock) -> MagicMock:
|
||||
"""Return a factory for creating mock Lyrion Media Servers with arbitrary number of players."""
|
||||
return lambda player_count, uuid: mock_pysqueezebox_server(
|
||||
player_factory, player_count, uuid
|
||||
)
|
||||
lms.async_get_players = AsyncMock(return_value=[player])
|
||||
lms.async_query = AsyncMock(return_value={"uuid": format_mac(TEST_MAC)})
|
||||
lms.async_status = AsyncMock(return_value={"uuid": format_mac(TEST_MAC)})
|
||||
return lms
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def lms(player_factory: MagicMock) -> MagicMock:
|
||||
"""Mock a Lyrion Media Server with one mock player attached."""
|
||||
return mock_pysqueezebox_server(player_factory, 1, uuid=TEST_MAC[0])
|
||||
|
||||
|
||||
def mock_pysqueezebox_server(
|
||||
player_factory: MagicMock, player_count: int, uuid: str
|
||||
) -> MagicMock:
|
||||
"""Create a mock Lyrion Media Server with the given number of mock players attached."""
|
||||
with patch("homeassistant.components.squeezebox.Server", autospec=True) as mock_lms:
|
||||
players = [player_factory(TEST_MAC[index]) for index in range(player_count)]
|
||||
mock_lms.async_get_players = AsyncMock(return_value=players)
|
||||
|
||||
mock_lms.uuid = uuid
|
||||
mock_lms.name = TEST_SERVER_NAME
|
||||
mock_lms.async_query = AsyncMock(return_value={"uuid": format_mac(uuid)})
|
||||
mock_lms.async_status = AsyncMock(return_value={"uuid": format_mac(uuid)})
|
||||
return mock_lms
|
||||
|
||||
|
||||
async def configure_squeezebox_media_player_platform(
|
||||
hass: HomeAssistant,
|
||||
config_entry: MockConfigEntry,
|
||||
lms: MagicMock,
|
||||
) -> None:
|
||||
"""Configure a squeezebox config entry with appropriate mocks for media_player."""
|
||||
with (
|
||||
patch("homeassistant.components.squeezebox.PLATFORMS", [Platform.MEDIA_PLAYER]),
|
||||
patch("homeassistant.components.squeezebox.Server", return_value=lms),
|
||||
):
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def configured_player(
|
||||
hass: HomeAssistant, config_entry: MockConfigEntry, lms: MagicMock
|
||||
) -> MagicMock:
|
||||
"""Fixture mocking calls to pysqueezebox Player from a configured squeezebox."""
|
||||
await configure_squeezebox_media_player_platform(hass, config_entry, lms)
|
||||
return (await lms.async_get_players())[0]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def configured_players(
|
||||
hass: HomeAssistant, config_entry: MockConfigEntry, lms_factory: MagicMock
|
||||
) -> list[MagicMock]:
|
||||
"""Fixture mocking calls to two pysqueezebox Players from a configured squeezebox."""
|
||||
lms = lms_factory(2, uuid=SERVER_UUIDS[0])
|
||||
await configure_squeezebox_media_player_platform(hass, config_entry, lms)
|
||||
return await lms.async_get_players()
|
||||
|
|
|
|||
99
tests/components/squeezebox/snapshots/test_media_player.ambr
Normal file
99
tests/components/squeezebox/snapshots/test_media_player.ambr
Normal file
|
|
@ -0,0 +1,99 @@
|
|||
# serializer version: 1
|
||||
# name: test_device_registry
|
||||
DeviceRegistryEntrySnapshot({
|
||||
'area_id': None,
|
||||
'config_entries': <ANY>,
|
||||
'configuration_url': None,
|
||||
'connections': set({
|
||||
tuple(
|
||||
'mac',
|
||||
'aa:bb:cc:dd:ee:ff',
|
||||
),
|
||||
}),
|
||||
'disabled_by': None,
|
||||
'entry_type': <DeviceEntryType.SERVICE: 'service'>,
|
||||
'hw_version': None,
|
||||
'id': <ANY>,
|
||||
'identifiers': set({
|
||||
tuple(
|
||||
'squeezebox',
|
||||
'aa:bb:cc:dd:ee:ff',
|
||||
),
|
||||
}),
|
||||
'is_new': False,
|
||||
'labels': set({
|
||||
}),
|
||||
'manufacturer': 'https://lyrion.org/',
|
||||
'model': 'Lyrion Music Server',
|
||||
'model_id': None,
|
||||
'name': 'Test Player',
|
||||
'name_by_user': None,
|
||||
'primary_config_entry': <ANY>,
|
||||
'serial_number': None,
|
||||
'suggested_area': None,
|
||||
'sw_version': None,
|
||||
'via_device_id': <ANY>,
|
||||
})
|
||||
# ---
|
||||
# name: test_entity_registry[media_player.test_player-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': set({
|
||||
}),
|
||||
'area_id': None,
|
||||
'capabilities': dict({
|
||||
}),
|
||||
'config_entry_id': <ANY>,
|
||||
'device_class': None,
|
||||
'device_id': <ANY>,
|
||||
'disabled_by': None,
|
||||
'domain': 'media_player',
|
||||
'entity_category': None,
|
||||
'entity_id': 'media_player.test_player',
|
||||
'has_entity_name': True,
|
||||
'hidden_by': None,
|
||||
'icon': None,
|
||||
'id': <ANY>,
|
||||
'labels': set({
|
||||
}),
|
||||
'name': None,
|
||||
'options': dict({
|
||||
}),
|
||||
'original_device_class': None,
|
||||
'original_icon': None,
|
||||
'original_name': None,
|
||||
'platform': 'squeezebox',
|
||||
'previous_unique_id': None,
|
||||
'supported_features': <MediaPlayerEntityFeature: 3077055>,
|
||||
'translation_key': None,
|
||||
'unique_id': 'aa:bb:cc:dd:ee:ff',
|
||||
'unit_of_measurement': None,
|
||||
})
|
||||
# ---
|
||||
# name: test_entity_registry[media_player.test_player-state]
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
'friendly_name': 'Test Player',
|
||||
'group_members': list([
|
||||
]),
|
||||
'is_volume_muted': True,
|
||||
'media_album_name': 'None',
|
||||
'media_artist': 'None',
|
||||
'media_channel': 'None',
|
||||
'media_duration': 1,
|
||||
'media_position': 1,
|
||||
'media_title': 'None',
|
||||
'query_result': dict({
|
||||
}),
|
||||
'repeat': <RepeatMode.OFF: 'off'>,
|
||||
'shuffle': False,
|
||||
'supported_features': <MediaPlayerEntityFeature: 3077055>,
|
||||
'volume_level': 0.01,
|
||||
}),
|
||||
'context': <ANY>,
|
||||
'entity_id': 'media_player.test_player',
|
||||
'last_changed': <ANY>,
|
||||
'last_reported': <ANY>,
|
||||
'last_updated': <ANY>,
|
||||
'state': 'idle',
|
||||
})
|
||||
# ---
|
||||
|
|
@ -1,22 +1,21 @@
|
|||
"""Test squeezebox binary sensors."""
|
||||
|
||||
import copy
|
||||
from copy import deepcopy
|
||||
from unittest.mock import patch
|
||||
|
||||
from homeassistant.const import Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from . import FAKE_QUERY_RESPONSE, setup_mocked_integration
|
||||
from .conftest import FAKE_QUERY_RESPONSE
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_binary_sensor(
|
||||
hass: HomeAssistant,
|
||||
entity_registry: er.EntityRegistry,
|
||||
config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test binary sensor states and attributes."""
|
||||
|
||||
# Setup component
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.squeezebox.PLATFORMS",
|
||||
|
|
@ -24,11 +23,13 @@ async def test_binary_sensor(
|
|||
),
|
||||
patch(
|
||||
"homeassistant.components.squeezebox.Server.async_query",
|
||||
return_value=copy.deepcopy(FAKE_QUERY_RESPONSE),
|
||||
return_value=deepcopy(FAKE_QUERY_RESPONSE),
|
||||
),
|
||||
):
|
||||
await setup_mocked_integration(hass)
|
||||
state = hass.states.get("binary_sensor.fakelib_library_rescan")
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
state = hass.states.get("binary_sensor.fakelib_needs_restart")
|
||||
|
||||
assert state is not None
|
||||
assert state.state == "on"
|
||||
assert state.state == "off"
|
||||
|
|
|
|||
815
tests/components/squeezebox/test_media_player.py
Normal file
815
tests/components/squeezebox/test_media_player.py
Normal file
|
|
@ -0,0 +1,815 @@
|
|||
"""Tests for the squeezebox media player component."""
|
||||
|
||||
from datetime import timedelta
|
||||
import json
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from freezegun.api import FrozenDateTimeFactory
|
||||
import pytest
|
||||
from syrupy import SnapshotAssertion
|
||||
|
||||
from homeassistant.components.media_player import (
|
||||
ATTR_GROUP_MEMBERS,
|
||||
ATTR_MEDIA_CONTENT_ID,
|
||||
ATTR_MEDIA_CONTENT_TYPE,
|
||||
ATTR_MEDIA_ENQUEUE,
|
||||
ATTR_MEDIA_POSITION,
|
||||
ATTR_MEDIA_POSITION_UPDATED_AT,
|
||||
ATTR_MEDIA_REPEAT,
|
||||
ATTR_MEDIA_SEEK_POSITION,
|
||||
ATTR_MEDIA_SHUFFLE,
|
||||
ATTR_MEDIA_VOLUME_LEVEL,
|
||||
ATTR_MEDIA_VOLUME_MUTED,
|
||||
DOMAIN as MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_CLEAR_PLAYLIST,
|
||||
SERVICE_JOIN,
|
||||
SERVICE_PLAY_MEDIA,
|
||||
SERVICE_UNJOIN,
|
||||
MediaPlayerEnqueue,
|
||||
MediaPlayerState,
|
||||
MediaType,
|
||||
RepeatMode,
|
||||
)
|
||||
from homeassistant.components.squeezebox.const import DOMAIN, SENSOR_UPDATE_INTERVAL
|
||||
from homeassistant.components.squeezebox.media_player import (
|
||||
ATTR_PARAMETERS,
|
||||
DISCOVERY_INTERVAL,
|
||||
SERVICE_CALL_METHOD,
|
||||
SERVICE_CALL_QUERY,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
ATTR_COMMAND,
|
||||
ATTR_ENTITY_ID,
|
||||
SERVICE_MEDIA_NEXT_TRACK,
|
||||
SERVICE_MEDIA_PAUSE,
|
||||
SERVICE_MEDIA_PLAY,
|
||||
SERVICE_MEDIA_PLAY_PAUSE,
|
||||
SERVICE_MEDIA_PREVIOUS_TRACK,
|
||||
SERVICE_MEDIA_SEEK,
|
||||
SERVICE_MEDIA_STOP,
|
||||
SERVICE_REPEAT_SET,
|
||||
SERVICE_SHUFFLE_SET,
|
||||
SERVICE_TURN_OFF,
|
||||
SERVICE_TURN_ON,
|
||||
SERVICE_VOLUME_DOWN,
|
||||
SERVICE_VOLUME_MUTE,
|
||||
SERVICE_VOLUME_SET,
|
||||
SERVICE_VOLUME_UP,
|
||||
STATE_UNAVAILABLE,
|
||||
STATE_UNKNOWN,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ServiceValidationError
|
||||
from homeassistant.helpers.device_registry import DeviceRegistry
|
||||
from homeassistant.helpers.entity_registry import EntityRegistry
|
||||
from homeassistant.util.dt import utcnow
|
||||
|
||||
from .conftest import FAKE_VALID_ITEM_ID, TEST_MAC
|
||||
|
||||
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
|
||||
|
||||
|
||||
async def test_device_registry(
|
||||
hass: HomeAssistant,
|
||||
device_registry: DeviceRegistry,
|
||||
configured_player: MagicMock,
|
||||
snapshot: SnapshotAssertion,
|
||||
) -> None:
|
||||
"""Test squeezebox device registered in the device registry."""
|
||||
reg_device = device_registry.async_get_device(identifiers={(DOMAIN, TEST_MAC[0])})
|
||||
assert reg_device is not None
|
||||
assert reg_device == snapshot
|
||||
|
||||
|
||||
async def test_entity_registry(
|
||||
hass: HomeAssistant,
|
||||
entity_registry: EntityRegistry,
|
||||
configured_player: MagicMock,
|
||||
snapshot: SnapshotAssertion,
|
||||
config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test squeezebox media_player entity registered in the entity registry."""
|
||||
await snapshot_platform(hass, entity_registry, snapshot, config_entry.entry_id)
|
||||
|
||||
|
||||
async def test_squeezebox_player_rediscovery(
|
||||
hass: HomeAssistant, configured_player: MagicMock, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
"""Test rediscovery of a squeezebox player."""
|
||||
|
||||
assert hass.states.get("media_player.test_player").state == MediaPlayerState.IDLE
|
||||
|
||||
# Make the player appear unavailable
|
||||
configured_player.connected = False
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_ENTITY_ID: "media_player.test_player"},
|
||||
blocking=True,
|
||||
)
|
||||
assert hass.states.get("media_player.test_player").state == STATE_UNAVAILABLE
|
||||
|
||||
# Make the player available again
|
||||
configured_player.connected = True
|
||||
freezer.tick(timedelta(seconds=DISCOVERY_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get("media_player.test_player").state == MediaPlayerState.IDLE
|
||||
|
||||
|
||||
async def test_squeezebox_turn_on(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test turn on service call."""
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_ENTITY_ID: "media_player.test_player"},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_set_power.assert_called_once_with(True)
|
||||
|
||||
|
||||
async def test_squeezebox_turn_off(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test turn off service call."""
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_TURN_OFF,
|
||||
{ATTR_ENTITY_ID: "media_player.test_player"},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_set_power.assert_called_once_with(False)
|
||||
|
||||
|
||||
async def test_squeezebox_state(
|
||||
hass: HomeAssistant, configured_player: MagicMock, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
"""Test determining the MediaPlayerState."""
|
||||
|
||||
configured_player.power = True
|
||||
configured_player.mode = "stop"
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get("media_player.test_player").state == MediaPlayerState.IDLE
|
||||
|
||||
configured_player.mode = "play"
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get("media_player.test_player").state == MediaPlayerState.PLAYING
|
||||
|
||||
configured_player.mode = "pause"
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get("media_player.test_player").state == MediaPlayerState.PAUSED
|
||||
|
||||
configured_player.power = False
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get("media_player.test_player").state == MediaPlayerState.OFF
|
||||
|
||||
|
||||
async def test_squeezebox_volume_up(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test volume up service call."""
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_VOLUME_UP,
|
||||
{ATTR_ENTITY_ID: "media_player.test_player"},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_set_volume.assert_called_once_with("+5")
|
||||
|
||||
|
||||
async def test_squeezebox_volume_down(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test volume down service call."""
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_VOLUME_DOWN,
|
||||
{ATTR_ENTITY_ID: "media_player.test_player"},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_set_volume.assert_called_once_with("-5")
|
||||
|
||||
|
||||
async def test_squeezebox_volume_set(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test volume set service call."""
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_VOLUME_SET,
|
||||
{ATTR_ENTITY_ID: "media_player.test_player", ATTR_MEDIA_VOLUME_LEVEL: 0.5},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_set_volume.assert_called_once_with("50")
|
||||
|
||||
|
||||
async def test_squeezebox_volume_property(
|
||||
hass: HomeAssistant, configured_player: MagicMock, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
"""Test volume property."""
|
||||
|
||||
configured_player.volume = 50
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert (
|
||||
hass.states.get("media_player.test_player").attributes[ATTR_MEDIA_VOLUME_LEVEL]
|
||||
== 0.5
|
||||
)
|
||||
|
||||
configured_player.volume = None
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert (
|
||||
ATTR_MEDIA_VOLUME_LEVEL
|
||||
not in hass.states.get("media_player.test_player").attributes
|
||||
)
|
||||
|
||||
|
||||
async def test_squeezebox_mute(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test mute service call."""
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_VOLUME_MUTE,
|
||||
{ATTR_ENTITY_ID: "media_player.test_player", ATTR_MEDIA_VOLUME_MUTED: True},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_set_muting.assert_called_once_with(True)
|
||||
|
||||
|
||||
async def test_squeezebox_unmute(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test unmute service call."""
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_VOLUME_MUTE,
|
||||
{ATTR_ENTITY_ID: "media_player.test_player", ATTR_MEDIA_VOLUME_MUTED: False},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_set_muting.assert_called_once_with(False)
|
||||
|
||||
|
||||
async def test_squeezebox_mute_property(
|
||||
hass: HomeAssistant, configured_player: MagicMock, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
"""Test the mute property."""
|
||||
|
||||
configured_player.muting = True
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert (
|
||||
hass.states.get("media_player.test_player").attributes[ATTR_MEDIA_VOLUME_MUTED]
|
||||
is True
|
||||
)
|
||||
|
||||
configured_player.muting = False
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert (
|
||||
hass.states.get("media_player.test_player").attributes[ATTR_MEDIA_VOLUME_MUTED]
|
||||
is False
|
||||
)
|
||||
|
||||
|
||||
async def test_squeezebox_repeat_mode(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test set repeat mode service call."""
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_REPEAT_SET,
|
||||
{
|
||||
ATTR_ENTITY_ID: "media_player.test_player",
|
||||
ATTR_MEDIA_REPEAT: RepeatMode.ALL,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_set_repeat.assert_called_once_with("playlist")
|
||||
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_REPEAT_SET,
|
||||
{
|
||||
ATTR_ENTITY_ID: "media_player.test_player",
|
||||
ATTR_MEDIA_REPEAT: RepeatMode.ONE,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_set_repeat.assert_called_with("song")
|
||||
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_REPEAT_SET,
|
||||
{
|
||||
ATTR_ENTITY_ID: "media_player.test_player",
|
||||
ATTR_MEDIA_REPEAT: RepeatMode.OFF,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_set_repeat.assert_called_with("none")
|
||||
|
||||
|
||||
async def test_squeezebox_repeat_mode_property(
|
||||
hass: HomeAssistant, configured_player: MagicMock, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
"""Test the repeat mode property."""
|
||||
configured_player.repeat = "playlist"
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert (
|
||||
hass.states.get("media_player.test_player").attributes[ATTR_MEDIA_REPEAT]
|
||||
== RepeatMode.ALL
|
||||
)
|
||||
|
||||
configured_player.repeat = "song"
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert (
|
||||
hass.states.get("media_player.test_player").attributes[ATTR_MEDIA_REPEAT]
|
||||
== RepeatMode.ONE
|
||||
)
|
||||
|
||||
configured_player.repeat = "none"
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert (
|
||||
hass.states.get("media_player.test_player").attributes[ATTR_MEDIA_REPEAT]
|
||||
== RepeatMode.OFF
|
||||
)
|
||||
|
||||
|
||||
async def test_squeezebox_shuffle(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test set shuffle service call."""
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_SHUFFLE_SET,
|
||||
{
|
||||
ATTR_ENTITY_ID: "media_player.test_player",
|
||||
ATTR_MEDIA_SHUFFLE: True,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_set_shuffle.assert_called_once_with("song")
|
||||
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_SHUFFLE_SET,
|
||||
{
|
||||
ATTR_ENTITY_ID: "media_player.test_player",
|
||||
ATTR_MEDIA_SHUFFLE: False,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_set_shuffle.assert_called_with("none")
|
||||
assert (
|
||||
hass.states.get("media_player.test_player").attributes[ATTR_MEDIA_SHUFFLE]
|
||||
is False
|
||||
)
|
||||
|
||||
|
||||
async def test_squeezebox_shuffle_property(
|
||||
hass: HomeAssistant, configured_player: MagicMock, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
"""Test the shuffle property."""
|
||||
|
||||
configured_player.shuffle = "song"
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert (
|
||||
hass.states.get("media_player.test_player").attributes[ATTR_MEDIA_SHUFFLE]
|
||||
is True
|
||||
)
|
||||
|
||||
configured_player.shuffle = "none"
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert (
|
||||
hass.states.get("media_player.test_player").attributes[ATTR_MEDIA_SHUFFLE]
|
||||
is False
|
||||
)
|
||||
|
||||
|
||||
async def test_squeezebox_play(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test play service call."""
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_MEDIA_PLAY,
|
||||
{ATTR_ENTITY_ID: "media_player.test_player"},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_play.assert_called_once()
|
||||
|
||||
|
||||
async def test_squeezebox_play_pause(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test play/pause service call."""
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_MEDIA_PLAY_PAUSE,
|
||||
{ATTR_ENTITY_ID: "media_player.test_player"},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_toggle_pause.assert_called_once()
|
||||
|
||||
|
||||
async def test_squeezebox_pause(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test pause service call."""
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_MEDIA_PAUSE,
|
||||
{ATTR_ENTITY_ID: "media_player.test_player"},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_pause.assert_called_once()
|
||||
|
||||
|
||||
async def test_squeezebox_seek(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test seek service call."""
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_PLAY_MEDIA,
|
||||
{
|
||||
ATTR_ENTITY_ID: "media_player.test_player",
|
||||
ATTR_MEDIA_CONTENT_ID: FAKE_VALID_ITEM_ID,
|
||||
ATTR_MEDIA_CONTENT_TYPE: MediaType.MUSIC,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_MEDIA_SEEK,
|
||||
{
|
||||
ATTR_ENTITY_ID: "media_player.test_player",
|
||||
ATTR_MEDIA_SEEK_POSITION: 100,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_time.assert_called_once_with(100)
|
||||
|
||||
|
||||
async def test_squeezebox_stop(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test stop service call."""
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_MEDIA_STOP,
|
||||
{ATTR_ENTITY_ID: "media_player.test_player"},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_stop.assert_called_once()
|
||||
|
||||
|
||||
async def test_squeezebox_load_playlist(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test load a playlist."""
|
||||
# load a playlist by number
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_PLAY_MEDIA,
|
||||
{
|
||||
ATTR_ENTITY_ID: "media_player.test_player",
|
||||
ATTR_MEDIA_CONTENT_ID: FAKE_VALID_ITEM_ID,
|
||||
ATTR_MEDIA_CONTENT_TYPE: MediaType.PLAYLIST,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
assert configured_player.async_load_playlist.call_count == 1
|
||||
|
||||
# load a list of urls
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_PLAY_MEDIA,
|
||||
{
|
||||
ATTR_ENTITY_ID: "media_player.test_player",
|
||||
ATTR_MEDIA_CONTENT_ID: json.dumps(
|
||||
{
|
||||
"urls": [
|
||||
{"url": FAKE_VALID_ITEM_ID},
|
||||
{"url": FAKE_VALID_ITEM_ID + "_2"},
|
||||
],
|
||||
"index": "0",
|
||||
}
|
||||
),
|
||||
ATTR_MEDIA_CONTENT_TYPE: MediaType.PLAYLIST,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
assert configured_player.async_load_playlist.call_count == 2
|
||||
|
||||
# clear the playlist
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_CLEAR_PLAYLIST,
|
||||
{ATTR_ENTITY_ID: "media_player.test_player"},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_clear_playlist.assert_called_once()
|
||||
|
||||
|
||||
async def test_squeezebox_enqueue(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test the various enqueue service calls."""
|
||||
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_PLAY_MEDIA,
|
||||
{
|
||||
ATTR_ENTITY_ID: "media_player.test_player",
|
||||
ATTR_MEDIA_CONTENT_ID: FAKE_VALID_ITEM_ID,
|
||||
ATTR_MEDIA_CONTENT_TYPE: MediaType.MUSIC,
|
||||
ATTR_MEDIA_ENQUEUE: MediaPlayerEnqueue.ADD,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_load_url.assert_called_once_with(FAKE_VALID_ITEM_ID, "add")
|
||||
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_PLAY_MEDIA,
|
||||
{
|
||||
ATTR_ENTITY_ID: "media_player.test_player",
|
||||
ATTR_MEDIA_CONTENT_ID: FAKE_VALID_ITEM_ID,
|
||||
ATTR_MEDIA_CONTENT_TYPE: MediaType.MUSIC,
|
||||
ATTR_MEDIA_ENQUEUE: MediaPlayerEnqueue.NEXT,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_load_url.assert_called_with(FAKE_VALID_ITEM_ID, "insert")
|
||||
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_PLAY_MEDIA,
|
||||
{
|
||||
ATTR_ENTITY_ID: "media_player.test_player",
|
||||
ATTR_MEDIA_CONTENT_ID: FAKE_VALID_ITEM_ID,
|
||||
ATTR_MEDIA_CONTENT_TYPE: MediaType.MUSIC,
|
||||
ATTR_MEDIA_ENQUEUE: MediaPlayerEnqueue.PLAY,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_load_url.assert_called_with(FAKE_VALID_ITEM_ID, "play_now")
|
||||
|
||||
|
||||
async def test_squeezebox_skip_tracks(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test track skipping service calls."""
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_PLAY_MEDIA,
|
||||
{
|
||||
ATTR_ENTITY_ID: "media_player.test_player",
|
||||
ATTR_MEDIA_CONTENT_ID: FAKE_VALID_ITEM_ID,
|
||||
ATTR_MEDIA_CONTENT_TYPE: MediaType.PLAYLIST,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_MEDIA_NEXT_TRACK,
|
||||
{ATTR_ENTITY_ID: "media_player.test_player"},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_index.assert_called_once_with("+1")
|
||||
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_MEDIA_PREVIOUS_TRACK,
|
||||
{ATTR_ENTITY_ID: "media_player.test_player"},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_index.assert_called_with("-1")
|
||||
|
||||
|
||||
async def test_squeezebox_call_query(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test query service call."""
|
||||
await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_CALL_QUERY,
|
||||
{
|
||||
ATTR_ENTITY_ID: "media_player.test_player",
|
||||
ATTR_COMMAND: "test_command",
|
||||
ATTR_PARAMETERS: ["param1", "param2"],
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_query.assert_called_once_with(
|
||||
"test_command", "param1", "param2"
|
||||
)
|
||||
|
||||
|
||||
async def test_squeezebox_call_method(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test method call service call."""
|
||||
await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_CALL_METHOD,
|
||||
{
|
||||
ATTR_ENTITY_ID: "media_player.test_player",
|
||||
ATTR_COMMAND: "test_command",
|
||||
ATTR_PARAMETERS: ["param1", "param2"],
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_query.assert_called_once_with(
|
||||
"test_command", "param1", "param2"
|
||||
)
|
||||
|
||||
|
||||
async def test_squeezebox_invalid_state(
|
||||
hass: HomeAssistant, configured_player: MagicMock, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
"""Test handling an unexpected state from pysqueezebox."""
|
||||
configured_player.mode = "invalid"
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get("media_player.test_player").state == STATE_UNKNOWN
|
||||
|
||||
|
||||
async def test_squeezebox_server_discovery(
|
||||
hass: HomeAssistant,
|
||||
lms: MagicMock,
|
||||
lms_factory: MagicMock,
|
||||
config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test discovery of a squeezebox server."""
|
||||
|
||||
async def mock_async_discover(callback):
|
||||
"""Mock the async_discover function of pysqueezebox."""
|
||||
return callback(lms_factory(2))
|
||||
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.squeezebox.Server",
|
||||
return_value=lms,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.squeezebox.media_player.async_discover",
|
||||
mock_async_discover,
|
||||
),
|
||||
):
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
# how do we check that a config flow started?
|
||||
|
||||
|
||||
async def test_squeezebox_join(hass: HomeAssistant, configured_players: list) -> None:
|
||||
"""Test joining a squeezebox player."""
|
||||
|
||||
# join a valid player
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_JOIN,
|
||||
{
|
||||
ATTR_ENTITY_ID: "media_player.test_player",
|
||||
ATTR_GROUP_MEMBERS: ["media_player.test_player_2"],
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
configured_players[0].async_sync.assert_called_once_with(
|
||||
configured_players[1].player_id
|
||||
)
|
||||
|
||||
# try to join an invalid player
|
||||
with pytest.raises(ServiceValidationError):
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_JOIN,
|
||||
{
|
||||
ATTR_ENTITY_ID: "media_player.test_player",
|
||||
ATTR_GROUP_MEMBERS: ["media_player.invalid"],
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
|
||||
async def test_squeezebox_unjoin(
|
||||
hass: HomeAssistant, configured_player: MagicMock
|
||||
) -> None:
|
||||
"""Test unjoining a squeezebox player."""
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_UNJOIN,
|
||||
{ATTR_ENTITY_ID: "media_player.test_player"},
|
||||
blocking=True,
|
||||
)
|
||||
configured_player.async_unsync.assert_called_once()
|
||||
|
||||
|
||||
async def test_squeezebox_media_content_properties(
|
||||
hass: HomeAssistant,
|
||||
configured_player: MagicMock,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Test media_content_id and media_content_type properties."""
|
||||
playlist_urls = [
|
||||
{"url": "test_title"},
|
||||
{"url": "test_title_2"},
|
||||
]
|
||||
configured_player.current_index = 0
|
||||
configured_player.playlist = playlist_urls
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get("media_player.test_player").attributes[
|
||||
ATTR_MEDIA_CONTENT_ID
|
||||
] == json.dumps({"index": 0, "urls": playlist_urls})
|
||||
assert (
|
||||
hass.states.get("media_player.test_player").attributes[ATTR_MEDIA_CONTENT_TYPE]
|
||||
== MediaType.PLAYLIST
|
||||
)
|
||||
|
||||
configured_player.url = "test_url"
|
||||
configured_player.playlist = [{"url": "test_url"}]
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert (
|
||||
hass.states.get("media_player.test_player").attributes[ATTR_MEDIA_CONTENT_ID]
|
||||
== "test_url"
|
||||
)
|
||||
assert (
|
||||
hass.states.get("media_player.test_player").attributes[ATTR_MEDIA_CONTENT_TYPE]
|
||||
== MediaType.MUSIC
|
||||
)
|
||||
|
||||
configured_player.playlist = None
|
||||
configured_player.url = None
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert (
|
||||
ATTR_MEDIA_CONTENT_ID
|
||||
not in hass.states.get("media_player.test_player").attributes
|
||||
)
|
||||
assert (
|
||||
ATTR_MEDIA_CONTENT_TYPE
|
||||
not in hass.states.get("media_player.test_player").attributes
|
||||
)
|
||||
|
||||
|
||||
async def test_squeezebox_media_position_property(
|
||||
hass: HomeAssistant, configured_player: MagicMock, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
"""Test media_position property."""
|
||||
configured_player.time = 100
|
||||
configured_player.async_update = AsyncMock(
|
||||
side_effect=lambda: setattr(configured_player, "time", 105)
|
||||
)
|
||||
last_update = utcnow()
|
||||
freezer.tick(timedelta(seconds=SENSOR_UPDATE_INTERVAL))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert (
|
||||
hass.states.get("media_player.test_player").attributes[ATTR_MEDIA_POSITION]
|
||||
== 105
|
||||
)
|
||||
assert (
|
||||
(
|
||||
hass.states.get("media_player.test_player").attributes[
|
||||
ATTR_MEDIA_POSITION_UPDATED_AT
|
||||
]
|
||||
)
|
||||
> last_update
|
||||
)
|
||||
|
|
@ -1,15 +1,18 @@
|
|||
"""Test squeezebox sensors."""
|
||||
|
||||
from copy import deepcopy
|
||||
from unittest.mock import patch
|
||||
|
||||
from homeassistant.const import Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from . import FAKE_QUERY_RESPONSE, setup_mocked_integration
|
||||
from .conftest import FAKE_QUERY_RESPONSE
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_sensor(hass: HomeAssistant) -> None:
|
||||
"""Test binary sensor states and attributes."""
|
||||
async def test_sensor(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
|
||||
"""Test sensor states and attributes."""
|
||||
|
||||
# Setup component
|
||||
with (
|
||||
|
|
@ -19,10 +22,12 @@ async def test_sensor(hass: HomeAssistant) -> None:
|
|||
),
|
||||
patch(
|
||||
"homeassistant.components.squeezebox.Server.async_query",
|
||||
return_value=FAKE_QUERY_RESPONSE,
|
||||
return_value=deepcopy(FAKE_QUERY_RESPONSE),
|
||||
),
|
||||
):
|
||||
await setup_mocked_integration(hass)
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
state = hass.states.get("sensor.fakelib_player_count")
|
||||
|
||||
assert state is not None
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue