jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-26 22:57:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor Plex tests using fixtures (#40260)

Browse Source 
* Refactor Plex tests using fixtures

* Avoid unnecessary coroutine declaration
This commit is contained in:
jjlawren 2020-09-22 22:14:41 -05:00 committed by GitHub
parent c6a48d3b61
commit df9634a41f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 121 additions and 439 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
tests/components/plex/conftest.py Normal file
View File

@ -0,0 +1,59 @@
"""Fixtures for Plex tests."""
import pytest
from homeassistant.components.plex.const import DOMAIN
from .const import DEFAULT_DATA, DEFAULT_OPTIONS
from .mock_classes import MockPlexAccount, MockPlexServer
from tests.async_mock import patch
from tests.common import MockConfigEntry
@pytest.fixture(name="entry")
def mock_config_entry():
"""Return the default mocked config entry."""
return MockConfigEntry(
domain=DOMAIN,
data=DEFAULT_DATA,
options=DEFAULT_OPTIONS,
unique_id=DEFAULT_DATA["server_id"],
)
@pytest.fixture
def mock_plex_account():
"""Mock the PlexAccount class and return the used instance."""
plex_account = MockPlexAccount()
with patch("plexapi.myplex.MyPlexAccount", return_value=plex_account):
yield plex_account
@pytest.fixture
def mock_websocket():
"""Mock the PlexWebsocket class."""
with patch("homeassistant.components.plex.PlexWebsocket", autospec=True) as ws:
yield ws
@pytest.fixture
def setup_plex_server(hass, entry, mock_plex_account, mock_websocket):
"""Set up and return a mocked Plex server instance."""
async def _wrapper(**kwargs):
"""Wrap the fixture to allow passing arguments to the MockPlexServer instance."""
config_entry = kwargs.get("config_entry", entry)
plex_server = MockPlexServer(**kwargs)
with patch("plexapi.server.PlexServer", return_value=plex_server):
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return plex_server
return _wrapper
@pytest.fixture
async def mock_plex_server(entry, setup_plex_server):
"""Init from a config entry and return a mocked PlexServer instance."""
return await setup_plex_server(config_entry=entry)

29
tests/components/plex/test_browse_media.py
View File

@ -3,39 +3,16 @@ from homeassistant.components.media_player.const import (
ATTR_MEDIA_CONTENT_ID, ATTR_MEDIA_CONTENT_ID,
ATTR_MEDIA_CONTENT_TYPE, ATTR_MEDIA_CONTENT_TYPE,
) )
from homeassistant.components.plex.const import CONF_SERVER_IDENTIFIER, DOMAIN from homeassistant.components.plex.const import CONF_SERVER_IDENTIFIER
from homeassistant.components.plex.media_browser import SPECIAL_METHODS from homeassistant.components.plex.media_browser import SPECIAL_METHODS
from homeassistant.components.websocket_api.const import ERR_UNKNOWN_ERROR, TYPE_RESULT from homeassistant.components.websocket_api.const import ERR_UNKNOWN_ERROR, TYPE_RESULT
from .const import DEFAULT_DATA, DEFAULT_OPTIONS from .const import DEFAULT_DATA
from .helpers import trigger_plex_update from .helpers import trigger_plex_update
from .mock_classes import MockPlexAccount, MockPlexServer
from tests.async_mock import patch
from tests.common import MockConfigEntry
async def test_browse_media(hass, hass_ws_client): async def test_browse_media(hass, hass_ws_client, mock_plex_server, mock_websocket):
"""Test getting Plex clients from plex.tv.""" """Test getting Plex clients from plex.tv."""
entry = MockConfigEntry(
domain=DOMAIN,
data=DEFAULT_DATA,
options=DEFAULT_OPTIONS,
unique_id=DEFAULT_DATA["server_id"],
)
mock_plex_server = MockPlexServer(config_entry=entry)
mock_plex_account = MockPlexAccount()
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
"plexapi.myplex.MyPlexAccount", return_value=mock_plex_account
), patch(
"homeassistant.components.plex.PlexWebsocket", autospec=True
) as mock_websocket:
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
websocket_client = await hass_ws_client(hass) websocket_client = await hass_ws_client(hass)
trigger_plex_update(mock_websocket) trigger_plex_update(mock_websocket)

93
tests/components/plex/test_config_flow.py
View File

@ -34,7 +34,7 @@ from homeassistant.const import (
CONF_VERIFY_SSL, CONF_VERIFY_SSL,
) )
from .const import DEFAULT_DATA, DEFAULT_OPTIONS, MOCK_SERVERS, MOCK_TOKEN from .const import DEFAULT_OPTIONS, MOCK_SERVERS, MOCK_TOKEN
from .helpers import trigger_plex_update from .helpers import trigger_plex_update
from .mock_classes import MockGDM, MockPlexAccount, MockPlexServer, MockResource from .mock_classes import MockGDM, MockPlexAccount, MockPlexServer, MockResource
@ -77,8 +77,6 @@ async def test_bad_credentials(hass):
async def test_bad_hostname(hass): async def test_bad_hostname(hass):
"""Test when an invalid address is provided.""" """Test when an invalid address is provided."""
mock_plex_account = MockPlexAccount()
await async_process_ha_core_config( await async_process_ha_core_config(
hass, hass,
{"internal_url": "http://example.local:8123"}, {"internal_url": "http://example.local:8123"},
@ -91,7 +89,7 @@ async def test_bad_hostname(hass):
assert result["step_id"] == "user" assert result["step_id"] == "user"
with patch( with patch(
"plexapi.myplex.MyPlexAccount", return_value=mock_plex_account "plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()
), patch.object( ), patch.object(
MockResource, "connect", side_effect=requests.exceptions.ConnectionError MockResource, "connect", side_effect=requests.exceptions.ConnectionError
), patch( ), patch(
@ -363,24 +361,8 @@ async def test_all_available_servers_configured(hass):
assert result["reason"] == "all_configured" assert result["reason"] == "all_configured"
async def test_option_flow(hass): async def test_option_flow(hass, entry, mock_plex_server):
"""Test config options flow selection.""" """Test config options flow selection."""
mock_plex_server = MockPlexServer()
entry = MockConfigEntry(
domain=DOMAIN,
data=DEFAULT_DATA,
options=DEFAULT_OPTIONS,
unique_id=DEFAULT_DATA["server_id"],
)
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()
), patch("homeassistant.components.plex.PlexWebsocket", autospec=True):
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(DOMAIN)) == 1 assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert entry.state == ENTRY_STATE_LOADED assert entry.state == ENTRY_STATE_LOADED
@ -411,24 +393,8 @@ async def test_option_flow(hass):
} }
async def test_missing_option_flow(hass): async def test_missing_option_flow(hass, entry, mock_plex_server):
"""Test config options flow selection when no options stored.""" """Test config options flow selection when no options stored."""
mock_plex_server = MockPlexServer()
entry = MockConfigEntry(
domain=DOMAIN,
data=DEFAULT_DATA,
options=None,
unique_id=DEFAULT_DATA["server_id"],
)
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()
), patch("homeassistant.components.plex.PlexWebsocket", autospec=True):
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(DOMAIN)) == 1 assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert entry.state == ENTRY_STATE_LOADED assert entry.state == ENTRY_STATE_LOADED
@ -459,29 +425,15 @@ async def test_missing_option_flow(hass):
} }
async def test_option_flow_new_users_available(hass, caplog): async def test_option_flow_new_users_available(
hass, caplog, entry, mock_websocket, setup_plex_server
):
"""Test config options multiselect defaults when new Plex users are seen.""" """Test config options multiselect defaults when new Plex users are seen."""
OPTIONS_OWNER_ONLY = copy.deepcopy(DEFAULT_OPTIONS) OPTIONS_OWNER_ONLY = copy.deepcopy(DEFAULT_OPTIONS)
OPTIONS_OWNER_ONLY[MP_DOMAIN][CONF_MONITORED_USERS] = {"Owner": {"enabled": True}} OPTIONS_OWNER_ONLY[MP_DOMAIN][CONF_MONITORED_USERS] = {"Owner": {"enabled": True}}
entry.options = OPTIONS_OWNER_ONLY
entry = MockConfigEntry( mock_plex_server = await setup_plex_server(config_entry=entry)
domain=DOMAIN,
data=DEFAULT_DATA,
options=OPTIONS_OWNER_ONLY,
unique_id=DEFAULT_DATA["server_id"],
)
mock_plex_server = MockPlexServer(config_entry=entry)
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()
), patch(
"homeassistant.components.plex.PlexWebsocket", autospec=True
) as mock_websocket:
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
trigger_plex_update(mock_websocket) trigger_plex_update(mock_websocket)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -734,29 +686,12 @@ async def test_manual_config_with_token(hass):
assert result["data"][PLEX_SERVER_CONFIG][CONF_TOKEN] == MOCK_TOKEN assert result["data"][PLEX_SERVER_CONFIG][CONF_TOKEN] == MOCK_TOKEN
async def test_setup_with_limited_credentials(hass): async def test_setup_with_limited_credentials(hass, entry, setup_plex_server):
"""Test setup with a user with limited permissions.""" """Test setup with a user with limited permissions."""
mock_plex_server = MockPlexServer() with patch.object(
MockPlexServer, "systemAccounts", side_effect=plexapi.exceptions.Unauthorized
entry = MockConfigEntry( ) as mock_accounts:
domain=DOMAIN, mock_plex_server = await setup_plex_server()
data=DEFAULT_DATA,
options=DEFAULT_OPTIONS,
unique_id=DEFAULT_DATA["server_id"],
)
with patch(
"plexapi.server.PlexServer", return_value=mock_plex_server
), patch.object(
mock_plex_server, "systemAccounts", side_effect=plexapi.exceptions.Unauthorized
) as mock_accounts, patch(
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()
), patch(
"homeassistant.components.plex.PlexWebsocket", autospec=True
):
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert mock_accounts.called assert mock_accounts.called

143
tests/components/plex/test_init.py
View File

@ -24,27 +24,8 @@ from tests.async_mock import patch
from tests.common import MockConfigEntry, async_fire_time_changed from tests.common import MockConfigEntry, async_fire_time_changed
async def test_set_config_entry_unique_id(hass): async def test_set_config_entry_unique_id(hass, entry, mock_plex_server):
"""Test updating missing unique_id from config entry.""" """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(
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()
), 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 len(hass.config_entries.async_entries(const.DOMAIN)) == 1
assert entry.state == ENTRY_STATE_LOADED assert entry.state == ENTRY_STATE_LOADED
@ -54,16 +35,8 @@ async def test_set_config_entry_unique_id(hass):
) )
async def test_setup_config_entry_with_error(hass): async def test_setup_config_entry_with_error(hass, entry):
"""Test setup component from config entry with errors.""" """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( with patch(
"homeassistant.components.plex.PlexServer.connect", "homeassistant.components.plex.PlexServer.connect",
side_effect=requests.exceptions.ConnectionError, side_effect=requests.exceptions.ConnectionError,
@ -87,91 +60,38 @@ async def test_setup_config_entry_with_error(hass):
assert entry.state == ENTRY_STATE_SETUP_ERROR assert entry.state == ENTRY_STATE_SETUP_ERROR
async def test_setup_with_insecure_config_entry(hass): async def test_setup_with_insecure_config_entry(hass, entry, setup_plex_server):
"""Test setup component with config.""" """Test setup component with config."""
mock_plex_server = MockPlexServer()
INSECURE_DATA = copy.deepcopy(DEFAULT_DATA) INSECURE_DATA = copy.deepcopy(DEFAULT_DATA)
INSECURE_DATA[const.PLEX_SERVER_CONFIG][CONF_VERIFY_SSL] = False INSECURE_DATA[const.PLEX_SERVER_CONFIG][CONF_VERIFY_SSL] = False
entry.data = INSECURE_DATA
entry = MockConfigEntry( await setup_plex_server(config_entry=entry)
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(
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()
), 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 len(hass.config_entries.async_entries(const.DOMAIN)) == 1
assert entry.state == ENTRY_STATE_LOADED assert entry.state == ENTRY_STATE_LOADED
async def test_unload_config_entry(hass): async def test_unload_config_entry(hass, entry, mock_plex_server):
"""Test unloading a config entry.""" """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) config_entries = hass.config_entries.async_entries(const.DOMAIN)
assert len(config_entries) == 1 assert len(config_entries) == 1
assert entry is config_entries[0] assert entry is config_entries[0]
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()
), 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 assert entry.state == ENTRY_STATE_LOADED
server_id = mock_plex_server.machineIdentifier server_id = mock_plex_server.machineIdentifier
loaded_server = hass.data[const.DOMAIN][const.SERVERS][server_id] loaded_server = hass.data[const.DOMAIN][const.SERVERS][server_id]
assert loaded_server.plex_server == mock_plex_server assert loaded_server.plex_server == mock_plex_server
with patch("homeassistant.components.plex.PlexWebsocket.close") as mock_close: websocket = hass.data[const.DOMAIN][const.WEBSOCKETS][server_id]
await hass.config_entries.async_unload(entry.entry_id) await hass.config_entries.async_unload(entry.entry_id)
assert mock_close.called assert websocket.close.called
assert entry.state == ENTRY_STATE_NOT_LOADED assert entry.state == ENTRY_STATE_NOT_LOADED
async def test_setup_with_photo_session(hass): async def test_setup_with_photo_session(hass, entry, mock_websocket, setup_plex_server):
"""Test setup component with config.""" """Test setup component with config."""
mock_plex_server = await setup_plex_server(config_entry=entry, session_type="photo")
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(
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()
), patch(
"homeassistant.components.plex.PlexWebsocket", autospec=True
) as mock_websocket:
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 len(hass.config_entries.async_entries(const.DOMAIN)) == 1
assert entry.state == ENTRY_STATE_LOADED assert entry.state == ENTRY_STATE_LOADED
@ -186,7 +106,7 @@ async def test_setup_with_photo_session(hass):
assert sensor.state == str(len(mock_plex_server.accounts)) assert sensor.state == str(len(mock_plex_server.accounts))
async def test_setup_when_certificate_changed(hass): async def test_setup_when_certificate_changed(hass, entry):
"""Test setup component when the Plex certificate has changed.""" """Test setup component when the Plex certificate has changed."""
old_domain = "1-2-3-4.1234567890abcdef1234567890abcdef.plex.direct" old_domain = "1-2-3-4.1234567890abcdef1234567890abcdef.plex.direct"
@ -210,8 +130,6 @@ async def test_setup_when_certificate_changed(hass):
unique_id=DEFAULT_DATA["server_id"], unique_id=DEFAULT_DATA["server_id"],
) )
new_entry = MockConfigEntry(domain=const.DOMAIN, data=DEFAULT_DATA)
# Test with account failure # Test with account failure
with patch( with patch(
"plexapi.server.PlexServer", side_effect=WrongCertHostnameException "plexapi.server.PlexServer", side_effect=WrongCertHostnameException
@ -247,49 +165,23 @@ async def test_setup_when_certificate_changed(hass):
assert ( assert (
old_entry.data[const.PLEX_SERVER_CONFIG][CONF_URL] old_entry.data[const.PLEX_SERVER_CONFIG][CONF_URL]
== new_entry.data[const.PLEX_SERVER_CONFIG][CONF_URL] == entry.data[const.PLEX_SERVER_CONFIG][CONF_URL]
) )
async def test_tokenless_server(hass): async def test_tokenless_server(hass, entry, mock_websocket, setup_plex_server):
"""Test setup with a server with token auth disabled.""" """Test setup with a server with token auth disabled."""
mock_plex_server = MockPlexServer()
TOKENLESS_DATA = copy.deepcopy(DEFAULT_DATA) TOKENLESS_DATA = copy.deepcopy(DEFAULT_DATA)
TOKENLESS_DATA[const.PLEX_SERVER_CONFIG].pop(CONF_TOKEN, None) TOKENLESS_DATA[const.PLEX_SERVER_CONFIG].pop(CONF_TOKEN, None)
entry.data = TOKENLESS_DATA
entry = MockConfigEntry( await setup_plex_server(config_entry=entry)
domain=const.DOMAIN,
data=TOKENLESS_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", autospec=True
) as mock_websocket:
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ENTRY_STATE_LOADED assert entry.state == ENTRY_STATE_LOADED
trigger_plex_update(mock_websocket)
await hass.async_block_till_done()
async def test_bad_token_with_tokenless_server(hass, entry):
async def test_bad_token_with_tokenless_server(hass):
"""Test setup with a bad token and a server with token auth disabled.""" """Test setup with a bad token and a server with token auth disabled."""
mock_plex_server = MockPlexServer() with patch("plexapi.server.PlexServer", return_value=MockPlexServer()), patch(
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(
"plexapi.myplex.MyPlexAccount", side_effect=plexapi.exceptions.Unauthorized "plexapi.myplex.MyPlexAccount", side_effect=plexapi.exceptions.Unauthorized
), patch( ), patch(
"homeassistant.components.plex.PlexWebsocket", autospec=True "homeassistant.components.plex.PlexWebsocket", autospec=True
@ -300,5 +192,6 @@ async def test_bad_token_with_tokenless_server(hass):
assert entry.state == ENTRY_STATE_LOADED assert entry.state == ENTRY_STATE_LOADED
# Ensure updates that rely on account return nothing
trigger_plex_update(mock_websocket) trigger_plex_update(mock_websocket)
await hass.async_block_till_done() await hass.async_block_till_done()

38
tests/components/plex/test_media_players.py
View File

@ -3,32 +3,12 @@ from plexapi.exceptions import NotFound
from homeassistant.components.plex.const import DOMAIN, SERVERS from homeassistant.components.plex.const import DOMAIN, SERVERS
from .const import DEFAULT_DATA, DEFAULT_OPTIONS
from .mock_classes import MockPlexAccount, MockPlexServer
from tests.async_mock import patch from tests.async_mock import patch
from tests.common import MockConfigEntry
async def test_plex_tv_clients(hass): async def test_plex_tv_clients(hass, entry, mock_plex_account, setup_plex_server):
"""Test getting Plex clients from plex.tv.""" """Test getting Plex clients from plex.tv."""
entry = MockConfigEntry( mock_plex_server = await setup_plex_server()
domain=DOMAIN,
data=DEFAULT_DATA,
options=DEFAULT_OPTIONS,
unique_id=DEFAULT_DATA["server_id"],
)
mock_plex_server = MockPlexServer(config_entry=entry)
mock_plex_account = MockPlexAccount()
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
"plexapi.myplex.MyPlexAccount", return_value=mock_plex_account
), 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 server_id = mock_plex_server.machineIdentifier
plex_server = hass.data[DOMAIN][SERVERS][server_id] plex_server = hass.data[DOMAIN][SERVERS][server_id]
@ -46,12 +26,7 @@ async def test_plex_tv_clients(hass):
# Ensure one more client is discovered # Ensure one more client is discovered
await hass.config_entries.async_unload(entry.entry_id) await hass.config_entries.async_unload(entry.entry_id)
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch( mock_plex_server = await setup_plex_server(config_entry=entry)
"plexapi.myplex.MyPlexAccount", return_value=mock_plex_account
), patch("homeassistant.components.plex.PlexWebsocket.listen"):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
plex_server = hass.data[DOMAIN][SERVERS][server_id] plex_server = hass.data[DOMAIN][SERVERS][server_id]
await plex_server._async_update_platforms() await plex_server._async_update_platforms()
@ -63,15 +38,10 @@ async def test_plex_tv_clients(hass):
# Ensure only plex.tv resource client is found # Ensure only plex.tv resource client is found
await hass.config_entries.async_unload(entry.entry_id) await hass.config_entries.async_unload(entry.entry_id)
mock_plex_server = await setup_plex_server(config_entry=entry)
mock_plex_server.clear_clients() mock_plex_server.clear_clients()
mock_plex_server.clear_sessions() mock_plex_server.clear_sessions()
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
"plexapi.myplex.MyPlexAccount", return_value=mock_plex_account
), patch("homeassistant.components.plex.PlexWebsocket.listen"):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
plex_server = hass.data[DOMAIN][SERVERS][server_id] plex_server = hass.data[DOMAIN][SERVERS][server_id]
await plex_server._async_update_platforms() await plex_server._async_update_platforms()

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

@ -10,32 +10,11 @@ from homeassistant.components.plex.const import DOMAIN, SERVERS, SERVICE_PLAY_ON
from homeassistant.const import ATTR_ENTITY_ID from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError
from .const import DEFAULT_DATA, DEFAULT_OPTIONS
from .mock_classes import MockPlexAccount, MockPlexServer
from tests.async_mock import patch from tests.async_mock import patch
from tests.common import MockConfigEntry
async def test_sonos_playback(hass): async def test_sonos_playback(hass, mock_plex_server):
"""Test playing media on a Sonos speaker.""" """Test playing media on a Sonos speaker."""
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(
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()
), 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 server_id = mock_plex_server.machineIdentifier
loaded_server = hass.data[DOMAIN][SERVERS][server_id] loaded_server = hass.data[DOMAIN][SERVERS][server_id]

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

@ -40,33 +40,16 @@ from .mock_classes import (
) )
from tests.async_mock import patch from tests.async_mock import patch
from tests.common import MockConfigEntry
async def test_new_users_available(hass): async def test_new_users_available(hass, entry, mock_websocket, setup_plex_server):
"""Test setting up when new users available on Plex server.""" """Test setting up when new users available on Plex server."""
MONITORED_USERS = {"Owner": {"enabled": True}} MONITORED_USERS = {"Owner": {"enabled": True}}
OPTIONS_WITH_USERS = copy.deepcopy(DEFAULT_OPTIONS) OPTIONS_WITH_USERS = copy.deepcopy(DEFAULT_OPTIONS)
OPTIONS_WITH_USERS[MP_DOMAIN][CONF_MONITORED_USERS] = MONITORED_USERS OPTIONS_WITH_USERS[MP_DOMAIN][CONF_MONITORED_USERS] = MONITORED_USERS
entry.options = OPTIONS_WITH_USERS
entry = MockConfigEntry( mock_plex_server = await setup_plex_server(config_entry=entry)
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(
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()
), patch(
"homeassistant.components.plex.PlexWebsocket", autospec=True
) as mock_websocket:
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 server_id = mock_plex_server.machineIdentifier
@ -83,31 +66,17 @@ async def test_new_users_available(hass):
assert sensor.state == str(len(mock_plex_server.accounts)) assert sensor.state == str(len(mock_plex_server.accounts))
async def test_new_ignored_users_available(hass, caplog): async def test_new_ignored_users_available(
hass, caplog, entry, mock_websocket, setup_plex_server
):
"""Test setting up when new users available on Plex server but are ignored.""" """Test setting up when new users available on Plex server but are ignored."""
MONITORED_USERS = {"Owner": {"enabled": True}} MONITORED_USERS = {"Owner": {"enabled": True}}
OPTIONS_WITH_USERS = copy.deepcopy(DEFAULT_OPTIONS) OPTIONS_WITH_USERS = copy.deepcopy(DEFAULT_OPTIONS)
OPTIONS_WITH_USERS[MP_DOMAIN][CONF_MONITORED_USERS] = MONITORED_USERS OPTIONS_WITH_USERS[MP_DOMAIN][CONF_MONITORED_USERS] = MONITORED_USERS
OPTIONS_WITH_USERS[MP_DOMAIN][CONF_IGNORE_NEW_SHARED_USERS] = True OPTIONS_WITH_USERS[MP_DOMAIN][CONF_IGNORE_NEW_SHARED_USERS] = True
entry.options = OPTIONS_WITH_USERS
entry = MockConfigEntry( mock_plex_server = await setup_plex_server(config_entry=entry)
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(
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()
), patch(
"homeassistant.components.plex.PlexWebsocket", autospec=True
) as mock_websocket:
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 server_id = mock_plex_server.machineIdentifier
@ -134,26 +103,10 @@ async def test_new_ignored_users_available(hass, caplog):
assert sensor.state == str(len(mock_plex_server.accounts)) assert sensor.state == str(len(mock_plex_server.accounts))
async def test_network_error_during_refresh(hass, caplog): async def test_network_error_during_refresh(
hass, caplog, mock_plex_server, mock_websocket
):
"""Test network failures during refreshes.""" """Test network failures during refreshes."""
entry = MockConfigEntry(
domain=DOMAIN,
data=DEFAULT_DATA,
options=DEFAULT_OPTIONS,
unique_id=DEFAULT_DATA["server_id"],
)
mock_plex_server = MockPlexServer()
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()
), patch(
"homeassistant.components.plex.PlexWebsocket", autospec=True
) as mock_websocket:
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 server_id = mock_plex_server.machineIdentifier
loaded_server = hass.data[DOMAIN][SERVERS][server_id] loaded_server = hass.data[DOMAIN][SERVERS][server_id]
@ -172,26 +125,8 @@ async def test_network_error_during_refresh(hass, caplog):
) )
async def test_mark_sessions_idle(hass): async def test_mark_sessions_idle(hass, mock_plex_server, mock_websocket):
"""Test marking media_players as idle when sessions end.""" """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()
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()
), patch(
"homeassistant.components.plex.PlexWebsocket", autospec=True
) as mock_websocket:
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 server_id = mock_plex_server.machineIdentifier
loaded_server = hass.data[DOMAIN][SERVERS][server_id] loaded_server = hass.data[DOMAIN][SERVERS][server_id]
@ -211,26 +146,17 @@ async def test_mark_sessions_idle(hass):
assert sensor.state == "0" assert sensor.state == "0"
async def test_ignore_plex_web_client(hass): async def test_ignore_plex_web_client(hass, entry, mock_websocket):
"""Test option to ignore Plex Web clients.""" """Test option to ignore Plex Web clients."""
OPTIONS = copy.deepcopy(DEFAULT_OPTIONS) OPTIONS = copy.deepcopy(DEFAULT_OPTIONS)
OPTIONS[MP_DOMAIN][CONF_IGNORE_PLEX_WEB_CLIENTS] = True OPTIONS[MP_DOMAIN][CONF_IGNORE_PLEX_WEB_CLIENTS] = True
entry.options = OPTIONS
entry = MockConfigEntry(
domain=DOMAIN,
data=DEFAULT_DATA,
options=OPTIONS,
unique_id=DEFAULT_DATA["server_id"],
)
mock_plex_server = MockPlexServer(config_entry=entry) mock_plex_server = MockPlexServer(config_entry=entry)
with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch( with patch("plexapi.server.PlexServer", return_value=mock_plex_server), patch(
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount(players=0) "plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount(players=0)
), patch( ):
"homeassistant.components.plex.PlexWebsocket", autospec=True
) as mock_websocket:
entry.add_to_hass(hass) entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id) assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -246,27 +172,8 @@ async def test_ignore_plex_web_client(hass):
assert len(media_players) == int(sensor.state) - 1 assert len(media_players) == int(sensor.state) - 1
async def test_media_lookups(hass): async def test_media_lookups(hass, mock_plex_server, mock_websocket):
"""Test media lookups to Plex server.""" """Test media lookups to Plex server."""
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(
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()
), patch(
"homeassistant.components.plex.PlexWebsocket", autospec=True
) as mock_websocket:
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 server_id = mock_plex_server.machineIdentifier
loaded_server = hass.data[DOMAIN][SERVERS][server_id] loaded_server = hass.data[DOMAIN][SERVERS][server_id]

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

@ -15,31 +15,15 @@ from homeassistant.const import (
CONF_VERIFY_SSL, CONF_VERIFY_SSL,
) )
from .const import DEFAULT_DATA, DEFAULT_OPTIONS, MOCK_SERVERS, MOCK_TOKEN from .const import MOCK_SERVERS, MOCK_TOKEN
from .mock_classes import MockPlexAccount, MockPlexLibrarySection, MockPlexServer from .mock_classes import MockPlexLibrarySection
from tests.async_mock import patch from tests.async_mock import patch
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
async def test_refresh_library(hass): async def test_refresh_library(hass, mock_plex_server, setup_plex_server):
"""Test refresh_library service call.""" """Test refresh_library service call."""
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(
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()
), patch("homeassistant.components.plex.PlexWebsocket", autospec=True):
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# Test with non-existent server # Test with non-existent server
with patch.object(MockPlexLibrarySection, "update") as mock_update: with patch.object(MockPlexLibrarySection, "update") as mock_update:
assert await hass.services.async_call( assert await hass.services.async_call(
@ -84,13 +68,7 @@ async def test_refresh_library(hass):
}, },
) )
mock_plex_server_2 = MockPlexServer(config_entry=entry_2) await setup_plex_server(config_entry=entry_2)
with patch("plexapi.server.PlexServer", return_value=mock_plex_server_2), patch(
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()
), patch("homeassistant.components.plex.PlexWebsocket", autospec=True):
entry_2.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry_2.entry_id)
await hass.async_block_till_done()
# Test multiple servers available but none specified # Test multiple servers available but none specified
with patch.object(MockPlexLibrarySection, "update") as mock_update: with patch.object(MockPlexLibrarySection, "update") as mock_update:
@ -103,24 +81,8 @@ async def test_refresh_library(hass):
assert not mock_update.called assert not mock_update.called
async def test_scan_clients(hass): async def test_scan_clients(hass, mock_plex_server):
"""Test scan_for_clients service call.""" """Test scan_for_clients service call."""
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(
"plexapi.myplex.MyPlexAccount", return_value=MockPlexAccount()
), patch("homeassistant.components.plex.PlexWebsocket", autospec=True):
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert await hass.services.async_call( assert await hass.services.async_call(
DOMAIN, DOMAIN,
SERVICE_SCAN_CLIENTS, SERVICE_SCAN_CLIENTS,