import os
import gzip
import tempfile
from unittest.mock import patch, MagicMock
from django.test import TestCase
from django.utils import timezone
from apps.epg.models import EPGSource, EPGData
from apps.epg.tasks import (
find_current_program_for_tvg_id,
build_programme_index,
build_programme_index_task,
)
FIXTURE_DIR = os.path.join(os.path.dirname(__file__), "fixtures")
FIXTURE_XML = os.path.join(FIXTURE_DIR, "test_epg.xml")
class FindCurrentProgramTests(TestCase):
def setUp(self):
self.now = timezone.now()
self.source = EPGSource.objects.create(
name="Test Source",
source_type="xmltv",
url="http://example.com/epg.xml",
file_path=FIXTURE_XML,
)
self.epg = EPGData.objects.create(
tvg_id="channel.current",
name="Current Channel",
epg_source=self.source,
)
def test_returns_none_for_dummy_source(self):
dummy = EPGSource.objects.create(name="Dummy", source_type="dummy")
epg = EPGData.objects.create(
tvg_id="x", name="X", epg_source=dummy
)
self.assertIsNone(find_current_program_for_tvg_id(epg))
def test_returns_none_for_schedules_direct_source(self):
sd = EPGSource.objects.create(
name="SD", source_type="schedules_direct"
)
epg = EPGData.objects.create(
tvg_id="x", name="X", epg_source=sd
)
self.assertIsNone(find_current_program_for_tvg_id(epg))
def test_returns_none_when_tvg_id_empty(self):
epg = EPGData.objects.create(
tvg_id="", name="Empty", epg_source=self.source
)
self.assertIsNone(find_current_program_for_tvg_id(epg))
def test_returns_none_when_tvg_id_none(self):
epg = EPGData.objects.create(
tvg_id=None, name="None", epg_source=self.source
)
self.assertIsNone(find_current_program_for_tvg_id(epg))
def test_byte_offset_index_hit(self):
# Build the index from the fixture
build_programme_index(self.source.id)
self.source.refresh_from_db()
self.assertIsNotNone(self.source.programme_index)
# "Always On Show" spans 2000-2099, so should always be current
result = find_current_program_for_tvg_id(self.epg)
self.assertIsNotNone(result)
self.assertEqual(result["title"], "Always On Show")
self.assertEqual(result["sub_title"], "The eternal broadcast")
self.assertEqual(
result["description"],
"This programme spans a very long time for testing",
)
self.assertIn("start_time", result)
self.assertIn("end_time", result)
def test_byte_offset_index_miss(self):
# Build index, then query for a tvg_id that exists in the index
# but has no programme airing now
build_programme_index(self.source.id)
self.source.refresh_from_db()
epg_past = EPGData.objects.create(
tvg_id="channel.past",
name="Past Channel",
epg_source=self.source,
)
result = find_current_program_for_tvg_id(epg_past)
self.assertIsNone(result)
def test_index_miss_tvg_id_not_in_index(self):
# tvg_id not in index at all
build_programme_index(self.source.id)
self.source.refresh_from_db()
epg_unknown = EPGData.objects.create(
tvg_id="channel.nonexistent",
name="Nonexistent",
epg_source=self.source,
)
result = find_current_program_for_tvg_id(epg_unknown)
self.assertIsNone(result)
def test_accepts_integer_id(self):
# find_current_program_for_tvg_id accepts an int (EPGData PK)
build_programme_index(self.source.id)
result = find_current_program_for_tvg_id(self.epg.id)
self.assertIsNotNone(result)
self.assertEqual(result["title"], "Always On Show")
def test_returns_none_for_nonexistent_id(self):
result = find_current_program_for_tvg_id(99999)
self.assertIsNone(result)
def test_multi_block_file(self):
# Create an XML where programmes for the same channel appear in
# multiple non-contiguous blocks (A, B, A, B pattern).
# The index records multiple offsets per channel so the lookup
# scans all blocks.
xml = (
'\n'
"\n"
' \n'
' \n'
' \n'
" A Morning\n"
" \n"
' \n'
" B Morning\n"
" \n"
# Second block for A — current programme lives here
' \n'
" A Current\n"
" \n"
' \n'
" B Current\n"
" \n"
"\n"
)
with tempfile.NamedTemporaryFile(
mode="w", suffix=".xml", delete=False
) as f:
f.write(xml)
tmp_path = f.name
try:
src = EPGSource.objects.create(
name="MultiBlock",
source_type="xmltv",
file_path=tmp_path,
)
build_programme_index(src.id)
src.refresh_from_db()
self.assertIsNotNone(src.programme_index)
epg_a = EPGData.objects.create(
tvg_id="A", name="A", epg_source=src
)
result = find_current_program_for_tvg_id(epg_a)
self.assertIsNotNone(result)
self.assertEqual(result["title"], "A Current")
finally:
os.unlink(tmp_path)
def test_channel_id_entities_and_whitespace_match_tvg_id(self):
# programme@channel carries an XML entity and surrounding whitespace;
# EPGData.tvg_id holds the lxml-decoded, stripped form. The index key
# and lookup must canonicalize to the same value.
xml = (
'\n'
"\n"
' \n'
' \n'
" A and E Now\n"
" \n"
"\n"
)
with tempfile.NamedTemporaryFile(
mode="w", suffix=".xml", delete=False
) as f:
f.write(xml)
tmp_path = f.name
try:
src = EPGSource.objects.create(
name="Entities", source_type="xmltv", file_path=tmp_path
)
build_programme_index(src.id)
src.refresh_from_db()
self.assertIn("A&E.us", src.programme_index["channels"])
epg = EPGData.objects.create(
tvg_id="A&E.us", name="A&E", epg_source=src
)
result = find_current_program_for_tvg_id(epg)
self.assertIsNotNone(result)
self.assertEqual(result["title"], "A and E Now")
finally:
os.unlink(tmp_path)
def test_offset_lookup_resolves_named_html_entities_in_programme_text(self):
xml = (
'\n'
"\n"
' \n'
' \n'
" Café Live\n"
" \n"
"\n"
)
with tempfile.NamedTemporaryFile(
mode="w", suffix=".xml", delete=False
) as f:
f.write(xml)
tmp_path = f.name
try:
src = EPGSource.objects.create(
name="Named Entities", source_type="xmltv", file_path=tmp_path
)
build_programme_index(src.id)
epg = EPGData.objects.create(
tvg_id="entity.channel", name="Entity Channel", epg_source=src
)
result = find_current_program_for_tvg_id(epg)
self.assertIsNotNone(result)
self.assertEqual(result["title"], "Caf\u00e9 Live")
finally:
os.unlink(tmp_path)
def test_epgshare_fr_style_programme_with_channel_first_and_apostrophe_entities(self):
# Based on epgshare01 FR feeds: channel is the first programme attr,
# ids/text include non-ASCII plus XML entities, and sub-title is common.
tvg_id = "France.3.-.C\u00f4te.d'Azur.fr"
xml = (
'\n'
'\n'
' \n'
' France 3 - C\u00f4te d'Azur\n'
" \n"
' \n'
" La p'tite librairie\n"
" Le Lys de Brooklyn, de Betty Smith\n"
" \n"
"\n"
)
with tempfile.NamedTemporaryFile(
mode="w", suffix=".xml", encoding="utf-8", delete=False
) as f:
f.write(xml)
tmp_path = f.name
try:
src = EPGSource.objects.create(
name="EPGShare FR", source_type="xmltv", file_path=tmp_path
)
build_programme_index(src.id)
src.refresh_from_db()
self.assertIn(tvg_id, src.programme_index["channels"])
epg = EPGData.objects.create(
tvg_id=tvg_id, name="France 3 Cote d'Azur", epg_source=src
)
result = find_current_program_for_tvg_id(epg)
self.assertIsNotNone(result)
self.assertEqual(result["title"], "La p'tite librairie")
self.assertEqual(
result["sub_title"], "Le Lys de Brooklyn, de Betty Smith"
)
finally:
os.unlink(tmp_path)
def test_epgshare_fr_style_description_decodes_predefined_entities(self):
xml = (
'\n'
"\n"
' \n'
' \n'
" Chasse & p\u00eache, le mag\n"
" Au sommaire : "La r\u00e9gion" <HD> & bonus.\n"
" \n"
"\n"
)
with tempfile.NamedTemporaryFile(
mode="w", suffix=".xml", encoding="utf-8", delete=False
) as f:
f.write(xml)
tmp_path = f.name
try:
src = EPGSource.objects.create(
name="EPGShare FR Entities", source_type="xmltv", file_path=tmp_path
)
build_programme_index(src.id)
epg = EPGData.objects.create(
tvg_id="Chasse.et.P\u00eache.fr",
name="Chasse et Peche",
epg_source=src,
)
result = find_current_program_for_tvg_id(epg)
self.assertIsNotNone(result)
self.assertEqual(result["title"], "Chasse & p\u00eache, le mag")
self.assertEqual(
result["description"],
'Au sommaire : "La r\u00e9gion" & bonus.',
)
finally:
os.unlink(tmp_path)
def test_epgshare_all_sources_style_start_stop_before_channel(self):
# The all-sources EPG contains both programme attr orders:
# channel/start/stop and start/stop/channel.
tvg_id = "Atfal.&.Mawaheb.ae"
xml = (
'\n'
"\n"
' \n'
' \n'
" Kids & Talent\n"
" \n"
"\n"
)
with tempfile.NamedTemporaryFile(
mode="w", suffix=".xml", encoding="utf-8", delete=False
) as f:
f.write(xml)
tmp_path = f.name
try:
src = EPGSource.objects.create(
name="EPGShare All Sources", source_type="xmltv", file_path=tmp_path
)
build_programme_index(src.id)
src.refresh_from_db()
self.assertIn(tvg_id, src.programme_index["channels"])
epg = EPGData.objects.create(
tvg_id=tvg_id, name="Atfal and Mawaheb", epg_source=src
)
result = find_current_program_for_tvg_id(epg)
self.assertIsNotNone(result)
self.assertEqual(result["title"], "Kids & Talent")
finally:
os.unlink(tmp_path)
def test_jesmann_fullguide_style_numeric_channel_id(self):
# FullGuide.xml.gz uses numeric channel ids and consistently orders
# programme attrs as start/stop/channel.
xml = (
'\n'
"\n"
' \n'
" Sample Numeric Channel\n"
" \n"
' \n'
" Breaking Basics\n"
" Tobi visits the "Flying Steps" in Berlin.\n"
" \n"
"\n"
)
with tempfile.NamedTemporaryFile(
mode="w", suffix=".xml", encoding="utf-8", delete=False
) as f:
f.write(xml)
tmp_path = f.name
try:
src = EPGSource.objects.create(
name="Jesmann FullGuide", source_type="xmltv", file_path=tmp_path
)
build_programme_index(src.id)
src.refresh_from_db()
self.assertIn("123958", src.programme_index["channels"])
epg = EPGData.objects.create(
tvg_id="123958", name="Numeric Channel", epg_source=src
)
result = find_current_program_for_tvg_id(epg)
self.assertIsNotNone(result)
self.assertEqual(result["title"], "Breaking Basics")
self.assertEqual(
result["description"],
'Tobi visits the "Flying Steps" in Berlin.',
)
finally:
os.unlink(tmp_path)
def test_offset_lookup_accepts_single_quoted_channel_attribute(self):
xml = (
'\n'
"\n"
" \n"
" \n"
" Single Quote Current\n"
" \n"
"\n"
)
with tempfile.NamedTemporaryFile(
mode="w", suffix=".xml", encoding="utf-8", delete=False
) as f:
f.write(xml)
tmp_path = f.name
try:
src = EPGSource.objects.create(
name="Single Quotes", source_type="xmltv", file_path=tmp_path
)
build_programme_index(src.id)
src.refresh_from_db()
self.assertIn("single.quote.channel", src.programme_index["channels"])
epg = EPGData.objects.create(
tvg_id="single.quote.channel",
name="Single Quote Channel",
epg_source=src,
)
result = find_current_program_for_tvg_id(epg)
self.assertIsNotNone(result)
self.assertEqual(result["title"], "Single Quote Current")
finally:
os.unlink(tmp_path)
def test_offset_lookup_resolves_html_named_entity_not_predefined_by_xml(self):
xml = (
'\n'
"\n"
' \n'
' \n'
" Café Society\n"
" \n"
"\n"
)
with tempfile.NamedTemporaryFile(
mode="w", suffix=".xml", encoding="utf-8", delete=False
) as f:
f.write(xml)
tmp_path = f.name
try:
src = EPGSource.objects.create(
name="HTML Entities", source_type="xmltv", file_path=tmp_path
)
build_programme_index(src.id)
epg = EPGData.objects.create(
tvg_id="html.entity.channel",
name="HTML Entity Channel",
epg_source=src,
)
result = find_current_program_for_tvg_id(epg)
self.assertIsNotNone(result)
self.assertEqual(result["title"], "Caf\u00e9\u00a0Society")
finally:
os.unlink(tmp_path)
def test_offset_lookup_handles_programme_element_larger_than_read_chunk(self):
long_desc = "x" * (2 * 1024 * 1024 + 1024)
xml = (
'\n'
"\n"
' \n'
' \n'
" Large Programme Current\n"
f" {long_desc}\n"
" \n"
"\n"
)
with tempfile.NamedTemporaryFile(
mode="w", suffix=".xml", encoding="utf-8", delete=False
) as f:
f.write(xml)
tmp_path = f.name
try:
src = EPGSource.objects.create(
name="Large Programme", source_type="xmltv", file_path=tmp_path
)
build_programme_index(src.id)
epg = EPGData.objects.create(
tvg_id="large.programme.channel",
name="Large Programme Channel",
epg_source=src,
)
result = find_current_program_for_tvg_id(epg)
self.assertIsNotNone(result)
self.assertEqual(result["title"], "Large Programme Current")
self.assertEqual(len(result["description"]), len(long_desc))
finally:
os.unlink(tmp_path)
@patch("apps.epg.tasks.build_programme_index_task")
def test_no_index_dispatches_build_and_returns_timeout(self, mock_build_task):
# Source with no index and file on disk
src = EPGSource.objects.create(
name="No Index",
source_type="xmltv",
file_path=FIXTURE_XML,
)
epg = EPGData.objects.create(
tvg_id="channel.current",
name="Current",
epg_source=src,
)
result = find_current_program_for_tvg_id(epg)
self.assertEqual(result, "timeout")
mock_build_task.delay.assert_called_once_with(src.id)
class BuildProgrammeIndexTests(TestCase):
def test_builds_index_from_fixture(self):
source = EPGSource.objects.create(
name="Index Test",
source_type="xmltv",
file_path=FIXTURE_XML,
)
build_programme_index(source.id)
source.refresh_from_db()
index = source.programme_index
self.assertIsNotNone(index)
channels = index["channels"]
self.assertIn("channel.current", channels)
self.assertIn("channel.past", channels)
# channel.empty has no programmes
self.assertNotIn("channel.empty", channels)
# Small fixture has no interleaved channels
self.assertEqual(index["interleaved_channels"], [])
def test_builds_index_when_channel_attribute_has_valid_xml_spacing(self):
xml = (
'\n'
"\n"
' \n'
' \n'
" Spaced Attribute Current\n"
" \n"
"\n"
)
with tempfile.NamedTemporaryFile(
mode="w", suffix=".xml", delete=False
) as f:
f.write(xml)
tmp_path = f.name
try:
src = EPGSource.objects.create(
name="Spaced Attribute", source_type="xmltv", file_path=tmp_path
)
build_programme_index(src.id)
src.refresh_from_db()
self.assertIn(
"spaced.channel", src.programme_index["channels"]
)
finally:
os.unlink(tmp_path)
def test_builds_index_from_extracted_file_path_for_gz_source(self):
xml = (
'\n'
"\n"
' \n'
' \n'
" GZ Current\n"
" \n"
"\n"
)
gz_path = None
xml_path = None
try:
with tempfile.NamedTemporaryFile(
mode="wb", suffix=".xml.gz", delete=False
) as gz_file:
gz_path = gz_file.name
with gzip.GzipFile(fileobj=gz_file, mode="wb") as compressed:
compressed.write(b"not the file the index should scan")
with tempfile.NamedTemporaryFile(
mode="w", suffix=".xml", encoding="utf-8", delete=False
) as xml_file:
xml_file.write(xml)
xml_path = xml_file.name
src = EPGSource.objects.create(
name="Extracted GZ",
source_type="xmltv",
file_path=gz_path,
extracted_file_path=xml_path,
)
build_programme_index(src.id)
src.refresh_from_db()
self.assertIn("gz.channel", src.programme_index["channels"])
finally:
if gz_path:
os.unlink(gz_path)
if xml_path:
os.unlink(xml_path)
def test_builds_index_ignores_elements_whose_name_only_starts_with_programme(self):
xml = (
'\n'
"\n"
' \n'
' \n'
" Not a Programme\n"
" \n"
' \n'
" Real Programme\n"
" \n"
"\n"
)
with tempfile.NamedTemporaryFile(
mode="w", suffix=".xml", encoding="utf-8", delete=False
) as f:
f.write(xml)
tmp_path = f.name
try:
src = EPGSource.objects.create(
name="Programme Prefix", source_type="xmltv", file_path=tmp_path
)
build_programme_index(src.id)
src.refresh_from_db()
self.assertIn("real.channel", src.programme_index["channels"])
self.assertNotIn("not.a.programme", src.programme_index["channels"])
finally:
os.unlink(tmp_path)
def test_nonexistent_source_does_not_raise(self):
# Should log error but not raise
build_programme_index(99999)
@patch("apps.epg.tasks.build_programme_index")
def test_task_builds_and_releases_lock_when_free(self, mock_build):
mock_redis = MagicMock()
mock_redis.set.return_value = True # lock acquired
with patch("core.utils.RedisClient.get_client", return_value=mock_redis):
build_programme_index_task(42)
mock_build.assert_called_once_with(42)
mock_redis.set.assert_called_once()
self.assertEqual(
mock_redis.set.call_args.args[0], "building_programme_index_42"
)
mock_redis.delete.assert_called_once_with("building_programme_index_42")
@patch("apps.epg.tasks.build_programme_index")
def test_task_skips_when_lock_held(self, mock_build):
mock_redis = MagicMock()
mock_redis.set.return_value = False # another build in flight
with patch("core.utils.RedisClient.get_client", return_value=mock_redis):
build_programme_index_task(42)
mock_build.assert_not_called()
mock_redis.delete.assert_not_called()
@patch("apps.epg.tasks.build_programme_index", side_effect=RuntimeError("boom"))
def test_task_releases_lock_on_failure(self, mock_build):
mock_redis = MagicMock()
mock_redis.set.return_value = True
with patch("core.utils.RedisClient.get_client", return_value=mock_redis):
with self.assertRaises(RuntimeError):
build_programme_index_task(42)
mock_redis.delete.assert_called_once_with("building_programme_index_42")
def test_per_channel_interleaved_marking(self):
xml = (
'\n'
"\n"
' \n'
' \n'
' \n'
" A Current\n"
" \n"
' \n'
" B Current\n"
" \n"
' \n'
" A Old\n"
" \n"
"\n"
)
with tempfile.NamedTemporaryFile(
mode="w", suffix=".xml", delete=False
) as f:
f.write(xml)
tmp_path = f.name
try:
src = EPGSource.objects.create(
name="Interleaved", source_type="xmltv", file_path=tmp_path
)
with patch("apps.epg.tasks._OFFSET_CAP", 1):
build_programme_index(src.id)
src.refresh_from_db()
index = src.programme_index
self.assertEqual(index["interleaved_channels"], ["A"])
epg_b = EPGData.objects.create(
tvg_id="B", name="B", epg_source=src
)
with patch(
"apps.epg.tasks._scan_from_offset_for_tvg_id"
) as mock_scan:
result_b = find_current_program_for_tvg_id(epg_b)
self.assertIsNotNone(result_b)
self.assertEqual(result_b["title"], "B Current")
mock_scan.assert_not_called()
epg_a = EPGData.objects.create(
tvg_id="A", name="A", epg_source=src
)
result_a = find_current_program_for_tvg_id(epg_a)
self.assertIsNotNone(result_a)
self.assertEqual(result_a["title"], "A Current")
finally:
os.unlink(tmp_path)