mirror of
https://github.com/home-assistant/core.git
synced 2025-07-24 13:47:35 +00:00
Rewrite and add Plex tests (#32633)
* Rewrite and add Plex tests * Remove unnecessary mocks * Explicitly import constants for readability
This commit is contained in:
jjlawren
GitHub
parent
f7ddbc7e1e
commit
44c7743351
@ -542,10 +542,8 @@ omit =
|
||||
homeassistant/components/pioneer/media_player.py
|
||||
homeassistant/components/pjlink/media_player.py
|
||||
homeassistant/components/plaato/*
|
||||
homeassistant/components/plex/__init__.py
|
||||
homeassistant/components/plex/media_player.py
|
||||
homeassistant/components/plex/sensor.py
|
||||
homeassistant/components/plex/server.py
|
||||
homeassistant/components/plugwise/*
|
||||
homeassistant/components/plum_lightpad/*
|
||||
homeassistant/components/pocketcasts/sensor.py
|
||||
|
||||
@ -333,6 +333,7 @@ class PlexMediaPlayer(MediaPlayerDevice):
|
||||
|
||||
def force_idle(self):
|
||||
"""Force client to idle."""
|
||||
self._player_state = STATE_IDLE
|
||||
self._state = STATE_IDLE
|
||||
self.session = None
|
||||
self._clear_media_details()
|
||||
|
||||
52
tests/components/plex/const.py
Normal file
52
tests/components/plex/const.py
Normal file
@ -0,0 +1,52 @@
|
||||
"""Constants used by Plex tests."""
|
||||
from homeassistant.components.media_player import DOMAIN as MP_DOMAIN
|
||||
from homeassistant.components.plex import const
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_PORT,
|
||||
CONF_TOKEN,
|
||||
CONF_URL,
|
||||
CONF_VERIFY_SSL,
|
||||
)
|
||||
|
||||
MOCK_SERVERS = [
|
||||
{
|
||||
CONF_HOST: "1.2.3.4",
|
||||
CONF_PORT: 32400,
|
||||
const.CONF_SERVER: "Plex Server 1",
|
||||
const.CONF_SERVER_IDENTIFIER: "unique_id_123",
|
||||
},
|
||||
{
|
||||
CONF_HOST: "4.3.2.1",
|
||||
CONF_PORT: 32400,
|
||||
const.CONF_SERVER: "Plex Server 2",
|
||||
const.CONF_SERVER_IDENTIFIER: "unique_id_456",
|
||||
},
|
||||
]
|
||||
|
||||
MOCK_USERS = {
|
||||
"Owner": {"enabled": True},
|
||||
"b": {"enabled": True},
|
||||
"c": {"enabled": True},
|
||||
}
|
||||
|
||||
MOCK_TOKEN = "secret_token"
|
||||
|
||||
DEFAULT_DATA = {
|
||||
const.CONF_SERVER: MOCK_SERVERS[0][const.CONF_SERVER],
|
||||
const.PLEX_SERVER_CONFIG: {
|
||||
const.CONF_CLIENT_IDENTIFIER: "00000000-0000-0000-0000-000000000000",
|
||||
CONF_TOKEN: MOCK_TOKEN,
|
||||
CONF_URL: f"https://{MOCK_SERVERS[0][CONF_HOST]}:{MOCK_SERVERS[0][CONF_PORT]}",
|
||||
CONF_VERIFY_SSL: True,
|
||||
},
|
||||
const.CONF_SERVER_IDENTIFIER: MOCK_SERVERS[0][const.CONF_SERVER_IDENTIFIER],
|
||||
}
|
||||
|
||||
DEFAULT_OPTIONS = {
|
||||
MP_DOMAIN: {
|
||||
const.CONF_IGNORE_NEW_SHARED_USERS: False,
|
||||
const.CONF_MONITORED_USERS: MOCK_USERS,
|
||||
const.CONF_USE_EPISODE_ART: False,
|
||||
}
|
||||
}
|
||||
@ -1,29 +1,12 @@
|
||||
"""Mock classes used in tests."""
|
||||
import itertools
|
||||
from homeassistant.components.plex.const import (
|
||||
CONF_SERVER,
|
||||
CONF_SERVER_IDENTIFIER,
|
||||
PLEX_SERVER_CONFIG,
|
||||
)
|
||||
from homeassistant.const import CONF_URL
|
||||
|
||||
from homeassistant.components.plex.const import CONF_SERVER, CONF_SERVER_IDENTIFIER
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT
|
||||
|
||||
MOCK_SERVERS = [
|
||||
{
|
||||
CONF_HOST: "1.2.3.4",
|
||||
CONF_PORT: 32400,
|
||||
CONF_SERVER: "Plex Server 1",
|
||||
CONF_SERVER_IDENTIFIER: "unique_id_123",
|
||||
},
|
||||
{
|
||||
CONF_HOST: "4.3.2.1",
|
||||
CONF_PORT: 32400,
|
||||
CONF_SERVER: "Plex Server 2",
|
||||
CONF_SERVER_IDENTIFIER: "unique_id_456",
|
||||
},
|
||||
]
|
||||
|
||||
MOCK_MONITORED_USERS = {
|
||||
"a": {"enabled": True},
|
||||
"b": {"enabled": False},
|
||||
"c": {"enabled": True},
|
||||
}
|
||||
from .const import DEFAULT_DATA, MOCK_SERVERS, MOCK_USERS
|
||||
|
||||
|
||||
class MockResource:
|
||||
@ -64,10 +47,11 @@ class MockPlexAccount:
|
||||
class MockPlexSystemAccount:
|
||||
"""Mock a PlexSystemAccount instance."""
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self, index):
|
||||
"""Initialize the object."""
|
||||
self.name = "Dummy"
|
||||
self.accountID = 1
|
||||
# Start accountIDs at 1 to set proper owner.
|
||||
self.name = list(MOCK_USERS)[index]
|
||||
self.accountID = index + 1
|
||||
|
||||
|
||||
class MockPlexServer:
|
||||
@ -76,68 +60,179 @@ class MockPlexServer:
|
||||
def __init__(
|
||||
self,
|
||||
index=0,
|
||||
ssl=True,
|
||||
load_users=True,
|
||||
num_users=len(MOCK_MONITORED_USERS),
|
||||
ignore_new_users=False,
|
||||
config_entry=None,
|
||||
num_users=len(MOCK_USERS),
|
||||
session_type="video",
|
||||
):
|
||||
"""Initialize the object."""
|
||||
host = MOCK_SERVERS[index][CONF_HOST]
|
||||
port = MOCK_SERVERS[index][CONF_PORT]
|
||||
self.friendlyName = MOCK_SERVERS[index][ # pylint: disable=invalid-name
|
||||
CONF_SERVER
|
||||
if config_entry:
|
||||
self._data = config_entry.data
|
||||
else:
|
||||
self._data = DEFAULT_DATA
|
||||
|
||||
self._baseurl = self._data[PLEX_SERVER_CONFIG][CONF_URL]
|
||||
self.friendlyName = self._data[CONF_SERVER]
|
||||
self.machineIdentifier = self._data[CONF_SERVER_IDENTIFIER]
|
||||
|
||||
self._systemAccounts = list(map(MockPlexSystemAccount, range(num_users)))
|
||||
|
||||
self._clients = []
|
||||
self._sessions = []
|
||||
self.set_clients(num_users)
|
||||
self.set_sessions(num_users, session_type)
|
||||
|
||||
def set_clients(self, num_clients):
|
||||
"""Set up mock PlexClients for this PlexServer."""
|
||||
self._clients = [MockPlexClient(self._baseurl, x) for x in range(num_clients)]
|
||||
|
||||
def set_sessions(self, num_sessions, session_type):
|
||||
"""Set up mock PlexSessions for this PlexServer."""
|
||||
self._sessions = [
|
||||
MockPlexSession(self._clients[x], mediatype=session_type, index=x)
|
||||
for x in range(num_sessions)
|
||||
]
|
||||
self.machineIdentifier = MOCK_SERVERS[index][ # pylint: disable=invalid-name
|
||||
CONF_SERVER_IDENTIFIER
|
||||
]
|
||||
prefix = "https" if ssl else "http"
|
||||
self._baseurl = f"{prefix}://{host}:{port}"
|
||||
self._systemAccount = MockPlexSystemAccount()
|
||||
self._ignore_new_users = ignore_new_users
|
||||
self._load_users = load_users
|
||||
self._num_users = num_users
|
||||
|
||||
def clear_clients(self):
|
||||
"""Clear all active PlexClients."""
|
||||
self._clients = []
|
||||
|
||||
def clear_sessions(self):
|
||||
"""Clear all active PlexSessions."""
|
||||
self._sessions = []
|
||||
|
||||
def clients(self):
|
||||
"""Mock the clients method."""
|
||||
return self._clients
|
||||
|
||||
def sessions(self):
|
||||
"""Mock the sessions method."""
|
||||
return self._sessions
|
||||
|
||||
def systemAccounts(self):
|
||||
"""Mock the systemAccounts lookup method."""
|
||||
return [self._systemAccount]
|
||||
return self._systemAccounts
|
||||
|
||||
def url(self, path, includeToken=False):
|
||||
"""Mock method to generate a server URL."""
|
||||
return f"{self._baseurl}{path}"
|
||||
|
||||
@property
|
||||
def accounts(self):
|
||||
"""Mock the accounts property."""
|
||||
return set(["a", "b", "c"])
|
||||
|
||||
@property
|
||||
def owner(self):
|
||||
"""Mock the owner property."""
|
||||
return "a"
|
||||
|
||||
@property
|
||||
def url_in_use(self):
|
||||
"""Return URL used by PlexServer."""
|
||||
return self._baseurl
|
||||
return set(MOCK_USERS)
|
||||
|
||||
@property
|
||||
def version(self):
|
||||
"""Mock version of PlexServer."""
|
||||
return "1.0"
|
||||
|
||||
@property
|
||||
def option_monitored_users(self):
|
||||
"""Mock loaded config option for monitored users."""
|
||||
userdict = dict(itertools.islice(MOCK_MONITORED_USERS.items(), self._num_users))
|
||||
return userdict if self._load_users else {}
|
||||
|
||||
class MockPlexClient:
|
||||
"""Mock a PlexClient instance."""
|
||||
|
||||
def __init__(self, url, index=0):
|
||||
"""Initialize the object."""
|
||||
self.machineIdentifier = f"client-{index+1}"
|
||||
self._baseurl = url
|
||||
|
||||
def url(self, key):
|
||||
"""Mock the url method."""
|
||||
return f"{self._baseurl}{key}"
|
||||
|
||||
@property
|
||||
def option_ignore_new_shared_users(self):
|
||||
"""Mock loaded config option for ignoring new users."""
|
||||
return self._ignore_new_users
|
||||
def device(self):
|
||||
"""Mock the device attribute."""
|
||||
return "DEVICE"
|
||||
|
||||
@property
|
||||
def option_show_all_controls(self):
|
||||
"""Mock loaded config option for showing all controls."""
|
||||
return False
|
||||
def platform(self):
|
||||
"""Mock the platform attribute."""
|
||||
return "PLATFORM"
|
||||
|
||||
@property
|
||||
def option_use_episode_art(self):
|
||||
"""Mock loaded config option for using episode art."""
|
||||
return False
|
||||
def product(self):
|
||||
"""Mock the product attribute."""
|
||||
return "PRODUCT"
|
||||
|
||||
@property
|
||||
def protocolCapabilities(self):
|
||||
"""Mock the protocolCapabilities attribute."""
|
||||
return ["player"]
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""Mock the state attribute."""
|
||||
return "playing"
|
||||
|
||||
@property
|
||||
def title(self):
|
||||
"""Mock the title attribute."""
|
||||
return "TITLE"
|
||||
|
||||
@property
|
||||
def version(self):
|
||||
"""Mock the version attribute."""
|
||||
return "1.0"
|
||||
|
||||
|
||||
class MockPlexSession:
|
||||
"""Mock a PlexServer.sessions() instance."""
|
||||
|
||||
def __init__(self, player, mediatype, index=0):
|
||||
"""Initialize the object."""
|
||||
self.TYPE = mediatype
|
||||
self.usernames = [list(MOCK_USERS)[index]]
|
||||
self.players = [player]
|
||||
self._section = MockPlexLibrarySection()
|
||||
|
||||
@property
|
||||
def duration(self):
|
||||
"""Mock the duration attribute."""
|
||||
return 10000000
|
||||
|
||||
@property
|
||||
def ratingKey(self):
|
||||
"""Mock the ratingKey attribute."""
|
||||
return 123
|
||||
|
||||
def section(self):
|
||||
"""Mock the section method."""
|
||||
return self._section
|
||||
|
||||
@property
|
||||
def summary(self):
|
||||
"""Mock the summary attribute."""
|
||||
return "SUMMARY"
|
||||
|
||||
@property
|
||||
def thumbUrl(self):
|
||||
"""Mock the thumbUrl attribute."""
|
||||
return "http://1.2.3.4/thumb"
|
||||
|
||||
@property
|
||||
def title(self):
|
||||
"""Mock the title attribute."""
|
||||
return "TITLE"
|
||||
|
||||
@property
|
||||
def type(self):
|
||||
"""Mock the type attribute."""
|
||||
return "movie"
|
||||
|
||||
@property
|
||||
def viewOffset(self):
|
||||
"""Mock the viewOffset attribute."""
|
||||
return 0
|
||||
|
||||
@property
|
||||
def year(self):
|
||||
"""Mock the year attribute."""
|
||||
return 2020
|
||||
|
||||
|
||||
class MockPlexLibrarySection:
|
||||
"""Mock a Plex LibrarySection instance."""
|
||||
|
||||
def __init__(self, library="Movies"):
|
||||
"""Initialize the object."""
|
||||
self.title = library
|
||||
|
||||
@ -1,55 +1,46 @@
|
||||
"""Tests for Plex config flow."""
|
||||
import copy
|
||||
from unittest.mock import patch
|
||||
|
||||
import asynctest
|
||||
from asynctest import patch
|
||||
import plexapi.exceptions
|
||||
import requests.exceptions
|
||||
|
||||
from homeassistant.components.media_player import DOMAIN as MP_DOMAIN
|
||||
from homeassistant.components.plex import config_flow
|
||||
from homeassistant.components.plex.const import (
|
||||
CONF_IGNORE_NEW_SHARED_USERS,
|
||||
CONF_MONITORED_USERS,
|
||||
CONF_SERVER,
|
||||
CONF_SERVER_IDENTIFIER,
|
||||
CONF_USE_EPISODE_ART,
|
||||
DOMAIN,
|
||||
PLEX_SERVER_CONFIG,
|
||||
PLEX_UPDATE_PLATFORMS_SIGNAL,
|
||||
SERVERS,
|
||||
)
|
||||
from homeassistant.config_entries import ENTRY_STATE_LOADED
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_TOKEN, CONF_URL
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_send
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from .mock_classes import MOCK_SERVERS, MockPlexAccount, MockPlexServer
|
||||
from .const import DEFAULT_DATA, DEFAULT_OPTIONS, MOCK_SERVERS, MOCK_TOKEN
|
||||
from .mock_classes import MockPlexAccount, MockPlexServer
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
MOCK_TOKEN = "secret_token"
|
||||
MOCK_FILE_CONTENTS = {
|
||||
f"{MOCK_SERVERS[0][CONF_HOST]}:{MOCK_SERVERS[0][CONF_PORT]}": {
|
||||
"ssl": False,
|
||||
"token": MOCK_TOKEN,
|
||||
"verify": True,
|
||||
}
|
||||
}
|
||||
|
||||
DEFAULT_OPTIONS = {
|
||||
config_flow.MP_DOMAIN: {
|
||||
config_flow.CONF_USE_EPISODE_ART: False,
|
||||
config_flow.CONF_IGNORE_NEW_SHARED_USERS: False,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def init_config_flow(hass):
|
||||
"""Init a configuration flow."""
|
||||
flow = config_flow.PlexFlowHandler()
|
||||
flow.hass = hass
|
||||
return flow
|
||||
|
||||
|
||||
async def test_bad_credentials(hass):
|
||||
"""Test when provided credentials are rejected."""
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": "user"}
|
||||
DOMAIN, context={"source": "user"}
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "start_website_auth"
|
||||
|
||||
with patch(
|
||||
"plexapi.myplex.MyPlexAccount", side_effect=plexapi.exceptions.Unauthorized
|
||||
), asynctest.patch("plexauth.PlexAuth.initiate_auth"), asynctest.patch(
|
||||
), patch("plexauth.PlexAuth.initiate_auth"), patch(
|
||||
"plexauth.PlexAuth.token", return_value="BAD TOKEN"
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"])
|
||||
@ -72,7 +63,7 @@ async def test_import_success(hass):
|
||||
|
||||
with patch("plexapi.server.PlexServer", return_value=mock_plex_server):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN,
|
||||
DOMAIN,
|
||||
context={"source": "import"},
|
||||
data={
|
||||
CONF_TOKEN: MOCK_TOKEN,
|
||||
@ -82,16 +73,10 @@ async def test_import_success(hass):
|
||||
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["title"] == mock_plex_server.friendlyName
|
||||
assert result["data"][config_flow.CONF_SERVER] == mock_plex_server.friendlyName
|
||||
assert (
|
||||
result["data"][config_flow.CONF_SERVER_IDENTIFIER]
|
||||
== mock_plex_server.machineIdentifier
|
||||
)
|
||||
assert (
|
||||
result["data"][config_flow.PLEX_SERVER_CONFIG][CONF_URL]
|
||||
== mock_plex_server.url_in_use
|
||||
)
|
||||
assert result["data"][config_flow.PLEX_SERVER_CONFIG][CONF_TOKEN] == MOCK_TOKEN
|
||||
assert result["data"][CONF_SERVER] == mock_plex_server.friendlyName
|
||||
assert result["data"][CONF_SERVER_IDENTIFIER] == mock_plex_server.machineIdentifier
|
||||
assert result["data"][PLEX_SERVER_CONFIG][CONF_URL] == mock_plex_server._baseurl
|
||||
assert result["data"][PLEX_SERVER_CONFIG][CONF_TOKEN] == MOCK_TOKEN
|
||||
|
||||
|
||||
async def test_import_bad_hostname(hass):
|
||||
@ -101,7 +86,7 @@ async def test_import_bad_hostname(hass):
|
||||
"plexapi.server.PlexServer", side_effect=requests.exceptions.ConnectionError
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN,
|
||||
DOMAIN,
|
||||
context={"source": "import"},
|
||||
data={
|
||||
CONF_TOKEN: MOCK_TOKEN,
|
||||
@ -116,14 +101,14 @@ async def test_unknown_exception(hass):
|
||||
"""Test when an unknown exception is encountered."""
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": "user"}
|
||||
DOMAIN, context={"source": "user"}
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "start_website_auth"
|
||||
|
||||
with patch("plexapi.myplex.MyPlexAccount", side_effect=Exception), asynctest.patch(
|
||||
with patch("plexapi.myplex.MyPlexAccount", side_effect=Exception), patch(
|
||||
"plexauth.PlexAuth.initiate_auth"
|
||||
), asynctest.patch("plexauth.PlexAuth.token", return_value="MOCK_TOKEN"):
|
||||
), patch("plexauth.PlexAuth.token", return_value="MOCK_TOKEN"):
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"])
|
||||
assert result["type"] == "external"
|
||||
|
||||
@ -141,14 +126,14 @@ async def test_no_servers_found(hass):
|
||||
await async_setup_component(hass, "http", {"http": {}})
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": "user"}
|
||||
DOMAIN, context={"source": "user"}
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "start_website_auth"
|
||||
|
||||
with patch(
|
||||
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount(servers=0)
|
||||
), asynctest.patch("plexauth.PlexAuth.initiate_auth"), asynctest.patch(
|
||||
), patch("plexauth.PlexAuth.initiate_auth"), patch(
|
||||
"plexauth.PlexAuth.token", return_value=MOCK_TOKEN
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"])
|
||||
@ -171,14 +156,14 @@ async def test_single_available_server(hass):
|
||||
await async_setup_component(hass, "http", {"http": {}})
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": "user"}
|
||||
DOMAIN, context={"source": "user"}
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "start_website_auth"
|
||||
|
||||
with patch("plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()), patch(
|
||||
"plexapi.server.PlexServer", return_value=mock_plex_server
|
||||
), asynctest.patch("plexauth.PlexAuth.initiate_auth"), asynctest.patch(
|
||||
), patch("plexauth.PlexAuth.initiate_auth"), patch(
|
||||
"plexauth.PlexAuth.token", return_value=MOCK_TOKEN
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"])
|
||||
@ -190,16 +175,12 @@ async def test_single_available_server(hass):
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"])
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["title"] == mock_plex_server.friendlyName
|
||||
assert result["data"][config_flow.CONF_SERVER] == mock_plex_server.friendlyName
|
||||
assert result["data"][CONF_SERVER] == mock_plex_server.friendlyName
|
||||
assert (
|
||||
result["data"][config_flow.CONF_SERVER_IDENTIFIER]
|
||||
== mock_plex_server.machineIdentifier
|
||||
result["data"][CONF_SERVER_IDENTIFIER] == mock_plex_server.machineIdentifier
|
||||
)
|
||||
assert (
|
||||
result["data"][config_flow.PLEX_SERVER_CONFIG][CONF_URL]
|
||||
== mock_plex_server.url_in_use
|
||||
)
|
||||
assert result["data"][config_flow.PLEX_SERVER_CONFIG][CONF_TOKEN] == MOCK_TOKEN
|
||||
assert result["data"][PLEX_SERVER_CONFIG][CONF_URL] == mock_plex_server._baseurl
|
||||
assert result["data"][PLEX_SERVER_CONFIG][CONF_TOKEN] == MOCK_TOKEN
|
||||
|
||||
|
||||
async def test_multiple_servers_with_selection(hass):
|
||||
@ -210,18 +191,16 @@ async def test_multiple_servers_with_selection(hass):
|
||||
await async_setup_component(hass, "http", {"http": {}})
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": "user"}
|
||||
DOMAIN, context={"source": "user"}
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "start_website_auth"
|
||||
|
||||
with patch(
|
||||
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount(servers=2)
|
||||
), patch(
|
||||
"plexapi.server.PlexServer", return_value=mock_plex_server
|
||||
), asynctest.patch(
|
||||
), patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
|
||||
"plexauth.PlexAuth.initiate_auth"
|
||||
), asynctest.patch(
|
||||
), patch(
|
||||
"plexauth.PlexAuth.token", return_value=MOCK_TOKEN
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"])
|
||||
@ -235,23 +214,16 @@ async def test_multiple_servers_with_selection(hass):
|
||||
assert result["step_id"] == "select_server"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
config_flow.CONF_SERVER: MOCK_SERVERS[0][config_flow.CONF_SERVER]
|
||||
},
|
||||
result["flow_id"], user_input={CONF_SERVER: MOCK_SERVERS[0][CONF_SERVER]},
|
||||
)
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["title"] == mock_plex_server.friendlyName
|
||||
assert result["data"][config_flow.CONF_SERVER] == mock_plex_server.friendlyName
|
||||
assert result["data"][CONF_SERVER] == mock_plex_server.friendlyName
|
||||
assert (
|
||||
result["data"][config_flow.CONF_SERVER_IDENTIFIER]
|
||||
== mock_plex_server.machineIdentifier
|
||||
result["data"][CONF_SERVER_IDENTIFIER] == mock_plex_server.machineIdentifier
|
||||
)
|
||||
assert (
|
||||
result["data"][config_flow.PLEX_SERVER_CONFIG][CONF_URL]
|
||||
== mock_plex_server.url_in_use
|
||||
)
|
||||
assert result["data"][config_flow.PLEX_SERVER_CONFIG][CONF_TOKEN] == MOCK_TOKEN
|
||||
assert result["data"][PLEX_SERVER_CONFIG][CONF_URL] == mock_plex_server._baseurl
|
||||
assert result["data"][PLEX_SERVER_CONFIG][CONF_TOKEN] == MOCK_TOKEN
|
||||
|
||||
|
||||
async def test_adding_last_unconfigured_server(hass):
|
||||
@ -262,28 +234,24 @@ async def test_adding_last_unconfigured_server(hass):
|
||||
await async_setup_component(hass, "http", {"http": {}})
|
||||
|
||||
MockConfigEntry(
|
||||
domain=config_flow.DOMAIN,
|
||||
domain=DOMAIN,
|
||||
data={
|
||||
config_flow.CONF_SERVER_IDENTIFIER: MOCK_SERVERS[1][
|
||||
config_flow.CONF_SERVER_IDENTIFIER
|
||||
],
|
||||
config_flow.CONF_SERVER: MOCK_SERVERS[1][config_flow.CONF_SERVER],
|
||||
CONF_SERVER_IDENTIFIER: MOCK_SERVERS[1][CONF_SERVER_IDENTIFIER],
|
||||
CONF_SERVER: MOCK_SERVERS[1][CONF_SERVER],
|
||||
},
|
||||
).add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": "user"}
|
||||
DOMAIN, context={"source": "user"}
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "start_website_auth"
|
||||
|
||||
with patch(
|
||||
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount(servers=2)
|
||||
), patch(
|
||||
"plexapi.server.PlexServer", return_value=mock_plex_server
|
||||
), asynctest.patch(
|
||||
), patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
|
||||
"plexauth.PlexAuth.initiate_auth"
|
||||
), asynctest.patch(
|
||||
), patch(
|
||||
"plexauth.PlexAuth.token", return_value=MOCK_TOKEN
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"])
|
||||
@ -295,16 +263,12 @@ async def test_adding_last_unconfigured_server(hass):
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"])
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["title"] == mock_plex_server.friendlyName
|
||||
assert result["data"][config_flow.CONF_SERVER] == mock_plex_server.friendlyName
|
||||
assert result["data"][CONF_SERVER] == mock_plex_server.friendlyName
|
||||
assert (
|
||||
result["data"][config_flow.CONF_SERVER_IDENTIFIER]
|
||||
== mock_plex_server.machineIdentifier
|
||||
result["data"][CONF_SERVER_IDENTIFIER] == mock_plex_server.machineIdentifier
|
||||
)
|
||||
assert (
|
||||
result["data"][config_flow.PLEX_SERVER_CONFIG][CONF_URL]
|
||||
== mock_plex_server.url_in_use
|
||||
)
|
||||
assert result["data"][config_flow.PLEX_SERVER_CONFIG][CONF_TOKEN] == MOCK_TOKEN
|
||||
assert result["data"][PLEX_SERVER_CONFIG][CONF_URL] == mock_plex_server._baseurl
|
||||
assert result["data"][PLEX_SERVER_CONFIG][CONF_TOKEN] == MOCK_TOKEN
|
||||
|
||||
|
||||
async def test_already_configured(hass):
|
||||
@ -313,23 +277,19 @@ async def test_already_configured(hass):
|
||||
mock_plex_server = MockPlexServer()
|
||||
|
||||
MockConfigEntry(
|
||||
domain=config_flow.DOMAIN,
|
||||
domain=DOMAIN,
|
||||
data={
|
||||
config_flow.CONF_SERVER: MOCK_SERVERS[0][config_flow.CONF_SERVER],
|
||||
config_flow.CONF_SERVER_IDENTIFIER: MOCK_SERVERS[0][
|
||||
config_flow.CONF_SERVER_IDENTIFIER
|
||||
],
|
||||
CONF_SERVER: MOCK_SERVERS[0][CONF_SERVER],
|
||||
CONF_SERVER_IDENTIFIER: MOCK_SERVERS[0][CONF_SERVER_IDENTIFIER],
|
||||
},
|
||||
unique_id=MOCK_SERVERS[0][config_flow.CONF_SERVER_IDENTIFIER],
|
||||
unique_id=MOCK_SERVERS[0][CONF_SERVER_IDENTIFIER],
|
||||
).add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
"plexapi.server.PlexServer", return_value=mock_plex_server
|
||||
), asynctest.patch("plexauth.PlexAuth.initiate_auth"), asynctest.patch(
|
||||
"plexauth.PlexAuth.token", return_value=MOCK_TOKEN
|
||||
):
|
||||
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
|
||||
"plexauth.PlexAuth.initiate_auth"
|
||||
), patch("plexauth.PlexAuth.token", return_value=MOCK_TOKEN):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN,
|
||||
DOMAIN,
|
||||
context={"source": "import"},
|
||||
data={
|
||||
CONF_TOKEN: MOCK_TOKEN,
|
||||
@ -346,34 +306,30 @@ async def test_all_available_servers_configured(hass):
|
||||
await async_setup_component(hass, "http", {"http": {}})
|
||||
|
||||
MockConfigEntry(
|
||||
domain=config_flow.DOMAIN,
|
||||
domain=DOMAIN,
|
||||
data={
|
||||
config_flow.CONF_SERVER_IDENTIFIER: MOCK_SERVERS[0][
|
||||
config_flow.CONF_SERVER_IDENTIFIER
|
||||
],
|
||||
config_flow.CONF_SERVER: MOCK_SERVERS[0][config_flow.CONF_SERVER],
|
||||
CONF_SERVER_IDENTIFIER: MOCK_SERVERS[0][CONF_SERVER_IDENTIFIER],
|
||||
CONF_SERVER: MOCK_SERVERS[0][CONF_SERVER],
|
||||
},
|
||||
).add_to_hass(hass)
|
||||
|
||||
MockConfigEntry(
|
||||
domain=config_flow.DOMAIN,
|
||||
domain=DOMAIN,
|
||||
data={
|
||||
config_flow.CONF_SERVER_IDENTIFIER: MOCK_SERVERS[1][
|
||||
config_flow.CONF_SERVER_IDENTIFIER
|
||||
],
|
||||
config_flow.CONF_SERVER: MOCK_SERVERS[1][config_flow.CONF_SERVER],
|
||||
CONF_SERVER_IDENTIFIER: MOCK_SERVERS[1][CONF_SERVER_IDENTIFIER],
|
||||
CONF_SERVER: MOCK_SERVERS[1][CONF_SERVER],
|
||||
},
|
||||
).add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": "user"}
|
||||
DOMAIN, context={"source": "user"}
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "start_website_auth"
|
||||
|
||||
with patch(
|
||||
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount(servers=2)
|
||||
), asynctest.patch("plexauth.PlexAuth.initiate_auth"), asynctest.patch(
|
||||
), patch("plexauth.PlexAuth.initiate_auth"), patch(
|
||||
"plexauth.PlexAuth.token", return_value=MOCK_TOKEN
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"])
|
||||
@ -389,20 +345,26 @@ async def test_all_available_servers_configured(hass):
|
||||
|
||||
async def test_option_flow(hass):
|
||||
"""Test config options flow selection."""
|
||||
|
||||
mock_plex_server = MockPlexServer(load_users=False)
|
||||
|
||||
MOCK_SERVER_ID = MOCK_SERVERS[0][config_flow.CONF_SERVER_IDENTIFIER]
|
||||
hass.data[config_flow.DOMAIN] = {
|
||||
config_flow.SERVERS: {MOCK_SERVER_ID: mock_plex_server}
|
||||
}
|
||||
mock_plex_server = MockPlexServer()
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=config_flow.DOMAIN,
|
||||
data={config_flow.CONF_SERVER_IDENTIFIER: MOCK_SERVER_ID},
|
||||
domain=DOMAIN,
|
||||
data=DEFAULT_DATA,
|
||||
options=DEFAULT_OPTIONS,
|
||||
unique_id=DEFAULT_DATA["server_id"],
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
|
||||
"homeassistant.components.plex.PlexWebsocket.listen"
|
||||
) as mock_listen:
|
||||
entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_listen.called
|
||||
|
||||
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
|
||||
assert entry.state == ENTRY_STATE_LOADED
|
||||
|
||||
result = await hass.config_entries.options.async_init(
|
||||
entry.entry_id, context={"source": "test"}, data=None
|
||||
@ -413,112 +375,69 @@ async def test_option_flow(hass):
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
config_flow.CONF_USE_EPISODE_ART: True,
|
||||
config_flow.CONF_IGNORE_NEW_SHARED_USERS: True,
|
||||
config_flow.CONF_MONITORED_USERS: list(mock_plex_server.accounts),
|
||||
CONF_USE_EPISODE_ART: True,
|
||||
CONF_IGNORE_NEW_SHARED_USERS: True,
|
||||
CONF_MONITORED_USERS: list(mock_plex_server.accounts),
|
||||
},
|
||||
)
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["data"] == {
|
||||
config_flow.MP_DOMAIN: {
|
||||
config_flow.CONF_USE_EPISODE_ART: True,
|
||||
config_flow.CONF_IGNORE_NEW_SHARED_USERS: True,
|
||||
config_flow.CONF_MONITORED_USERS: {
|
||||
MP_DOMAIN: {
|
||||
CONF_USE_EPISODE_ART: True,
|
||||
CONF_IGNORE_NEW_SHARED_USERS: True,
|
||||
CONF_MONITORED_USERS: {
|
||||
user: {"enabled": True} for user in mock_plex_server.accounts
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
async def test_option_flow_loading_saved_users(hass):
|
||||
"""Test config options flow selection when loading existing user config."""
|
||||
async def test_option_flow_new_users_available(hass, caplog):
|
||||
"""Test config options multiselect defaults when new Plex users are seen."""
|
||||
|
||||
mock_plex_server = MockPlexServer(load_users=True)
|
||||
|
||||
MOCK_SERVER_ID = MOCK_SERVERS[0][config_flow.CONF_SERVER_IDENTIFIER]
|
||||
hass.data[config_flow.DOMAIN] = {
|
||||
config_flow.SERVERS: {MOCK_SERVER_ID: mock_plex_server}
|
||||
}
|
||||
OPTIONS_OWNER_ONLY = copy.deepcopy(DEFAULT_OPTIONS)
|
||||
OPTIONS_OWNER_ONLY[MP_DOMAIN][CONF_MONITORED_USERS] = {"Owner": {"enabled": True}}
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=config_flow.DOMAIN,
|
||||
data={config_flow.CONF_SERVER_IDENTIFIER: MOCK_SERVER_ID},
|
||||
options=DEFAULT_OPTIONS,
|
||||
domain=DOMAIN,
|
||||
data=DEFAULT_DATA,
|
||||
options=OPTIONS_OWNER_ONLY,
|
||||
unique_id=DEFAULT_DATA["server_id"],
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
mock_plex_server = MockPlexServer(config_entry=entry)
|
||||
|
||||
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
|
||||
"homeassistant.components.plex.PlexWebsocket.listen"
|
||||
):
|
||||
entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
server_id = mock_plex_server.machineIdentifier
|
||||
|
||||
async_dispatcher_send(hass, PLEX_UPDATE_PLATFORMS_SIGNAL.format(server_id))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
monitored_users = hass.data[DOMAIN][SERVERS][server_id].option_monitored_users
|
||||
|
||||
new_users = [x for x in mock_plex_server.accounts if x not in monitored_users]
|
||||
assert len(monitored_users) == 1
|
||||
assert len(new_users) == 2
|
||||
|
||||
sensor = hass.states.get("sensor.plex_plex_server_1")
|
||||
assert sensor.state == str(len(mock_plex_server.accounts))
|
||||
|
||||
result = await hass.config_entries.options.async_init(
|
||||
entry.entry_id, context={"source": "test"}, data=None
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "plex_mp_settings"
|
||||
multiselect_defaults = result["data_schema"].schema["monitored_users"].options
|
||||
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
config_flow.CONF_USE_EPISODE_ART: True,
|
||||
config_flow.CONF_IGNORE_NEW_SHARED_USERS: True,
|
||||
config_flow.CONF_MONITORED_USERS: list(mock_plex_server.accounts),
|
||||
},
|
||||
)
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["data"] == {
|
||||
config_flow.MP_DOMAIN: {
|
||||
config_flow.CONF_USE_EPISODE_ART: True,
|
||||
config_flow.CONF_IGNORE_NEW_SHARED_USERS: True,
|
||||
config_flow.CONF_MONITORED_USERS: {
|
||||
user: {"enabled": True} for user in mock_plex_server.accounts
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
async def test_option_flow_new_users_available(hass):
|
||||
"""Test config options flow selection when new Plex accounts available."""
|
||||
|
||||
mock_plex_server = MockPlexServer(load_users=True, num_users=2)
|
||||
|
||||
MOCK_SERVER_ID = MOCK_SERVERS[0][config_flow.CONF_SERVER_IDENTIFIER]
|
||||
hass.data[config_flow.DOMAIN] = {
|
||||
config_flow.SERVERS: {MOCK_SERVER_ID: mock_plex_server}
|
||||
}
|
||||
|
||||
OPTIONS_WITH_USERS = copy.deepcopy(DEFAULT_OPTIONS)
|
||||
OPTIONS_WITH_USERS[config_flow.MP_DOMAIN][config_flow.CONF_MONITORED_USERS] = {
|
||||
"a": {"enabled": True}
|
||||
}
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=config_flow.DOMAIN,
|
||||
data={config_flow.CONF_SERVER_IDENTIFIER: MOCK_SERVER_ID},
|
||||
options=OPTIONS_WITH_USERS,
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.options.async_init(
|
||||
entry.entry_id, context={"source": "test"}, data=None
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "plex_mp_settings"
|
||||
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
config_flow.CONF_USE_EPISODE_ART: True,
|
||||
config_flow.CONF_IGNORE_NEW_SHARED_USERS: True,
|
||||
config_flow.CONF_MONITORED_USERS: list(mock_plex_server.accounts),
|
||||
},
|
||||
)
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["data"] == {
|
||||
config_flow.MP_DOMAIN: {
|
||||
config_flow.CONF_USE_EPISODE_ART: True,
|
||||
config_flow.CONF_IGNORE_NEW_SHARED_USERS: True,
|
||||
config_flow.CONF_MONITORED_USERS: {
|
||||
user: {"enabled": True} for user in mock_plex_server.accounts
|
||||
},
|
||||
}
|
||||
}
|
||||
assert "[Owner]" in multiselect_defaults["Owner"]
|
||||
for user in new_users:
|
||||
assert "[New]" in multiselect_defaults[user]
|
||||
|
||||
|
||||
async def test_external_timed_out(hass):
|
||||
@ -527,12 +446,12 @@ async def test_external_timed_out(hass):
|
||||
await async_setup_component(hass, "http", {"http": {}})
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": "user"}
|
||||
DOMAIN, context={"source": "user"}
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "start_website_auth"
|
||||
|
||||
with asynctest.patch("plexauth.PlexAuth.initiate_auth"), asynctest.patch(
|
||||
with patch("plexauth.PlexAuth.initiate_auth"), patch(
|
||||
"plexauth.PlexAuth.token", return_value=None
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"])
|
||||
@ -552,12 +471,12 @@ async def test_callback_view(hass, aiohttp_client):
|
||||
await async_setup_component(hass, "http", {"http": {}})
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": "user"}
|
||||
DOMAIN, context={"source": "user"}
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "start_website_auth"
|
||||
|
||||
with asynctest.patch("plexauth.PlexAuth.initiate_auth"), asynctest.patch(
|
||||
with patch("plexauth.PlexAuth.initiate_auth"), patch(
|
||||
"plexauth.PlexAuth.token", return_value=MOCK_TOKEN
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"])
|
||||
@ -575,13 +494,11 @@ async def test_multiple_servers_with_import(hass):
|
||||
|
||||
with patch(
|
||||
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount(servers=2)
|
||||
), asynctest.patch("plexauth.PlexAuth.initiate_auth"), asynctest.patch(
|
||||
), patch("plexauth.PlexAuth.initiate_auth"), patch(
|
||||
"plexauth.PlexAuth.token", return_value=MOCK_TOKEN
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN,
|
||||
context={"source": "import"},
|
||||
data={CONF_TOKEN: MOCK_TOKEN},
|
||||
DOMAIN, context={"source": "import"}, data={CONF_TOKEN: MOCK_TOKEN},
|
||||
)
|
||||
assert result["type"] == "abort"
|
||||
assert result["reason"] == "non-interactive"
|
||||
|
||||
302
tests/components/plex/test_init.py
Normal file
302
tests/components/plex/test_init.py
Normal file
@ -0,0 +1,302 @@
|
||||
"""Tests for Plex setup."""
|
||||
import copy
|
||||
from datetime import timedelta
|
||||
|
||||
from asynctest import patch
|
||||
import plexapi
|
||||
import requests
|
||||
|
||||
from homeassistant.components.media_player import DOMAIN as MP_DOMAIN
|
||||
import homeassistant.components.plex.const as const
|
||||
from homeassistant.config_entries import (
|
||||
ENTRY_STATE_LOADED,
|
||||
ENTRY_STATE_NOT_LOADED,
|
||||
ENTRY_STATE_SETUP_ERROR,
|
||||
ENTRY_STATE_SETUP_RETRY,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_PORT,
|
||||
CONF_SSL,
|
||||
CONF_TOKEN,
|
||||
CONF_VERIFY_SSL,
|
||||
)
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_send
|
||||
from homeassistant.setup import async_setup_component
|
||||
import homeassistant.util.dt as dt_util
|
||||
|
||||
from .const import DEFAULT_DATA, DEFAULT_OPTIONS, MOCK_SERVERS, MOCK_TOKEN
|
||||
from .mock_classes import MockPlexServer
|
||||
|
||||
from tests.common import MockConfigEntry, async_fire_time_changed
|
||||
|
||||
|
||||
async def test_setup_with_config(hass):
|
||||
"""Test setup component with config."""
|
||||
config = {
|
||||
const.DOMAIN: {
|
||||
CONF_HOST: MOCK_SERVERS[0][CONF_HOST],
|
||||
CONF_PORT: MOCK_SERVERS[0][CONF_PORT],
|
||||
CONF_TOKEN: MOCK_TOKEN,
|
||||
CONF_SSL: True,
|
||||
CONF_VERIFY_SSL: True,
|
||||
MP_DOMAIN: {
|
||||
const.CONF_IGNORE_NEW_SHARED_USERS: False,
|
||||
const.CONF_USE_EPISODE_ART: False,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
mock_plex_server = MockPlexServer()
|
||||
|
||||
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
|
||||
"homeassistant.components.plex.PlexWebsocket.listen"
|
||||
) as mock_listen:
|
||||
assert await async_setup_component(hass, const.DOMAIN, config) is True
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_listen.called
|
||||
assert len(hass.config_entries.async_entries(const.DOMAIN)) == 1
|
||||
entry = hass.config_entries.async_entries(const.DOMAIN)[0]
|
||||
assert entry.state == ENTRY_STATE_LOADED
|
||||
|
||||
server_id = mock_plex_server.machineIdentifier
|
||||
loaded_server = hass.data[const.DOMAIN][const.SERVERS][server_id]
|
||||
|
||||
assert loaded_server.plex_server == mock_plex_server
|
||||
|
||||
assert server_id in hass.data[const.DOMAIN][const.DISPATCHERS]
|
||||
assert server_id in hass.data[const.DOMAIN][const.WEBSOCKETS]
|
||||
assert (
|
||||
hass.data[const.DOMAIN][const.PLATFORMS_COMPLETED][server_id] == const.PLATFORMS
|
||||
)
|
||||
|
||||
|
||||
async def test_setup_with_config_entry(hass):
|
||||
"""Test setup component with config."""
|
||||
|
||||
mock_plex_server = MockPlexServer()
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=const.DOMAIN,
|
||||
data=DEFAULT_DATA,
|
||||
options=DEFAULT_OPTIONS,
|
||||
unique_id=DEFAULT_DATA["server_id"],
|
||||
)
|
||||
|
||||
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
|
||||
"homeassistant.components.plex.PlexWebsocket.listen"
|
||||
) as mock_listen:
|
||||
entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_listen.called
|
||||
|
||||
assert len(hass.config_entries.async_entries(const.DOMAIN)) == 1
|
||||
assert entry.state == ENTRY_STATE_LOADED
|
||||
|
||||
server_id = mock_plex_server.machineIdentifier
|
||||
loaded_server = hass.data[const.DOMAIN][const.SERVERS][server_id]
|
||||
|
||||
assert loaded_server.plex_server == mock_plex_server
|
||||
|
||||
assert server_id in hass.data[const.DOMAIN][const.DISPATCHERS]
|
||||
assert server_id in hass.data[const.DOMAIN][const.WEBSOCKETS]
|
||||
assert (
|
||||
hass.data[const.DOMAIN][const.PLATFORMS_COMPLETED][server_id] == const.PLATFORMS
|
||||
)
|
||||
|
||||
async_dispatcher_send(hass, const.PLEX_UPDATE_PLATFORMS_SIGNAL.format(server_id))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
sensor = hass.states.get("sensor.plex_plex_server_1")
|
||||
assert sensor.state == str(len(mock_plex_server.accounts))
|
||||
|
||||
async_dispatcher_send(hass, const.PLEX_UPDATE_PLATFORMS_SIGNAL.format(server_id))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
with patch.object(
|
||||
mock_plex_server, "clients", side_effect=plexapi.exceptions.BadRequest
|
||||
):
|
||||
async_dispatcher_send(
|
||||
hass, const.PLEX_UPDATE_PLATFORMS_SIGNAL.format(server_id)
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
with patch.object(
|
||||
mock_plex_server, "clients", side_effect=requests.exceptions.RequestException
|
||||
):
|
||||
async_dispatcher_send(
|
||||
hass, const.PLEX_UPDATE_PLATFORMS_SIGNAL.format(server_id)
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
async def test_set_config_entry_unique_id(hass):
|
||||
"""Test updating missing unique_id from config entry."""
|
||||
|
||||
mock_plex_server = MockPlexServer()
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=const.DOMAIN, data=DEFAULT_DATA, options=DEFAULT_OPTIONS, unique_id=None,
|
||||
)
|
||||
|
||||
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
|
||||
"homeassistant.components.plex.PlexWebsocket.listen"
|
||||
) as mock_listen:
|
||||
entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_listen.called
|
||||
|
||||
assert len(hass.config_entries.async_entries(const.DOMAIN)) == 1
|
||||
assert entry.state == ENTRY_STATE_LOADED
|
||||
|
||||
assert (
|
||||
hass.config_entries.async_entries(const.DOMAIN)[0].unique_id
|
||||
== mock_plex_server.machineIdentifier
|
||||
)
|
||||
|
||||
|
||||
async def test_setup_config_entry_with_error(hass):
|
||||
"""Test setup component from config entry with errors."""
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=const.DOMAIN,
|
||||
data=DEFAULT_DATA,
|
||||
options=DEFAULT_OPTIONS,
|
||||
unique_id=DEFAULT_DATA["server_id"],
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.plex.PlexServer.connect",
|
||||
side_effect=requests.exceptions.ConnectionError,
|
||||
):
|
||||
entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id) is False
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.config_entries.async_entries(const.DOMAIN)) == 1
|
||||
assert entry.state == ENTRY_STATE_SETUP_RETRY
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.plex.PlexServer.connect",
|
||||
side_effect=plexapi.exceptions.BadRequest,
|
||||
):
|
||||
next_update = dt_util.utcnow() + timedelta(seconds=30)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.config_entries.async_entries(const.DOMAIN)) == 1
|
||||
assert entry.state == ENTRY_STATE_SETUP_ERROR
|
||||
|
||||
|
||||
async def test_setup_with_insecure_config_entry(hass):
|
||||
"""Test setup component with config."""
|
||||
|
||||
mock_plex_server = MockPlexServer()
|
||||
|
||||
INSECURE_DATA = copy.deepcopy(DEFAULT_DATA)
|
||||
INSECURE_DATA[const.PLEX_SERVER_CONFIG][CONF_VERIFY_SSL] = False
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=const.DOMAIN,
|
||||
data=INSECURE_DATA,
|
||||
options=DEFAULT_OPTIONS,
|
||||
unique_id=DEFAULT_DATA["server_id"],
|
||||
)
|
||||
|
||||
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
|
||||
"homeassistant.components.plex.PlexWebsocket.listen"
|
||||
) as mock_listen:
|
||||
entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_listen.called
|
||||
|
||||
assert len(hass.config_entries.async_entries(const.DOMAIN)) == 1
|
||||
assert entry.state == ENTRY_STATE_LOADED
|
||||
|
||||
|
||||
async def test_unload_config_entry(hass):
|
||||
"""Test unloading a config entry."""
|
||||
mock_plex_server = MockPlexServer()
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=const.DOMAIN,
|
||||
data=DEFAULT_DATA,
|
||||
options=DEFAULT_OPTIONS,
|
||||
unique_id=DEFAULT_DATA["server_id"],
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
config_entries = hass.config_entries.async_entries(const.DOMAIN)
|
||||
assert len(config_entries) == 1
|
||||
assert entry is config_entries[0]
|
||||
|
||||
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
|
||||
"homeassistant.components.plex.PlexWebsocket.listen"
|
||||
) as mock_listen:
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert mock_listen.called
|
||||
|
||||
assert entry.state == ENTRY_STATE_LOADED
|
||||
|
||||
server_id = mock_plex_server.machineIdentifier
|
||||
loaded_server = hass.data[const.DOMAIN][const.SERVERS][server_id]
|
||||
|
||||
assert loaded_server.plex_server == mock_plex_server
|
||||
|
||||
assert server_id in hass.data[const.DOMAIN][const.DISPATCHERS]
|
||||
assert server_id in hass.data[const.DOMAIN][const.WEBSOCKETS]
|
||||
assert (
|
||||
hass.data[const.DOMAIN][const.PLATFORMS_COMPLETED][server_id] == const.PLATFORMS
|
||||
)
|
||||
|
||||
with patch("homeassistant.components.plex.PlexWebsocket.close") as mock_close:
|
||||
await hass.config_entries.async_unload(entry.entry_id)
|
||||
assert mock_close.called
|
||||
|
||||
assert entry.state == ENTRY_STATE_NOT_LOADED
|
||||
|
||||
assert server_id not in hass.data[const.DOMAIN][const.SERVERS]
|
||||
assert server_id not in hass.data[const.DOMAIN][const.DISPATCHERS]
|
||||
assert server_id not in hass.data[const.DOMAIN][const.WEBSOCKETS]
|
||||
|
||||
|
||||
async def test_setup_with_photo_session(hass):
|
||||
"""Test setup component with config."""
|
||||
|
||||
mock_plex_server = MockPlexServer(session_type="photo")
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=const.DOMAIN,
|
||||
data=DEFAULT_DATA,
|
||||
options=DEFAULT_OPTIONS,
|
||||
unique_id=DEFAULT_DATA["server_id"],
|
||||
)
|
||||
|
||||
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
|
||||
"homeassistant.components.plex.PlexWebsocket.listen"
|
||||
):
|
||||
entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.config_entries.async_entries(const.DOMAIN)) == 1
|
||||
assert entry.state == ENTRY_STATE_LOADED
|
||||
|
||||
server_id = mock_plex_server.machineIdentifier
|
||||
|
||||
async_dispatcher_send(hass, const.PLEX_UPDATE_PLATFORMS_SIGNAL.format(server_id))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
media_player = hass.states.get("media_player.plex_product_title")
|
||||
assert media_player.state == "idle"
|
||||
|
||||
sensor = hass.states.get("sensor.plex_plex_server_1")
|
||||
assert sensor.state == str(len(mock_plex_server.accounts))
|
||||
134
tests/components/plex/test_server.py
Normal file
134
tests/components/plex/test_server.py
Normal file
@ -0,0 +1,134 @@
|
||||
"""Tests for Plex server."""
|
||||
import copy
|
||||
|
||||
from asynctest import patch
|
||||
|
||||
from homeassistant.components.media_player import DOMAIN as MP_DOMAIN
|
||||
from homeassistant.components.plex.const import (
|
||||
CONF_IGNORE_NEW_SHARED_USERS,
|
||||
CONF_MONITORED_USERS,
|
||||
DOMAIN,
|
||||
PLEX_UPDATE_PLATFORMS_SIGNAL,
|
||||
SERVERS,
|
||||
)
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_send
|
||||
|
||||
from .const import DEFAULT_DATA, DEFAULT_OPTIONS
|
||||
from .mock_classes import MockPlexServer
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_new_users_available(hass):
|
||||
"""Test setting up when new users available on Plex server."""
|
||||
|
||||
MONITORED_USERS = {"Owner": {"enabled": True}}
|
||||
OPTIONS_WITH_USERS = copy.deepcopy(DEFAULT_OPTIONS)
|
||||
OPTIONS_WITH_USERS[MP_DOMAIN][CONF_MONITORED_USERS] = MONITORED_USERS
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data=DEFAULT_DATA,
|
||||
options=OPTIONS_WITH_USERS,
|
||||
unique_id=DEFAULT_DATA["server_id"],
|
||||
)
|
||||
|
||||
mock_plex_server = MockPlexServer(config_entry=entry)
|
||||
|
||||
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
|
||||
"homeassistant.components.plex.PlexWebsocket.listen"
|
||||
):
|
||||
entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
server_id = mock_plex_server.machineIdentifier
|
||||
|
||||
async_dispatcher_send(hass, PLEX_UPDATE_PLATFORMS_SIGNAL.format(server_id))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
monitored_users = hass.data[DOMAIN][SERVERS][server_id].option_monitored_users
|
||||
|
||||
ignored_users = [x for x in monitored_users if not monitored_users[x]["enabled"]]
|
||||
assert len(monitored_users) == 1
|
||||
assert len(ignored_users) == 0
|
||||
|
||||
sensor = hass.states.get("sensor.plex_plex_server_1")
|
||||
assert sensor.state == str(len(mock_plex_server.accounts))
|
||||
|
||||
|
||||
async def test_new_ignored_users_available(hass, caplog):
|
||||
"""Test setting up when new users available on Plex server but are ignored."""
|
||||
|
||||
MONITORED_USERS = {"Owner": {"enabled": True}}
|
||||
OPTIONS_WITH_USERS = copy.deepcopy(DEFAULT_OPTIONS)
|
||||
OPTIONS_WITH_USERS[MP_DOMAIN][CONF_MONITORED_USERS] = MONITORED_USERS
|
||||
OPTIONS_WITH_USERS[MP_DOMAIN][CONF_IGNORE_NEW_SHARED_USERS] = True
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data=DEFAULT_DATA,
|
||||
options=OPTIONS_WITH_USERS,
|
||||
unique_id=DEFAULT_DATA["server_id"],
|
||||
)
|
||||
|
||||
mock_plex_server = MockPlexServer(config_entry=entry)
|
||||
|
||||
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
|
||||
"homeassistant.components.plex.PlexWebsocket.listen"
|
||||
):
|
||||
entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
server_id = mock_plex_server.machineIdentifier
|
||||
|
||||
async_dispatcher_send(hass, PLEX_UPDATE_PLATFORMS_SIGNAL.format(server_id))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
monitored_users = hass.data[DOMAIN][SERVERS][server_id].option_monitored_users
|
||||
|
||||
ignored_users = [x for x in mock_plex_server.accounts if x not in monitored_users]
|
||||
assert len(monitored_users) == 1
|
||||
assert len(ignored_users) == 2
|
||||
for ignored_user in ignored_users:
|
||||
assert f"Ignoring Plex client owned by {ignored_user}" in caplog.text
|
||||
|
||||
sensor = hass.states.get("sensor.plex_plex_server_1")
|
||||
assert sensor.state == str(len(mock_plex_server.accounts))
|
||||
|
||||
|
||||
async def test_mark_sessions_idle(hass):
|
||||
"""Test marking media_players as idle when sessions end."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data=DEFAULT_DATA,
|
||||
options=DEFAULT_OPTIONS,
|
||||
unique_id=DEFAULT_DATA["server_id"],
|
||||
)
|
||||
|
||||
mock_plex_server = MockPlexServer(config_entry=entry)
|
||||
|
||||
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
|
||||
"homeassistant.components.plex.PlexWebsocket.listen"
|
||||
):
|
||||
entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
server_id = mock_plex_server.machineIdentifier
|
||||
|
||||
async_dispatcher_send(hass, PLEX_UPDATE_PLATFORMS_SIGNAL.format(server_id))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
sensor = hass.states.get("sensor.plex_plex_server_1")
|
||||
assert sensor.state == str(len(mock_plex_server.accounts))
|
||||
|
||||
mock_plex_server.clear_clients()
|
||||
mock_plex_server.clear_sessions()
|
||||
|
||||
async_dispatcher_send(hass, PLEX_UPDATE_PLATFORMS_SIGNAL.format(server_id))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
sensor = hass.states.get("sensor.plex_plex_server_1")
|
||||
assert sensor.state == "0"
|
||||
Loading…
x
Reference in New Issue
Block a user