From 6bf126d432ebfa1f83d20077084448aacb51a3ac Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 23 Jul 2026 19:42:12 +0000 Subject: [PATCH] feat(epg): Enhance Schedules Direct token management and request handling This commit introduces a new `sd_authorized_request` function to streamline authenticated API requests to Schedules Direct. It handles token expiration by clearing the cached token and retrying the request with a fresh token when necessary. Additionally, the `sd_obtain_token` function has been updated to reuse cached tokens when valid, improving efficiency. The changes are accompanied by unit tests to ensure correct behavior during token reuse and request retries, enhancing the overall reliability of the Schedules Direct integration. --- CHANGELOG.md | 2 +- apps/epg/sd_api.py | 89 +++++++-------- apps/epg/sd_tasks.py | 74 ++++++------- apps/epg/sd_utils.py | 85 ++++++++++++++- apps/epg/tests/test_schedules_direct.py | 138 ++++++++++++++++++++++++ 5 files changed, 300 insertions(+), 88 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d63bab5c..a2f6f76c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,7 +20,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Schedules Direct poster proxy no longer ignores daily image download limits or missing-image errors.** SD returns codes 5002/5003 as HTTP 200 with a JSON body; the proxy previously only inspected HTTP 400 and kept requesting images after the limit, which can get accounts blocked. It now detects 5002/5003 (subscriber and trial), stops further image fetches for that source until the next midnight UTC (persisted on the EPG source so all workers honor it), and on explicit IMAGE_NOT_FOUND (code 5000) clears that program's `sd_icon` so the same dead URI is not retried. Transient bare HTTP 404s do not blacklist the URI. Successful SD posters use a dedicated nginx cache under `/data/cache/sd_posters` (14 days); channel/VOD logos remain at `/data/cache/logos` (24 hours). On startup, an existing `/data/logo_cache` directory is moved to `/data/cache/logos` once so warm logo entries are kept. API and XMLTV icon URLs include a `?v=` hash of the stored SD image URI so a new artwork link after refresh uses a new cache key without waiting for TTL expiry. Selecting programs that still need artwork no longer uses a negated JSON `AND` that dropped every row when `sd_icon_missing` / `sd_poster_style` keys were absent (Postgres NULL), which had prevented poster links from being saved after refresh. - **Schedules Direct auth and lineup change handling better match SD guidelines.** Token codes 3000/3001 (offline/busy) stop as idle with clear "do not retry" messaging instead of looking like credential failures; 4010 (too many unique IPs) and related account codes get explicit messages. Lineup add/delete respects a known daily change lockout before calling SD, handles 4100 on deletes, and the UI blocks remove when no changes remain (adds and deletes both count toward the 6/day limit). -- **Schedules Direct stops retrying `/token` on auth failures that will not clear by themselves.** Codes 4002/4003 (bad credentials), 4001/4005/4007/4008 (account state), 4009/4010 (login/IP limits), and 4004 (15-minute account lock) persist a cooldown on the EPG source. Soft codes 3000/3001 use a 1-hour idle cooldown. Lockouts clear when username/password change or the cooldown expires. Token auth is centralized in `sd_obtain_token` for refresh, lineup/form, and poster paths. Redis-cached SD tokens are bound to a credential fingerprint (and cleared when the EPG source username/password change) so switching accounts cannot reuse another session. +- **Schedules Direct stops retrying `/token` on auth failures that will not clear by themselves.** Codes 4002/4003 (bad credentials), 4001/4005/4007/4008 (account state), 4009/4010 (login/IP limits), and 4004 (15-minute account lock) persist a cooldown on the EPG source. Soft codes 3000/3001 use a 1-hour idle cooldown. Lockouts clear when username/password change or the cooldown expires. Token auth is centralized in `sd_obtain_token` for refresh, lineup/form, and poster paths. That helper reuses a Redis-cached token until near `tokenExpires` (bound to a credential fingerprint, and cleared when the EPG source username/password change) so opening the SD form, refreshing EPG, and serving posters do not mint a new token on every call. When an authenticated SD call returns HTTP 401/403 (SD documents TOKEN_EXPIRED as 403), the cached token is cleared and the request is retried once with a fresh `/token`. - **Schedules Direct code split out of general EPG modules.** Refresh pipeline and Celery tasks live in `apps/epg/sd_tasks.py`; lineup/poster API mixins in `apps/epg/sd_api.py`; shared protocol helpers remain in `sd_utils.py`. Progress WebSocket updates (`send_epg_update`) live in `apps/epg/utils.py` so XMLTV and SD tasks can import cleanly. `tasks.py` / `api_views.py` re-export or inherit so existing imports and Celery task names are unchanged. ## [0.28.1] - 2026-07-20 diff --git a/apps/epg/sd_api.py b/apps/epg/sd_api.py index 577d72bd..28d15df0 100644 --- a/apps/epg/sd_api.py +++ b/apps/epg/sd_api.py @@ -25,10 +25,8 @@ from apps.epg.sd_utils import ( SD_CODE_IMAGE_NOT_FOUND, SD_IMAGE_LIMIT_CODES, sd_auth_lockout_active, - sd_clear_cached_token, - sd_get_cached_token, + sd_authorized_request, sd_handle_2055, - sd_headers_for_source, sd_image_limit_active, sd_mark_icon_missing, sd_next_midnight_utc, @@ -161,14 +159,14 @@ class SchedulesDirectSourceMixin: if error: return error - headers = sd_headers_for_source(source, token=token) - if request.method == "GET": countries = self._fetch_sd_countries() try: - resp = http_requests.get( + resp, token = sd_authorized_request( + 'GET', f"{SD_BASE_URL}/lineups", - headers=headers, + source=source, + token=token, timeout=15, ) if resp.status_code == 400: @@ -219,9 +217,11 @@ class SchedulesDirectSourceMixin: }, status=status.HTTP_200_OK) try: - resp = http_requests.put( + resp, token = sd_authorized_request( + 'PUT', f"{SD_BASE_URL}/lineups/{lineup_id}", - headers=headers, + source=source, + token=token, timeout=15, ) sd_data = resp.json() @@ -302,9 +302,11 @@ class SchedulesDirectSourceMixin: }, status=status.HTTP_200_OK) try: - resp = http_requests.delete( + resp, token = sd_authorized_request( + 'DELETE', f"{SD_BASE_URL}/lineups/{lineup_id}", - headers=headers, + source=source, + token=token, timeout=15, ) if resp.status_code == 400: @@ -372,14 +374,14 @@ class SchedulesDirectSourceMixin: if error: return error - headers = sd_headers_for_source(source, token=token) - try: - resp = http_requests.get( + resp, token = sd_authorized_request( + 'GET', f"{SD_BASE_URL}/headends", - params={'country': country, 'postalcode': postalcode}, - headers=headers, + source=source, + token=token, timeout=15, + params={'country': country, 'postalcode': postalcode}, ) try: headends = resp.json() @@ -471,37 +473,35 @@ class SchedulesDirectPosterMixin: status=status.HTTP_503_SERVICE_UNAVAILABLE, ) - token = sd_get_cached_token( - source.id, username=source.username, password=source.password - ) - - if not token: - auth = sd_obtain_token(source, timeout=10) - if auth.debug_rejected: - return Response( - {'error': auth.message}, - status=status.HTTP_400_BAD_REQUEST, - ) - if not auth.ok: - # Lockout codes are persisted by sd_obtain_token. Other failures - # use a short process-local cache so workers do not hammer /token. - if not auth.lockout: - self._sd_poster_error_cache[source.id] = { - 'until': time.time() + (3600 if auth.code else 300), - 'reason': auth.message or 'Authentication failed', - } - return Response( - {'error': auth.message}, - status=status.HTTP_502_BAD_GATEWAY, - ) - token = auth.token - self._sd_poster_error_cache.pop(source.id, None) + auth = sd_obtain_token(source, timeout=10) + if auth.debug_rejected: + return Response( + {'error': auth.message}, + status=status.HTTP_400_BAD_REQUEST, + ) + if not auth.ok: + # Lockout codes are persisted by sd_obtain_token. Other failures + # use a short process-local cache so workers do not hammer /token. + if not auth.lockout: + self._sd_poster_error_cache[source.id] = { + 'until': time.time() + (3600 if auth.code else 300), + 'reason': auth.message or 'Authentication failed', + } + return Response( + {'error': auth.message}, + status=status.HTTP_502_BAD_GATEWAY, + ) + token = auth.token + self._sd_poster_error_cache.pop(source.id, None) try: - img_resp = http_requests.get( + img_resp, token = sd_authorized_request( + 'GET', poster_sd_url, - headers=sd_headers_for_source(source, token=token, content_type=None), + source=source, + token=token, timeout=15, + content_type=None, allow_redirects=True, ) err_code, err_data = sd_parse_response_payload(img_resp) @@ -539,7 +539,8 @@ class SchedulesDirectPosterMixin: return Response(status=status.HTTP_404_NOT_FOUND) if img_resp.status_code in (401, 403): - sd_clear_cached_token(source.id) + # Clear already happened inside sd_authorized_request; seed a short + # process cache only when the fresh-token retry still failed. self._sd_poster_error_cache[source.id] = { 'until': time.time() + 3600, 'reason': f'SD returned {img_resp.status_code}', diff --git a/apps/epg/sd_tasks.py b/apps/epg/sd_tasks.py index 5356400b..349b1b27 100644 --- a/apps/epg/sd_tasks.py +++ b/apps/epg/sd_tasks.py @@ -22,7 +22,7 @@ from django.utils import timezone from apps.channels.models import Channel from apps.epg.models import EPGData, EPGSource, ProgramData, SDProgramMD5, SDScheduleMD5 -from apps.epg.sd_utils import SD_BASE_URL, sd_headers_for_source, sd_obtain_token +from apps.epg.sd_utils import SD_BASE_URL, sd_authorized_request, sd_obtain_token from apps.epg.utils import send_epg_update from core.utils import ( acquire_task_lock, @@ -291,14 +291,10 @@ def _sd_program_ids_needing_posters(mapped_epg_ids, poster_style): ) -def _sd_fetch_lineup_country(token, sd_headers_fn): +def _sd_fetch_lineup_country(sd_req): """Return country code prefix from the first subscribed lineup (poster metadata).""" try: - lineups_response = requests.get( - f"{SD_BASE_URL}/lineups", - headers=sd_headers_fn(token), - timeout=30, - ) + lineups_response = sd_req('GET', f"{SD_BASE_URL}/lineups", timeout=30) if lineups_response.ok: for lineup in lineups_response.json().get('lineups', []): lid = lineup.get('lineupID') or lineup.get('lineup') or '' @@ -309,7 +305,7 @@ def _sd_fetch_lineup_country(token, sd_headers_fn): return None -def _sd_setup_single_epg_fetch(source, epg_id, token, sd_headers_fn): +def _sd_setup_single_epg_fetch(source, epg_id, sd_req): """Build station_map / epg_id_map for a single mapped EPG entry.""" epg = EPGData.objects.filter(id=epg_id, epg_source=source).first() if not epg or not epg.tvg_id: @@ -320,7 +316,7 @@ def _sd_setup_single_epg_fetch(source, epg_id, token, sd_headers_fn): send_epg_update(source.id, "parsing_programs", 100, status="error", error=msg) return None - sd_lineup_country = _sd_fetch_lineup_country(token, sd_headers_fn) + sd_lineup_country = _sd_fetch_lineup_country(sd_req) send_epg_update( source.id, "parsing_programs", 15, @@ -331,7 +327,7 @@ def _sd_setup_single_epg_fetch(source, epg_id, token, sd_headers_fn): return station_map, epg_id_map, sd_lineup_country, epg -def _sd_setup_mapped_guide_fetch(source, token, sd_headers_fn): +def _sd_setup_mapped_guide_fetch(source, sd_req): """Build station_map / epg_id_map for all channels mapped to this SD source.""" from apps.channels.models import Channel @@ -368,7 +364,7 @@ def _sd_setup_mapped_guide_fetch(source, token, sd_headers_fn): send_epg_update(source.id, "parsing_programs", 100, status="error", error=msg) return None - sd_lineup_country = _sd_fetch_lineup_country(token, sd_headers_fn) + sd_lineup_country = _sd_fetch_lineup_country(sd_req) send_epg_update( source.id, "parsing_programs", 15, message=f"Fetching guide data for {len(station_map)} mapped stations...", @@ -605,15 +601,10 @@ def fetch_schedules_direct( logger.info(f"SD source {source.id}: Force flag set, bypassing 2-hour refresh guard.") # ------------------------------------------------------------------------- - # Build SD-specific headers + # SD API helpers (authenticated requests defined after /token succeeds) # SD API spec requires the User-Agent to identify the application and version. # SergeantPanda confirmed Dispatcharr should identify itself properly. # ------------------------------------------------------------------------- - from apps.epg.sd_utils import sd_headers_for_source, sd_obtain_token - - def _sd_headers(token=None, content_type='application/json'): - return sd_headers_for_source(source, token=token, content_type=content_type) - def _sd_post_refresh_tasks(mapped_epg_ids, program_metadata, today): """Poster fetch, logo auto-apply, and pruning. Runs even when schedules are unchanged. @@ -663,10 +654,10 @@ def fetch_schedules_direct( for batch_idx in range(total_art_batches): batch = artwork_list[batch_idx * SD_ARTWORK_BATCH_SIZE:(batch_idx + 1) * SD_ARTWORK_BATCH_SIZE] try: - art_response = requests.post( + art_response = _sd_req( + 'POST', f"{SD_BASE_URL}/metadata/programs/", json=batch, - headers=_sd_headers(token), timeout=120, ) art_response.raise_for_status() @@ -814,15 +805,24 @@ def fetch_schedules_direct( token = auth.token logger.info("Schedules Direct authentication successful.") + def _sd_req(method, url, **kwargs): + nonlocal token + content_type = kwargs.pop('content_type', 'application/json') + resp, token = sd_authorized_request( + method, + url, + source=source, + token=token, + content_type=content_type, + **kwargs, + ) + return resp + # ------------------------------------------------------------------------- # Step 2: Check account status (respect OFFLINE system status) # ------------------------------------------------------------------------- try: - status_response = requests.get( - f"{SD_BASE_URL}/status", - headers=_sd_headers(token), - timeout=30, - ) + status_response = _sd_req('GET', f"{SD_BASE_URL}/status", timeout=30) status_response.raise_for_status() status_data = status_response.json() system_status = status_data.get('systemStatus', [{}])[0].get('status', 'Online') @@ -848,12 +848,12 @@ def fetch_schedules_direct( sd_lineup_country = None if epg_id_only is not None: - setup = _sd_setup_single_epg_fetch(source, epg_id_only, token, _sd_headers) + setup = _sd_setup_single_epg_fetch(source, epg_id_only, _sd_req) if setup is None: return station_map, epg_id_map, sd_lineup_country, _single_epg = setup elif mapped_guide_batch: - setup = _sd_setup_mapped_guide_fetch(source, token, _sd_headers) + setup = _sd_setup_mapped_guide_fetch(source, _sd_req) if setup is None: return station_map, epg_id_map, sd_lineup_country = setup @@ -863,11 +863,7 @@ def fetch_schedules_direct( # ------------------------------------------------------------------------- send_epg_update(source.id, "parsing_programs", 10, message="Fetching subscribed lineups...") try: - lineups_response = requests.get( - f"{SD_BASE_URL}/lineups", - headers=_sd_headers(token), - timeout=30, - ) + lineups_response = _sd_req('GET', f"{SD_BASE_URL}/lineups", timeout=30) # SD returns 400 with code 4102 when no lineups are configured. # This is a valid account state. The user needs to add lineups via # the Manage Lineups UI. Treat as idle rather than error. @@ -919,9 +915,9 @@ def fetch_schedules_direct( if not lineup_id: continue try: - detail_response = requests.get( + detail_response = _sd_req( + 'GET', f"{SD_BASE_URL}/lineups/{lineup_id}", - headers=_sd_headers(token), timeout=30, ) detail_response.raise_for_status() @@ -1130,10 +1126,10 @@ def fetch_schedules_direct( ] for batch in station_batches: try: - md5_response = requests.post( + md5_response = _sd_req( + 'POST', f"{SD_BASE_URL}/schedules/md5", json=[{'stationID': sid, 'date': date_list} for sid in batch], - headers=_sd_headers(token), timeout=120, ) md5_response.raise_for_status() @@ -1257,10 +1253,10 @@ def fetch_schedules_direct( if not request_body: continue try: - sched_response = requests.post( + sched_response = _sd_req( + 'POST', f"{SD_BASE_URL}/schedules", json=request_body, - headers=_sd_headers(token), timeout=120, ) sched_response.raise_for_status() @@ -1392,10 +1388,10 @@ def fetch_schedules_direct( pass batch = program_id_list[batch_idx * SD_PROGRAM_BATCH_SIZE:(batch_idx + 1) * SD_PROGRAM_BATCH_SIZE] try: - prog_response = requests.post( + prog_response = _sd_req( + 'POST', f"{SD_BASE_URL}/programs", json=batch, - headers=_sd_headers(token), timeout=120, ) prog_response.raise_for_status() diff --git a/apps/epg/sd_utils.py b/apps/epg/sd_utils.py index 3198bc5f..35072056 100644 --- a/apps/epg/sd_utils.py +++ b/apps/epg/sd_utils.py @@ -394,11 +394,12 @@ _DEBUG_REJECTED_MESSAGE = ( def sd_obtain_token(source, username=None, password=None, *, timeout=30): """ - Authenticate with Schedules Direct and return a session token. + Return a Schedules Direct session token, reusing a cached one when valid. - Shared by refresh, lineup/form auth, and the poster proxy. Honors persisted - lockouts, reads JSON ``code`` before HTTP status, and persists cooldowns for - codes that must not be retried until the user or cooldown clears them. + Shared by refresh, lineup/form auth, and the poster proxy. Checks Redis for + an unexpired token bound to the current credentials before POSTing /token. + Honors persisted lockouts, reads JSON ``code`` before HTTP status, and + persists cooldowns for codes that must not be retried until cleared. """ if source is None: return SDTokenAuthResult( @@ -434,6 +435,14 @@ def sd_obtain_token(source, username=None, password=None, *, timeout=30): lockout=True, ) + cached = sd_get_cached_token(source.id, username=username, password=password) + if cached: + return SDTokenAuthResult( + ok=True, + token=cached, + code=0, + ) + sha1_password = hashlib.sha1(password.encode('utf-8')).hexdigest() try: response = requests.post( @@ -517,6 +526,74 @@ def sd_obtain_token(source, username=None, password=None, *, timeout=30): ) +def sd_authorized_request( + method, + url, + *, + source, + token, + username=None, + password=None, + timeout=30, + content_type='application/json', + **kwargs, +): + """ + Perform an authenticated Schedules Direct HTTP request. + + On HTTP 401/403 (SD documents TOKEN_EXPIRED as 403 + code 4006), clears the + Redis token cache, obtains a fresh token, and retries the request once. + + Returns ``(response, token)`` where ``token`` may have been refreshed. + """ + method_upper = (method or 'GET').upper() + http_fn = { + 'GET': requests.get, + 'POST': requests.post, + 'PUT': requests.put, + 'DELETE': requests.delete, + 'HEAD': requests.head, + 'PATCH': requests.patch, + }.get(method_upper) + + def _once(current_token): + headers = sd_headers_for_source( + source, + token=current_token, + content_type=content_type, + ) + if http_fn is not None: + return http_fn(url, headers=headers, timeout=timeout, **kwargs) + return requests.request( + method_upper, url, headers=headers, timeout=timeout, **kwargs + ) + + response = _once(token) + if response.status_code not in (401, 403): + return response, token + + logger.warning( + "SD source %s: %s %s returned %s; clearing cached token and retrying once", + getattr(source, 'id', None), + method_upper, + url, + response.status_code, + ) + sd_clear_cached_token(getattr(source, 'id', None)) + auth_timeout = timeout if isinstance(timeout, (int, float)) else 30 + auth = sd_obtain_token( + source, + username=username, + password=password, + timeout=min(int(auth_timeout), 30) if auth_timeout else 30, + ) + if not auth.ok or not auth.token: + return response, token + + retry_response = _once(auth.token) + return retry_response, auth.token + + def sd_next_midnight_utc(): """Return the next Schedules Direct counter reset (00:00Z).""" now = timezone.now() diff --git a/apps/epg/tests/test_schedules_direct.py b/apps/epg/tests/test_schedules_direct.py index 195ee677..4a3abe5b 100644 --- a/apps/epg/tests/test_schedules_direct.py +++ b/apps/epg/tests/test_schedules_direct.py @@ -1942,6 +1942,112 @@ class SDUtilsTests(TestCase): sd_get_cached_token(source.id, 'account-b', 'pass-a') ) + @patch('apps.epg.sd_utils.requests.post') + def test_obtain_token_reuses_cached_token(self, mock_post): + """Second obtain must not POST /token while Redis still has a valid token.""" + from apps.epg.sd_utils import sd_clear_cached_token, sd_obtain_token + + source = EPGSource.objects.create( + name='SD Token Reuse', + source_type='schedules_direct', + username='sduser', + password='sdpass', + ) + sd_clear_cached_token(source.id) + mock_post.return_value = MagicMock( + status_code=200, + json=MagicMock(return_value={ + 'code': 0, + 'token': 'tok-live', + 'tokenExpires': time.time() + 3600, + }), + raise_for_status=MagicMock(), + headers={}, + content=b'{}', + ) + first = sd_obtain_token(source) + self.assertTrue(first.ok) + self.assertEqual(first.token, 'tok-live') + mock_post.assert_called_once() + mock_post.reset_mock() + + second = sd_obtain_token(source) + self.assertTrue(second.ok) + self.assertEqual(second.token, 'tok-live') + mock_post.assert_not_called() + + @patch('apps.epg.sd_utils.requests.get') + @patch('apps.epg.sd_utils.requests.post') + def test_authorized_request_retries_once_on_403(self, mock_post, mock_get): + """SD TOKEN_EXPIRED is HTTP 403; clear cache and retry once with a fresh token.""" + from apps.epg.sd_utils import ( + sd_authorized_request, + sd_clear_cached_token, + sd_get_cached_token, + sd_set_cached_token, + ) + + source = EPGSource.objects.create( + name='SD Token Retry', + source_type='schedules_direct', + username='sduser', + password='sdpass', + ) + sd_clear_cached_token(source.id) + sd_set_cached_token( + source.id, 'stale-tok', time.time() + 3600, + username='sduser', password='sdpass', + ) + + expired = MagicMock( + status_code=403, + headers={'Content-Type': 'application/json'}, + content=b'{"code":4006,"response":"TOKEN_EXPIRED"}', + ) + expired.json = MagicMock(return_value={ + 'code': 4006, + 'response': 'TOKEN_EXPIRED', + 'message': 'Token has expired. Request new token.', + }) + ok = MagicMock( + status_code=200, + headers={'Content-Type': 'application/json'}, + content=b'{"lineups":[]}', + ) + ok.json = MagicMock(return_value={'lineups': []}) + mock_get.side_effect = [expired, ok] + mock_post.return_value = MagicMock( + status_code=200, + json=MagicMock(return_value={ + 'code': 0, + 'token': 'fresh-tok', + 'tokenExpires': time.time() + 3600, + }), + raise_for_status=MagicMock(), + headers={}, + content=b'{}', + ) + + resp, token = sd_authorized_request( + 'GET', + 'https://json.schedulesdirect.org/20141201/lineups', + source=source, + token='stale-tok', + timeout=15, + ) + self.assertEqual(resp.status_code, 200) + self.assertEqual(token, 'fresh-tok') + self.assertEqual(mock_get.call_count, 2) + mock_post.assert_called_once() + self.assertEqual( + sd_get_cached_token(source.id, 'sduser', 'sdpass'), + 'fresh-tok', + ) + retry_headers = mock_get.call_args_list[1].kwargs.get('headers') + if retry_headers is None: + retry_headers = mock_get.call_args_list[1][1].get('headers') + self.assertEqual(retry_headers.get('token'), 'fresh-tok') + class SDPosterProxyErrorHandlingTests(TestCase): """Poster proxy must honor SD image error codes so accounts are not blocked.""" @@ -2140,3 +2246,35 @@ class SDPosterProxyErrorHandlingTests(TestCase): mock_post.assert_not_called() self.assertEqual(mock_get.call_count, 2) + @patch('requests.get') + @patch('requests.post') + def test_poster_401_clears_token_and_retries_once(self, mock_post, mock_get): + """Invalid image token must be dropped and the image fetch retried once.""" + from apps.epg.sd_utils import sd_get_cached_token, sd_set_cached_token + + sd_set_cached_token( + self.source.id, 'stale-poster', time.time() + 3600, + username='sduser', password='sdpass', + ) + unauthorized = MagicMock(status_code=401, headers={}, content=b'') + unauthorized.json = MagicMock(side_effect=ValueError('not json')) + img = MagicMock() + img.status_code = 200 + img.headers = {'Content-Type': 'image/jpeg'} + img.content = b'\xff\xd8\xffjpeg-bytes' + img.json = MagicMock(side_effect=ValueError('not json')) + mock_get.side_effect = [unauthorized, img] + mock_post.return_value = self._auth_ok() + + resp = self.client.get(self.url) + self.assertEqual(resp.status_code, 200) + self.assertEqual(resp.content, b'\xff\xd8\xffjpeg-bytes') + self.assertEqual(mock_get.call_count, 2) + mock_post.assert_called_once() + self.assertEqual( + sd_get_cached_token(self.source.id, 'sduser', 'sdpass'), + 'poster-tok', + ) + from apps.epg.api_views import ProgramViewSet + self.assertNotIn(self.source.id, ProgramViewSet._sd_poster_error_cache) +