diff --git a/apps/channels/tests/test_channel_api.py b/apps/channels/tests/test_channel_api.py index 052560a0..363f5925 100644 --- a/apps/channels/tests/test_channel_api.py +++ b/apps/channels/tests/test_channel_api.py @@ -1,5 +1,7 @@ from django.test import TestCase from django.contrib.auth import get_user_model +from django.utils import timezone +from datetime import timedelta from rest_framework.test import APIClient from rest_framework import status @@ -339,3 +341,166 @@ class ChannelManagerEffectiveValuesTests(TestCase): helper_row.effective_channel_group_id, shortcut_row.effective_channel_group_id, ) + + +class SeriesRuleAPITests(TestCase): + """API tests for series rule CRUD and bulk-remove endpoints.""" + + def setUp(self): + User = get_user_model() + self.admin = User.objects.create_user(username="admin_sr", password="pass") + self.admin.user_level = 10 + self.admin.save() + self.client = APIClient() + self.client.force_authenticate(user=self.admin) + + from core.models import CoreSettings + CoreSettings.set_dvr_series_rules([]) + + self.rules_url = "/api/channels/series-rules/" + self.bulk_remove_url = "/api/channels/series-rules/bulk-remove/" + + # --- POST (create/upsert) --- + + def test_create_rule_with_tvg_id(self): + resp = self.client.post(self.rules_url, { + "tvg_id": "some.channel", "title": "My Show", "mode": "all", + }, format="json") + self.assertEqual(resp.status_code, status.HTTP_200_OK) + self.assertEqual(len(resp.data["rules"]), 1) + self.assertEqual(resp.data["rules"][0]["tvg_id"], "some.channel") + + def test_create_title_only_rule_no_tvg_id(self): + """A rule with no tvg_id (title-only) is accepted when title is provided.""" + resp = self.client.post(self.rules_url, { + "tvg_id": "", "title": "Untethered Show", "mode": "all", + }, format="json") + self.assertEqual(resp.status_code, status.HTTP_200_OK) + rule = resp.data["rules"][0] + self.assertEqual(rule["tvg_id"], "") + self.assertEqual(rule["title"], "Untethered Show") + + def test_create_rule_requires_title_or_description(self): + resp = self.client.post(self.rules_url, { + "tvg_id": "some.channel", "title": "", "description": "", + }, format="json") + self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) + + def test_upsert_key_is_tvg_id_and_title(self): + """Two POST requests with same tvg_id but different titles create two rules.""" + self.client.post(self.rules_url, { + "tvg_id": "ch.1", "title": "Show A", "mode": "all", + }, format="json") + self.client.post(self.rules_url, { + "tvg_id": "ch.1", "title": "Show B", "mode": "all", + }, format="json") + resp = self.client.get(self.rules_url) + self.assertEqual(len(resp.data["rules"]), 2) + + def test_upsert_updates_existing_rule(self): + """POSTing with an existing (tvg_id, title) pair updates in place.""" + self.client.post(self.rules_url, { + "tvg_id": "ch.1", "title": "Show A", "mode": "all", + }, format="json") + self.client.post(self.rules_url, { + "tvg_id": "ch.1", "title": "Show A", "mode": "new", + }, format="json") + resp = self.client.get(self.rules_url) + self.assertEqual(len(resp.data["rules"]), 1) + self.assertEqual(resp.data["rules"][0]["mode"], "new") + + # --- DELETE (query params) --- + + def test_delete_rule_by_tvg_id_and_title(self): + self.client.post(self.rules_url, { + "tvg_id": "ch.1", "title": "Show A", "mode": "all", + }, format="json") + resp = self.client.delete( + self.rules_url + "?tvg_id=ch.1&title=Show+A" + ) + self.assertEqual(resp.status_code, status.HTTP_200_OK) + self.assertEqual(resp.data["rules"], []) + + def test_delete_title_only_rule(self): + """Title-only rules (tvg_id='') are deleted via empty tvg_id query param.""" + self.client.post(self.rules_url, { + "tvg_id": "", "title": "Untethered Show", "mode": "all", + }, format="json") + resp = self.client.delete( + self.rules_url + "?tvg_id=&title=Untethered+Show" + ) + self.assertEqual(resp.status_code, status.HTTP_200_OK) + self.assertEqual(resp.data["rules"], []) + + def test_delete_only_removes_matching_rule(self): + """Delete by (tvg_id, title) leaves other rules intact.""" + self.client.post(self.rules_url, {"tvg_id": "ch.1", "title": "Show A", "mode": "all"}, format="json") + self.client.post(self.rules_url, {"tvg_id": "ch.1", "title": "Show B", "mode": "all"}, format="json") + self.client.delete(self.rules_url + "?tvg_id=ch.1&title=Show+A") + resp = self.client.get(self.rules_url) + self.assertEqual(len(resp.data["rules"]), 1) + self.assertEqual(resp.data["rules"][0]["title"], "Show B") + + def test_delete_removes_future_recordings(self): + """DELETE cleans up future recordings that matched the rule.""" + from apps.channels.models import Recording + + group = ChannelGroup.objects.create(name="G") + channel = Channel.objects.create(channel_number=1, name="Ch", channel_group=group) + now = timezone.now() + Recording.objects.create( + channel=channel, + start_time=now + timedelta(hours=1), + end_time=now + timedelta(hours=2), + custom_properties={"program": {"tvg_id": "ch.1", "title": "Show A"}}, + ) + + self.client.post(self.rules_url, {"tvg_id": "ch.1", "title": "Show A", "mode": "all"}, format="json") + self.client.delete(self.rules_url + "?tvg_id=ch.1&title=Show+A") + self.assertEqual(Recording.objects.count(), 0) + + # --- POST bulk-remove --- + + def test_bulk_remove_with_tvg_id(self): + from apps.channels.models import Recording + + group = ChannelGroup.objects.create(name="G2") + channel = Channel.objects.create(channel_number=2, name="Ch2", channel_group=group) + now = timezone.now() + Recording.objects.create( + channel=channel, + start_time=now + timedelta(hours=1), + end_time=now + timedelta(hours=2), + custom_properties={"program": {"tvg_id": "ch.x", "title": "Show X"}}, + ) + resp = self.client.post(self.bulk_remove_url, {"tvg_id": "ch.x"}, format="json") + self.assertEqual(resp.status_code, status.HTTP_200_OK) + self.assertEqual(resp.data["removed"], 1) + self.assertEqual(Recording.objects.count(), 0) + + def test_bulk_remove_title_only_no_tvg_id(self): + """Bulk-remove accepts title alone (no tvg_id) for title-only rules.""" + from apps.channels.models import Recording + + group = ChannelGroup.objects.create(name="G3") + channel = Channel.objects.create(channel_number=3, name="Ch3", channel_group=group) + now = timezone.now() + Recording.objects.create( + channel=channel, + start_time=now + timedelta(hours=1), + end_time=now + timedelta(hours=2), + custom_properties={"program": {"tvg_id": "ch.a", "title": "Cross Show"}}, + ) + Recording.objects.create( + channel=channel, + start_time=now + timedelta(hours=3), + end_time=now + timedelta(hours=4), + custom_properties={"program": {"tvg_id": "ch.b", "title": "Cross Show"}}, + ) + resp = self.client.post(self.bulk_remove_url, {"title": "Cross Show"}, format="json") + self.assertEqual(resp.status_code, status.HTTP_200_OK) + self.assertEqual(resp.data["removed"], 2) + + def test_bulk_remove_requires_tvg_id_or_title(self): + resp = self.client.post(self.bulk_remove_url, {}, format="json") + self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) diff --git a/apps/channels/tests/test_series_rule_dedup.py b/apps/channels/tests/test_series_rule_dedup.py index 33aff695..9c498f99 100644 --- a/apps/channels/tests/test_series_rule_dedup.py +++ b/apps/channels/tests/test_series_rule_dedup.py @@ -716,3 +716,121 @@ class NonSeriesRecordingTests(SeriesRuleDedupBaseTestCase): prog = self._create_program(hours_from_now=2) result = evaluate_series_rules_impl() self.assertEqual(result["scheduled"], 1) + + +# --------------------------------------------------------------------------- +# Title-only rule tests: tvg_id is empty / omitted (cross-EPG matching) +# --------------------------------------------------------------------------- + +@patch("apps.channels.tasks.prefetch_recording_artwork") +@patch("apps.channels.signals.schedule_recording_task", return_value="mock-task-id") +class TitleOnlyRuleTests(SeriesRuleDedupBaseTestCase): + """Tests for series rules where tvg_id is omitted (searches all EPG channels).""" + + def setUp(self): + super().setUp() + _set_series_rules([{ + "tvg_id": "", + "mode": "all", + "title": "Test Show", + }]) + + @patch("apps.channels.tasks.acquire_task_lock", return_value=True) + @patch("apps.channels.tasks.release_task_lock") + def test_title_only_rule_schedules_recording(self, mock_release, mock_lock, + mock_schedule, mock_artwork): + """A rule with no tvg_id matches programs on any EPG channel by title.""" + from apps.channels.tasks import evaluate_series_rules_impl + + self._create_program(hours_from_now=2) + result = evaluate_series_rules_impl() + self.assertEqual(result["scheduled"], 1) + self.assertEqual(Recording.objects.count(), 1) + self.assertEqual(Recording.objects.first().channel, self.channel) + + @patch("apps.channels.tasks.acquire_task_lock", return_value=True) + @patch("apps.channels.tasks.release_task_lock") + def test_title_only_rule_matches_across_multiple_epg_channels( + self, mock_release, mock_lock, mock_schedule, mock_artwork + ): + """Title-only rule creates a recording per matching EPG channel (distinct programs).""" + from apps.channels.tasks import evaluate_series_rules_impl + + epg2 = EPGData.objects.create( + tvg_id="test.channel.2", name="Channel 2 EPG", + epg_source=self.epg_source, + ) + Channel.objects.create( + channel_number=2, name="Test Channel 2", epg_data=epg2 + ) + + start1 = self.now + timedelta(hours=2) + ProgramData.objects.create( + epg=self.epg, tvg_id="test.channel.1", + start_time=start1, end_time=start1 + timedelta(hours=1), + title="Test Show", sub_title="Episode 1", + ) + start2 = self.now + timedelta(hours=3) + ProgramData.objects.create( + epg=epg2, tvg_id="test.channel.2", + start_time=start2, end_time=start2 + timedelta(hours=1), + title="Test Show", sub_title="Episode 2", + ) + + result = evaluate_series_rules_impl() + self.assertEqual(result["scheduled"], 2) + self.assertEqual(Recording.objects.count(), 2) + + @patch("apps.channels.tasks.acquire_task_lock", return_value=True) + @patch("apps.channels.tasks.release_task_lock") + def test_title_only_rule_dedup_after_epg_refresh(self, mock_release, mock_lock, + mock_schedule, mock_artwork): + """Dedup works for title-only rules after EPG refresh reassigns program IDs.""" + from apps.channels.tasks import evaluate_series_rules_impl + + prog = self._create_program(hours_from_now=2) + evaluate_series_rules_impl() + self.assertEqual(Recording.objects.count(), 1) + + self._simulate_epg_refresh([self._program_data_for_refresh(prog)]) + result = evaluate_series_rules_impl() + self.assertEqual(Recording.objects.count(), 1) + self.assertEqual(result["scheduled"], 0) + + @patch("apps.channels.tasks.acquire_task_lock", return_value=True) + @patch("apps.channels.tasks.release_task_lock") + def test_invalid_rule_no_title_no_description_skipped( + self, mock_release, mock_lock, mock_schedule, mock_artwork + ): + """Rules with neither title nor description are skipped and flagged invalid.""" + from apps.channels.tasks import evaluate_series_rules_impl + + _set_series_rules([{"tvg_id": "", "mode": "all", "title": "", "description": ""}]) + self._create_program(hours_from_now=2) + result = evaluate_series_rules_impl() + self.assertEqual(result["scheduled"], 0) + statuses = [d.get("status") for d in result["details"]] + self.assertIn("invalid_rule", statuses) + + @patch("apps.channels.tasks.acquire_task_lock", return_value=True) + @patch("apps.channels.tasks.release_task_lock") + def test_program_on_epg_with_no_channel_skipped( + self, mock_release, mock_lock, mock_schedule, mock_artwork + ): + """Programs on an EPG source that has no Channel assigned are skipped gracefully.""" + from apps.channels.tasks import evaluate_series_rules_impl + + epg_orphan = EPGData.objects.create( + tvg_id="orphan.channel", name="Orphan EPG", + epg_source=self.epg_source, + ) + start = self.now + timedelta(hours=2) + ProgramData.objects.create( + epg=epg_orphan, tvg_id="orphan.channel", + start_time=start, end_time=start + timedelta(hours=1), + title="Test Show", sub_title="Episode 1", + ) + + result = evaluate_series_rules_impl() + self.assertEqual(result["scheduled"], 0) + self.assertEqual(Recording.objects.count(), 0) diff --git a/frontend/src/pages/__tests__/guideUtils.test.js b/frontend/src/pages/__tests__/guideUtils.test.js index 0c7f5cd4..18307016 100644 --- a/frontend/src/pages/__tests__/guideUtils.test.js +++ b/frontend/src/pages/__tests__/guideUtils.test.js @@ -1282,10 +1282,22 @@ describe('guideUtils', () => { }); describe('deleteSeriesRuleByTvgId', () => { - it('should delete series rule via API', async () => { + it('should delete series rule via API with tvg_id and title', async () => { + await guideUtils.deleteSeriesRuleByTvgId('tvg-1', 'My Show'); + + expect(API.deleteSeriesRule).toHaveBeenCalledWith('tvg-1', 'My Show'); + }); + + it('should forward undefined title when not provided', async () => { await guideUtils.deleteSeriesRuleByTvgId('tvg-1'); - expect(API.deleteSeriesRule).toHaveBeenCalledWith('tvg-1'); + expect(API.deleteSeriesRule).toHaveBeenCalledWith('tvg-1', undefined); + }); + + it('should work with empty tvg_id for title-only rules', async () => { + await guideUtils.deleteSeriesRuleByTvgId('', 'Title-Only Show'); + + expect(API.deleteSeriesRule).toHaveBeenCalledWith('', 'Title-Only Show'); }); }); diff --git a/frontend/src/utils/cards/__tests__/RecordingCardUtils.test.js b/frontend/src/utils/cards/__tests__/RecordingCardUtils.test.js index 4dad5900..2af98a1c 100644 --- a/frontend/src/utils/cards/__tests__/RecordingCardUtils.test.js +++ b/frontend/src/utils/cards/__tests__/RecordingCardUtils.test.js @@ -198,16 +198,28 @@ describe('RecordingCardUtils', () => { title: 'Test Series', scope: 'title', }); - expect(API.deleteSeriesRule).toHaveBeenCalledWith('series-123'); + expect(API.deleteSeriesRule).toHaveBeenCalledWith( + 'series-123', + 'Test Series' + ); }); - it('does nothing when tvg_id is not provided', async () => { - const seriesInfo = { title: 'Test Series' }; + it('works for title-only rules with no tvg_id', async () => { + API.bulkRemoveSeriesRecordings.mockResolvedValue(); + API.deleteSeriesRule.mockResolvedValue(); + const seriesInfo = { tvg_id: undefined, title: 'Title-Only Show' }; await deleteSeriesAndRule(seriesInfo); - expect(API.bulkRemoveSeriesRecordings).not.toHaveBeenCalled(); - expect(API.deleteSeriesRule).not.toHaveBeenCalled(); + expect(API.bulkRemoveSeriesRecordings).toHaveBeenCalledWith({ + tvg_id: '', + title: 'Title-Only Show', + scope: 'title', + }); + expect(API.deleteSeriesRule).toHaveBeenCalledWith( + undefined, + 'Title-Only Show' + ); }); it('handles bulk remove error gracefully', async () => {