diff --git a/apps/epg/tests/test_programme_index.py b/apps/epg/tests/test_programme_index.py index 7ddce9e0..5ece0ae1 100644 --- a/apps/epg/tests/test_programme_index.py +++ b/apps/epg/tests/test_programme_index.py @@ -1,4 +1,5 @@ import os +import gzip import tempfile from unittest.mock import patch, MagicMock @@ -235,6 +236,282 @@ class FindCurrentProgramTests(TestCase): 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 @@ -306,6 +583,82 @@ class BuildProgrammeIndexTests(TestCase): 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)