test_server.py

import re
import time
from urllib.parse import quote_plus

import pytest
from datetime import datetime
from PIL import Image
from plexapi.exceptions import BadRequest, NotFound
from plexapi.server import PlexServer
from plexapi.utils import download
from requests import Session

from . import conftest as utils
from .payloads import SERVER_RESOURCES, SERVER_TRANSCODE_SESSIONS


def test_server_attr(plex, account):
 assert plex._baseurl == utils.SERVER_BASEURL
 assert len(plex.friendlyName) >= 1
 assert len(plex.machineIdentifier) == 40
 assert plex.myPlex is True
 # if you run the tests very shortly after server creation the state in rare cases may be `unknown`
 assert plex.myPlexMappingState in ("mapped", "unknown")
 assert plex.myPlexSigninState == "ok"
 assert utils.is_int(plex.myPlexSubscription, gte=0)
 assert re.match(utils.REGEX_EMAIL, plex.myPlexUsername)
 assert plex.platform in ("Linux", "Windows")
 assert len(plex.platformVersion) >= 5
 assert plex._token == account.authenticationToken
 assert utils.is_int(plex.transcoderActiveVideoSessions, gte=0)
 assert utils.is_datetime(plex.updatedAt)
 assert len(plex.version) >= 5


def test_server_alert_listener(plex, movies):
 try:
 messages = []
 listener = plex.startAlertListener(messages.append)
 movies.refresh()
 utils.wait_until(lambda: len(messages) >= 3, delay=1, timeout=30)
 assert len(messages) >= 3
 finally:
 listener.stop()


@pytest.mark.req_client
def test_server_session():
 # TODO: Implement test_server_session
 pass


def test_server_library(plex):
 # TODO: Implement test_server_library
 assert plex.library


def test_server_url(plex):
 assert "ohno" in plex.url("ohno")


def test_server_transcodeImage(tmpdir, plex, movie):
 width, height = 500, 100
 background = "000000"
 blend = "FFFFFF"

 original_url = movie.thumbUrl
 resize_jpeg_url = plex.transcodeImage(original_url, height, width)
 no_minSize_png_url = plex.transcodeImage(original_url, height, width, minSize=False, imageFormat="png")
 grayscale_url = plex.transcodeImage(original_url, height, width, saturation=0)
 opacity_background_url = plex.transcodeImage(original_url, height, width, opacity=0, background=background, blur=100)
 blend_url = plex.transcodeImage(original_url, height, width, blendColor=blend, blur=1000)
 online_no_upscale_url = plex.transcodeImage(
 "https://raw.githubusercontent.com/pushingkarmaorg/python-plexapi/master/tests/data/cute_cat.jpg",
 1000,
 1000,
 upscale=False
 )

 original_img = download(
 original_url, plex._token, savepath=str(tmpdir), filename="original_img",
 )
 resized_jpeg_img = download(
 resize_jpeg_url, plex._token, savepath=str(tmpdir), filename="resized_jpeg_img"
 )
 no_minSize_png_img = download(
 no_minSize_png_url, plex._token, savepath=str(tmpdir), filename="no_minSize_png_img"
 )
 grayscale_img = download(
 grayscale_url, plex._token, savepath=str(tmpdir), filename="grayscale_img"
 )
 opacity_background_img = download(
 opacity_background_url, plex._token, savepath=str(tmpdir), filename="opacity_background_img"
 )
 blend_img = download(
 blend_url, plex._token, savepath=str(tmpdir), filename="blend_img"
 )
 online_no_upscale_img = download(
 online_no_upscale_url, plex._token, savepath=str(tmpdir), filename="online_no_upscale_img"
 )

 with Image.open(original_img) as image:
 assert image.size[0] != width
 assert image.size[1] != height
 with Image.open(resized_jpeg_img) as image:
 assert image.size[0] == width
 assert image.size[1] != height
 assert image.format == "JPEG"
 with Image.open(no_minSize_png_img) as image:
 assert image.size[0] != width
 assert image.size[1] == height
 assert image.format == "PNG"
 assert utils.detect_color_image(grayscale_img) == "grayscale"
 assert utils.detect_dominant_hexcolor(opacity_background_img) == background
 assert utils.detect_color_distance(utils.detect_dominant_hexcolor(blend_img), blend)
 with Image.open(online_no_upscale_img) as image1:
 with Image.open(utils.STUB_IMAGE_PATH) as image2:
 assert image1.size == image2.size


def test_server_fetchitem_notfound(plex):
 with pytest.raises(NotFound):
 plex.fetchItem(123456789)


def test_server_search(plex, movie):
 title = movie.title
 # this search seem to fail on my computer but not at travis, wtf.
 assert plex.search(title)
 results = plex.search(title, mediatype="movie")
 assert results[0] == movie
 # Test genre search
 genre = movie.genres[0]
 results = plex.search(genre.tag, mediatype="genre")
 hub_tag = results[0]
 assert utils.is_int(hub_tag.count)
 assert hub_tag.filter == f"genre={hub_tag.id}"
 assert utils.is_int(hub_tag.id)
 assert utils.is_metadata(
 hub_tag.key,
 prefix=hub_tag.librarySectionKey,
 contains=f"{hub_tag.librarySectionID}/all",
 suffix=hub_tag.filter)
 assert utils.is_int(hub_tag.librarySectionID)
 assert utils.is_metadata(hub_tag.librarySectionKey, prefix="/library/sections")
 assert hub_tag.librarySectionTitle == "Movies"
 assert hub_tag.librarySectionType == 1
 assert hub_tag.reason == "section"
 assert hub_tag.reasonID == hub_tag.librarySectionID
 assert hub_tag.reasonTitle == hub_tag.librarySectionTitle
 assert utils.is_float(hub_tag.score, gte=0.0)
 assert hub_tag.type == "tag"
 assert hub_tag.tag == genre.tag
 assert hub_tag.tagType == 1
 assert hub_tag.tagValue is None
 assert hub_tag.thumb is None
 assert movie in hub_tag.items()
 # Test director search
 director = movie.directors[0]
 assert plex.search(director.tag, mediatype="director")
 # Test actor search
 role = movie.roles[0]
 results = plex.search(role.tag, mediatype="actor")
 assert results
 hub_tag = results[0]
 assert hub_tag.tagKey


def test_server_playlist(plex, show):
 episodes = show.episodes()
 playlist = plex.createPlaylist("test_playlist", items=episodes[:3])
 try:
 assert playlist.title == "test_playlist"
 with pytest.raises(NotFound):
 plex.playlist("<playlist-not-found>")
 finally:
 playlist.delete()


def test_server_playlists(plex, show):
 playlists = plex.playlists()
 count = len(playlists)
 episodes = show.episodes()
 playlist = plex.createPlaylist("test_playlist", items=episodes[:3])
 try:
 playlists = plex.playlists()
 assert len(playlists) == count + 1
 assert playlist in plex.playlists(playlistType='video')
 assert playlist not in plex.playlists(playlistType='audio')
 finally:
 playlist.delete()


def test_server_Server_query(plex):
 assert plex.query("/")
 with pytest.raises(NotFound):
 assert plex.query("/asdf/1234/asdf", headers={"random_headers": "1234"})


def test_server_Server_session(account):
 # Mock Session
 class MySession(Session):
 def __init__(self):
 super(self.__class__, self).__init__()
 self.plexapi_session_test = True

 # Test Code
 plex = PlexServer(
 utils.SERVER_BASEURL, account.authenticationToken, session=MySession()
 )
 assert hasattr(plex._session, "plexapi_session_test")


@pytest.mark.authenticated
def test_server_token_in_headers(plex):
 headers = plex._headers()
 assert "X-Plex-Token" in headers
 assert len(headers["X-Plex-Token"]) >= 1


def test_server_createPlayQueue(plex, movie):
 playqueue = plex.createPlayQueue(movie, shuffle=1, repeat=1)
 assert "shuffle=1" in playqueue._initpath
 assert "repeat=1" in playqueue._initpath
 assert playqueue.playQueueShuffled is True


def test_server_client_not_found(plex):
 with pytest.raises(NotFound):
 plex.client("<This-client-should-not-be-found>")


def test_server_sessions(plex):
 assert len(plex.sessions()) >= 0


def test_server_butlerTasks(plex):
 assert len(plex.butlerTasks())


def test_server_runButlerTask(plex):
 assert plex.runButlerTask("CleanOldBundles")
 with pytest.raises(BadRequest):
 plex.runButlerTask("<This-task-should-not-exist>")


def test_server_isLatest(plex, mocker):
 from os import environ

 is_latest = plex.isLatest()
 if environ.get("PLEX_CONTAINER_TAG") and environ["PLEX_CONTAINER_TAG"] not in ("latest", "plexpass", "public"):
 assert not is_latest
 else:
 return pytest.skip(
 "Run with PLEX_CONTAINER_TAG != latest, plexpass, or public to ensure that update is available"
 )


def test_server_installUpdate(plex, mocker):
 m = mocker.MagicMock(release="aa")
 with utils.patch('plexapi.server.PlexServer.checkForUpdate', return_value=m):
 with utils.callable_http_patch():
 plex.installUpdate()


def test_server_checkForUpdate(plex, mocker):
 class R:
 def __init__(self, **kwargs):
 self.download_key = "plex.tv/release/1337"
 self.version = "1337"
 self.added = "gpu transcode"
 self.fixed = "fixed rare bug"
 self.downloadURL = "http://path-to-update"
 self.state = "downloaded"

 with utils.patch('plexapi.server.PlexServer.checkForUpdate', return_value=R()):
 rel = plex.checkForUpdate(force=False, download=True)
 assert rel.download_key == "plex.tv/release/1337"
 assert rel.version == "1337"
 assert rel.added == "gpu transcode"
 assert rel.fixed == "fixed rare bug"
 assert rel.downloadURL == "http://path-to-update"
 assert rel.state == "downloaded"


@pytest.mark.client
def test_server_clients(plex):
 assert len(plex.clients())
 client = plex.clients()[0]
 assert client._baseurl == utils.CLIENT_BASEURL
 assert client._server._baseurl == utils.SERVER_BASEURL
 assert client.protocol == 'plex'
 assert int(client.protocolVersion) in range(4)
 assert isinstance(client.machineIdentifier, str)
 assert client.deviceClass in ['phone', 'tablet', 'stb', 'tv', 'pc']
 assert set(client.protocolCapabilities).issubset({'timeline', 'playback', 'navigation', 'mirror', 'playqueues'})


@pytest.mark.authenticated
@pytest.mark.xfail(strict=False)
def test_server_account(plex):
 account = plex.account()
 assert account.authToken
 # TODO: Figure out why this is missing from time to time.
 # assert account.mappingError == 'publisherror'
 assert account.mappingErrorMessage is None
 assert account.mappingState == "mapped"
 if account.mappingError != "unreachable":
 if account.privateAddress is not None:
 # This seems to fail way to often..
 if len(account.privateAddress):
 assert re.match(utils.REGEX_IPADDR, account.privateAddress)
 else:
 assert account.privateAddress == ""

 assert int(account.privatePort) >= 1000
 assert re.match(utils.REGEX_IPADDR, account.publicAddress)
 assert int(account.publicPort) >= 1000
 else:
 assert account.privateAddress == ""
 assert int(account.privatePort) == 0
 assert account.publicAddress == ""
 assert int(account.publicPort) == 0
 assert account.signInState == "ok"
 assert isinstance(account.subscriptionActive, bool)
 if account.subscriptionActive:
 assert len(account.subscriptionFeatures)
 # Below check keeps failing.. it should go away.
 # else: assert sorted(account.subscriptionFeatures) == ['adaptive_bitrate',
 # 'download_certificates', 'federated-auth', 'news']
 assert (
 account.subscriptionState == "Active"
 if account.subscriptionActive
 else "Unknown"
 )
 assert re.match(utils.REGEX_EMAIL, account.username)


@pytest.mark.authenticated
def test_server_claim_unclaim(plex, account):
 server_account = plex.account()
 assert server_account.signInState == 'ok'
 result = plex.unclaim()
 assert result.signInState == 'none'
 result = plex.claim(account)
 assert result.signInState == 'ok'


def test_server_downloadLogs(tmpdir, plex):
 plex.downloadLogs(savepath=str(tmpdir), unpack=True)
 assert len(tmpdir.listdir()) > 1


def test_server_downloadDatabases(tmpdir, plex):
 plex.downloadDatabases(savepath=str(tmpdir), unpack=True)
 assert len(tmpdir.listdir()) > 1


def test_server_browse(plex, movies):
 movies_path = movies.locations[0]
 # browse root
 paths = plex.browse()
 assert len(paths)
 # browse the path of the movie library
 paths = plex.browse(movies_path)
 assert len(paths)
 # browse the path of the movie library without files
 paths = plex.browse(movies_path, includeFiles=False)
 assert not len([f for f in paths if f.TAG == 'File'])
 # walk the path of the movie library
 for path, paths, files in plex.walk(movies_path):
 assert path.startswith(movies_path)
 assert len(paths) or len(files)


def test_server_allowMediaDeletion(account):
 plex = PlexServer(utils.SERVER_BASEURL, account.authenticationToken)
 # Check server current allowMediaDeletion setting
 if plex.allowMediaDeletion:
 # If allowed then test disallowed
 plex._allowMediaDeletion(False)
 time.sleep(1)
 plex = PlexServer(utils.SERVER_BASEURL, account.authenticationToken)
 assert plex.allowMediaDeletion is None
 # Test redundant toggle
 with pytest.raises(BadRequest):
 plex._allowMediaDeletion(False)

 plex._allowMediaDeletion(True)
 time.sleep(1)
 plex = PlexServer(utils.SERVER_BASEURL, account.authenticationToken)
 assert plex.allowMediaDeletion is True
 # Test redundant toggle
 with pytest.raises(BadRequest):
 plex._allowMediaDeletion(True)
 else:
 # If disallowed then test allowed
 plex._allowMediaDeletion(True)
 time.sleep(1)
 plex = PlexServer(utils.SERVER_BASEURL, account.authenticationToken)
 assert plex.allowMediaDeletion is True
 # Test redundant toggle
 with pytest.raises(BadRequest):
 plex._allowMediaDeletion(True)

 plex._allowMediaDeletion(False)
 time.sleep(1)
 plex = PlexServer(utils.SERVER_BASEURL, account.authenticationToken)
 assert plex.allowMediaDeletion is None
 # Test redundant toggle
 with pytest.raises(BadRequest):
 plex._allowMediaDeletion(False)


def test_server_system_accounts(plex):
 accounts = plex.systemAccounts()
 assert len(accounts)
 account = accounts[0]
 assert utils.is_bool(account.autoSelectAudio)
 assert account.defaultAudioLanguage == "en"
 assert account.defaultSubtitleLanguage == "en"
 assert utils.is_int(account.id, gte=0)
 assert len(account.key)
 assert (account.name == "") if account.id == 0 else len(account.name)
 assert account.subtitleMode == 0
 assert account.thumb == ""
 assert account.accountID == account.id
 assert account.accountKey == account.key
 assert plex.systemAccount(account.id) == account


def test_server_system_devices(plex):
 devices = plex.systemDevices()
 assert len(devices)
 device = devices[-1]
 assert device.clientIdentifier or device.clientIdentifier == ""
 assert utils.is_datetime(device.createdAt)
 assert utils.is_int(device.id)
 assert len(device.key)
 assert len(device.name) or device.name == ""
 assert len(device.platform) or device.platform == ""
 assert plex.systemDevice(device.id) == device


@pytest.mark.authenticated
def test_server_dashboard_bandwidth(account_plexpass, plex):
 bandwidthData = plex.bandwidth()
 assert len(bandwidthData)
 bandwidth = bandwidthData[0]
 assert utils.is_int(bandwidth.accountID, gte=0)
 assert utils.is_datetime(bandwidth.at)
 assert utils.is_int(bandwidth.bytes)
 assert utils.is_int(bandwidth.deviceID)
 assert utils.is_bool(bandwidth.lan)
 assert bandwidth.timespan == 6 # Default seconds timespan
 account = bandwidth.account()
 assert utils.is_int(account.id, gte=0)
 device = bandwidth.device()
 assert utils.is_int(device.id)


@pytest.mark.authenticated
def test_server_dashboard_bandwidth_filters(account_plexpass, plex):
 at = datetime(2021, 1, 1)
 filters = {
 'at>': at,
 'bytes>': 1,
 'lan': True,
 'accountID': 1
 }
 bandwidthData = plex.bandwidth(timespan='hours', **filters)
 assert len(bandwidthData)
 bandwidth = bandwidthData[0]
 assert bandwidth.accountID == 1
 assert bandwidth.at >= at
 assert bandwidth.bytes >= 1
 assert bandwidth.lan is True
 assert bandwidth.timespan == 4
 with pytest.raises(BadRequest):
 plex.bandwidth(timespan='n/a')
 with pytest.raises(BadRequest):
 filters = {'n/a': None}
 plex.bandwidth(**filters)
 with pytest.raises(BadRequest):
 filters = {'at': 123456}
 plex.bandwidth(**filters)


@pytest.mark.authenticated
def test_server_dashboard_resources(plex, requests_mock):
 url = plex.url("/statistics/resources")
 requests_mock.get(url, text=SERVER_RESOURCES)
 resourceData = plex.resources()
 assert len(resourceData)
 resource = resourceData[0]
 assert utils.is_datetime(resource.at)
 assert utils.is_float(resource.hostCpuUtilization, gte=0.0)
 assert utils.is_float(resource.hostMemoryUtilization, gte=0.0)
 assert utils.is_float(resource.processCpuUtilization, gte=0.0)
 assert utils.is_float(resource.processMemoryUtilization, gte=0.0)
 assert resource.timespan == 6 # Default seconds timespan


def test_server_transcode_sessions(plex, requests_mock):
 url = plex.url("/transcode/sessions")
 requests_mock.get(url, text=SERVER_TRANSCODE_SESSIONS)
 transcode_sessions = plex.transcodeSessions()
 assert len(transcode_sessions)
 session = transcode_sessions[0]
 assert session.audioChannels == 2
 assert session.audioCodec in utils.CODECS
 assert session.audioDecision == "transcode"
 assert session.complete is False
 assert session.container in utils.CONTAINERS
 assert session.context == "streaming"
 assert utils.is_int(session.duration, gte=100000)
 assert utils.is_int(session.height, gte=480)
 assert len(session.key)
 assert utils.is_float(session.maxOffsetAvailable, gte=0.0)
 assert utils.is_float(session.minOffsetAvailable, gte=0.0)
 assert utils.is_float(session.progress)
 assert session.protocol == "dash"
 assert utils.is_int(session.remaining)
 assert utils.is_int(session.size)
 assert session.sourceAudioCodec in utils.CODECS
 assert session.sourceVideoCodec in utils.CODECS
 assert utils.is_float(session.speed)
 assert session.subtitleDecision is None
 assert session.throttled is False
 assert utils.is_float(session.timestamp, gte=1600000000)
 assert session.transcodeHwDecoding in utils.HW_DECODERS
 assert session.transcodeHwDecodingTitle == "Windows (DXVA2)"
 assert session.transcodeHwEncoding in utils.HW_ENCODERS
 assert session.transcodeHwEncodingTitle == "Intel (QuickSync)"
 assert session.transcodeHwFullPipeline is False
 assert session.transcodeHwRequested is True
 assert session.videoCodec in utils.CODECS
 assert session.videoDecision == "transcode"
 assert utils.is_int(session.width, gte=852)


def test_server_PlexWebURL(plex):
 url = plex.getWebURL()
 assert url.startswith('https://app.plex.tv/desktop')
 assert plex.machineIdentifier in url
 assert quote_plus('/hubs') in url
 assert 'pageType=hub' in url
 # Test a different base
 base = 'https://doesnotexist.com/plex'
 url = plex.getWebURL(base=base)
 assert url.startswith(base)


def test_server_PlexWebURL_playlists(plex):
 tab = 'audio'
 url = plex.getWebURL(playlistTab=tab)
 assert url.startswith('https://app.plex.tv/desktop')
 assert plex.machineIdentifier in url
 assert 'source=playlists' in url
 assert f'pivot=playlists.{tab}' in url


def test_server_agents(plex):
 agents = plex.agents()
 assert agents
 agent = next((a for a in agents if a.identifier == 'com.plexapp.agents.imdb'), None)
 assert agent
 settings = agent.settings()
 assert settings
 setting = next((s for s in settings if s.id == 'country'), None)
 assert setting
 assert setting.enumValues is not None


def test_server_identity(plex):
 identity = plex.identity()
 assert identity.machineIdentifier == plex.machineIdentifier