Fix sync tests when using a token

.github/workflows/ci.yaml

@@ -176,11 +176,6 @@ jobs:
 
     - name: Sync tests with ${{ matrix.plex }} server
       if: matrix.plex == 'claimed'
-      env:
-        PLEXAPI_HEADER_PROVIDES: 'controller,sync-target'
-        PLEXAPI_HEADER_PLATFORM: iOS
-        PLEXAPI_HEADER_PLATFORM_VERSION: 11.4.1
-        PLEXAPI_HEADER_DEVICE: iPhone
       run: |
         . venv/bin/activate
         pytest \

plexapi/base.py

@@ -95,7 +95,7 @@ def _buildDetailsKey(self, **kwargs):
             or disable each parameter individually by setting it to False or 0.
         """
         details_key = self.key
-        if hasattr(self, '_INCLUDES'):
+        if details_key and hasattr(self, '_INCLUDES'):
             includes = {}
             for k, v in self._INCLUDES.items():
                 value = kwargs.get(k, v)

plexapi/library.py

@@ -8,7 +8,7 @@
 from plexapi.media import MediaTag
 from plexapi.settings import Setting
 
-warnings.simplefilter('default')
+warnings.simplefilter('default', category=DeprecationWarning)
 
 
 class Library(PlexObject):

plexapi/myplex.py

@@ -153,14 +153,15 @@ def _loadData(self, data):
         self.services = None
         self.joined_at = None
 
-    def device(self, name):
+    def device(self, name=None, clientIdentifier=None):
         """ Returns the :class:`~plexapi.myplex.MyPlexDevice` that matches the name specified.
 
             Parameters:
                 name (str): Name to match against.
+                clientIdentifier (str): clientIdentifier to match against.
         """
         for device in self.devices():
-            if device.name.lower() == name.lower():
+            if (name and device.name.lower() == name.lower() or device.clientIdentifier == clientIdentifier):
                 return device
         raise NotFound('Unable to find device %s' % name)
 
@@ -1098,33 +1099,44 @@ class MyPlexPinLogin(object):
             requestTimeout (int): timeout in seconds on initial connect to plex.tv (default config.TIMEOUT).
 
         Attributes:
-            PINS (str): 'https://plex.tv/pins.xml'
-            CHECKPINS (str): 'https://plex.tv/pins/{pinid}.xml'
+            PINS (str): 'https://plex.tv/api/v2/pins'
+            CHECKPINS (str): 'https://plex.tv/api/v2/pins/{pinid}'
+            LINK (str): 'https://plex.tv/api/v2/pins/link'
             POLLINTERVAL (int): 1
             finished (bool): Whether the pin login has finished or not.
             expired (bool): Whether the pin login has expired or not.
             token (str): Token retrieved through the pin login.
             pin (str): Pin to use for the login on https://plex.tv/link.
     """
-    PINS = 'https://plex.tv/pins.xml'               # get
-    CHECKPINS = 'https://plex.tv/pins/{pinid}.xml'  # get
+    PINS = 'https://plex.tv/api/v2/pins'               # get
+    CHECKPINS = 'https://plex.tv/api/v2/pins/{pinid}'  # get
+    LINK = 'https://plex.tv/api/v2/pins/link'          # put
     POLLINTERVAL = 1
 
-    def __init__(self, session=None, requestTimeout=None):
+    def __init__(self, session=None, requestTimeout=None, headers=None):
         super(MyPlexPinLogin, self).__init__()
         self._session = session or requests.Session()
         self._requestTimeout = requestTimeout or TIMEOUT
+        self.headers = headers
 
         self._loginTimeout = None
         self._callback = None
         self._thread = None
         self._abort = False
         self._id = None
+        self._code = None
 
         self.finished = False
         self.expired = False
         self.token = None
-        self.pin = self._getPin()
+
+    @property
+    def pin(self):
+        if self._code:
+            return self._code
+
+        self._getCode()
+        return self._code
 
     def run(self, callback=None, timeout=None):
         """ Starts the thread which monitors the PIN login state.
@@ -1187,21 +1199,39 @@ def checkLogin(self):
 
         return False
 
-    def _getPin(self):
-        if self.pin:
-            return self.pin
+    def link(self, code=None, token=None):
+        if code is None:
+            code = self.pin
+
+        url = self.LINK
+        headers = BASE_HEADERS.copy()
+        headers.update({
+            'Content-Type': 'application/x-www-form-urlencoded',
+            'X-Plex-Product': 'Plex SSO',
+        })
+
+        token = token or CONFIG.get('auth.server_token')
+        if token:
+            headers['X-Plex-Token'] = token
+
+        data = {'code': code}
+        self._query(url, self._session.put, headers=headers, data=data)
 
+    def _getCode(self):
         url = self.PINS
         response = self._query(url, self._session.post)
         if not response:
             return None
 
-        self._id = response.find('id').text
-        self.pin = response.find('code').text
+        self._id = response.attrib.get('id')
+        self._code = response.attrib.get('code')
 
-        return self.pin
+        return self._code
 
     def _checkLogin(self):
+        if not self._code:
+            self._getCode()
+
         if not self._id:
             return False
 
@@ -1213,7 +1243,7 @@ def _checkLogin(self):
         if not response:
             return False
 
-        token = response.find('auth_token').text
+        token = response.attrib.get('authToken')
         if not token:
             return False
 
@@ -1241,11 +1271,19 @@ def _pollLogin(self):
         finally:
             self.finished = True
 
-    def _query(self, url, method=None):
+    def _headers(self, **kwargs):
+        """ Returns dict containing base headers for all requests for pin login. """
+        headers = BASE_HEADERS.copy()
+        if self.headers:
+            headers.update(self.headers)
+        headers.update(kwargs)
+        return headers
+
+    def _query(self, url, method=None, headers=None, **kwargs):
         method = method or self._session.get
         log.debug('%s %s', method.__name__.upper(), url)
-        headers = BASE_HEADERS.copy()
-        response = method(url, headers=headers, timeout=self._requestTimeout)
+        headers = headers or self._headers()
+        response = method(url, headers=headers, timeout=self._requestTimeout, **kwargs)
         if not response.ok:  # pragma: no cover
             codename = codes.get(response.status_code)[0]
             errtext = response.text.replace('\n', ' ')

plexapi/utils.py

@@ -11,7 +11,7 @@
 from urllib.parse import quote
 
 import requests
-from plexapi.exceptions import NotFound
+from plexapi.exceptions import BadRequest, NotFound
 
 try:
     from tqdm import tqdm
@@ -379,6 +379,31 @@ def getMyPlexAccount(opts=None):  # pragma: no cover
     return MyPlexAccount(username, password)
 
 
+def createMyPlexDevice(headers, timeout=None):  # pragma: no cover
+    """ Helper function to create a new MyPlexDevice.
+
+        Parameters:
+            headers (dict): Provide the X-Plex- headers for the new device.
+                A unique X-Plex-Client-Identifier is required.
+            timeout (int): Timeout in seconds to wait for device login.
+    """
+    from plexapi.myplex import MyPlexPinLogin
+
+    if 'X-Plex-Client-Identifier' not in headers:
+        raise BadRequest('The X-Plex-Client-Identifier header is required.')
+
+    clientIdentifier = headers['X-Plex-Client-Identifier']
+
+    pinlogin = MyPlexPinLogin(headers=headers)
+    pinlogin.run(timeout=timeout)
+    pinlogin.link()
+    pinlogin.waitForLogin()
+
+    account = getMyPlexAccount()
+    device = account.device(clientIdentifier=clientIdentifier)
+    return device
+
+
 def choose(msg, items, attr):  # pragma: no cover
     """ Command line helper to display a list of choices, asking the
         user to choose one of the options.

tests/conftest.py

@@ -8,8 +8,10 @@
 import pytest
 import requests
 from plexapi.client import PlexClient
+from plexapi.exceptions import NotFound
 from plexapi.myplex import MyPlexAccount
 from plexapi.server import PlexServer
+from plexapi.utils import createMyPlexDevice
 
 from .payloads import ACCOUNT_XML
 
@@ -47,6 +49,15 @@
     "windows",
     "windows_phone",
 }
+SYNC_DEVICE_IDENTIFIER = "test-sync-client-%s" % plexapi.X_PLEX_IDENTIFIER
+SYNC_DEVICE_HEADERS = {
+    "X-Plex-Provides": "sync-target",
+    "X-Plex-Platform": "iOS",
+    "X-Plex-Platform-Version": "11.4.1",
+    "X-Plex-Device": "iPhone",
+    "X-Plex-Device-Name": "Test Sync Device",
+    "X-Plex-Client-Identifier": SYNC_DEVICE_IDENTIFIER
+}
 
 TEST_AUTHENTICATED = "authenticated"
 TEST_ANONYMOUSLY = "anonymously"
@@ -125,19 +136,10 @@ def account_plexpass(account):
 
 @pytest.fixture(scope="session")
 def account_synctarget(account_plexpass):
-    assert "sync-target" in plexapi.X_PLEX_PROVIDES, (
-        "You have to set env var " "PLEXAPI_HEADER_PROVIDES=sync-target,controller"
-    )
-    assert "sync-target" in plexapi.BASE_HEADERS["X-Plex-Provides"]
-    assert (
-        "iOS" == plexapi.X_PLEX_PLATFORM
-    ), "You have to set env var PLEXAPI_HEADER_PLATFORM=iOS"
-    assert (
-        "11.4.1" == plexapi.X_PLEX_PLATFORM_VERSION
-    ), "You have to set env var PLEXAPI_HEADER_PLATFORM_VERSION=11.4.1"
-    assert (
-        "iPhone" == plexapi.X_PLEX_DEVICE
-    ), "You have to set env var PLEXAPI_HEADER_DEVICE=iPhone"
+    assert "sync-target" in SYNC_DEVICE_HEADERS["X-Plex-Provides"]
+    assert "iOS" == SYNC_DEVICE_HEADERS["X-Plex-Platform"]
+    assert "11.4.1" == SYNC_DEVICE_HEADERS["X-Plex-Platform-Version"]
+    assert "iPhone" == SYNC_DEVICE_HEADERS["X-Plex-Device"]
     return account_plexpass
 
 
@@ -159,24 +161,23 @@ def plex(request):
 
 
 @pytest.fixture()
-def device(account):
-    d = None
-    for device in account.devices():
-        if device.clientIdentifier == plexapi.X_PLEX_IDENTIFIER:
-            d = device
-            break
-
-    assert d
-    return d
+def sync_device(account_synctarget):
+    try:
+        device = account_synctarget.device(clientIdentifier=SYNC_DEVICE_IDENTIFIER)
+    except NotFound:
+        device = createMyPlexDevice(SYNC_DEVICE_HEADERS, timeout=10)
+
+    assert device
+    return device
 
 
 @pytest.fixture()
-def clear_sync_device(device, account_synctarget, plex):
-    sync_items = account_synctarget.syncItems(clientId=device.clientIdentifier)
+def clear_sync_device(sync_device, plex):
+    sync_items = sync_device.syncItems()
     for item in sync_items.items:
         item.delete()
     plex.refreshSync()
-    return device
+    return sync_device
 
 
 @pytest.fixture

tests/test_myplex.py

@@ -56,10 +56,7 @@ def test_myplex_devices(account):
 
 
 def test_myplex_device(account, plex):
-    from plexapi import X_PLEX_DEVICE_NAME
-
     assert account.device(plex.friendlyName)
-    assert account.device(X_PLEX_DEVICE_NAME)
 
 
 def _test_myplex_connect_to_device(account):