python-plexapi/tests/test_myplex.py

# -*- coding: utf-8 -*-
import pytest
from plexapi.exceptions import BadRequest, NotFound, Unauthorized
from plexapi.myplex import MyPlexInvite

from . import conftest as utils
from .payloads import MYPLEX_INVITE


def test_myplex_accounts(account, plex):
    assert account, "Must specify username, password & resource to run this test."
    print("MyPlexAccount:")
    print(f"username: {account.username}")
    print(f"email: {account.email}")
    print(f"home: {account.home}")
    assert account.username, "Account has no username"
    assert account.authenticationToken, "Account has no authenticationToken"
    assert account.email, "Account has no email"
    assert account.home is not None, "Account has no home"
    account = plex.account()
    print("Local PlexServer.account():")
    print(f"username: {account.username}")
    # print('authToken: %s' % account.authToken)
    print(f"signInState: {account.signInState}")
    assert account.username, "Account has no username"
    assert account.authToken, "Account has no authToken"
    assert account.signInState, "Account has no signInState"


def test_myplex_resources(account):
    assert account, "Must specify username, password & resource to run this test."
    resources = account.resources()
    for resource in resources:
        name = resource.name or "Unknown"
        connections = [c.uri for c in resource.connections]
        connections = ", ".join(connections) if connections else "None"
        print(f"{name} ({resource.product}): {connections}")
    assert resources, f"No resources found for account: {account.username}"


def test_myplex_connect_to_resource(plex, account):
    servername = plex.friendlyName
    for resource in account.resources():
        if resource.name == servername:
            break
    assert resource.connect(timeout=10)


def test_myplex_devices(account):
    devices = account.devices()
    for device in devices:
        name = device.name or "Unknown"
        connections = ", ".join(device.connections) if device.connections else "None"
        print(f"{name} ({device.product}): {connections}")
    assert devices, f"No devices found for account: {account.name}"


def test_myplex_device(account, plex):
    assert account.device(plex.friendlyName)


def _test_myplex_connect_to_device(account):
    devices = account.devices()
    for device in devices:
        if device.name == "some client name" and len(device.connections):
            break
    client = device.connect()
    assert client, "Unable to connect to device"


def test_myplex_users(account):
    users = account.users()
    if not len(users):
        return pytest.skip("You have to add a shared account into your MyPlex")
    print(f"Found {len(users)} users.")
    user = account.user(users[0].title)
    print(f"Found user: {user}")
    assert user, f"Could not find user {users[0].title}"

    try:
        users[0].servers[0]
    except IndexError:
        return pytest.skip(f"{users[0].title} shared user does not have access to any servers")

    assert (
        len(users[0].servers[0].sections()) > 0
    ), "Couldn't info about the shared libraries"


def test_myplex_resource(account, plex):
    assert account.resource(plex.friendlyName)


def test_myplex_webhooks(account):
    if account.subscriptionActive:
        assert isinstance(account.webhooks(), list)
    else:
        with pytest.raises(BadRequest):
            account.webhooks()


def test_myplex_addwebhooks(account):
    if account.subscriptionActive:
        assert "http://example.com" in account.addWebhook("http://example.com")
    else:
        with pytest.raises(BadRequest):
            account.addWebhook("http://example.com")


def test_myplex_deletewebhooks(account):
    if account.subscriptionActive:
        assert "http://example.com" not in account.deleteWebhook("http://example.com")
    else:
        with pytest.raises(BadRequest):
            account.deleteWebhook("http://example.com")


def test_myplex_optout(account_once):
    def enabled():
        ele = account_once.query("https://plex.tv/api/v2/user/privacy")
        lib = ele.attrib.get("optOutLibraryStats")
        play = ele.attrib.get("optOutPlayback")
        return bool(int(lib)), bool(int(play))

    account_once.optOut(library=True, playback=True)
    utils.wait_until(lambda: enabled() == (True, True))
    account_once.optOut(library=False, playback=False)
    utils.wait_until(lambda: enabled() == (False, False))


@pytest.mark.authenticated
@pytest.mark.xfail(reason="Test account is missing online media sources?")
def test_myplex_onlineMediaSources_optOut(account):
    onlineMediaSources = account.onlineMediaSources()
    for optOut in onlineMediaSources:
        if optOut.key == 'tv.plex.provider.news':
            # News is no longer available
            continue

        optOutValue = optOut.value
        optOut.optIn()
        assert optOut.value == 'opt_in'
        optOut.optOut()
        assert optOut.value == 'opt_out'
        if optOut.key == 'tv.plex.provider.music':
            with pytest.raises(BadRequest):
                optOut.optOutManaged()
        else:
            optOut.optOutManaged()
            assert optOut.value == 'opt_out_managed'
        # Reset original value
        optOut._updateOptOut(optOutValue)

    with pytest.raises(NotFound):
        onlineMediaSources[0]._updateOptOut('unknown')


def test_myplex_inviteFriend(account, plex, mocker):
    inv_user = "hellowlol"
    vid_filter = {"contentRating": ["G"], "label": ["foo"]}
    secs = plex.library.sections()

    ids = account._getSectionIds(plex.machineIdentifier, secs)
    mocker.patch.object(account, "_getSectionIds", return_value=ids)
    with utils.callable_http_patch():
        account.inviteFriend(
            inv_user,
            plex,
            secs,
            allowSync=True,
            allowCameraUpload=True,
            allowChannels=False,
            filterMovies=vid_filter,
            filterTelevision=vid_filter,
            filterMusic={"label": ["foo"]},
        )

        assert inv_user not in [u.title for u in account.users()]


def test_myplex_acceptInvite(account, requests_mock):
    url = MyPlexInvite.REQUESTS
    requests_mock.get(url, text=MYPLEX_INVITE)
    invite = account.pendingInvite('testuser', includeSent=False)
    with utils.callable_http_patch():
        account.acceptInvite(invite)


def test_myplex_cancelInvite(account, requests_mock):
    url = MyPlexInvite.REQUESTED
    requests_mock.get(url, text=MYPLEX_INVITE)
    invite = account.pendingInvite('testuser', includeReceived=False)
    with utils.callable_http_patch():
        account.cancelInvite(invite)


def test_myplex_updateFriend(account, plex, mocker, shared_username):
    vid_filter = {"contentRating": ["G"], "label": ["foo"]}
    secs = plex.library.sections()
    user = account.user(shared_username)

    ids = account._getSectionIds(plex.machineIdentifier, secs)
    mocker.patch.object(account, "_getSectionIds", return_value=ids)
    mocker.patch.object(account, "user", return_value=user)
    with utils.callable_http_patch():
        account.updateFriend(
            shared_username,
            plex,
            secs,
            allowSync=True,
            removeSections=True,
            allowCameraUpload=True,
            allowChannels=False,
            filterMovies=vid_filter,
            filterTelevision=vid_filter,
            filterMusic={"label": ["foo"]},
        )

        with utils.callable_http_patch():
            account.removeFriend(shared_username)


def test_myplex_createExistingUser(account, plex, shared_username):
    user = account.user(shared_username)
    url = f"https://plex.tv/api/invites/requested/{user.id}?friend=0&server=0&home=1"

    account.createExistingUser(user, plex)
    assert shared_username in [u.username for u in account.users() if u.home is True]
    # Remove Home invite
    account.query(url, account._session.delete)
    # Confirm user was removed from home and has returned to friend
    assert shared_username not in [
        u.username for u in plex.myPlexAccount().users() if u.home is True
    ]
    assert shared_username in [
        u.username for u in plex.myPlexAccount().users() if u.home is False
    ]


@pytest.mark.skip(reason="broken test?")
def test_myplex_createHomeUser_remove(account, plex):
    homeuser = "New Home User"
    account.createHomeUser(homeuser, plex)
    assert homeuser in [u.title for u in plex.myPlexAccount().users() if u.home is True]
    account.removeHomeUser(homeuser)
    assert homeuser not in [
        u.title for u in plex.myPlexAccount().users() if u.home is True
    ]


def test_myplex_plexpass_attributes(account_plexpass):
    assert account_plexpass.subscriptionActive
    assert account_plexpass.subscriptionStatus == "Active"
    assert account_plexpass.subscriptionPlan
    assert "sync" in account_plexpass.subscriptionFeatures
    assert "premium_music_metadata" in account_plexpass.subscriptionFeatures
    assert "plexpass" in account_plexpass.roles
    assert utils.ENTITLEMENTS <= set(account_plexpass.entitlements)


def test_myplex_claimToken(account):
    assert account.claimToken().startswith("claim-")


def test_myplex_watchlist(account, movie, show, artist):
    # Ensure watchlist is cleared before tests
    for item in account.watchlist():
        account.removeFromWatchlist(item)
    assert not account.watchlist()

    # Add to watchlist from account
    account.addToWatchlist(movie)
    assert account.onWatchlist(movie)

    # Add to watchlist from object
    show.addToWatchlist(account)
    assert show.onWatchlist(account)

    # Remove from watchlist from account
    account.removeFromWatchlist(show)
    assert not account.onWatchlist(show)

    # Remove from watchlist from object
    movie.removeFromWatchlist(account)
    assert not movie.onWatchlist(account)

    # Add multiple items to watchlist
    account.addToWatchlist([movie, show])
    assert movie.onWatchlist(account) and show.onWatchlist(account)

    # Filter and sort watchlist
    watchlist = account.watchlist(filter='released', sort='titleSort', libtype='movie')
    guids = [i.guid for i in watchlist]
    assert movie.guid in guids and show.guid not in guids

    # Test adding existing item to watchlist
    with pytest.raises(BadRequest):
        account.addToWatchlist(movie)

    # Test retrieving maxresults from watchlist
    watchlist = account.watchlist(maxresults=1)
    assert len(watchlist) == 1

    # Remove multiple items from watchlist
    account.removeFromWatchlist([movie, show])
    assert not movie.onWatchlist(account) and not show.onWatchlist(account)

    # Test removing non-existing item from watchlist
    with pytest.raises(BadRequest):
        account.removeFromWatchlist(movie)

    # Test adding invalid item to watchlist
    with pytest.raises(BadRequest):
        account.addToWatchlist(artist)


def test_myplex_searchDiscover(account, movie, show):
    guids = lambda x: [r.guid for r in x]

    results = account.searchDiscover(movie.title)
    assert movie.guid in guids(results)
    results = account.searchDiscover(movie.title, libtype="show")
    assert movie.guid not in guids(results)

    results = account.searchDiscover(show.title)
    assert show.guid in [r.guid for r in results]
    results = account.searchDiscover(show.title, libtype="movie")
    assert show.guid not in guids(results)


@pytest.mark.authenticated
def test_myplex_viewStateSync(account):
    account.enableViewStateSync()
    assert account.viewStateSync is True
    account.disableViewStateSync()
    assert account.viewStateSync is False


@pytest.mark.authenticated
def test_myplex_pin(account, plex):
    assert account.pin is None

    account.setPin("0000")

    with pytest.raises(Unauthorized):
        account.setPin("1111")
    account.setPin("1111", currentPin="0000")

    with pytest.raises(Unauthorized):
        account.removePin("0000")
    account.removePin("1111")

    homeuser = "Test PIN User"
    try:
        account.createHomeUser(homeuser, plex)
        account.setManagedUserPin(homeuser, "0000")
        account.removeManagedUserPin(homeuser)
    finally:
        account.removeHomeUser(homeuser)


def test_myplex_geoip(account):
    assert account.geoip(account.publicIP())


def test_myplex_ping(account):
    assert account.ping()