mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-17 16:50:53 +00:00
tests: Add API tests for series rule CRUD operations and enhance delete functionality for title-only rules
This commit is contained in:
SergeantPanda
parent
8f97688bb1
commit
fe9f83a6af
4 changed files with 314 additions and 7 deletions
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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');
|
||||
});
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -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 () => {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue