Mirrors/Dispatcharr
1
0
Fork 1
mirror of https://github.com/Dispatcharr/Dispatcharr.git synced 2026-07-17 16:50:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Implement schedule MD5 delta detection and backfill logic for improved cache management

Browse source 
- Added functions to compute schedule changes based on MD5 hashes and backfill missing dates for mapped stations.
- Enhanced handling of orphaned ProgramData records for unmapped EPG entries during schedule fetch.
- Introduced unit tests to validate MD5 comparison, backfill logic, and metadata fetching requirements.
This commit is contained in:
SergeantPanda 2026-06-11 12:31:30 -05:00
parent bfa7019324
commit 205deb150b
2 changed files with 675 additions and 108 deletions
Download patch file Download diff file Expand all files Collapse all files

309
apps/epg/tasks.py
View file

@ -35,6 +35,73 @@ SD_BASE_URL = 'https://json.schedulesdirect.org/20141201'
SD_DAYS_TO_FETCH = 20
SD_PROGRAM_BATCH_SIZE = 5000
def _sd_compute_schedule_changes_from_md5(server_md5s, cached_md5s, date_list):
"""Return station_id -> [date_str] for dates whose schedule MD5 differs from cache."""
changed_by_station = {}
for (sid, date_str), server_info in server_md5s.items():
if date_str not in date_list:
continue
cached = cached_md5s.get((sid, date_str))
if cached != server_info['md5']:
changed_by_station.setdefault(sid, []).append(date_str)
return changed_by_station
def _sd_backfill_schedule_dates_without_data(
changed_by_station,
server_md5s,
date_list,
mapped_station_ids,
epg_id_map,
dates_with_data,
cached_md5s,
stations_without_any_data,
):
"""
Add fetch-window dates that lack ProgramData to changed_by_station.
Dates with a cached schedule MD5 are treated as already fetched (e.g. legitimately
empty airings). Stations with zero ProgramData still backfill all missing dates
so stale cache from unmapped lineup refreshes cannot block guide population.
"""
from datetime import date as date_type
stations_without_any_data = set(stations_without_any_data)
backfilled_count = 0
for sid in mapped_station_ids:
epg_db_id = epg_id_map.get(sid)
if not epg_db_id:
continue
force_despite_cache = sid in stations_without_any_data
already_changing = set(changed_by_station.get(sid, []))
for ds in date_list:
if ds in already_changing or (sid, ds) not in server_md5s:
continue
if (epg_db_id, date_type.fromisoformat(ds)) in dates_with_data:
continue
if (sid, ds) in cached_md5s and not force_despite_cache:
continue
changed_by_station.setdefault(sid, []).append(ds)
backfilled_count += 1
return backfilled_count
def _sd_programs_needing_metadata(
program_ids_needed,
schedule_program_md5s,
cached_prog_md5s,
programs_with_data,
):
"""Return programIDs that need metadata download from Schedules Direct."""
programs_with_data = set(programs_with_data)
return {
pid for pid in program_ids_needed
if schedule_program_md5s.get(pid) != cached_prog_md5s.get(pid)
or pid not in programs_with_data
}
SD_POSTER_CATEGORIES = (
'Iconic', 'Banner-L1', 'Banner-L2', 'Banner-L3', 'Banner',
'Staple', 'Poster Art', 'Box Art',
@ -2070,6 +2137,8 @@ def parse_programs_for_source(epg_source, tvg_id=None):
logger.info(f"[parse_programs_for_source] Final memory usage: {final_memory:.2f} MB difference: {final_memory - initial_memory:.2f} MB")
# Explicitly clear the process object to prevent potential memory leaks
process = None
@shared_task(bind=True)
def fetch_schedules_direct_stations(self, source_id):
"""
@ -2299,6 +2368,24 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
from apps.channels.utils import maybe_auto_apply_epg_logos
maybe_auto_apply_epg_logos(source)
try:
unmapped_epg_ids = list(
EPGData.objects.filter(epg_source=source).exclude(
id__in=mapped_epg_ids,
).values_list('id', flat=True)
)
if unmapped_epg_ids:
orphaned_count = ProgramData.objects.filter(
epg_id__in=unmapped_epg_ids,
).delete()[0]
if orphaned_count:
logger.info(
f"Cleaned up {orphaned_count} orphaned ProgramData records "
f"for {len(unmapped_epg_ids)} unmapped EPG entries."
)
except Exception as prune_err:
logger.warning(f"Failed to clean up orphaned SD ProgramData: {prune_err}")
today_utc = datetime(today.year, today.month, today.day, tzinfo=dt_timezone.utc)
try:
expired_count = ProgramData.objects.filter(epg_id__in=mapped_epg_ids, end_time__lt=today_utc).delete()[0]
@ -2598,32 +2685,77 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
# -------------------------------------------------------------------------
# Step 5: MD5-delta schedule fetch
# First fetch MD5 hashes for all stations/dates. Compare against our
# locally cached hashes to determine which schedules have changed.
# Only download schedules that have actually changed; this minimises
# API calls against SD's rate-limited endpoints.
# Only mapped channels need guide data. Fetch MD5 hashes and schedules for
# mapped stations only; never cache schedule MD5s for unmapped lineup entries.
# -------------------------------------------------------------------------
from apps.epg.models import SDScheduleMD5
from django.utils.dateparse import parse_datetime
send_epg_update(source.id, "parsing_programs", 33, message=f"Checking schedule MD5s for {len(station_map)} stations over {SD_DAYS_TO_FETCH} days...")
station_ids = list(station_map.keys())
today = date.today()
date_list = [(today + timedelta(days=i)).strftime('%Y-%m-%d') for i in range(SD_DAYS_TO_FETCH)]
# Prune SDScheduleMD5 records whose dates have rolled off the fetch window.
# These accumulate one row per station per day and are never useful once past.
pruned_sched_md5_count = SDScheduleMD5.objects.filter(epg_source=source, date__lt=today).delete()[0]
mapped_epg_ids = set(
Channel.objects.filter(
epg_data__epg_source=source,
epg_data__isnull=False,
).values_list('epg_data_id', flat=True)
)
mapped_tvg_ids = set(
EPGData.objects.filter(
id__in=mapped_epg_ids,
epg_source=source,
).values_list('tvg_id', flat=True)
)
mapped_station_ids = [sid for sid in station_ids if sid in mapped_tvg_ids]
# Prune expired schedule MD5s and drop cache for unmapped stations.
pruned_sched_md5_count = SDScheduleMD5.objects.filter(
epg_source=source, date__lt=today,
).delete()[0]
if pruned_sched_md5_count:
logger.info(f"Pruned {pruned_sched_md5_count} expired SDScheduleMD5 records (before {today}).")
# Fetch MD5 hashes for all stations in batches of 5000
if mapped_tvg_ids:
unmapped_cache_pruned = SDScheduleMD5.objects.filter(
epg_source=source,
).exclude(station_id__in=mapped_tvg_ids).delete()[0]
else:
unmapped_cache_pruned = SDScheduleMD5.objects.filter(epg_source=source).delete()[0]
if unmapped_cache_pruned:
logger.info(f"Pruned {unmapped_cache_pruned} SDScheduleMD5 records for unmapped lineup stations.")
if not mapped_station_ids:
logger.info("No channels mapped to this SD source; skipping schedule MD5 check and downloads.")
_sd_post_refresh_tasks(mapped_epg_ids, {}, today)
success_msg = (
f"{len(station_map)} lineup stations synced. "
"Map channels to EPG entries, then refresh to populate guide data."
)
source.status = EPGSource.STATUS_SUCCESS
source.last_message = success_msg
source.updated_at = timezone.now()
source.save(update_fields=['status', 'last_message', 'updated_at'])
send_epg_update(source.id, "parsing_programs", 100, status="success", message=success_msg)
return
send_epg_update(
source.id, "parsing_programs", 33,
message=f"Checking schedule MD5s for {len(mapped_station_ids)} mapped stations over {SD_DAYS_TO_FETCH} days...",
)
# Fetch MD5 hashes for mapped stations in batches of 5000
STATION_BATCH_SIZE = 5000
server_md5s = {} # (station_id, date) -> {md5, last_modified}
logger.info(f"Fetching schedule MD5s for {len(station_ids)} stations over {SD_DAYS_TO_FETCH} days.")
logger.info(
f"Fetching schedule MD5s for {len(mapped_station_ids)} mapped stations "
f"(of {len(station_ids)} lineup stations) over {SD_DAYS_TO_FETCH} days."
)
station_batches = [station_ids[i:i + STATION_BATCH_SIZE] for i in range(0, len(station_ids), STATION_BATCH_SIZE)]
station_batches = [
mapped_station_ids[i:i + STATION_BATCH_SIZE]
for i in range(0, len(mapped_station_ids), STATION_BATCH_SIZE)
]
for batch in station_batches:
try:
md5_response = requests.post(
@ -2644,84 +2776,65 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
except requests.exceptions.RequestException as e:
logger.warning(f"Failed to fetch schedule MD5s: {e}")
# Load our cached MD5s from DB
# Load our cached MD5s from DB (mapped stations only)
cached_md5s = {
(r.station_id, r.date.strftime('%Y-%m-%d')): r.md5
for r in SDScheduleMD5.objects.filter(epg_source=source, station_id__in=station_ids)
for r in SDScheduleMD5.objects.filter(
epg_source=source, station_id__in=mapped_station_ids,
)
}
# Determine which station/date combinations need downloading
changed_by_station = {} # station_id -> [date_str, ...]
for (sid, date_str), server_info in server_md5s.items():
if date_str not in date_list:
continue
cached = cached_md5s.get((sid, date_str))
if cached != server_info['md5']:
changed_by_station.setdefault(sid, []).append(date_str)
changed_by_station = _sd_compute_schedule_changes_from_md5(
server_md5s, cached_md5s, date_list,
)
window_start = datetime(today.year, today.month, today.day, tzinfo=dt_timezone.utc)
window_end = window_start + timedelta(days=SD_DAYS_TO_FETCH)
dates_with_data = set()
if mapped_epg_ids:
for epg_id, start_time in ProgramData.objects.filter(
epg_id__in=mapped_epg_ids,
start_time__gte=window_start,
start_time__lt=window_end,
).values_list('epg_id', 'start_time'):
dates_with_data.add((epg_id, start_time.date()))
stations_without_any_data = mapped_tvg_ids - set(
ProgramData.objects.filter(epg_id__in=mapped_epg_ids)
.values_list('tvg_id', flat=True).distinct()
)
backfilled_count = _sd_backfill_schedule_dates_without_data(
changed_by_station,
server_md5s,
date_list,
mapped_station_ids,
epg_id_map,
dates_with_data,
cached_md5s,
stations_without_any_data,
)
if backfilled_count:
logger.info(
f"Backfilling {backfilled_count} station/date combinations with no ProgramData "
f"in the {SD_DAYS_TO_FETCH}-day fetch window."
)
total_changed = sum(len(v) for v in changed_by_station.values())
total_possible = len(station_ids) * len(date_list)
logger.info(f"Schedule MD5 check: {len(server_md5s)} hashes checked, {total_changed} station/date combinations changed (of {total_possible} possible).")
total_possible = len(mapped_station_ids) * len(date_list)
logger.info(
f"Schedule MD5 check: {len(server_md5s)} hashes checked, "
f"{total_changed} station/date combinations to fetch (of {total_possible} possible)."
)
send_epg_update(source.id, "parsing_programs", 38,
message=f"MD5 check complete: {len(changed_by_station)} stations have schedule updates.")
# Force-fetch for mapped stations with no existing ProgramData.
# The MD5 cache stores hashes for all lineup stations, including unmapped ones.
# If a station was in the lineup (MD5 cached) but not yet mapped to a channel
# (no ProgramData created), a later mapping causes those cached dates to appear
# "unchanged", skipping the fetch and leaving the channel with no guide data.
mapped_epg_ids_early = set(
Channel.objects.filter(
epg_data__epg_source=source, epg_data__isnull=False
).values_list('epg_data_id', flat=True)
)
mapped_tvg_ids_early = set(
EPGData.objects.filter(
id__in=mapped_epg_ids_early, epg_source=source
).values_list('tvg_id', flat=True)
)
newly_mapped = set()
if mapped_tvg_ids_early:
stations_with_data = set(
ProgramData.objects.filter(
epg__epg_source=source,
tvg_id__in=mapped_tvg_ids_early,
).values_list('tvg_id', flat=True).distinct()
)
newly_mapped = mapped_tvg_ids_early - stations_with_data
if newly_mapped:
server_dates_by_sid = {}
for (sid, ds) in server_md5s:
if sid in newly_mapped:
server_dates_by_sid.setdefault(sid, set()).add(ds)
forced_count = 0
for sid in newly_mapped:
available = server_dates_by_sid.get(sid, set())
already_changing = set(changed_by_station.get(sid, []))
force_dates = [ds for ds in date_list if ds in available and ds not in already_changing]
if force_dates:
changed_by_station.setdefault(sid, []).extend(force_dates)
forced_count += len(force_dates)
if forced_count:
logger.info(
f"Force-fetching {forced_count} station/date combinations for "
f"{len(newly_mapped)} stations with no existing ProgramData (newly mapped channels)."
)
# schedules_by_station: stationID -> list of {programID, airDateTime, duration, ...}
schedules_by_station = {sid: [] for sid in station_ids}
schedules_by_station = {sid: [] for sid in mapped_station_ids}
program_ids_needed = set()
if not changed_by_station:
logger.info("No schedule changes detected, skipping schedule and program downloads.")
from apps.channels.models import Channel as ChannelModel
mapped_epg_ids_no_change = set(
ChannelModel.objects.filter(
epg_data__epg_source=source,
epg_data__isnull=False,
).values_list('epg_data_id', flat=True)
)
_sd_post_refresh_tasks(mapped_epg_ids_no_change, {}, today)
_sd_post_refresh_tasks(mapped_epg_ids, {}, today)
send_epg_update(source.id, "parsing_programs", 100, status="success",
message="No schedule changes detected since last refresh. Guide data is up to date.")
source.status = EPGSource.STATUS_SUCCESS
@ -2857,21 +2970,21 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
).only('program_id', 'md5')
}
# For newly mapped stations with no ProgramData, invalidate cached program MD5s so
# metadata gets re-fetched. Without this, programs whose MD5s were cached when the
# station was unmapped would be skipped, producing "No Title" records.
if newly_mapped:
for sid in newly_mapped:
for airing in schedules_by_station.get(sid, []):
pid = airing.get('programID')
if pid:
cached_prog_md5s.pop(pid, None)
programs_with_data = set()
if program_ids_needed:
programs_with_data = set(
ProgramData.objects.filter(
epg__epg_source=source,
program_id__in=program_ids_needed,
).values_list('program_id', flat=True).distinct()
)
# Only fetch programs where MD5 differs from our cached value
programs_to_fetch = {
pid for pid in program_ids_needed
if schedule_program_md5s.get(pid) != cached_prog_md5s.get(pid)
}
programs_to_fetch = _sd_programs_needing_metadata(
program_ids_needed,
schedule_program_md5s,
cached_prog_md5s,
programs_with_data,
)
logger.info(
f"Program MD5 delta: {len(program_ids_needed)} programs in schedules, "
@ -2928,22 +3041,6 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
logger.info("Building program records...")
send_epg_update(source.id, "parsing_programs", 80)
# Only process stations that are mapped to channels to match the existing
# XMLTV flow (parse_programs_for_source only processes mapped channels).
from apps.channels.models import Channel as ChannelModel
mapped_epg_ids = set(
ChannelModel.objects.filter(
epg_data__epg_source=source,
epg_data__isnull=False,
).values_list('epg_data_id', flat=True)
)
mapped_tvg_ids = set(
EPGData.objects.filter(
id__in=mapped_epg_ids,
epg_source=source,
).values_list('tvg_id', flat=True)
)
# Cache existing program data for unchanged programs BEFORE surgical delete.
# When a station/date schedule MD5 changes, ALL airings are re-fetched, but only
# programs with changed program MD5s get metadata re-downloaded. The surgical delete

474
apps/epg/tests/test_schedules_direct.py
View file

@ -7,19 +7,26 @@ Covers:
- fetch_schedules_direct: credential validation
- fetch_schedules_direct: SHA1 password hashing and token exchange
- fetch_schedules_direct: graceful error handling on auth failure
- fetch_schedules_direct: schedule MD5 delta, backfill, and cache invalidation
- parse_schedules_direct_time: correct UTC parsing
- EPG signals: SD sources skip the XMLTV program parser
"""
import hashlib
from datetime import datetime
from datetime import date, datetime, timedelta, timezone as dt_timezone
from unittest.mock import MagicMock, patch
from django.test import TestCase
from django.utils import timezone
from apps.epg.models import EPGSource, EPGData
from apps.channels.models import Channel
from apps.epg.models import EPGSource, EPGData, ProgramData, SDScheduleMD5
from apps.epg.serializers import EPGSourceSerializer
from apps.epg.tasks import (
_sd_backfill_schedule_dates_without_data,
_sd_compute_schedule_changes_from_md5,
_sd_programs_needing_metadata,
)
# ---------------------------------------------------------------------------
@ -270,6 +277,469 @@ class FetchSchedulesDirectStationsOnlyTests(TestCase):
self.assertEqual(complete_call[1]['channels_count'], 1)
# ---------------------------------------------------------------------------
# Schedule MD5 delta, backfill, and cache tests
# ---------------------------------------------------------------------------
class SDScheduleMd5DeltaTests(TestCase):
"""Pure-function tests for schedule MD5 comparison."""
DATE_LIST = ['2026-06-11', '2026-06-12', '2026-06-13']
def test_detects_changed_md5(self):
server_md5s = {
('10001', '2026-06-11'): {'md5': 'abc', 'last_modified': ''},
('10001', '2026-06-12'): {'md5': 'def', 'last_modified': ''},
}
cached_md5s = {
('10001', '2026-06-11'): 'abc',
('10001', '2026-06-12'): 'old',
}
changed = _sd_compute_schedule_changes_from_md5(
server_md5s, cached_md5s, self.DATE_LIST,
)
self.assertEqual(changed['10001'], ['2026-06-12'])
def test_missing_cache_treated_as_changed(self):
server_md5s = {
('10001', '2026-06-11'): {'md5': 'abc', 'last_modified': ''},
}
changed = _sd_compute_schedule_changes_from_md5(server_md5s, {}, self.DATE_LIST)
self.assertEqual(changed['10001'], ['2026-06-11'])
def test_ignores_dates_outside_fetch_window(self):
server_md5s = {
('10001', '2026-06-01'): {'md5': 'abc', 'last_modified': ''},
}
changed = _sd_compute_schedule_changes_from_md5(server_md5s, {}, self.DATE_LIST)
self.assertEqual(changed, {})
class SDScheduleBackfillTests(TestCase):
"""Backfill must fix stale-cache gaps without re-fetching empty cached days."""
DATE_LIST = ['2026-06-11', '2026-06-12', '2026-06-13']
EPG_ID = 42
EPG_ID_MAP = {'10001': 42}
def _server_md5s(self):
return {
(sid, ds): {'md5': 'hash', 'last_modified': ''}
for sid in ('10001', '10002')
for ds in self.DATE_LIST
}
def test_backfills_when_no_cache_and_no_program_data(self):
changed = {}
count = _sd_backfill_schedule_dates_without_data(
changed,
self._server_md5s(),
self.DATE_LIST,
['10001'],
self.EPG_ID_MAP,
set(),
{},
{'10001'},
)
self.assertEqual(count, 3)
self.assertEqual(len(changed['10001']), 3)
def test_stale_cache_ignored_when_station_has_zero_program_data(self):
"""Newly mapped channel: stale MD5 cache must not block backfill."""
changed = {}
cached_md5s = {
(sid, ds): 'hash'
for sid in ('10001',)
for ds in self.DATE_LIST
}
count = _sd_backfill_schedule_dates_without_data(
changed,
self._server_md5s(),
self.DATE_LIST,
['10001'],
self.EPG_ID_MAP,
set(),
cached_md5s,
{'10001'},
)
self.assertEqual(count, 3)
self.assertEqual(len(changed['10001']), 3)
def test_skips_cached_empty_day_when_station_has_other_program_data(self):
"""Legitimately empty schedule day must not be re-fetched every refresh."""
changed = {}
cached_md5s = {('10001', '2026-06-12'): 'hash'}
dates_with_data = {(self.EPG_ID, date(2026, 6, 11))}
count = _sd_backfill_schedule_dates_without_data(
changed,
self._server_md5s(),
self.DATE_LIST,
['10001'],
self.EPG_ID_MAP,
dates_with_data,
cached_md5s,
set(),
)
self.assertEqual(count, 1)
self.assertEqual(changed['10001'], ['2026-06-13'])
def test_does_not_duplicate_dates_already_marked_changed(self):
changed = {'10001': ['2026-06-11']}
count = _sd_backfill_schedule_dates_without_data(
changed,
self._server_md5s(),
self.DATE_LIST,
['10001'],
self.EPG_ID_MAP,
set(),
{},
{'10001'},
)
self.assertEqual(count, 2)
self.assertEqual(sorted(changed['10001']), self.DATE_LIST)
class SDProgramMetadataDeltaTests(TestCase):
def test_fetches_when_md5_changed(self):
needed = _sd_programs_needing_metadata(
{'EP0001'},
{'EP0001': 'new'},
{'EP0001': 'old'},
{'EP0001'},
)
self.assertEqual(needed, {'EP0001'})
def test_fetches_when_no_local_program_data(self):
needed = _sd_programs_needing_metadata(
{'EP0001', 'EP0002'},
{'EP0001': 'same', 'EP0002': 'same'},
{'EP0001': 'same', 'EP0002': 'same'},
{'EP0001'},
)
self.assertEqual(needed, {'EP0002'})
def test_skips_when_md5_matches_and_program_data_exists(self):
needed = _sd_programs_needing_metadata(
{'EP0001'},
{'EP0001': 'same'},
{'EP0001': 'same'},
{'EP0001'},
)
self.assertEqual(needed, set())
class SDScheduleDeltaIntegrationTests(TestCase):
"""DB-backed tests for cache pruning and mapped-only MD5 API calls."""
MAPPED_STATION = '10001'
UNMAPPED_STATION = '10002'
def _make_sd_source(self):
return EPGSource.objects.create(
name='SD Integration',
source_type='schedules_direct',
username='sduser',
password='sdpass',
)
def _lineup_get_side_effect(self, url, **kwargs):
if url.endswith('/status'):
return MagicMock(
status_code=200,
json=MagicMock(return_value={'systemStatus': [{'status': 'Online'}]}),
)
if url.endswith('/lineups'):
return MagicMock(
status_code=200,
json=MagicMock(return_value={'lineups': [{'lineupID': 'USA-TEST-X'}]}),
)
if '/lineups/USA-TEST-X' in url:
return MagicMock(
status_code=200,
json=MagicMock(return_value={
'stations': [
{
'stationID': self.MAPPED_STATION,
'name': 'Mapped Station',
'callsign': 'MAP',
},
{
'stationID': self.UNMAPPED_STATION,
'name': 'Unmapped Station',
'callsign': 'UNM',
},
],
}),
)
raise AssertionError(f'Unexpected GET URL: {url}')
def _build_date_list(self, days=3):
today = date.today()
return [(today + timedelta(days=i)).strftime('%Y-%m-%d') for i in range(days)]
def _seed_full_window_program_data(self, epg, days=3):
today = date.today()
for i in range(days):
day = today + timedelta(days=i)
start = datetime(day.year, day.month, day.day, 12, 0, tzinfo=dt_timezone.utc)
ProgramData.objects.create(
epg=epg,
start_time=start,
end_time=start + timedelta(hours=1),
title='Show',
tvg_id=epg.tvg_id,
)
@patch('apps.epg.tasks.SD_DAYS_TO_FETCH', 3)
@patch('apps.epg.tasks.send_epg_update')
@patch('apps.epg.tasks.requests.get')
@patch('apps.epg.tasks.requests.post')
def test_md5_api_only_requests_mapped_stations(
self, mock_post, mock_get, mock_send_epg_update,
):
from apps.epg.tasks import fetch_schedules_direct
source = self._make_sd_source()
mapped_epg = EPGData.objects.create(
tvg_id=self.MAPPED_STATION,
name='Mapped',
epg_source=source,
)
Channel.objects.create(name='Mapped Ch', epg_data=mapped_epg)
date_list = self._build_date_list(3)
self._seed_full_window_program_data(mapped_epg, days=3)
today = date.today()
for i, ds in enumerate(date_list):
SDScheduleMD5.objects.create(
epg_source=source,
station_id=self.MAPPED_STATION,
date=today + timedelta(days=i),
md5=f'md5-{ds}',
last_modified=timezone.now(),
)
SDScheduleMD5.objects.create(
epg_source=source,
station_id=self.UNMAPPED_STATION,
date=today,
md5='unmapped-stale',
last_modified=timezone.now(),
)
mock_get.side_effect = self._lineup_get_side_effect
md5_response_payload = {
self.MAPPED_STATION: {
ds: {'code': 0, 'md5': f'md5-{ds}', 'lastModified': '2026-06-11T00:00:00Z'}
for ds in date_list
},
}
def post_side_effect(url, **kwargs):
if url.endswith('/token'):
return MagicMock(
status_code=200,
json=MagicMock(return_value={'code': 0, 'token': 'tok'}),
)
if url.endswith('/schedules/md5'):
return MagicMock(
status_code=200,
json=MagicMock(return_value=md5_response_payload),
)
raise AssertionError(f'Unexpected POST URL: {url}')
mock_post.side_effect = post_side_effect
fetch_schedules_direct(source, force=True)
md5_calls = [c for c in mock_post.call_args_list if c[0][0].endswith('/schedules/md5')]
self.assertEqual(len(md5_calls), 1)
request_body = md5_calls[0][1]['json']
station_ids_in_request = {entry['stationID'] for entry in request_body}
self.assertEqual(station_ids_in_request, {self.MAPPED_STATION})
self.assertFalse(
SDScheduleMD5.objects.filter(
epg_source=source,
station_id=self.UNMAPPED_STATION,
).exists()
)
schedule_calls = [c for c in mock_post.call_args_list if c[0][0].endswith('/schedules')]
self.assertEqual(len(schedule_calls), 0)
source.refresh_from_db()
self.assertEqual(source.status, EPGSource.STATUS_SUCCESS)
@patch('apps.epg.tasks.SD_DAYS_TO_FETCH', 3)
@patch('apps.epg.tasks.send_epg_update')
@patch('apps.epg.tasks.requests.get')
@patch('apps.epg.tasks.requests.post')
def test_newly_mapped_station_fetches_despite_stale_cache(
self, mock_post, mock_get, mock_send_epg_update,
):
from apps.epg.tasks import fetch_schedules_direct
source = self._make_sd_source()
mapped_epg = EPGData.objects.create(
tvg_id=self.MAPPED_STATION,
name='Mapped',
epg_source=source,
)
Channel.objects.create(name='Mapped Ch', epg_data=mapped_epg)
date_list = self._build_date_list(3)
today = date.today()
for i, ds in enumerate(date_list):
SDScheduleMD5.objects.create(
epg_source=source,
station_id=self.MAPPED_STATION,
date=today + timedelta(days=i),
md5=f'md5-{ds}',
last_modified=timezone.now(),
)
mock_get.side_effect = self._lineup_get_side_effect
schedule_payload = [{
'stationID': self.MAPPED_STATION,
'metadata': {'startDate': date_list[0], 'md5': 'md5-' + date_list[0], 'modified': '2026-06-11T00:00:00Z'},
'programs': [{
'programID': 'EP000000000001',
'airDateTime': f'{date_list[0]}T12:00:00Z',
'duration': 3600,
'md5': 'prog-md5-1',
}],
}]
program_payload = [{
'programID': 'EP000000000001',
'titles': [{'title120': 'Test Show'}],
}]
def post_side_effect(url, **kwargs):
if url.endswith('/token'):
return MagicMock(
status_code=200,
json=MagicMock(return_value={'code': 0, 'token': 'tok'}),
)
if url.endswith('/schedules/md5'):
return MagicMock(
status_code=200,
json=MagicMock(return_value={
self.MAPPED_STATION: {
ds: {'code': 0, 'md5': f'md5-{ds}', 'lastModified': '2026-06-11T00:00:00Z'}
for ds in date_list
},
}),
)
if url.endswith('/schedules'):
return MagicMock(status_code=200, json=MagicMock(return_value=schedule_payload))
if url.endswith('/programs'):
return MagicMock(status_code=200, json=MagicMock(return_value=program_payload))
raise AssertionError(f'Unexpected POST URL: {url}')
mock_post.side_effect = post_side_effect
fetch_schedules_direct(source, force=True)
schedule_calls = [c for c in mock_post.call_args_list if c[0][0].endswith('/schedules')]
self.assertGreaterEqual(len(schedule_calls), 1)
self.assertEqual(
ProgramData.objects.filter(epg=mapped_epg).count(),
1,
)
@patch('apps.epg.tasks.SD_DAYS_TO_FETCH', 3)
@patch('apps.epg.tasks.send_epg_update')
@patch('apps.epg.tasks.requests.get')
@patch('apps.epg.tasks.requests.post')
def test_orphan_program_data_removed_on_post_refresh(
self, mock_post, mock_get, mock_send_epg_update,
):
from apps.epg.tasks import fetch_schedules_direct
source = self._make_sd_source()
mapped_epg = EPGData.objects.create(
tvg_id=self.MAPPED_STATION,
name='Mapped',
epg_source=source,
)
orphan_epg = EPGData.objects.create(
tvg_id=self.UNMAPPED_STATION,
name='Orphan',
epg_source=source,
)
Channel.objects.create(name='Mapped Ch', epg_data=mapped_epg)
date_list = self._build_date_list(3)
self._seed_full_window_program_data(mapped_epg, days=3)
start = timezone.now()
ProgramData.objects.create(
epg=orphan_epg,
start_time=start,
end_time=start + timedelta(hours=1),
title='Orphan Show',
tvg_id=orphan_epg.tvg_id,
)
today = date.today()
for i, ds in enumerate(date_list):
SDScheduleMD5.objects.create(
epg_source=source,
station_id=self.MAPPED_STATION,
date=today + timedelta(days=i),
md5=f'md5-{ds}',
last_modified=timezone.now(),
)
mock_get.side_effect = self._lineup_get_side_effect
def post_side_effect(url, **kwargs):
if url.endswith('/token'):
return MagicMock(
status_code=200,
json=MagicMock(return_value={'code': 0, 'token': 'tok'}),
)
if url.endswith('/schedules/md5'):
return MagicMock(
status_code=200,
json=MagicMock(return_value={
self.MAPPED_STATION: {
ds: {'code': 0, 'md5': f'md5-{ds}', 'lastModified': '2026-06-11T00:00:00Z'}
for ds in date_list
},
}),
)
raise AssertionError(f'Unexpected POST URL: {url}')
mock_post.side_effect = post_side_effect
fetch_schedules_direct(source, force=True)
self.assertFalse(ProgramData.objects.filter(epg=orphan_epg).exists())
def test_stale_program_md5_fetched_when_no_program_data(self):
programs_with_data = set()
needed = _sd_programs_needing_metadata(
{'EP0001'},
{'EP0001': 'cached-md5'},
{'EP0001': 'cached-md5'},
programs_with_data,
)
self.assertEqual(needed, {'EP0001'})
def test_shared_program_skips_metadata_when_cached(self):
needed = _sd_programs_needing_metadata(
{'EP0001'},
{'EP0001': 'cached-md5'},
{'EP0001': 'cached-md5'},
{'EP0001'},
)
self.assertEqual(needed, set())
# ---------------------------------------------------------------------------
# parse_schedules_direct_time tests
# ---------------------------------------------------------------------------