mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-21 01:05:30 +00:00
7086f41d64
- Moved matching logic to a dedicated module for better organization and testability. - Made single-channel auto-matching asynchronous, allowing for larger EPG libraries without hitting HTTP timeouts. - Enhanced memory management and throughput during EPG matching, including optimizations for fuzzy matching and bulk processing. - Fixed various reliability issues in the auto-matching process, ensuring accurate channel assignments and improved UI feedback. - Updated API views and frontend components to reflect changes in the matching process and provide real-time notifications. - Added tests for EPG matching functionality and name normalization. - Single-channel and selected-channel auto-match always run, even when the channel already has EPG assigned; match-all (no channel IDs) still only processes channels without EPG.
58 lines
2 KiB
Python
58 lines
2 KiB
Python
"""Tests for applying EPG auto-match results to channels."""
|
|
from unittest.mock import patch
|
|
|
|
from django.test import TestCase
|
|
|
|
from apps.channels.epg_matching import apply_matched_epg_to_channels
|
|
from apps.channels.models import Channel
|
|
from apps.epg.models import EPGData, EPGSource
|
|
|
|
|
|
class ApplyMatchedEpgToChannelsTests(TestCase):
|
|
def setUp(self):
|
|
self.source = EPGSource.objects.create(
|
|
name="XML EPG",
|
|
source_type="xmltv",
|
|
url="http://example.com/epg.xml",
|
|
)
|
|
self.epg_one = EPGData.objects.create(
|
|
tvg_id="ch.one",
|
|
name="Channel One",
|
|
epg_source=self.source,
|
|
)
|
|
self.epg_two = EPGData.objects.create(
|
|
tvg_id="ch.two",
|
|
name="Channel Two",
|
|
epg_source=self.source,
|
|
)
|
|
self.channel = Channel.objects.create(
|
|
channel_number=1,
|
|
name="Channel One",
|
|
tvg_id="ch.one",
|
|
epg_data=self.epg_one,
|
|
)
|
|
|
|
@patch("apps.epg.tasks.parse_programs_for_tvg_id.delay")
|
|
def test_skips_unchanged_assignment(self, mock_delay):
|
|
changed = apply_matched_epg_to_channels(
|
|
[{"id": self.channel.id, "epg_data_id": self.epg_one.id}]
|
|
)
|
|
|
|
self.assertEqual(changed, [])
|
|
mock_delay.assert_not_called()
|
|
self.channel.refresh_from_db()
|
|
self.assertEqual(self.channel.epg_data_id, self.epg_one.id)
|
|
|
|
@patch("apps.epg.tasks.parse_programs_for_tvg_id.delay")
|
|
def test_updates_changed_assignment_and_dispatches_parse(self, mock_delay):
|
|
changed = apply_matched_epg_to_channels(
|
|
[{"id": self.channel.id, "epg_data_id": self.epg_two.id}]
|
|
)
|
|
|
|
self.assertEqual(
|
|
changed,
|
|
[{"channel_id": self.channel.id, "epg_data_id": self.epg_two.id}],
|
|
)
|
|
mock_delay.assert_called_once_with(self.epg_two.id)
|
|
self.channel.refresh_from_db()
|
|
self.assertEqual(self.channel.epg_data_id, self.epg_two.id)
|