diff --git a/resources/language/resource.language.de_de/strings.po b/resources/language/resource.language.de_de/strings.po
index 1bca4933..b2ea0916 100644
--- a/resources/language/resource.language.de_de/strings.po
+++ b/resources/language/resource.language.de_de/strings.po
@@ -1177,6 +1177,18 @@ msgctxt "#30332"
msgid "Automatic (best playable)"
msgstr "Automatisch (beste abspielbare Qualität)"
+msgctxt "#30333"
+msgid "Search method"
+msgstr "Suchmethode"
+
+msgctxt "#30334"
+msgid "Website (twitch.tv, fuzzy)"
+msgstr "Website (twitch.tv, fehlertolerant)"
+
+msgctxt "#30335"
+msgid "Helix API"
+msgstr "Helix-API"
+
#~ msgctxt "#30132"
#~ msgid "OAuth token is required for access to authorized user functions."
#~ msgstr "Für den Zugriff auf geschützte Benutzerfunktionen wird ein OAuth-Token benötigt."
diff --git a/resources/language/resource.language.en_gb/strings.po b/resources/language/resource.language.en_gb/strings.po
index 44882fc6..d3a3d664 100644
--- a/resources/language/resource.language.en_gb/strings.po
+++ b/resources/language/resource.language.en_gb/strings.po
@@ -1224,3 +1224,15 @@ msgstr ""
msgctxt "#30332"
msgid "Automatic (best playable)"
msgstr ""
+
+msgctxt "#30333"
+msgid "Search method"
+msgstr ""
+
+msgctxt "#30334"
+msgid "Website (twitch.tv, fuzzy)"
+msgstr ""
+
+msgctxt "#30335"
+msgid "Helix API"
+msgstr ""
diff --git a/resources/lib/twitch_addon/addon/api.py b/resources/lib/twitch_addon/addon/api.py
index ee06e602..7fa69b69 100644
--- a/resources/lib/twitch_addon/addon/api.py
+++ b/resources/lib/twitch_addon/addon/api.py
@@ -13,7 +13,7 @@
import sys
import os
-from . import cache, utils
+from . import cache, gql_search, utils
from .common import kodi, log_utils
from .common.cache import invalidate_cache_for_function
from .constants import Keys, SCOPES
@@ -160,14 +160,14 @@ def valid_token(self, client_id, token, scopes): # client_id, token used for un
if self.client_id and token_check['client_id'] != self.client_id:
log_utils.log('Token client_id mismatch: token=%s, configured=%s. Clearing stale token.' % (
token_check['client_id'], self.client_id), log_utils.LOGWARNING)
- # Token was obtained with a different client_id — it won't work with Helix
+ # Token was obtained with a different client_id, so it won't work with Helix
kodi.set_setting('oauth_token_helix', '')
kodi.set_setting('device_refresh_token', '')
kodi.set_setting('device_token_expires_at', '')
kodi.set_setting('is_device_authenticated', 'false')
return False
elif not self.client_id:
- # No client_id configured — adopt the token's client_id
+ # No client_id configured, so adopt the token's client_id
log_utils.log('No client_id configured, adopting from token: %s' % token_check['client_id'], log_utils.LOGDEBUG)
self.client_id = token_check['client_id']
self.queries.CLIENT_ID = self.client_id
@@ -304,22 +304,46 @@ def get_game_streams(self, game_id=None, language=Language.ALL, after='MA==', be
return self.error_check(results)
@api_error_handler
- @cache.cache_method(cache_limit=cache.limit)
def get_channel_search(self, search_query, after='MA==', first=20):
+ backend = utils.get_search_backend()
+ return self._get_channel_search(search_query, after, first, backend)
+
+ @cache.cache_method(cache_limit=cache.limit)
+ def _get_channel_search(self, search_query, after, first, backend):
+ if backend == 0 and after == 'MA==':
+ results = gql_search.search(search_query, 'channels')
+ if results is not None:
+ return results
results = self.api.search.get_channels(search_query=search_query, after=after, first=first,
live_only=Boolean.FALSE)
return self.error_check(results)
@api_error_handler
- @cache.cache_method(cache_limit=cache.limit)
def get_stream_search(self, search_query, after='MA==', first=20):
+ backend = utils.get_search_backend()
+ return self._get_stream_search(search_query, after, first, backend)
+
+ @cache.cache_method(cache_limit=cache.limit)
+ def _get_stream_search(self, search_query, after, first, backend):
+ if backend == 0 and after == 'MA==':
+ results = gql_search.search(search_query, 'streams')
+ if results is not None:
+ return results
results = self.api.search.get_channels(search_query=search_query, after=after, first=first,
live_only=Boolean.TRUE)
return self.error_check(results)
@api_error_handler
- @cache.cache_method(cache_limit=cache.limit)
def get_game_search(self, search_query, after='MA==', first=20):
+ backend = utils.get_search_backend()
+ return self._get_game_search(search_query, after, first, backend)
+
+ @cache.cache_method(cache_limit=cache.limit)
+ def _get_game_search(self, search_query, after, first, backend):
+ if backend == 0 and after == 'MA==':
+ results = gql_search.search(search_query, 'games')
+ if results is not None:
+ return results
results = self.api.search.get_categories(search_query=search_query, after=after, first=first)
return self.error_check(results)
diff --git a/resources/lib/twitch_addon/addon/device_auth.py b/resources/lib/twitch_addon/addon/device_auth.py
index 0e767e8a..452b2909 100644
--- a/resources/lib/twitch_addon/addon/device_auth.py
+++ b/resources/lib/twitch_addon/addon/device_auth.py
@@ -428,7 +428,7 @@ def auto_refresh_token():
error_msg = str(e)
log_utils.log('Token auto-refresh failed: %s' % error_msg, log_utils.LOGWARNING)
# Only clear refresh token if Twitch says it's definitively invalid.
- # Do NOT clear on network errors — next startup might succeed.
+ # Do NOT clear on network errors because the next startup might succeed.
if any(keyword in error_msg.lower() for keyword in ['invalid refresh token', 'invalid_grant', 'invalid grant']):
log_utils.log('Refresh token is invalid, clearing device tokens', log_utils.LOGWARNING)
kodi.set_setting('device_refresh_token', '')
diff --git a/resources/lib/twitch_addon/addon/gql_search.py b/resources/lib/twitch_addon/addon/gql_search.py
new file mode 100644
index 00000000..b5b023de
--- /dev/null
+++ b/resources/lib/twitch_addon/addon/gql_search.py
@@ -0,0 +1,173 @@
+# -*- coding: utf-8 -*-
+"""
+ Website search via Twitch's GQL backend.
+
+ Adapted from anxdpanic/plugin.video.twitch PR #706.
+
+ Results are adapted to the existing Helix search shape so routes and
+ converters remain unchanged. Returns None on failure so callers can fall
+ back to Helix.
+
+ SPDX-License-Identifier: GPL-3.0-only
+ See LICENSES/GPL-3.0-only for more information.
+"""
+
+import requests
+
+from . import utils
+from .common import log_utils
+from .constants import Keys
+
+
+GQL_URL = 'https://gql.twitch.tv/gql'
+TIMEOUT = 15
+
+_CHANNEL_QUERY = (
+ 'query Search($q: String!) {'
+ ' searchFor(userQuery: $q, platform: "web", target: {index: CHANNEL}) {'
+ ' channels { edges { item { ... on User {'
+ ' id login displayName'
+ ' broadcastSettings { language title }'
+ ' profileImageURL(width: 300)'
+ ' stream { id viewersCount previewImageURL game { id name displayName } }'
+ ' } } } }'
+ ' }'
+ '}'
+)
+
+_GAME_QUERY = (
+ 'query Search($q: String!) {'
+ ' searchFor(userQuery: $q, platform: "web", target: {index: GAME}) {'
+ ' games { edges { item { ... on Game {'
+ ' id name displayName boxArtURL(width: 285, height: 380)'
+ ' } } } }'
+ ' }'
+ '}'
+)
+
+
+def _post(query, search_query):
+ body = [{
+ 'operationName': 'Search',
+ 'query': query,
+ 'variables': {'q': search_query},
+ }]
+ response = requests.post(
+ GQL_URL,
+ json=body,
+ headers={'Client-ID': utils.get_private_client_id()},
+ timeout=TIMEOUT,
+ )
+ response.raise_for_status()
+ envelope = response.json()
+ if isinstance(envelope, list):
+ if len(envelope) != 1:
+ return None
+ envelope = envelope[0]
+ if not isinstance(envelope, dict) or envelope.get('errors'):
+ return None
+ data = envelope.get('data')
+ if not isinstance(data, dict):
+ return None
+ search_for = data.get('searchFor')
+ if not isinstance(search_for, dict):
+ return None
+ return search_for
+
+
+def _channel_item(item):
+ if not isinstance(item, dict):
+ return None
+ if not item.get('id') or not item.get('login') or not item.get('displayName'):
+ return None
+
+ settings = item.get('broadcastSettings')
+ if not isinstance(settings, dict):
+ settings = {}
+ stream = item.get('stream')
+ if not isinstance(stream, dict):
+ stream = {}
+ game = stream.get('game')
+ if not isinstance(game, dict):
+ game = {}
+ profile = item.get('profileImageURL') or ''
+
+ return {
+ Keys.ID: item['id'],
+ Keys.BROADCASTER_LOGIN: item['login'],
+ Keys.DISPLAY_NAME: item['displayName'],
+ Keys.BROADCASTER_LANGUAGE: settings.get('language') or '',
+ Keys.TITLE: settings.get('title') or '',
+ Keys.OFFLINE_IMAGE_URL: profile,
+ Keys.THUMBNAIL_URL: stream.get('previewImageURL') or profile,
+ Keys.VIEWER_COUNT: stream.get('viewersCount') or 0,
+ Keys.GAME_NAME: game.get('name') or game.get('displayName') or '',
+ Keys.GAME_ID: game.get('id') or '',
+ }
+
+
+def _game_item(item):
+ if not isinstance(item, dict):
+ return None
+ name = item.get('name') or item.get('displayName')
+ if not item.get('id') or not name:
+ return None
+ return {
+ Keys.ID: item['id'],
+ Keys.NAME: name,
+ Keys.BOX_ART_URL: item.get('boxArtURL') or '',
+ }
+
+
+def _get_edges(search_for, container_name):
+ container = search_for.get(container_name)
+ if not isinstance(container, dict):
+ return None
+ edges = container.get('edges')
+ if not isinstance(edges, list) or not edges:
+ return None
+ return edges
+
+
+def search(search_query, kind):
+ """Return Helix-shaped search data, or None to request Helix fallback."""
+ try:
+ if kind == 'games':
+ search_for = _post(_GAME_QUERY, search_query)
+ container_name = 'games'
+ adapter = _game_item
+ elif kind in ('channels', 'streams'):
+ search_for = _post(_CHANNEL_QUERY, search_query)
+ container_name = 'channels'
+ adapter = _channel_item
+ else:
+ return None
+
+ if search_for is None:
+ return None
+ edges = _get_edges(search_for, container_name)
+ if edges is None:
+ return None
+
+ items = []
+ for edge in edges:
+ if not isinstance(edge, dict):
+ continue
+ source = edge.get('item')
+ if kind == 'streams' and (
+ not isinstance(source, dict)
+ or not isinstance(source.get('stream'), dict)
+ or not source.get('stream')):
+ continue
+ item = adapter(source)
+ if item is not None:
+ items.append(item)
+ if not items:
+ return None
+ return {Keys.DATA: items}
+ except Exception as error:
+ log_utils.log(
+ 'gql_search failed: %s' % error.__class__.__name__,
+ log_utils.LOGWARNING,
+ )
+ return None
diff --git a/resources/lib/twitch_addon/addon/utils.py b/resources/lib/twitch_addon/addon/utils.py
index 0afff22c..67636846 100644
--- a/resources/lib/twitch_addon/addon/utils.py
+++ b/resources/lib/twitch_addon/addon/utils.py
@@ -333,6 +333,13 @@ def get_search_history_size():
return int(kodi.get_setting('search_history_size'))
+def get_search_backend():
+ backend = kodi.get_setting('search_backend')
+ if isinstance(backend, bool):
+ return 0
+ return 1 if backend in (1, '1') else 0
+
+
def get_search_history(search_type):
history = None
history_size = get_search_history_size()
diff --git a/resources/settings.xml b/resources/settings.xml
index a09fddf3..dc371d30 100644
--- a/resources/settings.xml
+++ b/resources/settings.xml
@@ -380,6 +380,17 @@
+
+ 0
+ 0
+
+
+
+
+
+
+
+
0
50
diff --git a/tests/test_gql_search.py b/tests/test_gql_search.py
new file mode 100644
index 00000000..daca4e7e
--- /dev/null
+++ b/tests/test_gql_search.py
@@ -0,0 +1,232 @@
+import importlib.util
+import sys
+import types
+import unittest
+from pathlib import Path
+from unittest.mock import MagicMock
+
+
+ROOT = Path(__file__).resolve().parents[1]
+GQL_SEARCH = ROOT / 'resources' / 'lib' / 'twitch_addon' / 'addon' / 'gql_search.py'
+
+
+class _Names(object):
+ def __getattr__(self, name):
+ return name.lower()
+
+
+def load_gql_search():
+ for name in ('twitch_addon', 'twitch_addon.addon'):
+ package = types.ModuleType(name)
+ package.__path__ = []
+ sys.modules[name] = package
+
+ constants = types.ModuleType('twitch_addon.addon.constants')
+ constants.Keys = _Names()
+ sys.modules[constants.__name__] = constants
+
+ log_utils = types.ModuleType('twitch_addon.addon.common.log_utils')
+ log_utils.LOGWARNING = 2
+ log_utils.log = MagicMock()
+ common = types.ModuleType('twitch_addon.addon.common')
+ common.log_utils = log_utils
+ sys.modules[common.__name__] = common
+ sys.modules[log_utils.__name__] = log_utils
+
+ utils = types.ModuleType('twitch_addon.addon.utils')
+ utils.get_private_client_id = MagicMock(return_value='website-client-id')
+ sys.modules[utils.__name__] = utils
+
+ requests = types.ModuleType('requests')
+ requests.RequestException = RequestException
+ requests.post = MagicMock()
+ sys.modules['requests'] = requests
+
+ spec = importlib.util.spec_from_file_location(
+ 'twitch_addon.addon.gql_search', GQL_SEARCH
+ )
+ module = importlib.util.module_from_spec(spec)
+ sys.modules[spec.name] = module
+ spec.loader.exec_module(module)
+ return module, requests, utils, log_utils
+
+
+class RequestException(Exception):
+ pass
+
+
+class FakeResponse(object):
+ def __init__(self, payload=None, status_error=None, json_error=None):
+ self.payload = payload
+ self.status_error = status_error
+ self.json_error = json_error
+
+ def raise_for_status(self):
+ if self.status_error:
+ raise self.status_error
+
+ def json(self):
+ if self.json_error:
+ raise self.json_error
+ return self.payload
+
+
+def channel_payload(edges):
+ return [{
+ 'data': {
+ 'searchFor': {
+ 'channels': {'edges': edges},
+ },
+ },
+ }]
+
+
+def game_payload(edges):
+ return {
+ 'data': {
+ 'searchFor': {
+ 'games': {'edges': edges},
+ },
+ },
+ }
+
+
+LIVE_CHANNEL = {
+ 'id': '1',
+ 'login': 'live_login',
+ 'displayName': 'Live Name',
+ 'broadcastSettings': {'language': 'en', 'title': 'Live title'},
+ 'profileImageURL': 'profile.jpg',
+ 'stream': {
+ 'id': 'stream-1',
+ 'viewersCount': 42,
+ 'previewImageURL': 'preview.jpg',
+ 'game': {'id': 'game-1', 'name': 'Game'},
+ },
+}
+OFFLINE_CHANNEL = {
+ 'id': '2',
+ 'login': 'offline_login',
+ 'displayName': 'Offline Name',
+ 'stream': None,
+}
+
+
+class GqlSearchTests(unittest.TestCase):
+ def setUp(self):
+ self.module, self.requests, self.utils, self.log_utils = load_gql_search()
+
+ def respond(self, payload):
+ self.requests.post.return_value = FakeResponse(payload)
+
+ def test_channel_search_maps_live_and_offline_converter_fields(self):
+ self.respond(channel_payload([
+ {'item': LIVE_CHANNEL},
+ {'item': OFFLINE_CHANNEL},
+ ]))
+
+ result = self.module.search('query', 'channels')
+
+ self.assertNotIn('pagination', result)
+ self.assertEqual(2, len(result['data']))
+ live, offline = result['data']
+ self.assertEqual('1', live['id'])
+ self.assertEqual('live_login', live['broadcaster_login'])
+ self.assertEqual('Live Name', live['display_name'])
+ self.assertEqual('en', live['broadcaster_language'])
+ self.assertEqual('Live title', live['title'])
+ self.assertEqual('profile.jpg', live['offline_image_url'])
+ self.assertEqual('preview.jpg', live['thumbnail_url'])
+ self.assertEqual(42, live['viewer_count'])
+ self.assertEqual('Game', live['game_name'])
+ self.assertEqual('game-1', live['game_id'])
+ self.assertEqual('', offline['title'])
+ self.assertEqual('', offline['broadcaster_language'])
+ self.assertEqual('', offline['thumbnail_url'])
+ self.assertEqual(0, offline['viewer_count'])
+ self.assertEqual('', offline['game_name'])
+ self.assertEqual('', offline['game_id'])
+
+ def test_stream_search_filters_offline_channels(self):
+ self.respond(channel_payload([
+ {'item': OFFLINE_CHANNEL},
+ {'item': dict(OFFLINE_CHANNEL, stream={})},
+ {'item': LIVE_CHANNEL},
+ ]))
+
+ result = self.module.search('query', 'streams')
+
+ self.assertEqual(['1'], [item['id'] for item in result['data']])
+
+ def test_game_search_maps_name_fallback_and_box_art(self):
+ self.respond(game_payload([
+ {'item': {'id': '10', 'name': 'First', 'boxArtURL': 'first.jpg'}},
+ {'item': {'id': '11', 'displayName': 'Second'}},
+ ]))
+
+ result = self.module.search('query', 'games')
+
+ self.assertEqual([
+ {'id': '10', 'name': 'First', 'box_art_url': 'first.jpg'},
+ {'id': '11', 'name': 'Second', 'box_art_url': ''},
+ ], result['data'])
+
+ def test_malformed_items_are_skipped_but_valid_items_remain(self):
+ self.respond(channel_payload([
+ {},
+ {'item': None},
+ {'item': {'id': '1', 'login': 'missing-display'}},
+ {'item': LIVE_CHANNEL},
+ ]))
+
+ result = self.module.search('query', 'channels')
+
+ self.assertEqual(['1'], [item['id'] for item in result['data']])
+
+ def test_request_is_anonymous_batched_and_bounded(self):
+ self.respond(channel_payload([{'item': LIVE_CHANNEL}]))
+
+ self.module.search('needle', 'channels')
+
+ args, kwargs = self.requests.post.call_args
+ self.assertEqual(('https://gql.twitch.tv/gql',), args)
+ self.assertEqual(15, kwargs['timeout'])
+ self.assertEqual({'Client-ID': 'website-client-id'}, kwargs['headers'])
+ self.assertNotIn('Authorization', kwargs['headers'])
+ self.assertIsInstance(kwargs['json'], list)
+ self.assertEqual('needle', kwargs['json'][0]['variables']['q'])
+ self.utils.get_private_client_id.assert_called_once_with()
+
+ def test_failure_shapes_return_none_for_helix_fallback(self):
+ failures = [
+ [],
+ 'not-an-envelope',
+ {'errors': [{'message': 'failed'}]},
+ {'data': None},
+ {'data': {'searchFor': []}},
+ {'data': {'searchFor': {'channels': []}}},
+ {'data': {'searchFor': {'channels': {'edges': 'bad'}}}},
+ channel_payload([]),
+ channel_payload([{'item': {'id': '1'}}]),
+ ]
+ for payload in failures:
+ with self.subTest(payload=payload):
+ self.respond(payload)
+ self.assertIsNone(self.module.search('query', 'channels'))
+
+ def test_request_http_and_json_errors_return_none(self):
+ errors = [
+ FakeResponse(status_error=RequestException('http')),
+ FakeResponse(json_error=ValueError('json')),
+ ]
+ for response in errors:
+ with self.subTest(response=response):
+ self.requests.post.return_value = response
+ self.assertIsNone(self.module.search('query', 'channels'))
+
+ self.requests.post.side_effect = RequestException('timeout')
+ self.assertIsNone(self.module.search('query', 'channels'))
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/tests/test_search_backend.py b/tests/test_search_backend.py
new file mode 100644
index 00000000..16630574
--- /dev/null
+++ b/tests/test_search_backend.py
@@ -0,0 +1,212 @@
+import ast
+import importlib.util
+import sys
+import types
+import unittest
+import xml.etree.ElementTree as ET
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+
+ROOT = Path(__file__).resolve().parents[1]
+API = ROOT / 'resources' / 'lib' / 'twitch_addon' / 'addon' / 'api.py'
+UTILS = ROOT / 'resources' / 'lib' / 'twitch_addon' / 'addon' / 'utils.py'
+SETTINGS = ROOT / 'resources' / 'settings.xml'
+
+
+class _Names(object):
+ def __getattr__(self, name):
+ return name.lower()
+
+
+class _Parameters(object):
+ FALSE = False
+ TRUE = True
+ ALL = ''
+ TIME = ''
+
+
+def identity_decorator(func=None, **kwargs):
+ if func is not None:
+ return func
+ return lambda wrapped: wrapped
+
+
+def load_api():
+ for name in ('twitch_addon', 'twitch_addon.addon'):
+ package = types.ModuleType(name)
+ package.__path__ = []
+ sys.modules[name] = package
+
+ cache = types.ModuleType('twitch_addon.addon.cache')
+ cache.limit = 1
+ cache.cache_method = lambda cache_limit: identity_decorator
+ cache.reset_cache = MagicMock()
+ sys.modules[cache.__name__] = cache
+
+ utils = types.ModuleType('twitch_addon.addon.utils')
+ utils.i18n = lambda value: value
+ utils.get_client_id = MagicMock(return_value='client-id')
+ utils.get_oauth_token = MagicMock(return_value='')
+ utils.get_search_backend = MagicMock(return_value=0)
+ sys.modules[utils.__name__] = utils
+
+ gql_search = types.ModuleType('twitch_addon.addon.gql_search')
+ gql_search.search = MagicMock()
+ sys.modules[gql_search.__name__] = gql_search
+
+ kodi = types.ModuleType('twitch_addon.addon.common.kodi')
+ log_utils = types.ModuleType('twitch_addon.addon.common.log_utils')
+ log_utils.LOGWARNING = 2
+ log_utils.LOGINFO = 1
+ log_utils.LOGDEBUG = 0
+ log_utils.log = MagicMock()
+ common = types.ModuleType('twitch_addon.addon.common')
+ common.kodi = kodi
+ common.log_utils = log_utils
+ sys.modules[common.__name__] = common
+ sys.modules[kodi.__name__] = kodi
+ sys.modules[log_utils.__name__] = log_utils
+
+ common_cache = types.ModuleType('twitch_addon.addon.common.cache')
+ common_cache.invalidate_cache_for_function = MagicMock()
+ sys.modules[common_cache.__name__] = common_cache
+
+ constants = types.ModuleType('twitch_addon.addon.constants')
+ constants.Keys = _Names()
+ constants.SCOPES = []
+ sys.modules[constants.__name__] = constants
+
+ error_handling = types.ModuleType('twitch_addon.addon.error_handling')
+ error_handling.api_error_handler = identity_decorator
+ sys.modules[error_handling.__name__] = error_handling
+
+ exceptions = types.ModuleType('twitch_addon.addon.twitch_exceptions')
+ exceptions.PlaybackFailed = type('PlaybackFailed', (Exception,), {})
+ exceptions.TwitchException = type('TwitchException', (Exception,), {})
+ sys.modules[exceptions.__name__] = exceptions
+
+ twitch_package = types.ModuleType('twitch')
+ twitch_package.__path__ = []
+ queries = types.ModuleType('twitch.queries')
+ oauth = types.ModuleType('twitch.oauth')
+ oauth.clients = types.SimpleNamespace(MobileClient=MagicMock())
+ twitch_package.queries = queries
+ twitch_package.oauth = oauth
+ sys.modules['twitch'] = twitch_package
+ sys.modules['twitch.queries'] = queries
+ sys.modules['twitch.oauth'] = oauth
+
+ twitch_api = types.ModuleType('twitch.api')
+ twitch_api.__path__ = []
+ usher = types.ModuleType('twitch.api.usher')
+ helix = types.ModuleType('twitch.api.helix')
+ helix.search = types.SimpleNamespace(
+ get_channels=MagicMock(return_value={'data': [{'id': 'helix'}]}),
+ get_categories=MagicMock(return_value={'data': [{'id': 'helix'}]}),
+ )
+ twitch_api.usher = usher
+ twitch_api.helix = helix
+ sys.modules['twitch.api'] = twitch_api
+ sys.modules['twitch.api.usher'] = usher
+ sys.modules['twitch.api.helix'] = helix
+
+ parameters = types.ModuleType('twitch.api.parameters')
+ for name in (
+ 'Language', 'Boolean', 'VideoSort', 'PeriodHelix'
+ ):
+ setattr(parameters, name, _Parameters)
+ sys.modules[parameters.__name__] = parameters
+
+ spec = importlib.util.spec_from_file_location('twitch_addon.addon.api', API)
+ module = importlib.util.module_from_spec(spec)
+ sys.modules[spec.name] = module
+ spec.loader.exec_module(module)
+ return module, utils, gql_search
+
+
+def load_get_search_backend(get_setting):
+ tree = ast.parse(UTILS.read_text())
+ function = next(
+ node for node in tree.body
+ if isinstance(node, ast.FunctionDef) and node.name == 'get_search_backend'
+ )
+ module = ast.Module(body=[function], type_ignores=[])
+ namespace = {'kodi': types.SimpleNamespace(get_setting=get_setting)}
+ exec(compile(module, str(UTILS), 'exec'), namespace)
+ return namespace['get_search_backend']
+
+
+class SearchBackendTests(unittest.TestCase):
+ def test_setting_only_exact_one_selects_helix(self):
+ for value, expected in (
+ ('1', 1), (1, 1), ('0', 0), (0, 0), ('', 0),
+ ('invalid', 0), ('2', 0), ('01', 0), (True, 0), (None, 0),
+ ):
+ with self.subTest(value=value):
+ backend = load_get_search_backend(lambda name, value=value: value)
+ self.assertEqual(expected, backend())
+
+ def test_search_backend_setting_uses_current_integer_spinner_schema(self):
+ setting = ET.parse(SETTINGS).find(".//setting[@id='search_backend']")
+
+ if setting is None:
+ self.fail('search_backend setting is missing')
+ control = setting.find('control')
+ if control is None:
+ self.fail('search_backend control is missing')
+ self.assertEqual('integer', setting.get('type'))
+ self.assertEqual('30333', setting.get('label'))
+ self.assertEqual('0', setting.findtext('default'))
+ self.assertEqual('spinner', control.get('type'))
+ self.assertEqual(
+ [('30334', '0'), ('30335', '1')],
+ [(option.get('label'), option.text)
+ for option in setting.findall('./constraints/options/option')],
+ )
+
+ def test_gql_success_is_first_page_only_and_has_backend_cache_identity(self):
+ module, utils, gql = load_api()
+ twitch = module.Twitch.__new__(module.Twitch)
+ gql.search.return_value = {'data': [{'id': 'gql'}]}
+
+ with patch.object(twitch, '_get_channel_search', wraps=twitch._get_channel_search) as cached:
+ first = twitch.get_channel_search('query')
+ utils.get_search_backend.return_value = 1
+ helix = twitch.get_channel_search('query')
+
+ self.assertEqual({'data': [{'id': 'gql'}]}, first)
+ self.assertEqual({'data': [{'id': 'helix'}]}, helix)
+ self.assertEqual([0, 1], [call.args[-1] for call in cached.call_args_list])
+ self.assertNotIn('pagination', first)
+
+ def test_gql_failure_falls_back_to_existing_helix_call(self):
+ module, utils, gql = load_api()
+ twitch = module.Twitch.__new__(module.Twitch)
+ gql.search.return_value = None
+
+ result = twitch.get_stream_search('query', first=15)
+
+ self.assertEqual({'data': [{'id': 'helix'}]}, result)
+ gql.search.assert_called_once_with('query', 'streams')
+ module.Twitch.api.search.get_channels.assert_called_once_with(
+ search_query='query', after='MA==', first=15,
+ live_only=_Parameters.TRUE,
+ )
+
+ def test_helix_setting_and_non_initial_cursor_bypass_gql(self):
+ module, utils, gql = load_api()
+ twitch = module.Twitch.__new__(module.Twitch)
+
+ utils.get_search_backend.return_value = 1
+ twitch.get_game_search('query')
+ utils.get_search_backend.return_value = 0
+ twitch.get_channel_search('query', after='next-cursor')
+
+ gql.search.assert_not_called()
+ module.Twitch.api.search.get_categories.assert_called_once()
+ module.Twitch.api.search.get_channels.assert_called_once()
+
+
+if __name__ == '__main__':
+ unittest.main()