jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 01:08:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Allow advanced Plex play_media search options (#56226)

Browse Source
This commit is contained in:
jjlawren 2021-10-24 16:22:16 -05:00 committed by GitHub
parent 6c01ed8d97
commit 0600a21e02
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 492 additions and 472 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
homeassistant/components/plex/errors.py
View File

@ -16,7 +16,3 @@ class ServerNotSpecified(PlexException):
class ShouldUpdateConfigEntry(PlexException):
"""Config entry data is out of date and should be updated."""
class MediaNotFound(PlexException):
"""Media lookup failed for a given search query."""

15
homeassistant/components/plex/media_player.py
View File

@ -468,10 +468,10 @@ class PlexMediaPlayer(MediaPlayerEntity):
def play_media(self, media_type, media_id, **kwargs):
"""Play a piece of media."""
if not (self.device and "playback" in self._device_protocol_capabilities):
_LOGGER.debug(
"Client is not currently accepting playback controls: %s", self.name
raise HomeAssistantError(
f"Client is not currently accepting playback controls: {self.name}"
)
return
if not self.plex_server.has_token:
_LOGGER.warning(
"Plex integration configured without a token, playback may fail"
@ -495,16 +495,17 @@ class PlexMediaPlayer(MediaPlayerEntity):
media = self.plex_server.lookup_media(media_type, **src)
if media is None:
_LOGGER.error("Media could not be found: %s", media_id)
return
raise HomeAssistantError(f"Media could not be found: {media_id}")
_LOGGER.debug("Attempting to play %s on %s", media, self.name)
playqueue = self.plex_server.create_playqueue(media, shuffle=shuffle)
try:
self.device.playMedia(playqueue)
except requests.exceptions.ConnectTimeout:
_LOGGER.error("Timed out playing on %s", self.name)
except requests.exceptions.ConnectTimeout as exc:
raise HomeAssistantError(
f"Request failed when playing on {self.name}"
) from exc
@property
def extra_state_attributes(self):

166
homeassistant/components/plex/media_search.py
View File

@ -3,114 +3,82 @@ import logging
from plexapi.exceptions import BadRequest, NotFound
from .errors import MediaNotFound
LEGACY_PARAM_MAPPING = {
"show_name": "show.title",
"season_number": "season.index",
"episode_name": "episode.title",
"episode_number": "episode.index",
"artist_name": "artist.title",
"album_name": "album.title",
"track_name": "track.title",
"track_number": "track.index",
"video_name": "movie.title",
}
PREFERRED_LIBTYPE_ORDER = (
"episode",
"season",
"show",
"track",
"album",
"artist",
)
_LOGGER = logging.getLogger(__name__)
def lookup_movie(library_section, **kwargs):
"""Find a specific movie and return a Plex media object."""
def search_media(media_type, library_section, allow_multiple=False, **kwargs):
"""Search for specified Plex media in the provided library section.
Returns a single media item or None.
If `allow_multiple` is `True`, return a list of matching items.
"""
search_query = {}
libtype = kwargs.pop("libtype", None)
# Preserve legacy service parameters
for legacy_key, key in LEGACY_PARAM_MAPPING.items():
if value := kwargs.pop(legacy_key, None):
_LOGGER.debug(
"Legacy parameter '%s' used, consider using '%s'", legacy_key, key
)
search_query[key] = value
search_query.update(**kwargs)
if not libtype:
# Default to a sane libtype if not explicitly provided
for preferred_libtype in PREFERRED_LIBTYPE_ORDER:
if any(key.startswith(preferred_libtype) for key in search_query):
libtype = preferred_libtype
break
search_query.update(libtype=libtype)
_LOGGER.debug("Processed search query: %s", search_query)
try:
title = kwargs["title"]
except KeyError:
_LOGGER.error("Must specify 'title' for this search")
results = library_section.search(**search_query)
except (BadRequest, NotFound) as exc:
_LOGGER.error("Problem in query %s: %s", search_query, exc)
return None
try:
movies = library_section.search(**kwargs, libtype="movie", maxresults=3)
except BadRequest as err:
_LOGGER.error("Invalid search payload provided: %s", err)
if not results:
return None
if not movies:
raise MediaNotFound(f"Movie {title}") from None
if len(results) > 1:
if allow_multiple:
return results
if len(movies) > 1:
exact_matches = [x for x in movies if x.title.lower() == title.lower()]
if len(exact_matches) == 1:
return exact_matches[0]
match_list = [f"{x.title} ({x.year})" for x in movies]
_LOGGER.warning("Multiple matches found during search: %s", match_list)
if title := search_query.get("title") or search_query.get("movie.title"):
exact_matches = [x for x in results if x.title.lower() == title.lower()]
if len(exact_matches) == 1:
return exact_matches[0]
_LOGGER.warning(
"Multiple matches, make content_id more specific or use `allow_multiple`: %s",
results,
)
return None
return movies[0]
def lookup_tv(library_section, **kwargs):
"""Find TV media and return a Plex media object."""
season_number = kwargs.get("season_number")
episode_number = kwargs.get("episode_number")
try:
show_name = kwargs["show_name"]
show = library_section.get(show_name)
except KeyError:
_LOGGER.error("Must specify 'show_name' for this search")
return None
except NotFound as err:
raise MediaNotFound(f"Show {show_name}") from err
if not season_number:
return show
try:
season = show.season(int(season_number))
except NotFound as err:
raise MediaNotFound(f"Season {season_number} of {show_name}") from err
if not episode_number:
return season
try:
return season.episode(episode=int(episode_number))
except NotFound as err:
episode = f"S{str(season_number).zfill(2)}E{str(episode_number).zfill(2)}"
raise MediaNotFound(f"Episode {episode} of {show_name}") from err
def lookup_music(library_section, **kwargs):
"""Search for music and return a Plex media object."""
album_name = kwargs.get("album_name")
track_name = kwargs.get("track_name")
track_number = kwargs.get("track_number")
try:
artist_name = kwargs["artist_name"]
artist = library_section.get(artist_name)
except KeyError:
_LOGGER.error("Must specify 'artist_name' for this search")
return None
except NotFound as err:
raise MediaNotFound(f"Artist {artist_name}") from err
if album_name:
try:
album = artist.album(album_name)
except NotFound as err:
raise MediaNotFound(f"Album {album_name} by {artist_name}") from err
if track_name:
try:
return album.track(track_name)
except NotFound as err:
raise MediaNotFound(
f"Track {track_name} on {album_name} by {artist_name}"
) from err
if track_number:
for track in album.tracks():
if int(track.index) == int(track_number):
return track
raise MediaNotFound(
f"Track {track_number} on {album_name} by {artist_name}"
) from None
return album
if track_name:
try:
return artist.get(track_name)
except NotFound as err:
raise MediaNotFound(f"Track {track_name} by {artist_name}") from err
return artist
return results[0]

38
homeassistant/components/plex/server.py
View File

@ -13,13 +13,7 @@ from requests import Session
import requests.exceptions
from homeassistant.components.media_player import DOMAIN as MP_DOMAIN
from homeassistant.components.media_player.const import (
MEDIA_TYPE_EPISODE,
MEDIA_TYPE_MOVIE,
MEDIA_TYPE_MUSIC,
MEDIA_TYPE_PLAYLIST,
MEDIA_TYPE_VIDEO,
)
from homeassistant.components.media_player.const import MEDIA_TYPE_PLAYLIST
from homeassistant.const import CONF_CLIENT_ID, CONF_TOKEN, CONF_URL, CONF_VERIFY_SSL
from homeassistant.core import callback
from homeassistant.helpers.debounce import Debouncer
@ -47,13 +41,8 @@ from .const import (
X_PLEX_PRODUCT,
X_PLEX_VERSION,
)
from .errors import (
MediaNotFound,
NoServersFound,
ServerNotSpecified,
ShouldUpdateConfigEntry,
)
from .media_search import lookup_movie, lookup_music, lookup_tv
from .errors import NoServersFound, ServerNotSpecified, ShouldUpdateConfigEntry
from .media_search import search_media
from .models import PlexSession
_LOGGER = logging.getLogger(__name__)
@ -652,26 +641,7 @@ class PlexServer:
_LOGGER.error("Library '%s' not found", library_name)
return None
try:
if media_type == MEDIA_TYPE_EPISODE:
return lookup_tv(library_section, **kwargs)
if media_type == MEDIA_TYPE_MOVIE:
return lookup_movie(library_section, **kwargs)
if media_type == MEDIA_TYPE_MUSIC:
return lookup_music(library_section, **kwargs)
if media_type == MEDIA_TYPE_VIDEO:
# Legacy method for compatibility
try:
video_name = kwargs["video_name"]
return library_section.get(video_name)
except KeyError:
_LOGGER.error("Must specify 'video_name' for this search")
return None
except NotFound as err:
raise MediaNotFound(f"Video {video_name}") from err
except MediaNotFound as failed_item:
_LOGGER.error("%s not found in %s", failed_item, library_name)
return None
return search_media(media_type, library_section, **kwargs)
@property
def sensor_attributes(self):

303
tests/components/plex/test_media_search.py Normal file
View File

@ -0,0 +1,303 @@
"""Tests for Plex server."""
from unittest.mock import patch
from plexapi.exceptions import BadRequest, NotFound
import pytest
from homeassistant.components.media_player import DOMAIN as MP_DOMAIN
from homeassistant.components.media_player.const import (
ATTR_MEDIA_CONTENT_ID,
ATTR_MEDIA_CONTENT_TYPE,
MEDIA_TYPE_EPISODE,
MEDIA_TYPE_MOVIE,
MEDIA_TYPE_MUSIC,
MEDIA_TYPE_PLAYLIST,
MEDIA_TYPE_VIDEO,
SERVICE_PLAY_MEDIA,
)
from homeassistant.components.plex.const import DOMAIN
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.exceptions import HomeAssistantError
async def test_media_lookups(
hass, mock_plex_server, requests_mock, playqueue_created, caplog
):
"""Test media lookups to Plex server."""
# Plex Key searches
media_player_id = hass.states.async_entity_ids("media_player")[0]
requests_mock.post("/playqueues", text=playqueue_created)
requests_mock.get("/player/playback/playMedia", status_code=200)
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: DOMAIN,
ATTR_MEDIA_CONTENT_ID: 1,
},
True,
)
with pytest.raises(HomeAssistantError) as excinfo:
with patch("plexapi.server.PlexServer.fetchItem", side_effect=NotFound):
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: DOMAIN,
ATTR_MEDIA_CONTENT_ID: 123,
},
True,
)
assert "Media could not be found: 123" in str(excinfo.value)
# TV show searches
with pytest.raises(HomeAssistantError) as excinfo:
payload = '{"library_name": "Not a Library", "show_name": "TV Show"}'
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_EPISODE,
ATTR_MEDIA_CONTENT_ID: payload,
},
True,
)
assert f"Media could not be found: {payload}" in str(excinfo.value)
with patch("plexapi.library.LibrarySection.search") as search:
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_EPISODE,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "TV Shows", "show_name": "TV Show"}',
},
True,
)
search.assert_called_with(**{"show.title": "TV Show", "libtype": "show"})
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_EPISODE,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "TV Shows", "episode_name": "An Episode"}',
},
True,
)
search.assert_called_with(
**{"episode.title": "An Episode", "libtype": "episode"}
)
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_EPISODE,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "TV Shows", "show_name": "TV Show", "season_number": 1}',
},
True,
)
search.assert_called_with(
**{"show.title": "TV Show", "season.index": 1, "libtype": "season"}
)
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_EPISODE,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "TV Shows", "show_name": "TV Show", "season_number": 1, "episode_number": 3}',
},
True,
)
search.assert_called_with(
**{
"show.title": "TV Show",
"season.index": 1,
"episode.index": 3,
"libtype": "episode",
}
)
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MUSIC,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "Music", "artist_name": "Artist"}',
},
True,
)
search.assert_called_with(**{"artist.title": "Artist", "libtype": "artist"})
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MUSIC,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "Music", "album_name": "Album"}',
},
True,
)
search.assert_called_with(**{"album.title": "Album", "libtype": "album"})
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MUSIC,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "Music", "artist_name": "Artist", "track_name": "Track 3"}',
},
True,
)
search.assert_called_with(
**{"artist.title": "Artist", "track.title": "Track 3", "libtype": "track"}
)
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MUSIC,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "Music", "artist_name": "Artist", "album_name": "Album"}',
},
True,
)
search.assert_called_with(
**{"artist.title": "Artist", "album.title": "Album", "libtype": "album"}
)
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MUSIC,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "Music", "artist_name": "Artist", "album_name": "Album", "track_number": 3}',
},
True,
)
search.assert_called_with(
**{
"artist.title": "Artist",
"album.title": "Album",
"track.index": 3,
"libtype": "track",
}
)
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MUSIC,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "Music", "artist_name": "Artist", "album_name": "Album", "track_name": "Track 3"}',
},
True,
)
search.assert_called_with(
**{
"artist.title": "Artist",
"album.title": "Album",
"track.title": "Track 3",
"libtype": "track",
}
)
# Movie searches
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_VIDEO,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "Movies", "video_name": "Movie 1"}',
},
True,
)
search.assert_called_with(**{"movie.title": "Movie 1", "libtype": None})
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MOVIE,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "Movies", "title": "Movie 1"}',
},
True,
)
search.assert_called_with(**{"title": "Movie 1", "libtype": None})
# TV show searches
with pytest.raises(HomeAssistantError) as excinfo:
payload = '{"library_name": "Movies", "title": "Not a Movie"}'
with patch("plexapi.library.LibrarySection.search", side_effect=BadRequest):
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_VIDEO,
ATTR_MEDIA_CONTENT_ID: payload,
},
True,
)
assert "Problem in query" in caplog.text
assert f"Media could not be found: {payload}" in str(excinfo.value)
# Playlist searches
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_PLAYLIST,
ATTR_MEDIA_CONTENT_ID: '{"playlist_name": "Playlist 1"}',
},
True,
)
with pytest.raises(HomeAssistantError) as excinfo:
payload = '{"playlist_name": "Not a Playlist"}'
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_PLAYLIST,
ATTR_MEDIA_CONTENT_ID: payload,
},
True,
)
assert "Playlist 'Not a Playlist' not found" in caplog.text
assert f"Media could not be found: {payload}" in str(excinfo.value)
with pytest.raises(HomeAssistantError) as excinfo:
payload = "{}"
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_PLAYLIST,
ATTR_MEDIA_CONTENT_ID: payload,
},
True,
)
assert "Must specify 'playlist_name' for this search" in caplog.text
assert f"Media could not be found: {payload}" in str(excinfo.value)

202
tests/components/plex/test_playback.py
View File

@ -2,20 +2,48 @@
from http import HTTPStatus
from unittest.mock import patch
import pytest
from homeassistant.components.media_player.const import (
ATTR_MEDIA_CONTENT_ID,
ATTR_MEDIA_CONTENT_TYPE,
DOMAIN as MP_DOMAIN,
MEDIA_TYPE_EPISODE,
MEDIA_TYPE_MOVIE,
MEDIA_TYPE_MUSIC,
SERVICE_PLAY_MEDIA,
)
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.exceptions import HomeAssistantError
class MockPlexMedia:
"""Minimal mock of plexapi media object."""
key = "key"
def __init__(self, title, mediatype):
"""Initialize the instance."""
self.listType = mediatype
self.title = title
self.type = mediatype
def section(self):
"""Return the LibrarySection."""
return MockPlexLibrarySection()
class MockPlexLibrarySection:
"""Minimal mock of plexapi LibrarySection."""
uuid = "00000000-0000-0000-0000-000000000000"
async def test_media_player_playback(
hass, setup_plex_server, requests_mock, playqueue_created, player_plexweb_resources
hass,
setup_plex_server,
requests_mock,
playqueue_created,
player_plexweb_resources,
caplog,
):
"""Test playing media on a Plex media_player."""
requests_mock.get("http://1.2.3.5:32400/resources", text=player_plexweb_resources)
@ -26,112 +54,88 @@ async def test_media_player_playback(
requests_mock.post("/playqueues", text=playqueue_created)
requests_mock.get("/player/playback/playMedia", status_code=HTTPStatus.OK)
# Test movie success
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MOVIE,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "Movies", "title": "Movie 1" }',
},
True,
)
# Test movie incomplete dict
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MOVIE,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "Movies"}',
},
True,
)
# Test movie failure with options
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MOVIE,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "Movies", "title": "Does not exist" }',
},
True,
)
# Test movie failure with nothing found
# Test media lookup failure
payload = '{"library_name": "Movies", "title": "Movie 1" }'
with patch("plexapi.library.LibrarySection.search", return_value=None):
with pytest.raises(HomeAssistantError) as excinfo:
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MOVIE,
ATTR_MEDIA_CONTENT_ID: payload,
},
True,
)
assert f"Media could not be found: {payload}" in str(excinfo.value)
movie1 = MockPlexMedia("Movie", "movie")
movie2 = MockPlexMedia("Movie II", "movie")
movie3 = MockPlexMedia("Movie III", "movie")
# Test movie success
movies = [movie1]
with patch("plexapi.library.LibrarySection.search", return_value=movies):
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MOVIE,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "Movies", "title": "Does not exist" }',
ATTR_MEDIA_CONTENT_ID: '{"library_name": "Movies", "title": "Movie 1" }',
},
True,
)
# Test movie success with dict
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MUSIC,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "Music", "artist_name": "Artist", "album_name": "Album"}',
},
True,
)
# Test multiple choices with exact match
movies = [movie1, movie2]
with patch("plexapi.library.LibrarySection.search", return_value=movies), patch(
"homeassistant.components.plex.server.PlexServer.create_playqueue"
) as mock_create_playqueue:
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MOVIE,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "Movies", "title": "Movie" }',
},
True,
)
assert mock_create_playqueue.call_args.args == (movie1,)
# Test TV show episoe lookup failure
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_EPISODE,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "TV Shows", "show_name": "TV Show", "season_number": 1, "episode_number": 99}',
},
True,
)
# Test multiple choices without exact match
movies = [movie2, movie3]
with pytest.raises(HomeAssistantError) as excinfo:
payload = '{"library_name": "Movies", "title": "Movie" }'
with patch("plexapi.library.LibrarySection.search", return_value=movies):
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MOVIE,
ATTR_MEDIA_CONTENT_ID: payload,
},
True,
)
assert f"Media could not be found: {payload}" in str(excinfo.value)
assert "Multiple matches, make content_id more specific" in caplog.text
# Test track name lookup failure
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MUSIC,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "Music", "artist_name": "Artist", "album_name": "Album", "track_name": "Not a track"}',
},
True,
)
# Test media lookup failure by key
requests_mock.get("/library/metadata/999", status_code=HTTPStatus.NOT_FOUND)
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MUSIC,
ATTR_MEDIA_CONTENT_ID: "999",
},
True,
)
# Test invalid Plex server requested
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MUSIC,
ATTR_MEDIA_CONTENT_ID: '{"plex_server": "unknown_plex_server", "library_name": "Music", "artist_name": "Artist", "album_name": "Album"}',
},
True,
)
# Test multiple choices with allow_multiple
movies = [movie1, movie2, movie3]
with patch("plexapi.library.LibrarySection.search", return_value=movies), patch(
"homeassistant.components.plex.server.PlexServer.create_playqueue"
) as mock_create_playqueue:
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player,
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MOVIE,
ATTR_MEDIA_CONTENT_ID: '{"library_name": "Movies", "title": "Movie", "allow_multiple": true }',
},
True,
)
assert mock_create_playqueue.call_args.args == (movies,)

225
tests/components/plex/test_server.py
View File

@ -1,22 +1,10 @@
"""Tests for Plex server."""
import copy
from http import HTTPStatus
from unittest.mock import patch
from plexapi.exceptions import BadRequest, NotFound
from requests.exceptions import ConnectionError, RequestException
from homeassistant.components.media_player import DOMAIN as MP_DOMAIN
from homeassistant.components.media_player.const import (
ATTR_MEDIA_CONTENT_ID,
ATTR_MEDIA_CONTENT_TYPE,
MEDIA_TYPE_EPISODE,
MEDIA_TYPE_MOVIE,
MEDIA_TYPE_MUSIC,
MEDIA_TYPE_PLAYLIST,
MEDIA_TYPE_VIDEO,
SERVICE_PLAY_MEDIA,
)
from homeassistant.components.plex.const import (
CONF_IGNORE_NEW_SHARED_USERS,
CONF_IGNORE_PLEX_WEB_CLIENTS,
@ -25,7 +13,6 @@ from homeassistant.components.plex.const import (
DOMAIN,
SERVERS,
)
from homeassistant.const import ATTR_ENTITY_ID
from .const import DEFAULT_DATA, DEFAULT_OPTIONS
from .helpers import trigger_plex_update, wait_for_debouncer
@ -179,215 +166,3 @@ async def test_ignore_plex_web_client(hass, entry, setup_plex_server):
media_players = hass.states.async_entity_ids("media_player")
assert len(media_players) == int(sensor.state) - 1
async def test_media_lookups(hass, mock_plex_server, requests_mock, playqueue_created):
"""Test media lookups to Plex server."""
server_id = mock_plex_server.machine_identifier
loaded_server = hass.data[DOMAIN][SERVERS][server_id]
# Plex Key searches
media_player_id = hass.states.async_entity_ids("media_player")[0]
requests_mock.post("/playqueues", text=playqueue_created)
requests_mock.get("/player/playback/playMedia", status_code=HTTPStatus.OK)
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: DOMAIN,
ATTR_MEDIA_CONTENT_ID: 1,
},
True,
)
with patch("plexapi.server.PlexServer.fetchItem", side_effect=NotFound):
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: DOMAIN,
ATTR_MEDIA_CONTENT_ID: 123,
},
True,
)
# TV show searches
assert (
loaded_server.lookup_media(
MEDIA_TYPE_EPISODE, library_name="Not a Library", show_name="TV Show"
)
is None
)
assert (
loaded_server.lookup_media(
MEDIA_TYPE_EPISODE, library_name="TV Shows", show_name="Not a TV Show"
)
is None
)
assert (
loaded_server.lookup_media(
MEDIA_TYPE_EPISODE, library_name="TV Shows", episode_name="An Episode"
)
is None
)
assert loaded_server.lookup_media(
MEDIA_TYPE_EPISODE, library_name="TV Shows", show_name="TV Show"
)
assert loaded_server.lookup_media(
MEDIA_TYPE_EPISODE,
library_name="TV Shows",
show_name="TV Show",
season_number=1,
)
assert loaded_server.lookup_media(
MEDIA_TYPE_EPISODE,
library_name="TV Shows",
show_name="TV Show",
season_number=1,
episode_number=3,
)
assert (
loaded_server.lookup_media(
MEDIA_TYPE_EPISODE,
library_name="TV Shows",
show_name="TV Show",
season_number=2,
)
is None
)
assert (
loaded_server.lookup_media(
MEDIA_TYPE_EPISODE,
library_name="TV Shows",
show_name="TV Show",
season_number=2,
episode_number=1,
)
is None
)
# Music searches
assert (
loaded_server.lookup_media(
MEDIA_TYPE_MUSIC, library_name="Music", album_name="Album"
)
is None
)
assert loaded_server.lookup_media(
MEDIA_TYPE_MUSIC, library_name="Music", artist_name="Artist"
)
assert loaded_server.lookup_media(
MEDIA_TYPE_MUSIC,
library_name="Music",
artist_name="Artist",
track_name="Track 3",
)
assert loaded_server.lookup_media(
MEDIA_TYPE_MUSIC,
library_name="Music",
artist_name="Artist",
album_name="Album",
)
assert (
loaded_server.lookup_media(
MEDIA_TYPE_MUSIC,
library_name="Music",
artist_name="Not an Artist",
album_name="Album",
)
is None
)
assert (
loaded_server.lookup_media(
MEDIA_TYPE_MUSIC,
library_name="Music",
artist_name="Artist",
album_name="Not an Album",
)
is None
)
assert (
loaded_server.lookup_media(
MEDIA_TYPE_MUSIC,
library_name="Music",
artist_name="Artist",
album_name=" Album",
track_name="Not a Track",
)
is None
)
assert (
loaded_server.lookup_media(
MEDIA_TYPE_MUSIC,
library_name="Music",
artist_name="Artist",
track_name="Not a Track",
)
is None
)
assert loaded_server.lookup_media(
MEDIA_TYPE_MUSIC,
library_name="Music",
artist_name="Artist",
album_name="Album",
track_number=3,
)
assert (
loaded_server.lookup_media(
MEDIA_TYPE_MUSIC,
library_name="Music",
artist_name="Artist",
album_name="Album",
track_number=30,
)
is None
)
assert loaded_server.lookup_media(
MEDIA_TYPE_MUSIC,
library_name="Music",
artist_name="Artist",
album_name="Album",
track_name="Track 3",
)
# Playlist searches
assert loaded_server.lookup_media(MEDIA_TYPE_PLAYLIST, playlist_name="Playlist 1")
assert loaded_server.lookup_media(MEDIA_TYPE_PLAYLIST) is None
assert (
loaded_server.lookup_media(MEDIA_TYPE_PLAYLIST, playlist_name="Not a Playlist")
is None
)
# Legacy Movie searches
assert loaded_server.lookup_media(MEDIA_TYPE_VIDEO, video_name="Movie") is None
assert loaded_server.lookup_media(MEDIA_TYPE_VIDEO, library_name="Movies") is None
assert loaded_server.lookup_media(
MEDIA_TYPE_VIDEO, library_name="Movies", video_name="Movie 1"
)
assert (
loaded_server.lookup_media(
MEDIA_TYPE_VIDEO, library_name="Movies", video_name="Not a Movie"
)
is None
)
# Movie searches
assert loaded_server.lookup_media(MEDIA_TYPE_MOVIE, title="Movie") is None
assert loaded_server.lookup_media(MEDIA_TYPE_MOVIE, library_name="Movies") is None
assert loaded_server.lookup_media(
MEDIA_TYPE_MOVIE, library_name="Movies", title="Movie 1"
)
with patch("plexapi.library.LibrarySection.search", side_effect=BadRequest):
assert (
loaded_server.lookup_media(
MEDIA_TYPE_MOVIE, library_name="Movies", title="Not a Movie"
)
is None
)
assert (
loaded_server.lookup_media(
MEDIA_TYPE_MOVIE, library_name="Movies", title="Movie"
)
) is None

11
tests/components/plex/test_services.py
View File

@ -180,10 +180,13 @@ async def test_sonos_play_media(
# Test with speakers available but media not found
content_id_bad_media = '{"library_name": "Music", "artist_name": "Not an Artist"}'
with pytest.raises(HomeAssistantError) as excinfo:
play_on_sonos(hass, MEDIA_TYPE_MUSIC, content_id_bad_media, sonos_speaker_name)
assert "Plex media not found" in str(excinfo.value)
assert playback_mock.call_count == 3
with patch("plexapi.library.LibrarySection.search", return_value=None):
with pytest.raises(HomeAssistantError) as excinfo:
play_on_sonos(
hass, MEDIA_TYPE_MUSIC, content_id_bad_media, sonos_speaker_name
)
assert "Plex media not found" in str(excinfo.value)
assert playback_mock.call_count == 3
# Test with speakers available and playqueue
requests_mock.get("https://1.2.3.4:32400/playQueues/1234", text=playqueue_1234)