fix(legacy agents): do not upload themes unless Themerr is enabled

Contents/Code/constants.py

@@ -62,6 +62,18 @@
     show='TV Shows'
 )
 
+# the explicit IPv4 address is used because `localhost` can resolve to ::1, which `websocket` rejects
+plex_url = 'http://127.0.0.1:32400'
+plex_token = os.environ.get('PLEXTOKEN')
+
+plex_section_type_settings_map = dict(
+    album=9,
+    artist=8,
+    movie=1,
+    photo=13,
+    show=2,
+)
+
 # issue url constants
 base_url = 'https://github.com/LizardByte/ThemerrDB/issues/new?assignees='
 issue_label = 'request-theme'

Contents/Code/general_helper.py

@@ -14,18 +14,113 @@
 else:  # the code is running outside of Plex
     from plexhints.core_kit import Core  # core kit
     from plexhints.log_kit import Log  # log kit
+    from plexhints.parse_kit import XML  # parse kit
     from plexhints.prefs_kit import Prefs  # prefs kit
 
 
 # local imports
-from constants import metadata_base_directory, metadata_type_map, themerr_data_directory
+from constants import (
+    contributes_to,
+    metadata_base_directory,
+    metadata_type_map,
+    plex_section_type_settings_map,
+    plex_url,
+    themerr_data_directory
+)
 
 # constants
 legacy_keys = [
     'downloaded_timestamp'
 ]
 
 
+def agent_enabled(item_agent, item_type):
+    # type: (str, str) -> bool
+    """
+    Check if the specified agent is enabled.
+
+    Parameters
+    ----------
+    item_agent : str
+        The agent to check.
+    item_type : str
+        The type of the item to check.
+
+    Returns
+    -------
+    py:class:`bool`
+        True if the agent is enabled, False otherwise.
+
+    Examples
+    --------
+    >>> agent_enabled(item_agent='com.plexapp.agents.imdb', item_type='movie')
+    True
+    >>> agent_enabled(item_agent='com.plexapp.agents.themoviedb', item_type='movie')
+    True
+    >>> agent_enabled(item_agent='com.plexapp.agents.themoviedb', item_type='show')
+    True
+    >>> agent_enabled(item_agent='com.plexapp.agents.thetvdb', item_type='show')
+    True
+    >>> agent_enabled(item_agent='dev.lizardbyte.retroarcher-plex', item_type='movie')
+    True
+    """
+    # get the settings for this agent
+    settings_url = '{}/system/agents/{}/config/{}'.format(
+        plex_url, item_agent, plex_section_type_settings_map[item_type])
+    settings_data = XML.ElementFromURL(
+        url=settings_url,
+        cacheTime=0
+    )
+    Log.Debug('settings data: {}'.format(settings_data))
+
+    themerr_plex_element = settings_data.find(".//Agent[@name='Themerr-plex']")
+
+    if themerr_plex_element.get('enabled') == '1':  # Plex is using a string
+        return True
+    else:
+        return False
+
+
+def continue_update(item_agent, item_type):
+    # type: (str, str) -> bool
+    """
+    Check if the specified agent should continue updating.
+
+    Parameters
+    ----------
+    item_agent : str
+        The agent to check.
+    item_type : str
+        The type of the item to check.
+
+    Returns
+    -------
+    py:class:`bool`
+        True if the agent should continue updating, False otherwise.
+
+    Examples
+    --------
+    >>> continue_update(item_agent='tv.plex.agents.movie', item_type='movie')
+    True
+    >>> continue_update(item_agent='com.plexapp.agents.imdb', item_type='movie')
+    True
+    >>> continue_update(item_agent='com.plexapp.agents.themoviedb', item_type='movie')
+    True
+    >>> continue_update(item_agent='com.plexapp.agents.themoviedb', item_type='show')
+    True
+    >>> continue_update(item_agent='com.plexapp.agents.thetvdb', item_type='show')
+    True
+    >>> continue_update(item_agent='dev.lizardbyte.retroarcher-plex', item_type='movie')
+    True
+    """
+    if item_agent == 'tv.plex.agents.movie':
+        return Prefs['bool_plex_movie_support']
+    elif item_agent in contributes_to:
+        return agent_enabled(item_agent=item_agent, item_type=item_type)
+    else:
+        return False
+
+
 def get_media_upload_path(item, media_type):
     # type: (any, str) -> str
     """

Contents/Code/plex_api_helper.py

@@ -12,7 +12,7 @@
     pass
 else:  # the code is running outside of Plex
     from plexhints.log_kit import Log  # log kit
-    from plexhints.parse_kit import JSON, XML  # parse kit
+    from plexhints.parse_kit import JSON  # parse kit
     from plexhints.prefs_kit import Prefs  # prefs kit
 
 # imports from Libraries\Shared
@@ -28,7 +28,7 @@
 from plexapi.utils import reverseSearchType
 
 # local imports
-from constants import contributes_to, guid_map, media_type_dict
+from constants import contributes_to, guid_map, media_type_dict, plex_token, plex_url
 import general_helper
 import lizardbyte_db_helper
 import themerr_db_helper
@@ -46,18 +46,6 @@
 # when accessing a missing field
 os.environ["PLEXAPI_PLEXAPI_AUTORELOAD"] = "false"
 
-# the explicit IPv4 address is used because `localhost` can resolve to ::1, which `websocket` rejects
-plex_url = 'http://127.0.0.1:32400'
-plex_token = os.environ.get('PLEXTOKEN')
-
-plex_section_type_settings_map = dict(
-    album=9,
-    artist=8,
-    movie=1,
-    photo=13,
-    show=2,
-)
-
 
 def setup_plexapi():
     """
@@ -692,28 +680,14 @@ def scheduled_update():
             # individual items that was matched with a supported agent...
             continue  # skip unsupported metadata agents
 
-        if section.agent == 'tv.plex.agents.movie':
-            if not Prefs['bool_plex_movie_support']:
-                continue
-        elif section.agent in contributes_to:
-            # check if the agent is enabled
-            if not plex_token:
-                Log.Error('Plex token not found in environment, cannot proceed.')
-                continue
-
-            # get the settings for this agent
-            settings_url = '{}/system/agents/{}/config/{}'.format(
-                plex_url, section.agent, plex_section_type_settings_map[section.type])
-            settings_data = XML.ElementFromURL(
-                url=settings_url,
-                cacheTime=0
-            )
-            Log.Debug('settings data: {}'.format(settings_data))
+        if not plex_token:
+            Log.Error('Plex token not found in environment, cannot proceed.')
+            continue
 
-            themerr_plex_element = settings_data.find(".//Agent[@name='Themerr-plex']")
-            if themerr_plex_element.get('enabled') != '1':  # Plex is using a string
-                Log.Debug('Themerr-plex is disabled for agent "{}"'.format(section.agent))
-                continue
+        # check if the agent is enabled
+        if not general_helper.continue_update(item_agent=section.agent, item_type=section.type):
+            Log.Debug('Themerr-plex is disabled for agent "{}"'.format(section.agent))
+            continue
 
         # get all the items in the section
         media_items = section.all() if Prefs['bool_auto_update_movie_themes'] else []

tests/unit/test_general_helper.py

@@ -12,6 +12,29 @@
 from Code import general_helper
 
 
+@pytest.mark.parametrize('item_agent, item_type, expected', [
+    ('com.plexapp.agents.imdb', 'movie', True),
+    ('com.plexapp.agents.themoviedb', 'movie', True),
+    # ('com.plexapp.agents.themoviedb', 'show', True),
+    # ('com.plexapp.agents.thetvdb', 'show', True),
+])
+def test_agent_enabled(item_agent, item_type, expected):
+    assert general_helper.agent_enabled(item_agent=item_agent, item_type=item_type) is expected
+
+
+@pytest.mark.parametrize('item_agent, item_type, expected', [
+    ('tv.plex.agents.movie', 'movie', True),
+    ('com.plexapp.agents.imdb', 'movie', True),
+    ('com.plexapp.agents.themoviedb', 'movie', True),
+    # ('tv.plex.agents.series', 'show', True),
+    # ('com.plexapp.agents.themoviedb', 'show', True),
+    # ('com.plexapp.agents.thetvdb', 'show', True),
+    ('invalid', 'invalid', False),
+])
+def test_continue_update(item_agent, item_type, expected):
+    assert general_helper.continue_update(item_agent=item_agent, item_type=item_type) is expected
+
+
 def test_get_media_upload_path(movies):
     test_items = [
         movies.all()[0]