import json from datetime import datetime, timedelta, timezone from unittest.mock import patch import feedparser from django.contrib.auth.models import User from django.test import TestCase from .models import EpisodeProgress, PodcastEpisode, PodcastFeed, PodcastQueue, PodcastSettings from .views import _refresh_feed _parse = feedparser.parse def _rss(items): """Items are (guid, date) or (guid, date, extra XML inside the ); the guid is the title too.""" entries = ''.join( f'{guid}{guid}{date}' f'{extra[0] if extra else ""}' f'' for guid, date, *extra in items ) return ('' f'Feed{entries}') class QueuePositionTests(TestCase): def setUp(self): self.user = User.objects.create_user(username='alice', password='pw12345678') self.client.force_login(self.user) self.feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/feed', title='Feed') base = datetime(2026, 1, 1, tzinfo=timezone.utc) self.eps = [ PodcastEpisode.objects.create( feed=self.feed, guid=f'e{i}', title=f'E{i}', audio_url=f'https://example.org/e{i}.mp3', pub_date=base + timedelta(days=i), ) for i in range(6) ] def queue(self): return list(PodcastQueue.objects.filter(user=self.user).order_by('position').values_list('episode_id', flat=True)) def add(self, *episodes, **extra): body = {'episode_ids': [e.id for e in episodes], **extra} return self.client.post('/podcasts/queue/add/', json.dumps(body), content_type='application/json') def test_default_adds_to_the_back(self): self.add(self.eps[0]) self.add(self.eps[1]) self.assertEqual(self.queue(), [self.eps[0].id, self.eps[1].id]) def test_front_from_setting(self): self.add(self.eps[0], self.eps[1]) PodcastSettings.objects.update_or_create(user=self.user, defaults={'enqueue_position': 'front'}) self.add(self.eps[2]) self.assertEqual(self.queue(), [self.eps[2].id, self.eps[0].id, self.eps[1].id]) def test_after_current_uses_client_hint(self): self.add(self.eps[0], self.eps[1], self.eps[2]) self.add(self.eps[3], where='after_current', current_episode_id=self.eps[1].id) self.assertEqual(self.queue(), [e.id for e in (self.eps[0], self.eps[1], self.eps[3], self.eps[2])]) def test_after_current_falls_back_to_last_listened(self): self.add(self.eps[0], self.eps[1], self.eps[2]) EpisodeProgress.objects.update_or_create(user=self.user, episode=self.eps[2], defaults={'position_seconds': 30}) self.add(self.eps[3], where='after_current') self.assertEqual(self.queue(), [e.id for e in (self.eps[0], self.eps[1], self.eps[2], self.eps[3])]) def test_after_current_without_current_goes_to_front(self): self.add(self.eps[0]) self.add(self.eps[1], where='after_current') self.assertEqual(self.queue(), [self.eps[1].id, self.eps[0].id]) def test_already_queued_stays_where_it_is(self): self.add(self.eps[0], self.eps[1]) r = self.add(self.eps[0], where='front') self.assertFalse(r.json()['created']) self.assertEqual(self.queue(), [self.eps[0].id, self.eps[1].id]) def test_single_episode_id_still_works(self): r = self.client.post('/podcasts/queue/add/', json.dumps({'episode_id': self.eps[4].id}), content_type='application/json') self.assertTrue(r.json()['created']) self.assertEqual(self.queue(), [self.eps[4].id]) def test_other_users_episode_is_not_found(self): bob = User.objects.create_user(username='bob', password='pw12345678') feed = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/b', title='B') ep = PodcastEpisode.objects.create(feed=feed, guid='x', title='X', audio_url='https://example.org/x.mp3') r = self.add(ep) self.assertEqual(r.status_code, 404) def test_unknown_position_is_rejected(self): self.assertEqual(self.add(self.eps[0], where='middle').status_code, 400) def test_settings_roundtrip(self): r = self.client.get('/podcasts/settings/') self.assertEqual(r.json()['enqueue_position'], 'back') self.assertEqual(r.json()['auto_queue_position'], 'front') r = self.client.post('/podcasts/settings/', json.dumps({'auto_queue_position': 'after_current'}), content_type='application/json') self.assertEqual(r.json()['auto_queue_position'], 'after_current') self.assertEqual(PodcastSettings.objects.get(user=self.user).auto_queue_position, 'after_current') bad = self.client.post('/podcasts/settings/', json.dumps({'enqueue_position': 'nowhere'}), content_type='application/json') self.assertEqual(bad.status_code, 400) class InboxRuleTests(TestCase): """Queued or started episodes leave the inbox; played means the last 30 s.""" def setUp(self): self.user = User.objects.create_user(username='alice', password='pw12345678') self.client.force_login(self.user) feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/f', title='F') self.eps = [PodcastEpisode.objects.create(feed=feed, guid=f'g{i}', title=f'E{i}', duration_seconds=3600, audio_url=f'https://example.org/{i}.mp3') for i in range(3)] def inbox(self): return {e['id'] for e in self.client.get('/podcasts/inbox/').json()['episodes']} def post(self, url, body): return self.client.post(url, json.dumps(body), content_type='application/json') def test_queued_and_started_leave_the_inbox(self): self.assertEqual(self.inbox(), {e.id for e in self.eps}) self.post('/podcasts/queue/add/', {'episode_ids': [self.eps[0].id]}) self.post('/podcasts/progress/save/', {'episode_id': self.eps[1].id, 'position_seconds': 12}) self.assertEqual(self.inbox(), {self.eps[2].id}) p = EpisodeProgress.objects.get(user=self.user, episode=self.eps[0]) self.assertFalse(p.played) def test_played_within_the_last_30_seconds(self): r = self.post('/podcasts/progress/save/', {'episode_id': self.eps[0].id, 'position_seconds': 3500}) self.assertFalse(r.json()['played']) # 90 % but 100 s left: not yet r = self.post('/podcasts/progress/save/', {'episode_id': self.eps[0].id, 'position_seconds': 3575}) self.assertTrue(r.json()['played']) class AutoQueueTests(TestCase): """New episodes of an auto_queue feed land where the user's setting says.""" def setUp(self): self.user = User.objects.create_user(username='alice', password='pw12345678') self.feed = PodcastFeed.objects.create( user=self.user, rss_url='https://example.org/feed', title='Feed', auto_queue=True, last_refreshed_at=datetime(2026, 1, 1, tzinfo=timezone.utc), # not the first fetch ) other = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/o', title='O') self.old = [ PodcastEpisode.objects.create(feed=other, guid=f'o{i}', title=f'O{i}', audio_url=f'https://example.org/o{i}.mp3') for i in range(3) ] for i, ep in enumerate(self.old): PodcastQueue.objects.create(user=self.user, episode=ep, position=i) self.rss = _rss([ ('new1', 'Mon, 05 Jan 2026 10:00:00 GMT'), ('new2', 'Tue, 06 Jan 2026 10:00:00 GMT'), ]) def refresh(self): with patch('podcasts.views._fetch_feed', lambda url: _parse(self.rss)): _refresh_feed(self.feed) order = PodcastQueue.objects.filter(user=self.user).order_by('position') return [q.episode.guid for q in order] def test_front_is_the_default(self): self.assertEqual(self.refresh(), ['new2', 'new1', 'o0', 'o1', 'o2']) def test_back_appends_oldest_first(self): PodcastSettings.objects.create(user=self.user, auto_queue_position='back') self.assertEqual(self.refresh(), ['o0', 'o1', 'o2', 'new1', 'new2']) def test_after_current(self): PodcastSettings.objects.create(user=self.user, auto_queue_position='after_current') EpisodeProgress.objects.create(user=self.user, episode=self.old[1], position_seconds=12) self.assertEqual(self.refresh(), ['o0', 'o1', 'new2', 'new1', 'o2']) def test_feed_without_auto_queue_queues_nothing(self): self.feed.auto_queue = False self.feed.save() self.assertEqual(self.refresh(), ['o0', 'o1', 'o2']) class FirstFetchTests(TestCase): """A new subscription doesn't flood the inbox with its back catalogue.""" def setUp(self): self.user = User.objects.create_user(username='alice', password='pw12345678') self.client.force_login(self.user) self.rss = _rss([(f'e{i}', f'Mon, {i + 1:02d} Jan 2026 10:00:00 GMT') for i in range(5)]) def subscribe(self): with patch('podcasts.views._fetch_feed', lambda url: _parse(self.rss)): return self.client.post('/podcasts/feeds/add/', json.dumps({'rss_url': 'https://example.org/f'}), content_type='application/json') def test_only_the_newest_is_new(self): self.assertEqual(self.subscribe().json()['new_episodes'], 5) inbox = self.client.get('/podcasts/inbox/').json()['episodes'] self.assertEqual([e['title'] for e in inbox], ['e4']) # The others are still listed for the podcast, not marked played. feed = PodcastFeed.objects.get(user=self.user) eps = self.client.get(f'/podcasts/feeds/{feed.id}/episodes/').json()['episodes'] self.assertEqual(len(eps), 5) self.assertFalse(any(e['played'] for e in eps)) def test_later_fetches_keep_every_new_episode(self): self.subscribe() feed = PodcastFeed.objects.get(user=self.user) self.rss = _rss([(f'e{i}', f'Mon, {i + 1:02d} Jan 2026 10:00:00 GMT') for i in range(8)]) with patch('podcasts.views._fetch_feed', lambda url: _parse(self.rss)): _refresh_feed(feed) inbox = self.client.get('/podcasts/inbox/').json()['episodes'] self.assertEqual([e['title'] for e in inbox], ['e7', 'e6', 'e5', 'e4']) class RefreshAllTests(TestCase): """One request refreshes all feeds on the server; one request returns the state.""" def setUp(self): self.user = User.objects.create_user(username='alice', password='pw12345678') self.client.force_login(self.user) self.a = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/a', title='A') self.b = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/b', title='B') bob = User.objects.create_user(username='bob', password='pw12345678') self.bobs = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/a', title='A') self.fetched = [] def fake_fetch(self, url): self.fetched.append(url) if url.endswith('/b'): raise ValueError('broken feed') return _parse(_rss([(url[-1] + '1', 'Mon, 05 Jan 2026 10:00:00 GMT')])) def test_refresh_all_reports_counts_and_skips_recent(self): with patch('podcasts.views._fetch_feed', self.fake_fetch): r = self.client.post('/podcasts/feeds/refresh-all/', '{}', content_type='application/json') self.assertEqual(r.json(), {'ok': True, 'refreshed': 1, 'failed': 1, 'new_episodes': 1}) self.assertEqual(sorted(self.fetched), ['https://example.org/a', 'https://example.org/b']) # A was just refreshed: a second call within max_age only retries B. self.fetched.clear() with patch('podcasts.views._fetch_feed', self.fake_fetch): self.client.post('/podcasts/feeds/refresh-all/', '{}', content_type='application/json') self.assertEqual(self.fetched, ['https://example.org/b']) def test_a_shared_feed_url_is_fetched_once(self): from .views import _refresh_feeds with patch('podcasts.views._fetch_feed', self.fake_fetch): refreshed, failed, new = _refresh_feeds([self.a, self.bobs]) self.assertEqual(self.fetched, ['https://example.org/a']) self.assertEqual((refreshed, failed, new), (2, 0, 2)) def test_state_in_one_response(self): ep = PodcastEpisode.objects.create(feed=self.a, guid='x', title='X', audio_url='https://example.org/x.mp3') PodcastQueue.objects.create(user=self.user, episode=ep, position=0) r = self.client.get('/podcasts/state/') data = r.json() self.assertEqual({f['title'] for f in data['feeds']}, {'A', 'B'}) self.assertEqual([q['episode__id'] for q in data['queue']], [ep.id]) self.assertEqual([e['id'] for e in data['inbox']], [ep.id]) self.assertTrue(data['inbox'][0]['in_queue']) # Shownotes stay out of the one-request state (they can be large). self.assertNotIn('description', data['inbox'][0]) self.assertEqual(data['settings'], {'enqueue_position': 'back', 'auto_queue_position': 'front', 'keep_sorted': ''}) # --------------------------------------------------------------------------- # Shownotes, chapters, transcripts # --------------------------------------------------------------------------- NS = ( 'xmlns:podcast="https://podcastindex.org/namespace/1.0" ' 'xmlns:psc="http://podlove.org/simple-chapters" ' 'xmlns:content="http://purl.org/rss/1.0/modules/content/" ' 'xmlns:itunes="http://www.itunes.com/dtds/podcast-1.0.dtd"' ) def _rich_rss(items, ns=NS): """Like _rss, but each item is (guid, extra XML inside the ).""" entries = ''.join( f'{guid}{guid}Mon, 05 Jan 2026 10:00:00 GMT' f'{extra}' for guid, extra in items ) return f'Feed{entries}' PSC = ( '' '' '' '' '' '' ) TRANSCRIPTS = ( '' '' '' ) CHAPTERS_JSON = { 'version': '1.2.0', 'chapters': [ {'startTime': 95.5, 'title': 'Zweites', 'url': 'https://x.org/b', 'img': 'https://x.org/b.jpg'}, {'startTime': 0, 'title': 'Erstes'}, {'startTime': 50, 'title': 'Nur Bild', 'toc': False}, {'startTime': 'kaputt', 'title': 'Ungültig'}, {'startTime': 120, 'title': 'Link', 'url': 'javascript:alert(1)'}, ], } class FeedExtrasParsingTests(TestCase): """What feedparser (6.0.x) hands us for these tags, and what we store.""" def setUp(self): self.user = User.objects.create_user(username='alice', password='pw12345678') self.feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/feed', title='Feed') def refresh(self, rss): with patch('podcasts.views._fetch_feed', lambda url: _parse(rss)): _refresh_feed(self.feed) def test_feedparser_keys(self): import podcasts.views # noqa: F401 -- installs the feedparser extension rss = _rich_rss([('a', PSC + TRANSCRIPTS + '')]) entry = _parse(rss).entries[0] self.assertEqual(entry['podcast_chapters']['url'], 'https://x.org/c.json') self.assertEqual([t['url'] for t in entry['podcast_transcripts']], ['https://example.org/a.vtt', 'https://example.org/a.srt', 'javascript:alert(1)']) self.assertEqual(entry['psc_chapters']['chapters'][1]['start'], '00:01:02.500') def test_other_prefix_for_the_podcast_namespace(self): ns = NS.replace('xmlns:podcast=', 'xmlns:pc=') rss = _rich_rss([('a', '' '')], ns) self.refresh(rss) ep = PodcastEpisode.objects.get() self.assertEqual(ep.chapters_url, 'https://x.org/c.json') self.assertEqual(ep.transcripts, [{'url': 'https://example.org/a.vtt', 'type': 'text/vtt', 'language': ''}]) def test_psc_chapters(self): self.refresh(_rich_rss([('a', PSC)])) ep = PodcastEpisode.objects.get() self.assertEqual(ep.chapters, [ {'start': 0.0, 'title': 'Intro'}, {'start': 62.5, 'title': 'Teil & zwei', 'url': 'https://x.org/2', 'img': 'https://x.org/2.jpg'}, {'start': 725.0, 'title': 'Kurz'}, {'start': 3723.0, 'title': 'Drei'}, ]) self.assertEqual(ep.chapters_url, '') def test_transcripts_and_chapters_url(self): self.refresh(_rich_rss([('a', TRANSCRIPTS + '')])) ep = PodcastEpisode.objects.get() self.assertEqual(ep.transcripts, [ {'url': 'https://example.org/a.vtt', 'type': 'text/vtt', 'language': 'de'}, {'url': 'https://example.org/a.srt', 'type': 'application/srt', 'language': ''}, ]) self.assertEqual(ep.chapters_url, 'https://x.org/c.json') self.assertEqual(ep.chapters, []) self.assertIsNone(ep.chapters_fetched_at) def test_content_encoded_is_preferred(self): self.refresh(_rich_rss([ ('a', 'Kurz' 'Lang Link

]]>'), ('b', 'Nur kurz'), ])) self.assertIn('Lang', PodcastEpisode.objects.get(guid='a').description) self.assertEqual(PodcastEpisode.objects.get(guid='b').description, 'Nur kurz') def test_existing_episodes_are_updated(self): self.refresh(_rich_rss([('a', 'Alt'), ('b', 'Gleich')])) a, b = PodcastEpisode.objects.get(guid='a'), PodcastEpisode.objects.get(guid='b') rss = _rich_rss([ ('a', 'Neu' + PSC + TRANSCRIPTS), ('b', 'Gleich'), ]) with patch('podcasts.views.PodcastEpisode.objects.bulk_update', wraps=PodcastEpisode.objects.bulk_update) as bulk: self.refresh(rss) # Only the episode that changed is written. self.assertEqual([e.guid for e in bulk.call_args.args[0]], ['a']) a.refresh_from_db() self.assertEqual(a.description, 'Neu') self.assertEqual(len(a.chapters), 4) self.assertEqual(len(a.transcripts), 2) self.assertEqual(PodcastEpisode.objects.count(), 2) # Nothing changed: nothing written. with patch('podcasts.views.PodcastEpisode.objects.bulk_update') as bulk: self.refresh(rss) bulk.assert_not_called() def test_fetched_chapters_survive_a_refresh_until_the_url_changes(self): tag = '' self.refresh(_rich_rss([('a', tag.format('c1'))])) fetched = datetime(2026, 1, 2, tzinfo=timezone.utc) PodcastEpisode.objects.update(chapters=[{'start': 0.0, 'title': 'X'}], chapters_fetched_at=fetched) self.refresh(_rich_rss([('a', tag.format('c1'))])) ep = PodcastEpisode.objects.get() self.assertEqual(ep.chapters, [{'start': 0.0, 'title': 'X'}]) self.assertEqual(ep.chapters_fetched_at, fetched) self.refresh(_rich_rss([('a', tag.format('c2'))])) ep.refresh_from_db() self.assertEqual((ep.chapters_url, ep.chapters, ep.chapters_fetched_at), ('https://x.org/c2.json', [], None)) class ShownotesTests(TestCase): def test_sanitizer(self): from .shownotes import sanitize_html dirty = ( '

Hallo Link' '' 'böseauch' '' 'Text bleibtoffen' ) self.assertEqual( sanitize_html(dirty), '

Hallo Link' 'böseauch' 'Text bleibtoffen

', ) self.assertEqual(sanitize_html('1 < 2 && 3'), '1 < 2 && 3') self.assertEqual(sanitize_html('Mail'), 'Mail') def test_timestamps(self): from .shownotes import extract_timestamps notes = ( '

Kapitel:
00:00 Intro
(12:34) - Thema eins
1:02:03 Zum Schluss

' '
  • [05:07] Liste
  • Um 20:15 geht es los
  • 99:99 kaputt
' ) self.assertEqual(extract_timestamps(notes), [ {'seconds': 0, 'label': 'Intro'}, {'seconds': 754, 'label': 'Thema eins'}, {'seconds': 3723, 'label': 'Zum Schluss'}, {'seconds': 307, 'label': 'Liste'}, ]) self.assertEqual(len(extract_timestamps(notes, duration_seconds=1000)), 3) class _FakeResponse: def __init__(self, body=b'', status=200, headers=None): self.body, self.status_code, self.headers = body, status, headers or {} self.raw = None def iter_content(self, chunk_size=1): yield self.body def close(self): pass class EpisodeDetailTests(TestCase): def setUp(self): self.user = User.objects.create_user(username='alice', password='pw12345678') self.client.force_login(self.user) self.feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/feed', title='Feed', artwork_url='https://example.org/art.jpg') self.ep = PodcastEpisode.objects.create( feed=self.feed, guid='a', title='A', audio_url='https://example.org/a.mp3', duration_seconds=3600, pub_date=datetime(2026, 1, 5, 10, tzinfo=timezone.utc), description='

Hallo
01:02 Thema x

', transcripts=[{'url': 'https://example.org/a.vtt', 'type': 'text/vtt', 'language': 'de'}], chapters_url='https://x.org/c.json', ) self.requests = [] def fake_get(self, url, **kwargs): self.requests.append(url) return _FakeResponse(json.dumps(CHAPTERS_JSON).encode()) def get(self, ep=None): with patch('podcasts.chapters.assert_safe_url'), patch('podcasts.chapters.requests.get', self.fake_get): return self.client.get(f'/podcasts/episodes/{(ep or self.ep).id}/') def test_detail(self): EpisodeProgress.objects.create(user=self.user, episode=self.ep, position_seconds=42) r = self.get() self.assertEqual(r.status_code, 200) data = r.json() self.assertEqual(data['description'], '

Hallo
01:02 Thema x

') self.assertEqual(data['timestamps'], [{'seconds': 62, 'label': 'Thema x'}]) self.assertEqual(data['transcripts'], self.ep.transcripts) self.assertEqual(data['feed'], {'id': self.feed.id, 'title': 'Feed', 'artwork_url': 'https://example.org/art.jpg'}) self.assertEqual((data['position_seconds'], data['played'], data['in_queue']), (42, False, False)) self.assertEqual(data['pub_date'], '2026-01-05T10:00:00+00:00') self.assertEqual(data['chapters'], [ {'start': 0.0, 'title': 'Erstes'}, {'start': 95.5, 'title': 'Zweites', 'url': 'https://x.org/b', 'img': 'https://x.org/b.jpg'}, {'start': 120.0, 'title': 'Link'}, ]) def test_chapters_are_fetched_once(self): self.get() self.get() self.assertEqual(self.requests, ['https://x.org/c.json']) self.ep.refresh_from_db() self.assertEqual(len(self.ep.chapters), 3) self.assertIsNotNone(self.ep.chapters_fetched_at) def test_inline_chapters_need_no_fetch(self): self.ep.chapters = [{'start': 0.0, 'title': 'Inline'}] self.ep.save() self.assertEqual(self.get().json()['chapters'], [{'start': 0.0, 'title': 'Inline'}]) self.assertEqual(self.requests, []) def test_redirect_is_followed_and_rechecked(self): def fake_get(url, **kwargs): self.requests.append(url) if url.endswith('c.json'): return _FakeResponse(status=302, headers={'Location': '/real.json'}) return _FakeResponse(json.dumps(CHAPTERS_JSON).encode()) with patch('podcasts.chapters.assert_safe_url') as check, patch('podcasts.chapters.requests.get', fake_get): r = self.client.get(f'/podcasts/episodes/{self.ep.id}/') self.assertEqual(len(r.json()['chapters']), 3) self.assertEqual([c.args[0] for c in check.call_args_list], ['https://x.org/c.json', 'https://x.org/real.json']) def test_private_chapters_url_is_refused_and_remembered(self): self.ep.chapters_url = 'http://127.0.0.1/c.json' self.ep.save() with patch('podcasts.chapters.requests.get', self.fake_get): r = self.client.get(f'/podcasts/episodes/{self.ep.id}/') self.assertEqual(r.json()['chapters'], []) self.assertEqual(self.requests, []) self.ep.refresh_from_db() self.assertIsNotNone(self.ep.chapters_fetched_at) def test_oversized_or_broken_chapters(self): from . import chapters with patch('podcasts.chapters.assert_safe_url'), patch( 'podcasts.chapters.requests.get', lambda url, **kw: _FakeResponse(b'x' * (chapters.CHAPTERS_MAX_BYTES + 1)), ): with self.assertRaises(chapters.ChaptersInvalid): chapters.fetch('https://x.org/big.json') with patch('podcasts.chapters.assert_safe_url'), patch( 'podcasts.chapters.requests.get', lambda url, **kw: _FakeResponse(b'{"chapters": 3}'), ): with self.assertRaises(chapters.ChaptersInvalid): chapters.fetch('https://x.org/odd.json') def test_server_error_is_retried_later(self): with patch('podcasts.chapters.assert_safe_url'), patch( 'podcasts.chapters.requests.get', lambda url, **kw: _FakeResponse(status=503), ): r = self.client.get(f'/podcasts/episodes/{self.ep.id}/') self.assertEqual(r.json()['chapters'], []) self.ep.refresh_from_db() self.assertIsNone(self.ep.chapters_fetched_at) def test_other_users_episode_is_not_found(self): bob = User.objects.create_user(username='bob', password='pw12345678') feed = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/b', title='B') ep = PodcastEpisode.objects.create(feed=feed, guid='x', title='X', audio_url='https://example.org/x.mp3', chapters_url='https://x.org/c.json') self.assertEqual(self.get(ep).status_code, 404) self.assertEqual(self.requests, []) def test_login_required(self): self.client.logout() self.assertEqual(self.get().status_code, 401) def test_favorite_and_last_played_at(self): r = self.get().json() self.assertEqual((r['favorite'], r['last_played_at']), (False, None)) p = EpisodeProgress.objects.create(user=self.user, episode=self.ep, position_seconds=5, favorite=True) EpisodeProgress.objects.filter(pk=p.pk).update(updated_at=datetime(2026, 2, 3, 8, tzinfo=timezone.utc)) r = self.get().json() self.assertEqual((r['favorite'], r['last_played_at']), (True, '2026-02-03T08:00:00+00:00')) # --------------------------------------------------------------------------- # Favorites, episode list, statistics, OPML export # --------------------------------------------------------------------------- def _at(month, day): return datetime(2026, month, day, 9, tzinfo=timezone.utc) def _progress(user, episode, updated_at, **fields): p = EpisodeProgress.objects.create(user=user, episode=episode, **fields) # auto_now would overwrite updated_at on save(); update() doesn't. EpisodeProgress.objects.filter(pk=p.pk).update(updated_at=updated_at) return p class FavoriteTests(TestCase): def setUp(self): self.user = User.objects.create_user(username='alice', password='pw12345678') self.client.force_login(self.user) feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/feed', title='Feed') self.ep = PodcastEpisode.objects.create(feed=feed, guid='a', title='A', audio_url='https://example.org/a.mp3') def post(self, body=None, ep=None): return self.client.post(f'/podcasts/episodes/{(ep or self.ep).id}/favorite/', '' if body is None else json.dumps(body), content_type='application/json') def test_toggle_creates_the_progress_row(self): self.assertFalse(EpisodeProgress.objects.exists()) self.assertEqual(self.post().json(), {'ok': True, 'favorite': True}) p = EpisodeProgress.objects.get(user=self.user, episode=self.ep) self.assertTrue(p.favorite) self.assertEqual((p.played, p.dismissed, p.position_seconds), (False, False, 0)) self.assertEqual(self.post({}).json(), {'ok': True, 'favorite': False}) self.assertFalse(EpisodeProgress.objects.get(pk=p.pk).favorite) def test_set_explicitly(self): self.assertEqual(self.post({'favorite': True}).json()['favorite'], True) self.assertEqual(self.post({'favorite': True}).json()['favorite'], True) self.assertTrue(EpisodeProgress.objects.get().favorite) self.assertEqual(self.post({'favorite': False}).json()['favorite'], False) self.assertEqual(self.post({'favorite': False}).json()['favorite'], False) self.assertFalse(EpisodeProgress.objects.get().favorite) def test_starring_keeps_the_listening_time(self): p = _progress(self.user, self.ep, _at(2, 1), position_seconds=30) self.post({'favorite': True}) p.refresh_from_db() self.assertTrue(p.favorite) self.assertEqual((p.position_seconds, p.updated_at), (30, _at(2, 1))) def test_owner_only(self): bob = User.objects.create_user(username='bob', password='pw12345678') feed = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/b', title='B') ep = PodcastEpisode.objects.create(feed=feed, guid='x', title='X', audio_url='https://example.org/x.mp3') self.assertEqual(self.post({'favorite': True}, ep).status_code, 404) self.assertFalse(EpisodeProgress.objects.exists()) self.assertEqual(self.client.post('/podcasts/episodes/999999/favorite/', '{}', content_type='application/json').status_code, 404) def test_login_and_post_required(self): self.assertEqual(self.client.get(f'/podcasts/episodes/{self.ep.id}/favorite/').status_code, 405) self.client.logout() self.assertEqual(self.post({'favorite': True}).status_code, 401) class EpisodeListTests(TestCase): """GET /podcasts/episodes/: all feeds, filters, history, search, pages.""" def setUp(self): self.user = User.objects.create_user(username='alice', password='pw12345678') self.client.force_login(self.user) self.fa = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/a', title='Alpha', artwork_url='https://example.org/a.jpg') self.fb = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/b', title='Beta') def ep(feed, guid, title, pub_date, **kw): return PodcastEpisode.objects.create(feed=feed, guid=guid, title=title, pub_date=pub_date, audio_url=f'https://example.org/{guid}.mp3', description='x' * 1000, **kw) self.a1 = ep(self.fa, 'a1', 'A1', _at(1, 1), duration_seconds=600) self.a2 = ep(self.fa, 'a2', 'A2', _at(1, 2)) self.a3 = ep(self.fa, 'a3', 'A3', _at(1, 3)) self.a4 = ep(self.fa, 'a4', 'A4 ohne Datum', None) self.b1 = ep(self.fb, 'b1', 'Bonus Talk', _at(1, 4)) self.b2 = ep(self.fb, 'b2', 'B2', _at(1, 5)) _progress(self.user, self.a1, _at(2, 1), played=True) _progress(self.user, self.a2, _at(2, 3), position_seconds=100) _progress(self.user, self.b1, _at(2, 2), position_seconds=50) _progress(self.user, self.a3, _at(2, 5), favorite=True) _progress(self.user, self.b2, _at(2, 4), dismissed=True) # e.g. back catalogue of a new subscription PodcastQueue.objects.create(user=self.user, episode=self.a2, position=0) # Someone else's listening must not show up. bob = User.objects.create_user(username='bob', password='pw12345678') fbob = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/a', title='Alpha') xb = PodcastEpisode.objects.create(feed=fbob, guid='a1', title='A1', audio_url='https://example.org/a1.mp3', pub_date=_at(1, 9)) _progress(bob, xb, _at(3, 1), played=True, favorite=True) _progress(bob, self.a3, _at(3, 1), played=True, position_seconds=10) # can't happen, but must not leak def get(self, **params): r = self.client.get('/podcasts/episodes/', params) self.assertEqual(r.status_code, 200) return r.json() def ids(self, **params): return [e['id'] for e in self.get(**params)['episodes']] def test_all_newest_first_without_date_last(self): data = self.get() self.assertEqual([e['id'] for e in data['episodes']], [e.id for e in (self.b2, self.b1, self.a3, self.a2, self.a1, self.a4)]) self.assertEqual((data['offset'], data['limit'], data['total']), (0, 100, 6)) self.assertEqual(self.ids(filter='all'), [e['id'] for e in data['episodes']]) def test_episode_fields(self): eps = {e['id']: e for e in self.get()['episodes']} self.assertEqual(eps[self.a2.id], { 'id': self.a2.id, 'title': 'A2', 'audio_url': 'https://example.org/a2.mp3', 'duration_seconds': 0, 'pub_date': '2026-01-02T09:00:00+00:00', 'artwork_url': '', 'feed__id': self.fa.id, 'feed__title': 'Alpha', 'feed__artwork_url': 'https://example.org/a.jpg', 'position_seconds': 100, 'played': False, 'favorite': False, 'in_queue': True, 'last_played_at': '2026-02-03T09:00:00+00:00', }) a4 = eps[self.a4.id] self.assertEqual((a4['pub_date'], a4['last_played_at'], a4['position_seconds'], a4['played'], a4['favorite'], a4['in_queue']), (None, None, 0, False, False, False)) self.assertEqual(eps[self.a3.id]['favorite'], True) self.assertEqual(eps[self.a3.id]['played'], False) # bob's row doesn't count self.assertEqual(eps[self.a1.id]['played'], True) def test_unplayed(self): # Dismissed episodes (b2) are still listed. self.assertEqual(self.ids(filter='unplayed'), [e.id for e in (self.b2, self.b1, self.a3, self.a2, self.a4)]) def test_in_progress_by_publication_date(self): self.assertEqual(self.ids(filter='in_progress'), [self.b1.id, self.a2.id]) def test_played(self): self.assertEqual(self.ids(filter='played'), [self.a1.id]) def test_favorites(self): self.assertEqual(self.ids(filter='favorites'), [self.a3.id]) def test_history_last_listened_first(self): data = self.get(filter='history') self.assertEqual([e['id'] for e in data['episodes']], [self.a2.id, self.b1.id, self.a1.id]) self.assertEqual([e['last_played_at'] for e in data['episodes']], ['2026-02-03T09:00:00+00:00', '2026-02-02T09:00:00+00:00', '2026-02-01T09:00:00+00:00']) self.assertEqual(data['total'], 3) # Listening again moves an episode to the top. self.client.post('/podcasts/progress/save/', json.dumps({'episode_id': self.a1.id, 'position_seconds': 20}), content_type='application/json') self.assertEqual(self.ids(filter='history'), [self.a1.id, self.a2.id, self.b1.id]) def test_feed_id_and_q(self): self.assertEqual(self.ids(feed_id=self.fb.id), [self.b2.id, self.b1.id]) self.assertEqual(self.ids(q='BONUS'), [self.b1.id]) self.assertEqual(self.ids(q='ohne dat'), [self.a4.id]) self.assertEqual(self.ids(filter='history', feed_id=self.fa.id), [self.a2.id, self.a1.id]) self.assertEqual(self.ids(filter='in_progress', q='a2'), [self.a2.id]) data = self.get(feed_id=self.fb.id, q='nothing') self.assertEqual((data['episodes'], data['total']), ([], 0)) bobs_feed = PodcastFeed.objects.get(user__username='bob') self.assertEqual(self.ids(feed_id=bobs_feed.id), []) def test_pages(self): data = self.get(offset=1, limit=2) self.assertEqual([e['id'] for e in data['episodes']], [self.b1.id, self.a3.id]) self.assertEqual((data['offset'], data['limit'], data['total']), (1, 2, 6)) data = self.get(offset=5, limit=2) self.assertEqual(([e['id'] for e in data['episodes']], data['total']), ([self.a4.id], 6)) data = self.get(offset=10) self.assertEqual((data['episodes'], data['total']), ([], 6)) self.assertEqual(self.get(limit=10000)['limit'], 500) self.assertEqual(self.get(offset=-3)['offset'], 0) data = self.get(filter='history', limit=1, offset=1) self.assertEqual(([e['id'] for e in data['episodes']], data['total']), ([self.b1.id], 3)) def test_query_count_does_not_grow_with_the_page(self): from .views import _episode_list_data for i in range(20): e = PodcastEpisode.objects.create(feed=self.fb, guid=f'n{i}', title=f'N{i}', audio_url=f'https://example.org/n{i}.mp3', pub_date=_at(1, 10)) _progress(self.user, e, _at(2, 10), position_seconds=i + 1) PodcastQueue.objects.create(user=self.user, episode=e, position=i + 1) with self.assertNumQueries(2): episodes, total = _episode_list_data(self.user, 'history') self.assertEqual((len(episodes), total), (23, 23)) def test_bad_parameters(self): self.assertEqual(self.client.get('/podcasts/episodes/', {'filter': 'new'}).status_code, 400) self.assertEqual(self.client.get('/podcasts/episodes/', {'limit': 'ten'}).status_code, 400) self.assertEqual(self.client.get('/podcasts/episodes/', {'feed_id': 'x'}).status_code, 400) def test_login_required(self): self.client.logout() self.assertEqual(self.client.get('/podcasts/episodes/').status_code, 401) class StatsTests(TestCase): def setUp(self): self.user = User.objects.create_user(username='alice', password='pw12345678') self.client.force_login(self.user) self.fa = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/a', title='Alpha', artwork_url='https://example.org/a.jpg') self.fb = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/b', title='Beta') self.fc = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/c', title='Gamma') def ep(feed, guid, duration): return PodcastEpisode.objects.create(feed=feed, guid=guid, title=guid, duration_seconds=duration, audio_url=f'https://example.org/{guid}.mp3') now = datetime.now(timezone.utc) _progress(self.user, ep(self.fa, 'a1', 3600), now, played=True, position_seconds=50) # 3600 _progress(self.user, ep(self.fa, 'a2', 1000), now, position_seconds=300) # 300, started _progress(self.user, ep(self.fa, 'a3', 0), now, played=True, position_seconds=120) # 120: no length _progress(self.user, ep(self.fb, 'b1', 900), now - timedelta(days=30), position_seconds=200) # 200, started _progress(self.user, ep(self.fc, 'c1', 500), now, dismissed=True, favorite=True) # 0 bob = User.objects.create_user(username='bob', password='pw12345678') fbob = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/a', title='Alpha') _progress(bob, ep(fbob, 'x', 999), now, played=True) def test_all_time(self): r = self.client.get('/podcasts/stats/') self.assertEqual(r.status_code, 200) self.assertEqual(r.json(), { 'total_seconds': 4220, 'episodes_played': 2, 'episodes_started': 2, 'since': None, 'feeds': [ {'feed_id': self.fa.id, 'title': 'Alpha', 'artwork_url': 'https://example.org/a.jpg', 'seconds': 4020, 'episodes_played': 2}, {'feed_id': self.fb.id, 'title': 'Beta', 'artwork_url': '', 'seconds': 200, 'episodes_played': 0}, ], }) def test_last_days(self): before = datetime.now(timezone.utc) data = self.client.get('/podcasts/stats/', {'days': 7}).json() self.assertEqual((data['total_seconds'], data['episodes_played'], data['episodes_started']), (4020, 2, 1)) self.assertEqual([f['feed_id'] for f in data['feeds']], [self.fa.id]) since = datetime.fromisoformat(data['since']) self.assertAlmostEqual((before - since).total_seconds(), 7 * 86400, delta=60) data = self.client.get('/podcasts/stats/', {'days': 60}).json() self.assertEqual(data['total_seconds'], 4220) self.assertEqual([f['feed_id'] for f in data['feeds']], [self.fa.id, self.fb.id]) def test_empty(self): EpisodeProgress.objects.filter(user=self.user).delete() self.assertEqual(self.client.get('/podcasts/stats/').json(), { 'total_seconds': 0, 'episodes_played': 0, 'episodes_started': 0, 'since': None, 'feeds': [], }) def test_bad_days_and_login(self): self.assertEqual(self.client.get('/podcasts/stats/', {'days': 'week'}).status_code, 400) self.assertEqual(self.client.get('/podcasts/stats/', {'days': -1}).status_code, 400) self.client.logout() self.assertEqual(self.client.get('/podcasts/stats/').status_code, 401) class OpmlExportTests(TestCase): def setUp(self): self.user = User.objects.create_user(username='alice', password='pw12345678') self.client.force_login(self.user) PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/feed?a=1&b=2', title='Tom & Jerry "Live" ', link='https://example.org/?x=1&y=2') PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/plain', title="Bob's Podcast") bob = User.objects.create_user(username='bob', password='pw12345678') PodcastFeed.objects.create(user=bob, rss_url='https://example.org/secret', title='Secret') def test_export(self): import xml.etree.ElementTree as ET r = self.client.get('/podcasts/feeds/export.opml') self.assertEqual(r.status_code, 200) self.assertEqual(r['Content-Type'], 'text/x-opml; charset=utf-8') self.assertEqual(r['Content-Disposition'], 'attachment; filename="diora-podcasts.opml"') raw = r.content.decode('utf-8') self.assertIn('Tom & Jerry "Live" <Folge>', raw) self.assertIn('xmlUrl="https://example.org/feed?a=1&b=2"', raw) self.assertNotIn('Secret', raw) root = ET.fromstring(r.content) self.assertEqual(root.tag, 'opml') self.assertEqual(root.get('version'), '2.0') self.assertEqual(root.findtext('head/title'), 'diora Podcasts') outlines = [o.attrib for o in root.iter('outline')] self.assertEqual(outlines, [ {'type': 'rss', 'text': "Bob's Podcast", 'title': "Bob's Podcast", 'xmlUrl': 'https://example.org/plain', 'htmlUrl': ''}, {'type': 'rss', 'text': 'Tom & Jerry "Live" ', 'title': 'Tom & Jerry "Live" ', 'xmlUrl': 'https://example.org/feed?a=1&b=2', 'htmlUrl': 'https://example.org/?x=1&y=2'}, ]) def test_round_trip_through_import(self): from django.core.files.uploadedfile import SimpleUploadedFile opml = self.client.get('/podcasts/feeds/export.opml').content carol = User.objects.create_user(username='carol', password='pw12345678') self.client.force_login(carol) r = self.client.post('/podcasts/feeds/import/', {'file': SimpleUploadedFile('diora-podcasts.opml', opml, 'text/x-opml')}) self.assertEqual(r.json(), {'ok': True, 'added': 2, 'skipped': 0}) mine = lambda u: set(PodcastFeed.objects.filter(user=u).values_list('title', 'rss_url')) self.assertEqual(mine(carol), mine(self.user)) # Importing the same file again adds nothing. r = self.client.post('/podcasts/feeds/import/', {'file': SimpleUploadedFile('diora-podcasts.opml', opml, 'text/x-opml')}) self.assertEqual(r.json(), {'ok': True, 'added': 0, 'skipped': 2}) def test_no_feeds(self): import xml.etree.ElementTree as ET PodcastFeed.objects.filter(user=self.user).delete() root = ET.fromstring(self.client.get('/podcasts/feeds/export.opml').content) self.assertEqual(list(root.iter('outline')), []) def test_login_required(self): self.client.logout() self.assertEqual(self.client.get('/podcasts/feeds/export.opml').status_code, 401) # --------------------------------------------------------------------------- # Per-feed settings, new-episode rules, keep the queue sorted # --------------------------------------------------------------------------- FEED_DEFAULTS = { 'playback_speed': None, 'skip_intro': 0, 'skip_outro': 0, 'volume': 1.0, 'new_episodes': 'inbox', 'include_words': '', 'exclude_words': '', 'min_duration': 0, 'auto_queue': False, } class FeedSettingsTests(TestCase): def setUp(self): self.user = User.objects.create_user(username='alice', password='pw12345678') self.client.force_login(self.user) self.feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/feed', title='Feed') def post(self, body, feed=None): return self.client.post(f'/podcasts/feeds/{(feed or self.feed).id}/settings/', json.dumps(body), content_type='application/json') def test_defaults(self): r = self.post({}) self.assertEqual(r.status_code, 200) self.assertEqual(r.json(), {'feed_id': self.feed.id, **FEED_DEFAULTS}) def test_change_some(self): r = self.post({'playback_speed': 1.5, 'skip_intro': 30, 'skip_outro': 600, 'volume': 2, 'include_words': ' News, Talk ', 'exclude_words': 'Teaser', 'min_duration': 10, 'unknown': 'ignored'}) self.assertEqual(r.json(), { 'feed_id': self.feed.id, 'playback_speed': 1.5, 'skip_intro': 30, 'skip_outro': 600, 'volume': 2.0, 'new_episodes': 'inbox', 'include_words': 'News, Talk', 'exclude_words': 'Teaser', 'min_duration': 10, 'auto_queue': False, }) self.feed.refresh_from_db() self.assertEqual((self.feed.playback_speed, self.feed.skip_intro, self.feed.volume, self.feed.min_duration), (1.5, 30, 2.0, 10)) # Only what is sent changes; null clears the speed. r = self.post({'playback_speed': None, 'include_words': ''}) self.assertEqual((r.json()['playback_speed'], r.json()['include_words'], r.json()['skip_intro']), (None, '', 30)) self.feed.refresh_from_db() self.assertIsNone(self.feed.playback_speed) def test_every_allowed_volume_and_the_speed_limits(self): for v in (0.2, 0.5, 1.0, 1.5, 2.0, 2.5): self.assertEqual(self.post({'volume': v}).json()['volume'], v) for s in (0.5, 3, 3.0): self.assertEqual(self.post({'playback_speed': s}).json()['playback_speed'], s) def test_invalid_values_name_the_field_and_change_nothing(self): bad = [ ('playback_speed', 0.4), ('playback_speed', 3.5), ('playback_speed', 'fast'), ('playback_speed', True), ('skip_intro', -1), ('skip_intro', 601), ('skip_intro', 1.5), ('skip_intro', '10'), ('skip_intro', None), ('skip_outro', 601), ('volume', 0.7), ('volume', '1.0'), ('volume', None), ('new_episodes', 'later'), ('new_episodes', None), ('include_words', 'x' * 501), ('include_words', 5), ('exclude_words', ['a']), ('min_duration', -1), ('min_duration', 601), ] for name, value in bad: with self.subTest(name=name, value=value): r = self.post({'skip_intro': 10, name: value} if name != 'skip_intro' else {name: value}) self.assertEqual(r.status_code, 400) self.assertIn(name, r.json()['error']) self.assertEqual(r.json()['field'], name) self.feed.refresh_from_db() self.assertEqual({k: getattr(self.feed, k) for k in FEED_DEFAULTS}, FEED_DEFAULTS) self.assertEqual(self.client.post(f'/podcasts/feeds/{self.feed.id}/settings/', '[1]', content_type='application/json').status_code, 400) def test_owner_only(self): bob = User.objects.create_user(username='bob', password='pw12345678') bobs = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/feed', title='Feed') self.assertEqual(self.post({'skip_intro': 10}, bobs).status_code, 404) bobs.refresh_from_db() self.assertEqual(bobs.skip_intro, 0) self.assertEqual(self.client.post('/podcasts/feeds/999999/settings/', '{}', content_type='application/json').status_code, 404) self.assertEqual(self.client.get(f'/podcasts/feeds/{self.feed.id}/settings/').status_code, 405) self.client.logout() self.assertEqual(self.post({'skip_intro': 10}).status_code, 401) def test_new_episodes_and_auto_queue_stay_in_step(self): self.assertEqual(self.post({'new_episodes': 'queue'}).json()['auto_queue'], True) self.feed.refresh_from_db() self.assertTrue(self.feed.auto_queue) self.assertEqual(self.post({'new_episodes': 'ignore'}).json()['auto_queue'], False) self.assertEqual(self.post({'new_episodes': 'inbox'}).json()['auto_queue'], False) def toggle(value=None): body = {} if value is None else {'auto_queue': value} r = self.client.post(f'/podcasts/feeds/{self.feed.id}/set-auto-queue/', json.dumps(body), content_type='application/json') self.feed.refresh_from_db() return r.json()['auto_queue'], self.feed.auto_queue, self.feed.new_episodes self.assertEqual(toggle(), (True, True, 'queue')) self.assertEqual(toggle(), (False, False, 'inbox')) self.post({'new_episodes': 'ignore'}) self.assertEqual(toggle(False), (False, False, 'ignore')) self.assertEqual(toggle(True), (True, True, 'queue')) # Also when a feed is created or saved in code. f = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/2', title='2', auto_queue=True) self.assertEqual(PodcastFeed.objects.get(pk=f.pk).new_episodes, 'queue') f.new_episodes = 'inbox' f.save() self.assertFalse(PodcastFeed.objects.get(pk=f.pk).auto_queue) def test_feeds_and_state_carry_the_settings(self): self.post({'playback_speed': 1.25, 'skip_outro': 20, 'volume': 0.5, 'new_episodes': 'queue', 'exclude_words': 'Trailer', 'min_duration': 5}) PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/2', title='Zweiter') expected = { self.feed.id: {'feed_id': self.feed.id, **FEED_DEFAULTS, 'playback_speed': 1.25, 'skip_outro': 20, 'volume': 0.5, 'new_episodes': 'queue', 'exclude_words': 'Trailer', 'min_duration': 5, 'auto_queue': True}, } for url in ('/podcasts/feeds/', '/podcasts/state/'): with self.subTest(url=url): feeds = {f['id']: f for f in self.client.get(url).json()['feeds']} self.assertEqual(len(feeds), 2) mine = feeds[self.feed.id] self.assertEqual({k: mine[k] for k in expected[self.feed.id]}, expected[self.feed.id]) other = next(f for f in feeds.values() if f['id'] != self.feed.id) self.assertEqual({k: other[k] for k in FEED_DEFAULTS}, FEED_DEFAULTS) self.assertEqual(other['feed_id'], other['id']) def test_migration_maps_auto_queue_to_new_episodes(self): import importlib from django.apps import apps migration = importlib.import_module('podcasts.migrations.0007_feed_settings') on = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/on', title='On') off = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/off', title='Off') # As the rows look right after the new columns are added. PodcastFeed.objects.filter(pk=on.pk).update(auto_queue=True, new_episodes='inbox') PodcastFeed.objects.filter(pk=off.pk).update(auto_queue=False, new_episodes='inbox') migration.auto_queue_to_new_episodes(apps, None) self.assertEqual(PodcastFeed.objects.get(pk=on.pk).new_episodes, 'queue') self.assertEqual(PodcastFeed.objects.get(pk=off.pk).new_episodes, 'inbox') self.assertEqual([op.__class__.__name__ for op in migration.Migration.operations][-1], 'RunPython') self.assertEqual(migration.Migration.dependencies, [('podcasts', '0006_episodeprogress_favorite')]) class NewEpisodeRuleTests(TestCase): """What happens to a feed's new episodes on a refresh: its filter, then inbox / queue / ignore.""" def setUp(self): self.user = User.objects.create_user(username='alice', password='pw12345678') self.client.force_login(self.user) self.feed = PodcastFeed.objects.create( user=self.user, rss_url='https://example.org/feed', title='Feed', last_refreshed_at=datetime(2026, 1, 1, tzinfo=timezone.utc), # not the first fetch ) other = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/o', title='O') self.old = PodcastEpisode.objects.create(feed=other, guid='o0', title='O0', audio_url='https://example.org/o0.mp3') PodcastQueue.objects.create(user=self.user, episode=self.old, position=0) dur = '{}' self.rss = _rss([ ('Talk-Folge', 'Mon, 05 Jan 2026 10:00:00 GMT', dur.format('15:00')), ('News-Teaser', 'Tue, 06 Jan 2026 10:00:00 GMT', dur.format('2:00')), ('Kurzer-Talk', 'Wed, 07 Jan 2026 10:00:00 GMT', dur.format('300')), ('Ohne-Laenge', 'Thu, 08 Jan 2026 10:00:00 GMT'), ]) def configure(self, **settings): PodcastFeed.objects.filter(pk=self.feed.pk).update(**settings) self.feed.refresh_from_db() def refresh(self): with patch('podcasts.views._fetch_feed', lambda url: _parse(self.rss)): return _refresh_feed(self.feed) def inbox(self): return [e['title'] for e in self.client.get('/podcasts/inbox/').json()['episodes'] if e['feed__id'] == self.feed.id] def queue(self): return [q.episode.guid for q in PodcastQueue.objects.filter(user=self.user).order_by('position')] def dismissed(self): return set(EpisodeProgress.objects.filter(user=self.user, dismissed=True, played=False) .values_list('episode__guid', flat=True)) def test_inbox_is_the_default(self): self.assertEqual(self.refresh(), 4) self.assertEqual(self.inbox(), ['Ohne-Laenge', 'Kurzer-Talk', 'News-Teaser', 'Talk-Folge']) self.assertEqual(self.queue(), ['o0']) self.assertEqual(self.dismissed(), set()) def test_queue(self): self.configure(new_episodes='queue', auto_queue=True) self.refresh() self.assertEqual(self.queue(), ['Ohne-Laenge', 'Kurzer-Talk', 'News-Teaser', 'Talk-Folge', 'o0']) self.assertEqual(self.inbox(), []) def test_queue_at_the_auto_queue_position(self): self.configure(new_episodes='queue', auto_queue=True) PodcastSettings.objects.create(user=self.user, auto_queue_position='back') self.refresh() self.assertEqual(self.queue(), ['o0', 'Talk-Folge', 'News-Teaser', 'Kurzer-Talk', 'Ohne-Laenge']) def test_ignore(self): self.configure(new_episodes='ignore') self.assertEqual(self.refresh(), 4) self.assertEqual(self.inbox(), []) self.assertEqual(self.queue(), ['o0']) self.assertEqual(self.dismissed(), {'Talk-Folge', 'News-Teaser', 'Kurzer-Talk', 'Ohne-Laenge'}) # Still listed for the podcast, not played. eps = self.client.get(f'/podcasts/feeds/{self.feed.id}/episodes/').json()['episodes'] self.assertEqual(len(eps), 4) self.assertFalse(any(e['played'] for e in eps)) def test_include_words(self): self.configure(include_words='talk, nothing ,') self.refresh() self.assertEqual(self.inbox(), ['Kurzer-Talk', 'Talk-Folge']) self.assertEqual(self.dismissed(), {'News-Teaser', 'Ohne-Laenge'}) def test_exclude_words(self): self.configure(exclude_words='TEASER,laenge') self.refresh() self.assertEqual(self.inbox(), ['Kurzer-Talk', 'Talk-Folge']) self.assertEqual(self.dismissed(), {'News-Teaser', 'Ohne-Laenge'}) def test_exclude_beats_include(self): self.configure(include_words='talk', exclude_words='kurz') self.refresh() self.assertEqual(self.inbox(), ['Talk-Folge']) def test_min_duration_keeps_unknown_lengths(self): self.configure(min_duration=5) # 300 s is just long enough, 120 s is not self.refresh() self.assertEqual(self.inbox(), ['Ohne-Laenge', 'Kurzer-Talk', 'Talk-Folge']) self.assertEqual(self.dismissed(), {'News-Teaser'}) PodcastEpisode.objects.filter(feed=self.feed).delete() self.configure(min_duration=6) self.refresh() self.assertEqual(self.inbox(), ['Ohne-Laenge', 'Talk-Folge']) def test_filtered_episodes_are_not_queued(self): self.configure(new_episodes='queue', auto_queue=True, exclude_words='teaser', min_duration=10) self.refresh() self.assertEqual(self.queue(), ['Ohne-Laenge', 'Talk-Folge', 'o0']) self.assertEqual(self.inbox(), []) self.assertFalse(EpisodeProgress.objects.filter(played=True).exists()) def test_only_new_episodes_are_affected(self): self.refresh() # all four in the inbox self.configure(new_episodes='ignore', exclude_words='talk') self.rss = _rss([('Talk-Neu', 'Fri, 09 Jan 2026 10:00:00 GMT')] + [ ('Talk-Folge', 'Mon, 05 Jan 2026 10:00:00 GMT')]) self.assertEqual(self.refresh(), 1) self.assertEqual(self.inbox(), ['Ohne-Laenge', 'Kurzer-Talk', 'News-Teaser', 'Talk-Folge']) def test_first_fetch_keeps_only_the_newest(self): # E.g. imported by OPML and set to queue before its first refresh. self.configure(last_refreshed_at=None, new_episodes='queue', auto_queue=True) self.refresh() self.assertEqual(self.queue(), ['Ohne-Laenge', 'o0']) self.assertEqual(self.inbox(), []) self.assertEqual(self.dismissed(), {'Talk-Folge', 'News-Teaser', 'Kurzer-Talk', 'Ohne-Laenge'}) class KeepSortedTests(TestCase): def setUp(self): self.user = User.objects.create_user(username='alice', password='pw12345678') self.client.force_login(self.user) beta = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/b', title='Beta') alpha = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/a', title='alpha') def ep(feed, guid, day, duration): return PodcastEpisode.objects.create(feed=feed, guid=guid, title=guid, duration_seconds=duration, audio_url=f'https://example.org/{guid}.mp3', pub_date=_at(1, day)) # remaining: e0 3000, e1 1000, e2 500 (2000 - 1500 heard), e3 600 self.e0 = ep(beta, 'e0', 1, 3000) self.e1 = ep(alpha, 'e1', 3, 1000) self.e2 = ep(beta, 'e2', 2, 2000) self.e3 = ep(alpha, 'e3', 4, 600) EpisodeProgress.objects.create(user=self.user, episode=self.e2, position_seconds=1500) for i, e in enumerate((self.e0, self.e1, self.e2)): PodcastQueue.objects.create(user=self.user, episode=e, position=i) def post(self, url, body): return self.client.post(url, json.dumps(body), content_type='application/json') def queue(self): return [q['episode__title'] for q in self.client.get('/podcasts/queue/').json()['queue']] def keep(self, rule): return self.post('/podcasts/settings/', {'keep_sorted': rule}) def test_setting_sorts_right_away(self): expected = { 'new': ['e1', 'e2', 'e0'], 'old': ['e0', 'e2', 'e1'], 'short': ['e2', 'e1', 'e0'], 'long': ['e0', 'e1', 'e2'], 'feed': ['e1', 'e2', 'e0'], } for rule, order in expected.items(): with self.subTest(rule=rule): r = self.keep(rule) self.assertEqual(r.status_code, 200) self.assertEqual(r.json()['keep_sorted'], rule) self.assertEqual(self.queue(), order) self.assertEqual(self.client.get('/podcasts/settings/').json()['keep_sorted'], 'feed') self.assertEqual(self.client.get('/podcasts/state/').json()['settings']['keep_sorted'], 'feed') def test_default_off_and_invalid(self): self.assertEqual(self.client.get('/podcasts/settings/').json()['keep_sorted'], '') self.assertEqual(self.keep('random').status_code, 400) self.assertEqual(self.keep(None).status_code, 400) self.assertEqual(PodcastSettings.for_user(self.user).keep_sorted, '') self.assertEqual(self.queue(), ['e0', 'e1', 'e2']) def test_off_keeps_the_users_order(self): self.keep('new') self.keep('') self.post('/podcasts/queue/reorder/', {'order': [self.e0.id, self.e2.id, self.e1.id]}) self.assertEqual(self.queue(), ['e0', 'e2', 'e1']) self.post('/podcasts/queue/add/', {'episode_ids': [self.e3.id], 'where': 'front'}) self.assertEqual(self.queue(), ['e3', 'e0', 'e2', 'e1']) def test_enqueue_is_sorted(self): self.keep('old') self.post('/podcasts/queue/add/', {'episode_ids': [self.e3.id], 'where': 'front'}) self.assertEqual(self.queue(), ['e0', 'e2', 'e1', 'e3']) PodcastQueue.objects.filter(episode=self.e3).delete() self.keep('short') self.post('/podcasts/queue/add/', {'episode_ids': [self.e3.id]}) self.assertEqual(self.queue(), ['e2', 'e3', 'e1', 'e0']) def test_feed_rule_newest_first_within_a_podcast(self): self.keep('feed') self.post('/podcasts/queue/add/', {'episode_ids': [self.e3.id]}) self.assertEqual(self.queue(), ['e3', 'e1', 'e2', 'e0']) def test_reorder_is_undone(self): self.keep('long') self.post('/podcasts/queue/reorder/', {'order': [self.e2.id, self.e1.id, self.e0.id]}) self.assertEqual(self.queue(), ['e0', 'e1', 'e2']) # Listening changes what is left: e0 now has the least left. EpisodeProgress.objects.create(user=self.user, episode=self.e0, position_seconds=2900) self.post('/podcasts/queue/reorder/', {'order': [self.e0.id, self.e1.id, self.e2.id]}) self.assertEqual(self.queue(), ['e1', 'e2', 'e0']) def test_auto_queued_episodes_are_sorted_too(self): self.keep('old') feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/n', title='N', new_episodes='queue', last_refreshed_at=_at(1, 1)) with patch('podcasts.views._fetch_feed', lambda url: _parse(_rss([('n1', 'Fri, 02 Jan 2026 12:00:00 GMT')]))): _refresh_feed(feed) self.assertEqual(self.queue(), ['e0', 'e2', 'n1', 'e1'])