mrwnslz/diora-web
1
0
Fork 0
Code Issues 4 Pull requests Projects Releases Packages Wiki Activity Actions
diora-web/podcasts/tests.py
mrwnslz a937ed8b3e Folgen haben einen Link: PodcastEpisode.link aus dem <link> des Feed-Eintrags (neu und beim Abruf nachgetragen), in episode_detail und allen Folgenlisten 
Grundlage fürs Teilen einer Folge; Migration 0027.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-09 09:32:13 +01:00

2797 lines
148 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import base64
import json
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
import feedparser
from django.contrib.auth.models import User
from django.core.cache import cache
from django.test import TestCase
from .models import EpisodeProgress, PodcastEpisode, PodcastFeed, PodcastQueue, PodcastSettings
from .views import FeedFetch, _refresh_feed
_parse = feedparser.parse
# The account's playback settings as a new user has them.
PLAYBACK_DEFAULTS = {
'skip_back': None, 'skip_forward': None, 'smart_played': None, 'speed_presets': None,
'keep_skipped': True, 'delete_dequeue': True, 'mobile_downloads': True, 'mobile_streaming': 'allow',
'times_by_speed': False,
}
def _ok(rss, **kw):
"""What _fetch_feed returns for a 200 with this RSS."""
return FeedFetch(200, _parse(rss), **kw)
def _rss(items):
"""Items are (guid, date) or (guid, date, extra XML inside the <item>);
the guid is the title too."""
entries = ''.join(
f'<item><guid>{guid}</guid><title>{guid}</title><pubDate>{date}</pubDate>'
f'<enclosure url="https://example.org/{guid}.mp3" type="audio/mpeg" length="1"/>{extra[0] if extra else ""}'
f'</item>'
for guid, date, *extra in items
)
return ('<?xml version="1.0"?><rss version="2.0" xmlns:itunes="http://www.itunes.com/dtds/podcast-1.0.dtd">'
f'<channel><title>Feed</title>{entries}</channel></rss>')
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, **kw: _ok(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, **kw: _ok(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, **kw: _ok(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 VideoEnclosureTests(TestCase):
"""Video podcasts are taken in (the players play their sound only)."""
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/f', title='Feed')
@staticmethod
def rss(*items):
entries = ''.join(
f'<item><guid>{guid}</guid><title>{guid}</title><pubDate>{date}</pubDate>{enclosures}</item>'
for guid, date, enclosures in items
)
return f'<?xml version="1.0"?><rss version="2.0"><channel><title>Feed</title>{entries}</channel></rss>'
def refresh(self, rss, **kw):
with patch('podcasts.views._fetch_feed', lambda url, **_: _ok(rss)):
return _refresh_feed(self.feed, **kw)
def test_video_only_and_audio_preferred(self):
self.refresh(self.rss(
('v', 'Mon, 05 Jan 2026 10:00:00 GMT', '<enclosure url="https://example.org/v.mp4" type="video/mp4" length="1"/>'),
('both', 'Tue, 06 Jan 2026 10:00:00 GMT',
'<enclosure url="https://example.org/b.mp4" type="video/mp4" length="1"/>'
'<enclosure url="https://example.org/b.m4a" type="audio/mp4" length="1"/>'),
('untyped', 'Wed, 07 Jan 2026 10:00:00 GMT', '<enclosure url="https://example.org/u.m4v?x=1" length="1"/>'),
('none', 'Thu, 08 Jan 2026 10:00:00 GMT', '<enclosure url="https://example.org/n.pdf" type="application/pdf"/>'),
))
urls = dict(PodcastEpisode.objects.filter(feed=self.feed).values_list('guid', 'audio_url'))
self.assertEqual(urls, {
'v': 'https://example.org/v.mp4',
'both': 'https://example.org/b.m4a',
'untyped': 'https://example.org/u.m4v?x=1',
})
def test_full_reread_keeps_old_found_episodes_out_of_the_inbox(self):
audio = '<enclosure url="https://example.org/a.mp3" type="audio/mpeg" length="1"/>'
video = '<enclosure url="https://example.org/{}.mp4" type="video/mp4" length="1"/>'
self.refresh(self.rss(('a', 'Mon, 05 Jan 2026 10:00:00 GMT', audio)))
self.feed.refresh_from_db()
later = (self.feed.last_refreshed_at + timedelta(days=1)).strftime('%a, %d %b %Y %H:%M:%S GMT')
self.refresh(self.rss(
('a', 'Mon, 05 Jan 2026 10:00:00 GMT', audio),
('old-video', 'Tue, 06 Jan 2026 10:00:00 GMT', video.format('o')),
('new-video', later, video.format('n')),
), full=True)
inbox = [e['title'] for e in self.client.get('/podcasts/inbox/').json()['episodes']]
self.assertIn('new-video', inbox)
self.assertNotIn('old-video', inbox)
old = PodcastEpisode.objects.get(feed=self.feed, guid='old-video')
progress = EpisodeProgress.objects.get(user=self.user, episode=old)
self.assertTrue(progress.dismissed)
self.assertFalse(progress.played)
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, **kw):
self.fetched.append(url)
if url.endswith('/b'):
raise ValueError('broken feed')
return _ok(_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, 'done': True, 'left': 0})
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_refresh_stops_at_its_deadline(self):
import time as _time
from .views import _refresh_feeds
def slow(url, **kw):
_time.sleep(.5)
return self.fake_fetch(url, **kw)
with patch('podcasts.views._fetch_feed', slow):
refreshed, failed, new = _refresh_feeds([self.a, self.b], deadline=_time.monotonic() + .05)
self.assertEqual((refreshed, failed), (0, 0))
self.a.refresh_from_db()
self.assertIsNone(self.a.last_refreshed_at) # not done: left as it was
def test_the_body_fingerprint_is_kept_and_sent_again(self):
from .views import _validators
with patch('podcasts.views._fetch_feed', lambda url, **kw: FeedFetch(200, _parse(_rss([('x', 'Mon, 05 Jan 2026 10:00:00 GMT')])), body_hash='abc')):
self.client.post('/podcasts/feeds/refresh-all/', '{"max_age":0}', content_type='application/json')
self.a.refresh_from_db()
self.assertEqual(self.a.http_body_hash, 'abc')
self.assertEqual(_validators([self.a]), {'body_hash': 'abc'})
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': '', 'auto_download_default': False,
**PLAYBACK_DEFAULTS})
def test_state_counts_the_whole_inbox(self):
# The list is capped (inbox_limit); inbox_total counts every new episode of this user.
eps = [PodcastEpisode.objects.create(feed=self.a, guid=f'n{i}', title=f'N{i}', audio_url=f'https://example.org/{i}.mp3',
pub_date=datetime.now(timezone.utc) - timedelta(days=i)) for i in range(5)]
EpisodeProgress.objects.create(user=self.user, episode=eps[0], played=True)
EpisodeProgress.objects.create(user=self.user, episode=eps[1], dismissed=True)
EpisodeProgress.objects.create(user=self.user, episode=eps[2], position_seconds=30)
PodcastEpisode.objects.create(feed=self.bobs, guid='b', title='Bobs', audio_url='https://example.org/b.mp3')
data = self.client.get('/podcasts/state/?inbox_limit=2').json()
self.assertEqual([e['id'] for e in data['inbox']], [eps[2].id, eps[3].id])
self.assertEqual((data['inbox_total'], data['inbox_limit']), (3, 2))
self.assertEqual(self.client.get('/podcasts/state/').json()['inbox_total'], 3)
# ---------------------------------------------------------------------------
# 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 <item>)."""
entries = ''.join(
f'<item><guid>{guid}</guid><title>{guid}</title><pubDate>Mon, 05 Jan 2026 10:00:00 GMT</pubDate>'
f'<enclosure url="https://example.org/{guid}.mp3" type="audio/mpeg" length="1"/>{extra}</item>'
for guid, extra in items
)
return f'<?xml version="1.0"?><rss version="2.0" {ns}><channel><title>Feed</title>{entries}</channel></rss>'
PSC = (
'<psc:chapters version="1.2">'
'<psc:chapter start="00:00:00" title="Intro"/>'
'<psc:chapter start="00:01:02.500" title="Teil &amp; zwei" href="https://x.org/2" image="https://x.org/2.jpg"/>'
'<psc:chapter start="1:02:03" title="Drei"/>'
'<psc:chapter start="12:05" title="Kurz"/>'
'</psc:chapters>'
)
TRANSCRIPTS = (
'<podcast:transcript url="https://example.org/a.vtt" type="text/vtt" language="de"/>'
'<podcast:transcript url="https://example.org/a.srt" type="application/srt" rel="captions"/>'
'<podcast:transcript url="javascript:alert(1)" type="text/html"/>'
)
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, **kw: _ok(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 + '<podcast:chapters url="https://x.org/c.json" '
'type="application/json+chapters"/>')])
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', '<pc:transcript url="https://example.org/a.vtt" type="text/vtt"/>'
'<pc:chapters url="https://x.org/c.json" type="application/json+chapters"/>')], 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 + '<podcast:chapters url="https://x.org/c.json" '
'type="application/json+chapters"/>')]))
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', '<description>Kurz</description>'
'<content:encoded><![CDATA[<p>Lang <a href="https://x.org">Link</a></p>]]></content:encoded>'),
('b', '<description>Nur kurz</description>'),
]))
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', '<description>Alt</description>'), ('b', '<description>Gleich</description>')]))
a, b = PodcastEpisode.objects.get(guid='a'), PodcastEpisode.objects.get(guid='b')
rss = _rich_rss([
('a', '<description>Neu</description>' + PSC + TRANSCRIPTS),
('b', '<description>Gleich</description>'),
])
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 = '<podcast:chapters url="https://x.org/{}.json" type="application/json+chapters"/>'
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 = (
'<p onclick="steal()">Hallo <a href="https://x.org/a?b=1&amp;c=2" onmouseover="x()" target="_top">Link</a>'
'<script>alert(1)</script><style>p{}</style><iframe src="https://evil.org">drin</iframe>'
'<a href="javascript:alert(1)">böse</a><a href=" JaVa\tScript:alert(1)">auch</a>'
'<img src="data:image/png;base64,AAA"><img src="https://x.org/i.png" onerror="x()">'
'<unknown>Text bleibt</unknown><b>offen'
)
self.assertEqual(
sanitize_html(dirty),
'<p>Hallo <a href="https://x.org/a?b=1&amp;c=2" rel="nofollow noopener noreferrer">Link</a>'
'<a rel="nofollow noopener noreferrer">böse</a><a rel="nofollow noopener noreferrer">auch</a>'
'<img src="https://x.org/i.png">Text bleibt<b>offen</b></p>',
)
self.assertEqual(sanitize_html('1 &lt; 2 &amp;&amp; <i>3</i>'), '1 &lt; 2 &amp;&amp; <i>3</i>')
self.assertEqual(sanitize_html('<a href="mailto:a@b.org">Mail</a>'),
'<a href="mailto:a@b.org" rel="nofollow noopener noreferrer">Mail</a>')
def test_timestamps(self):
from .shownotes import extract_timestamps
notes = (
'<p>Kapitel:<br>00:00 Intro<br>(12:34) - Thema eins<br>1:02:03 Zum Schluss</p>'
'<ul><li>[05:07] Liste</li><li>Um 20:15 geht es los</li><li>99:99 kaputt</li></ul>'
)
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='<p>Hallo<script>alert(1)</script><br>01:02 Thema <a href="https://x.org">x</a></p>',
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'], '<p>Hallo<br>01:02 Thema <a href="https://x.org" '
'rel="nofollow noopener noreferrer">x</a></p>')
self.assertEqual(data['timestamps'], [{'seconds': 62, 'label': 'Thema x'}])
self.assertEqual(data['transcripts'], self.ep.transcripts)
self.assertIs(data['has_transcript'], True)
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.audio_url = 'http://127.0.0.1/a.mp3' # the ID3 fallback is refused too
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):
cache.clear() # how often feeds come (_intervals) is kept per user id
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_folder_shows_its_feeds_episodes(self):
self.fa.folders = ['Politik']
self.fa.save()
self.assertEqual(sorted(self.ids(folder='Politik')), sorted([self.a1.id, self.a2.id, self.a3.id, self.a4.id]))
self.assertEqual(sorted(self.ids(folder='')), sorted([self.b1.id, self.b2.id])) # the ones in none
self.assertEqual(self.ids(folder='Sport'), [])
def test_state_says_how_often_a_feed_comes(self):
feeds = {f['id']: f for f in self.client.get('/podcasts/state/').json()['feeds']}
# A1-A3 a day apart (A4 has no date); Beta's two a day apart too.
self.assertAlmostEqual(feeds[self.fa.id]['interval_days'], (_at(1, 3) - _at(1, 1)).total_seconds() / 86400 / 2, places=1)
self.assertAlmostEqual(feeds[self.fb.id]['interval_days'], (_at(1, 5) - _at(1, 4)).total_seconds() / 86400, places=1)
def test_a_whole_archive_at_once_counts_as_seldom(self):
f = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/c', title='Archiv')
for k in range(5):
PodcastEpisode.objects.create(feed=f, guid=f'c{k}', title=f'C{k}', audio_url=f'https://example.org/c{k}.mp3',
pub_date=_at(1, 1) + timedelta(minutes=k))
feeds = {x['id']: x for x in self.client.get('/podcasts/state/').json()['feeds']}
self.assertGreater(feeds[f.id]['interval_days'], 30)
hourly = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/h', title='Stündlich')
for k in range(10):
PodcastEpisode.objects.create(feed=hourly, guid=f'h{k}', title=f'H{k}', audio_url=f'https://example.org/h{k}.mp3',
pub_date=_at(1, 1) + timedelta(hours=k))
cache.clear()
feeds = {x['id']: x for x in self.client.get('/podcasts/state/').json()['feeds']}
self.assertAlmostEqual(feeds[hourly.id]['interval_days'], 1 / 24, places=2)
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']}
a2 = dict(eps[self.a2.id])
self.assertTrue(a2.pop('discovered_at')) # when diora found it (the Strom's mark)
self.assertEqual(a2, {
'id': self.a2.id, 'title': 'A2', 'audio_url': 'https://example.org/a2.mp3', 'link': '', '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, 'in_later': False,
'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/', {'filter': 'unplayed,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)
self.assertEqual(self.client.get('/podcasts/episodes/', {'ids': '1,x'}).status_code, 400)
self.assertEqual(self.client.get('/podcasts/episodes/', {'sort': 'random'}).status_code, 400)
def test_queued(self):
self.assertEqual(self.ids(filter='queued'), [self.a2.id])
def test_filters_combine(self):
# Several filters, joined by commas, all have to match.
self.assertEqual(self.ids(filter='unplayed,queued'), [self.a2.id])
self.assertEqual(self.ids(filter='in_progress,queued'), [self.a2.id])
self.assertEqual(self.ids(filter='unplayed,favorites'), [self.a3.id])
self.assertEqual(self.ids(filter='played,favorites'), [])
self.assertEqual(self.ids(filter='history,in_progress'), [self.a2.id, self.b1.id])
self.assertEqual(self.ids(filter='all,favorites'), [self.a3.id])
self.assertEqual(self.ids(filter='unplayed,'), self.ids(filter='unplayed'))
self.assertEqual(self.get(filter='unplayed,queued')['total'], 1)
def test_sorting(self):
newest = [e.id for e in (self.b2, self.b1, self.a3, self.a2, self.a1, self.a4)]
self.assertEqual(self.ids(sort='new'), newest)
# Oldest first; without a date still last.
self.assertEqual(self.ids(sort='old'), [e.id for e in (self.a1, self.a2, self.a3, self.b1, self.b2, self.a4)])
PodcastEpisode.objects.filter(pk=self.a2.pk).update(duration_seconds=1200)
PodcastEpisode.objects.filter(pk=self.b1.pk).update(duration_seconds=300)
# By length; unknown lengths (0) last, newest first among them.
self.assertEqual(self.ids(sort='long')[:3], [self.a2.id, self.a1.id, self.b1.id])
self.assertEqual(self.ids(sort='short')[:3], [self.b1.id, self.a1.id, self.a2.id])
self.assertEqual(self.ids(sort='short')[3:], [self.b2.id, self.a3.id, self.a4.id])
self.assertEqual(self.ids(sort='title'), [e.id for e in (self.a1, self.a2, self.a3, self.a4, self.b2, self.b1)])
# The history sorts by date too when asked; by default by the last listening.
self.assertEqual(self.ids(filter='history', sort='new'), [self.b1.id, self.a2.id, self.a1.id])
self.assertEqual(self.ids(filter='unplayed', sort='played'), [e.id for e in (self.a3, self.b2, self.a2, self.b1, self.a4)])
def test_ids(self):
# The episodes downloaded on a device; someone else's stay out.
bobs = PodcastEpisode.objects.get(feed__user__username='bob')
ids = f'{self.a1.id},{self.b1.id},{bobs.id}'
self.assertEqual(self.ids(ids=ids), [self.b1.id, self.a1.id])
self.assertEqual(self.ids(ids=ids, filter='unplayed'), [self.b1.id])
self.assertEqual(self.ids(ids=''), [])
def test_search_in_shownotes(self):
PodcastEpisode.objects.filter(pk=self.a1.pk).update(description='<p>Heute mit dem Gast Bonusmaterial</p>')
# By title only unless asked; with notes=1 title matches first.
self.assertEqual(self.ids(q='bonus'), [self.b1.id])
self.assertEqual(self.ids(q='bonus', notes='1'), [self.b1.id, self.a1.id])
self.assertEqual(self.ids(q='gast', notes='1'), [self.a1.id])
self.assertEqual(self.ids(q='gast', notes='1', filter='unplayed'), [])
self.assertEqual(self.get(q='bonus', notes='1')['total'], 2)
# The notes never come along in the list.
self.assertNotIn('description', self.get(q='gast', notes='1')['episodes'][0])
def test_combined_query_count(self):
from .views import _episode_list_data
with self.assertNumQueries(2):
episodes, total = _episode_list_data(self.user, ['unplayed', 'queued'], q='a', notes=True,
sort='short', ids=[self.a2.id, self.a3.id])
self.assertEqual(([e['id'] for e in episodes], total), ([self.a2.id], 1))
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" <Folge>', 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 &amp; Jerry &quot;Live&quot; &lt;Folge&gt;', raw)
self.assertIn('xmlUrl="https://example.org/feed?a=1&amp;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" <Folge>', 'title': 'Tom & Jerry "Live" <Folge>',
'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, 'auto_download': False,
'notify': False, 'episode_sort': '', 'episode_filter': [], 'skip_silence': None, 'kind': 'podcast',
'folder': '', 'folders': [], 'refresh_interval': '', 'favorite': 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, 'auto_download': False, 'notify': False,
'episode_sort': '', 'episode_filter': [], 'skip_silence': None, 'kind': 'podcast', 'folder': '', 'folders': [],
'refresh_interval': '', 'favorite': 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, 3.5):
self.assertEqual(self.post({'playback_speed': s}).json()['playback_speed'], s)
def test_episode_list_and_skip_silence(self):
r = self.post({'episode_sort': 'old', 'episode_filter': ['unplayed', 'favorites', 'unplayed'],
'skip_silence': False})
self.assertEqual((r.json()['episode_sort'], r.json()['episode_filter'], r.json()['skip_silence']),
('old', ['unplayed', 'favorites'], False))
self.feed.refresh_from_db()
self.assertEqual((self.feed.episode_sort, self.feed.episode_filter, self.feed.skip_silence),
('old', ['unplayed', 'favorites'], False))
for sort in PodcastFeed.EPISODE_SORTS:
self.assertEqual(self.post({'episode_sort': sort}).json()['episode_sort'], sort)
# Back to the defaults: newest first, all episodes, as the device is set.
r = self.post({'episode_sort': '', 'episode_filter': [], 'skip_silence': None})
self.assertEqual((r.json()['episode_sort'], r.json()['episode_filter'], r.json()['skip_silence']), ('', [], None))
self.assertTrue(self.post({'skip_silence': True}).json()['skip_silence'])
self.assertEqual(self.post({'episode_filter': list(PodcastFeed.EPISODE_FILTERS)}).json()['episode_filter'],
list(PodcastFeed.EPISODE_FILTERS))
state = {f['id']: f for f in self.client.get('/podcasts/state/').json()['feeds']}[self.feed.id]
self.assertEqual((state['episode_sort'], state['episode_filter'], state['skip_silence']),
('', list(PodcastFeed.EPISODE_FILTERS), True))
def test_invalid_values_name_the_field_and_change_nothing(self):
bad = [
('playback_speed', 0.4), ('playback_speed', 3.6), ('playback_speed', 'fast'), ('playback_speed', True),
('episode_sort', 'played'), ('episode_sort', 1), ('episode_filter', 'unplayed'),
('episode_filter', ['history']), ('episode_filter', [1]), ('skip_silence', 1), ('skip_silence', '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),
('auto_download', 1), ('auto_download', 'true'), ('auto_download', None),
('notify', 1), ('notify', 'false'), ('notify', None),
]
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_auto_download_on_and_off(self):
self.assertIs(self.post({'auto_download': True}).json()['auto_download'], True)
self.feed.refresh_from_db()
self.assertTrue(self.feed.auto_download)
# Other settings leave it alone.
self.assertIs(self.post({'skip_intro': 5}).json()['auto_download'], True)
self.assertIs(self.post({'auto_download': False}).json()['auto_download'], False)
self.feed.refresh_from_db()
self.assertFalse(self.feed.auto_download)
def test_notify_on_and_off(self):
self.assertIs(self.post({'notify': True}).json()['notify'], True)
self.feed.refresh_from_db()
self.assertTrue(self.feed.notify)
# Other settings (auto_download too) leave it alone.
self.assertIs(self.post({'auto_download': True}).json()['notify'], True)
self.assertIs(self.post({'notify': False}).json()['notify'], False)
self.feed.refresh_from_db()
self.assertFalse(self.feed.notify)
self.assertTrue(self.feed.auto_download)
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, 'auto_download': True, 'notify': True})
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, 'auto_download': True, 'notify': 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 = '<itunes:duration>{}</itunes:duration>'
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, **kw: _ok(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, **kw: _ok(_rss([('n1', 'Fri, 02 Jan 2026 12:00:00 GMT')]))):
_refresh_feed(feed)
self.assertEqual(self.queue(), ['e0', 'e2', 'n1', 'e1'])
# ---------------------------------------------------------------------------
# Feed refresh: conditional GET, errors, permanent redirects, backoff
# ---------------------------------------------------------------------------
class _Hop:
def __init__(self, url, status):
self.url, self.status_code = url, status
class _HttpResponse:
"""Just enough of a requests.Response for _fetch_feed."""
def __init__(self, url, status=200, content=b'', headers=None, history=()):
self.url, self.status_code, self.content = url, status, content
self.headers = headers or {}
self.history = list(history)
class FetchFeedTests(TestCase):
"""_fetch_feed itself, with requests.get faked."""
RSS = _rss([('a', 'Mon, 05 Jan 2026 10:00:00 GMT')]).encode()
def fetch(self, response, url='https://example.org/feed', **kw):
calls = []
def fake_get(u, **kwargs):
calls.append(kwargs['headers'])
if isinstance(response, Exception):
raise response
return response
from .views import _fetch_feed
with patch('podcasts.views.requests.get', fake_get):
result = _fetch_feed(url, **kw)
return result, calls[0]
def error(self, response):
from .views import FeedFetchError
with self.assertRaises(FeedFetchError) as cm:
self.fetch(response)
return str(cm.exception)
def test_the_same_body_again_counts_as_unchanged_without_parsing(self):
first, _ = self.fetch(_HttpResponse('https://example.org/feed', content=self.RSS))
self.assertEqual(first.status, 200)
self.assertTrue(first.body_hash)
# The server ignores validators and sends the feed again: a 304, unparsed.
with patch('podcasts.views.feedparser.parse') as parse:
again, _ = self.fetch(_HttpResponse('https://example.org/feed', content=self.RSS), body_hash=first.body_hash)
parse.assert_not_called()
self.assertEqual((again.status, again.parsed, again.body_hash), (304, None, first.body_hash))
# Only the build date or a comment differs: still the same.
stamped = self.RSS.replace(b'<channel>', b'<channel><lastBuildDate>Sun, 04 Oct 2026 08:00:00 GMT</lastBuildDate><!-- 0.12 s -->', 1)
same, _ = self.fetch(_HttpResponse('https://example.org/feed', content=stamped), body_hash=first.body_hash)
self.assertEqual(same.status, 304)
# A new episode: parsed.
newer = _rss([('b', 'Tue, 06 Jan 2026 10:00:00 GMT'), ('a', 'Mon, 05 Jan 2026 10:00:00 GMT')]).encode()
changed, _ = self.fetch(_HttpResponse('https://example.org/feed', content=newer), body_hash=first.body_hash)
self.assertEqual(changed.status, 200)
self.assertNotEqual(changed.body_hash, first.body_hash)
def test_200_returns_parsed_feed_and_validators(self):
r, headers = self.fetch(_HttpResponse('https://example.org/feed', content=self.RSS, headers={
'ETag': '"v1"', 'Last-Modified': 'Mon, 05 Jan 2026 10:00:00 GMT'}))
self.assertEqual(r.status, 200)
self.assertEqual([e.title for e in r.parsed.entries], ['a'])
self.assertEqual((r.etag, r.last_modified), ('"v1"', 'Mon, 05 Jan 2026 10:00:00 GMT'))
self.assertEqual(r.permanent_url, 'https://example.org/feed')
self.assertNotIn('If-None-Match', headers)
self.assertNotIn('If-Modified-Since', headers)
def test_conditional_request_and_304(self):
r, headers = self.fetch(_HttpResponse('https://example.org/feed', status=304),
etag='"v1"', last_modified='Mon, 05 Jan 2026 10:00:00 GMT')
self.assertEqual(headers['If-None-Match'], '"v1"')
self.assertEqual(headers['If-Modified-Since'], 'Mon, 05 Jan 2026 10:00:00 GMT')
self.assertEqual(r.status, 304)
self.assertIsNone(r.parsed)
def test_304_without_validators_is_an_error(self):
self.assertEqual(self.error(_HttpResponse('https://example.org/feed', status=304)), 'HTTP 304')
def test_errors(self):
import requests
self.assertEqual(self.error(_HttpResponse('https://example.org/feed', status=404)), 'HTTP 404')
self.assertEqual(self.error(_HttpResponse('https://example.org/feed', status=503)), 'HTTP 503')
self.assertEqual(self.error(requests.Timeout()), 'timeout')
self.assertEqual(self.error(requests.ConnectionError()), 'connection failed')
self.assertEqual(self.error(requests.TooManyRedirects()), 'too many redirects')
self.assertEqual(self.error(_HttpResponse('https://example.org/feed', content=b'<html><body><p>Hi')),
'not a feed')
self.assertEqual(self.error(_HttpResponse('https://example.org/feed', content=b'\x00\xff garbage')),
'not a feed')
def test_permanent_redirect_chain(self):
r, _ = self.fetch(_HttpResponse('https://c.org/feed', content=self.RSS, history=[
_Hop('https://example.org/feed', 301), _Hop('https://b.org/feed', 308)]))
self.assertEqual(r.permanent_url, 'https://c.org/feed')
self.assertEqual(r.final_url, 'https://c.org/feed')
def test_temporary_hop_ends_the_permanent_part(self):
r, _ = self.fetch(_HttpResponse('https://c.org/feed', content=self.RSS, history=[
_Hop('https://example.org/feed', 301), _Hop('https://b.org/feed', 302)]))
self.assertEqual(r.permanent_url, 'https://b.org/feed')
for status in (302, 307):
r, _ = self.fetch(_HttpResponse('https://c.org/feed', content=self.RSS, history=[
_Hop('https://example.org/feed', status), _Hop('https://b.org/feed', 301)]))
self.assertEqual(r.permanent_url, 'https://example.org/feed')
class FeedRefreshTests(TestCase):
"""What a refresh stores: validators, 304s, errors, moved feeds."""
V = {'etag': '"v1"', 'last_modified': 'Mon, 05 Jan 2026 10:00:00 GMT'}
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),
http_etag=self.V['etag'], http_last_modified=self.V['last_modified'],
)
PodcastEpisode.objects.create(feed=self.feed, guid='old', title='Old', audio_url='https://example.org/old.mp3')
self.calls = []
self.answer = None
def fake_fetch(self, url, **kw):
self.calls.append((url, kw))
if isinstance(self.answer, Exception):
raise self.answer
return self.answer
def refresh(self, feeds=None):
from .views import _refresh_feeds
with patch('podcasts.views._fetch_feed', self.fake_fetch):
return _refresh_feeds(feeds or [self.feed])
def test_304_touches_no_episodes(self):
from .views import FeedFetch
self.feed.error_count, self.feed.last_error = 2, 'timeout'
self.feed.last_error_at = datetime(2026, 1, 2, tzinfo=timezone.utc)
self.feed.save()
self.answer = FeedFetch(304, None, **self.V)
with patch('podcasts.views.PodcastEpisode.objects.get_or_create') as create, \
patch('podcasts.views.PodcastEpisode.objects.bulk_update') as update:
self.assertEqual(self.refresh(), (1, 0, 0))
create.assert_not_called()
update.assert_not_called()
self.assertEqual(self.calls, [('https://example.org/feed', self.V)])
self.feed.refresh_from_db()
self.assertGreater(self.feed.last_refreshed_at, datetime(2026, 1, 1, tzinfo=timezone.utc))
self.assertEqual((self.feed.error_count, self.feed.last_error, self.feed.last_error_at), (0, '', None))
self.assertEqual((self.feed.http_etag, self.feed.http_last_modified), (self.V['etag'], self.V['last_modified']))
self.assertEqual(self.feed.episodes.count(), 1)
def test_200_stores_new_validators(self):
self.answer = _ok(_rss([('new', 'Mon, 05 Jan 2026 10:00:00 GMT')]), etag='"v2"', last_modified='')
self.assertEqual(self.refresh(), (1, 0, 1))
self.feed.refresh_from_db()
self.assertEqual((self.feed.http_etag, self.feed.http_last_modified), ('"v2"', ''))
def test_first_fetch_is_never_conditional(self):
self.feed.last_refreshed_at = None
self.feed.save()
self.answer = _ok(_rss([]))
self.refresh()
self.assertEqual(self.calls, [('https://example.org/feed', {})])
# Nothing stored yet either: unconditional.
self.calls.clear()
new = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/n', title='N')
self.refresh([new])
self.assertEqual(self.calls, [('https://example.org/n', {})])
def test_shared_url_validators_only_when_all_agree(self):
from .views import FeedFetch
bob = User.objects.create_user(username='bob', password='pw12345678')
bobs = PodcastFeed.objects.create(
user=bob, rss_url='https://example.org/feed', title='Feed',
last_refreshed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
http_etag=self.V['etag'], http_last_modified=self.V['last_modified'],
)
self.answer = FeedFetch(304, None, **self.V)
self.assertEqual(self.refresh([self.feed, bobs]), (2, 0, 0))
self.assertEqual(self.calls, [('https://example.org/feed', self.V)])
# Bob's copy has other validators: one full download for both.
bobs.http_etag = '"v0"'
bobs.save()
self.calls.clear()
self.answer = _ok(_rss([('new', 'Mon, 05 Jan 2026 10:00:00 GMT')]), etag='"v2"')
self.assertEqual(self.refresh([self.feed, bobs]), (2, 0, 2))
self.assertEqual(self.calls, [('https://example.org/feed', {})])
self.assertEqual(set(PodcastFeed.objects.values_list('http_etag', flat=True)), {'"v2"'})
# Carol just subscribed (never fetched): a full download as well.
carol = User.objects.create_user(username='carol', password='pw12345678')
carols = PodcastFeed.objects.create(user=carol, rss_url='https://example.org/feed', title='Feed',
http_etag='"v2"')
self.calls.clear()
self.refresh([self.feed, PodcastFeed.objects.get(pk=bobs.pk), carols])
self.assertEqual(self.calls, [('https://example.org/feed', {})])
self.assertEqual(PodcastEpisode.objects.filter(feed=carols).count(), 1)
def test_errors_are_counted_and_cleared(self):
from .views import FeedFetchError
self.answer = FeedFetchError('HTTP 404')
self.assertEqual(self.refresh(), (0, 1, 0))
self.feed.refresh_from_db()
self.assertEqual(self.refresh([self.feed]), (0, 1, 0))
self.feed.refresh_from_db()
self.assertEqual((self.feed.last_error, self.feed.error_count), ('HTTP 404', 2))
self.assertIsNotNone(self.feed.last_error_at)
# Unchanged otherwise: still the last good refresh.
self.assertEqual(self.feed.last_refreshed_at, datetime(2026, 1, 1, tzinfo=timezone.utc))
feeds = self.client.get('/podcasts/feeds/').json()['feeds']
self.assertEqual(feeds[0]['last_error'], 'HTTP 404')
self.assertEqual(feeds[0]['error_count'], 2)
self.assertEqual(feeds[0]['last_error_at'], self.feed.last_error_at.isoformat())
state = self.client.get('/podcasts/state/').json()['feeds']
self.assertEqual(state[0]['error_count'], 2)
self.answer = _ok(_rss([('new', 'Mon, 05 Jan 2026 10:00:00 GMT')]))
self.assertEqual(self.refresh([self.feed]), (1, 0, 1))
feeds = self.client.get('/podcasts/feeds/').json()['feeds']
self.assertEqual((feeds[0]['last_error'], feeds[0]['last_error_at'], feeds[0]['error_count']), ('', None, 0))
def test_refresh_now_reports_and_stores_the_error(self):
from .views import FeedFetchError
self.answer = FeedFetchError('not a feed')
with patch('podcasts.views._fetch_feed', self.fake_fetch):
r = self.client.post('/podcasts/feeds/refresh/', json.dumps({'feed_id': self.feed.id}),
content_type='application/json')
self.assertEqual(r.status_code, 502)
self.assertEqual(r.json()['error'], 'not a feed')
# By hand: the whole feed, without the validators (corrections to known episodes come in).
self.assertEqual(self.calls, [('https://example.org/feed', {})])
self.feed.refresh_from_db()
self.assertEqual((self.feed.last_error, self.feed.error_count), ('not a feed', 1))
def test_failed_subscription_is_not_kept(self):
from .views import FeedFetchError
self.answer = FeedFetchError('HTTP 404')
with patch('podcasts.views._fetch_feed', self.fake_fetch):
r = self.client.post('/podcasts/feeds/add/', json.dumps({'rss_url': 'https://example.org/gone'}),
content_type='application/json')
self.assertEqual(r.status_code, 400)
self.assertIn('HTTP 404', r.json()['error'])
self.assertEqual(self.calls, [('https://example.org/gone', {})])
self.assertFalse(PodcastFeed.objects.filter(rss_url='https://example.org/gone').exists())
def test_permanent_redirect_moves_the_subscription(self):
self.answer = _ok(_rss([('new', 'Mon, 05 Jan 2026 10:00:00 GMT')]),
final_url='https://new.org/feed', permanent_url='https://new.org/feed')
self.assertEqual(self.refresh(), (1, 0, 1))
self.feed.refresh_from_db()
self.assertEqual(self.feed.rss_url, 'https://new.org/feed')
self.assertEqual(self.feed.episodes.count(), 2)
def test_permanent_redirect_to_a_url_the_user_has_already(self):
other = PodcastFeed.objects.create(user=self.user, rss_url='https://new.org/feed', title='Other')
self.answer = _ok(_rss([('new', 'Mon, 05 Jan 2026 10:00:00 GMT')]), etag='"v2"',
final_url='https://new.org/feed', permanent_url='https://new.org/feed')
self.assertEqual(self.refresh(), (1, 0, 1))
self.feed.refresh_from_db()
self.assertEqual(self.feed.rss_url, 'https://example.org/feed')
self.assertEqual((self.feed.error_count, self.feed.http_etag), (0, '"v2"'))
self.assertEqual(PodcastFeed.objects.get(pk=other.pk).rss_url, 'https://new.org/feed')
def test_temporary_redirect_keeps_the_url(self):
self.answer = _ok(_rss([]), final_url='https://tmp.org/feed', permanent_url='https://example.org/feed')
self.refresh()
self.feed.refresh_from_db()
self.assertEqual(self.feed.rss_url, 'https://example.org/feed')
class FeedBackoffTests(TestCase):
"""The hourly job tries a feed failing 3+ times in a row less often."""
def setUp(self):
from django.utils import timezone as dj_tz
self.user = User.objects.create_user(username='alice', password='pw12345678')
self.client.force_login(self.user)
now = dj_tz.now()
long_ago = datetime(2026, 1, 1, tzinfo=timezone.utc)
# (name, error_count, hours since the last error) -> due in the hourly job?
self.cases = {
'ok': (0, None, True),
'two': (2, 0.2, True),
'three-early': (3, 1, False), # waits 2 h
'three-due': (3, 2.5, True),
'four-early': (4, 3, False), # waits 4 h
'four-due': (4, 4.5, True),
'six-early': (6, 15, False), # waits 16 h
'six-due': (6, 17, True),
'ten-early': (10, 23, False), # capped at 24 h
'ten-due': (10, 25, True),
}
self.feeds = {}
for name, (count, hours, _) in self.cases.items():
self.feeds[name] = PodcastFeed.objects.create(
user=self.user, rss_url=f'https://example.org/{name}', title=name,
last_refreshed_at=long_ago, error_count=count, last_error='timeout' if count else '',
last_error_at=now - timedelta(hours=hours) if hours is not None else None,
)
self.fetched = []
def fake_fetch(self, url, **kw):
self.fetched.append(url.rsplit('/', 1)[1])
return _ok(_rss([]))
def run_command(self, *args):
from io import StringIO
from django.core.management import call_command
out = StringIO()
with patch('podcasts.views._fetch_feed', self.fake_fetch):
call_command('refresh_feeds', *args, stdout=out)
return out.getvalue()
def test_hourly_job_skips_feeds_in_backoff(self):
out = self.run_command()
due = {name for name, (_, _, is_due) in self.cases.items() if is_due}
self.assertEqual(set(self.fetched), due)
self.assertIn(f'{len(due)} feeds refreshed, 0 failed', out)
def test_explicit_feed_id_ignores_backoff(self):
self.run_command('--feed-id', str(self.feeds['ten-early'].id))
self.assertEqual(self.fetched, ['ten-early'])
def test_user_refresh_all_tries_them_but_respects_max_age(self):
from django.utils import timezone as dj_tz
recent = self.feeds['three-early']
recent.last_refreshed_at = dj_tz.now() - timedelta(seconds=60)
recent.save()
with patch('podcasts.views._fetch_feed', self.fake_fetch):
r = self.client.post('/podcasts/feeds/refresh-all/', '{}', content_type='application/json')
self.assertEqual(set(self.fetched), set(self.cases) - {'three-early'})
self.assertEqual(r.json()['refreshed'], len(self.cases) - 1)
def test_backoff_hours(self):
from .views import _backoff_hours
self.assertEqual([_backoff_hours(n) for n in range(0, 9)], [0, 0, 0, 2, 4, 8, 16, 24, 24])
# ---------------------------------------------------------------------------
# ID3v2 chapters (CHAP frames in the audio file)
# ---------------------------------------------------------------------------
def _ss(n):
"""Synchsafe 4-byte integer."""
return bytes([(n >> 21) & 0x7F, (n >> 14) & 0x7F, (n >> 7) & 0x7F, n & 0x7F])
def _frame(frame_id, data, v=3, fmt=0, plain_size=False):
size = _ss(len(data)) if v == 4 and not plain_size else len(data).to_bytes(4, 'big')
return frame_id.encode() + size + bytes([0, fmt]) + data
def _text(s, enc=3):
codec = {0: 'latin-1', 1: 'utf-16', 3: 'utf-8'}[enc]
return bytes([enc]) + s.encode(codec) + (b'\0\0' if enc == 1 else b'\0')
def _chap(element_id, start_ms, end_ms, subframes=b''):
import struct
return element_id + b'\0' + struct.pack('>IIII', start_ms, end_ms, 0xFFFFFFFF, 0xFFFFFFFF) + subframes
def _ctoc(element_id, children, top=True):
return element_id + b'\0' + bytes([0x03 if top else 0x01, len(children)]) + b''.join(c + b'\0' for c in children)
def _unsync(data):
"""ID3 unsynchronisation: a $00 after every $FF followed by %111xxxxx or $00 (or at the end)."""
out = bytearray()
for i, b in enumerate(data):
out.append(b)
if b == 0xFF and (i + 1 == len(data) or data[i + 1] == 0 or data[i + 1] >= 0xE0):
out.append(0)
return bytes(out)
def _tag(body, v=3, flags=0, padding=64):
body = body + b'\0' * padding
return b'ID3' + bytes([v, 0, flags]) + _ss(len(body)) + body
_APIC = b'\0image/jpeg\0\x03\0' + b'\xff\xd8\xff\xe0' + b'p' * 300
# A v2.3 tag as podcast tools write it: title, cover, table of contents, chapters.
ID3V23 = _tag(
_frame('TIT2', _text('Folge 1'))
+ _frame('APIC', _APIC * 10)
+ _frame('CTOC', _ctoc(b'toc', [b'ch1', b'ch2', b'ch3']))
+ _frame('CHAP', _chap(b'ch1', 0, 60000, _frame('TIT2', _text('Intro', 0))))
+ _frame('CHAP', _chap(b'ch2', 60000, 125500,
_frame('TIT2', _text('Zweiter Teil – ü', 1))
+ _frame('WXXX', b'\x00Link\x00https://x.org/2')
+ _frame('APIC', _APIC)))
+ _frame('CHAP', _chap(b'ch3', 125500, 200000))
)
ID3V23_CHAPTERS = [
{'start': 0.0, 'title': 'Intro'},
{'start': 60.0, 'title': 'Zweiter Teil – ü', 'url': 'https://x.org/2'},
{'start': 125.5, 'title': ''},
]
class ID3ParserTests(TestCase):
def chapters(self, data):
from .chapters import _chapter, _finish
from .id3 import chapters
return _finish(_chapter(s / 1000, t, u) for s, _e, t, u in chapters(data))
def test_v23(self):
self.assertEqual(self.chapters(ID3V23), ID3V23_CHAPTERS)
def test_v23_unsynchronised_with_extended_header(self):
ext = (6).to_bytes(4, 'big') + b'\0\0' + b'\0\0\0\0'
frames = (
_frame('CHAP', _chap(b'a', 0xFFE0, 70000, _frame('TIT2', _text('Ünsynchron', 1))))
+ _frame('CHAP', _chap(b'b', 70000, 80000, _frame('TIT2', _text('Zwei')) + _frame('WOAF', b'https://x.org/w')))
)
body = _unsync(ext + frames + b'\0' * 16)
self.assertNotEqual(body, ext + frames + b'\0' * 16) # there really was something to unsynchronise
tag = b'ID3' + bytes([3, 0, 0xC0]) + _ss(len(body)) + body
self.assertEqual(self.chapters(tag), [
{'start': 65.504, 'title': 'Ünsynchron'},
{'start': 70.0, 'title': 'Zwei', 'url': 'https://x.org/w'},
])
def test_v24_synchsafe_sizes_toc_and_frame_flags(self):
# The cover makes the CHAP frame longer than 127 bytes, where synchsafe
# and plain sizes differ.
long_chap = _chap(b'c2', 30000, 60000, _frame('APIC', _APIC, v=4) + _frame('TIT2', _text('Mit Bild'), v=4))
self.assertGreater(len(long_chap), 127)
# Unsynchronised frame with a data length indicator (flags %00000011).
raw = _chap(b'c3', 0x0001FFE0, 200000, _frame('TIT2', _text('ÿÿ Flags', 0), v=4))
flagged = b'CHAP' + _ss(4 + len(_unsync(raw))) + bytes([0, 0x03]) + _ss(len(raw)) + _unsync(raw)
tag = _tag(
_frame('CTOC', _ctoc(b'top', [b'sub', b'c3']), v=4)
+ _frame('CTOC', _ctoc(b'sub', [b'c2', b'c1'], top=False), v=4)
+ _frame('CHAP', _chap(b'c1', 0, 30000, _frame('TIT2', _text('Eins'), v=4)), v=4)
+ _frame('CHAP', long_chap, v=4)
+ flagged
+ _frame('CHAP', _chap(b'extra', 0, 1000, _frame('TIT2', _text('Nicht im TOC'), v=4)), v=4),
v=4,
)
from .id3 import chapters
# Table-of-contents order (nested first), then what it doesn't list.
self.assertEqual([c[2] for c in chapters(tag)], ['Mit Bild', 'Eins', 'ÿÿ Flags', 'Nicht im TOC'])
self.assertEqual(self.chapters(tag), [
{'start': 0.0, 'title': 'Eins'},
{'start': 0.0, 'title': 'Nicht im TOC'},
{'start': 30.0, 'title': 'Mit Bild'},
{'start': 131.04, 'title': 'ÿÿ Flags'},
])
def test_v24_with_plain_frame_sizes(self):
# Some writers put v2.3-style sizes into v2.4 tags.
long_chap = _chap(b'c1', 5000, 60000, _frame('APIC', _APIC, v=4, plain_size=True)
+ _frame('TIT2', _text('Plain'), v=4, plain_size=True))
tag = _tag(_frame('CHAP', long_chap, v=4, plain_size=True)
+ _frame('CHAP', _chap(b'c2', 60000, 70000, _frame('TIT2', _text('Danach'), v=4)), v=4), v=4)
self.assertEqual(self.chapters(tag), [{'start': 5.0, 'title': 'Plain'}, {'start': 60.0, 'title': 'Danach'}])
def test_no_tag_or_no_chapters(self):
from .id3 import chapters, tag_size
self.assertEqual(chapters(b'\xff\xfb\x90\x00' * 100), [])
self.assertEqual(chapters(_tag(_frame('TIT2', _text('Nur Titel')))), [])
self.assertIsNone(tag_size(b'ID3\x03\x00\x00\x80\x00\x00\x00')) # not synchsafe
self.assertEqual(tag_size(_tag(b'', padding=300)), 310)
class _StreamResponse(_FakeResponse):
"""Hands out the body in the chunks asked for, counting what was taken."""
def __init__(self, body=b'', status=200, headers=None):
super().__init__(body, status, headers)
self.sent = 0
def iter_content(self, chunk_size=1):
for i in range(0, len(self.body), chunk_size):
chunk = self.body[i:i + chunk_size]
self.sent += len(chunk)
yield chunk
class ID3ChapterFetchTests(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')
self.ep = PodcastEpisode.objects.create(feed=self.feed, guid='a', title='A',
audio_url='https://example.org/a.mp3')
self.calls = []
self.responses = {}
def fake_get(self, url, **kwargs):
self.calls.append((url, kwargs.get('headers') or {}))
r = self.responses[url]
return r() if callable(r) else r
def get(self):
with patch('podcasts.chapters.assert_safe_url'), patch('podcasts.chapters.requests.get', self.fake_get):
return self.client.get(f'/podcasts/episodes/{self.ep.id}/')
def test_range_request_answered_with_206(self):
from .chapters import ID3_MAX_BYTES
self.responses['https://example.org/a.mp3'] = lambda: _StreamResponse(ID3V23 + b'\xff\xfb' * 100, status=206)
self.assertEqual(self.get().json()['chapters'], ID3V23_CHAPTERS)
self.assertEqual(self.calls[0][1]['Range'], f'bytes=0-{ID3_MAX_BYTES - 1}')
# Cached: the second detail doesn't touch the file again.
self.assertEqual(self.get().json()['chapters'], ID3V23_CHAPTERS)
self.assertEqual(len(self.calls), 1)
self.ep.refresh_from_db()
self.assertEqual(self.ep.chapters, ID3V23_CHAPTERS)
self.assertIsNotNone(self.ep.chapters_fetched_at)
def test_server_ignoring_range_is_read_only_as_far_as_the_tag(self):
from .chapters import ID3_CHUNK
whole_file = _StreamResponse(ID3V23 + b'\xff\xfb' * (3 * 1024 * 1024), status=200)
self.responses['https://example.org/a.mp3'] = whole_file
self.assertEqual(self.get().json()['chapters'], ID3V23_CHAPTERS)
self.assertLessEqual(whole_file.sent, ID3_CHUNK)
def test_tag_larger_than_the_cap(self):
from .chapters import ID3_CHUNK, ID3_MAX_BYTES
tag = _tag(
_frame('CHAP', _chap(b'ch1', 0, 1000, _frame('TIT2', _text('Vorne'))))
+ _frame('APIC', b'\0image/png\0\x03\0' + b'p' * (ID3_MAX_BYTES + 1024))
+ _frame('CHAP', _chap(b'ch2', 1000, 2000, _frame('TIT2', _text('Hinter der Grenze')))),
)
response = _StreamResponse(tag + b'\xff\xfb' * 1024, status=206)
self.responses['https://example.org/a.mp3'] = response
self.assertEqual(self.get().json()['chapters'], [{'start': 0.0, 'title': 'Vorne'}])
self.assertLessEqual(response.sent, ID3_MAX_BYTES + ID3_CHUNK)
def test_file_without_tag_is_remembered(self):
self.responses['https://example.org/a.mp3'] = lambda: _StreamResponse(b'\0\0\0\x20ftypM4A ' + b'x' * 5000)
self.assertEqual(self.get().json()['chapters'], [])
self.assertEqual(self.get().json()['chapters'], [])
self.assertEqual(len(self.calls), 1)
self.ep.refresh_from_db()
self.assertIsNotNone(self.ep.chapters_fetched_at)
def test_empty_json_chapters_fall_back_to_the_file(self):
self.ep.chapters_url = 'https://x.org/c.json'
self.ep.save()
self.responses['https://x.org/c.json'] = _FakeResponse(b'{"chapters": []}')
self.responses['https://example.org/a.mp3'] = _StreamResponse(ID3V23, status=206)
self.assertEqual(self.get().json()['chapters'], ID3V23_CHAPTERS)
self.assertEqual([c[0] for c in self.calls], ['https://x.org/c.json', 'https://example.org/a.mp3'])
def test_json_chapters_win(self):
self.ep.chapters_url = 'https://x.org/c.json'
self.ep.save()
self.responses['https://x.org/c.json'] = _FakeResponse(json.dumps(CHAPTERS_JSON).encode())
self.assertEqual(len(self.get().json()['chapters']), 3)
self.assertEqual([c[0] for c in self.calls], ['https://x.org/c.json'])
def test_redirect_is_followed_and_rechecked(self):
self.responses['https://example.org/a.mp3'] = _FakeResponse(
status=302, headers={'Location': 'https://cdn.example.org/a.mp3'})
self.responses['https://cdn.example.org/a.mp3'] = _StreamResponse(ID3V23, status=206)
with patch('podcasts.chapters.assert_safe_url') as check, patch('podcasts.chapters.requests.get', self.fake_get):
r = self.client.get(f'/podcasts/episodes/{self.ep.id}/')
self.assertEqual(r.json()['chapters'], ID3V23_CHAPTERS)
self.assertEqual([c.args[0] for c in check.call_args_list],
['https://example.org/a.mp3', 'https://cdn.example.org/a.mp3'])
self.assertEqual(self.calls[1][1]['Range'], self.calls[0][1]['Range'])
def test_temporary_failure_is_not_cached(self):
self.responses['https://example.org/a.mp3'] = _FakeResponse(status=503)
self.assertEqual(self.get().json()['chapters'], [])
self.ep.refresh_from_db()
self.assertIsNone(self.ep.chapters_fetched_at)
self.responses['https://example.org/a.mp3'] = _StreamResponse(ID3V23, status=206)
self.assertEqual(self.get().json()['chapters'], ID3V23_CHAPTERS)
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.calls, [])
class ChapterCacheRefreshTests(TestCase):
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, **kw: _ok(rss)):
_refresh_feed(self.feed)
def test_chapters_from_the_file_survive_a_refresh(self):
self.refresh(_rich_rss([('a', '')]))
fetched = datetime(2026, 1, 2, tzinfo=timezone.utc)
PodcastEpisode.objects.update(chapters=[{'start': 0.0, 'title': 'ID3'}], chapters_fetched_at=fetched)
self.refresh(_rich_rss([('a', '')]))
ep = PodcastEpisode.objects.get()
self.assertEqual((ep.chapters, ep.chapters_fetched_at), ([{'start': 0.0, 'title': 'ID3'}], fetched))
# A chapters URL turning up means fetching that.
self.refresh(_rich_rss([('a', '<podcast:chapters url="https://x.org/c.json" type="application/json+chapters"/>')]))
ep.refresh_from_db()
self.assertEqual((ep.chapters, ep.chapters_fetched_at), ([], None))
def test_inline_chapters_going_away_means_a_fetch(self):
self.refresh(_rich_rss([('a', PSC)]))
self.assertEqual(len(PodcastEpisode.objects.get().chapters), 4)
self.refresh(_rich_rss([('a', '')]))
ep = PodcastEpisode.objects.get()
self.assertEqual((ep.chapters, ep.chapters_fetched_at), ([], None))
# ---------------------------------------------------------------------------
# Transcripts
# ---------------------------------------------------------------------------
VTT = (
'WEBVTT\r\nKind: captions\r\n\r\n'
'NOTE Ein Kommentar\r\nüber zwei Zeilen\r\n\r\n'
'STYLE\r\n::cue { color: red }\r\n\r\n'
'1\r\n00:00.000 --> 00:02.000\r\n<v Anna>Hallo und</v>\r\n\r\n'
'intro-2\r\n00:02.000 --> 00:04.500 align:start position:10%\r\n<v Anna>willkommen &amp; so.</v>\r\n\r\n'
'00:00:04.500 --> 00:00:06.000\r\n<v.loud Ben>Danke!\r\n<v Anna>Gern.\r\n\r\n'
'01:00:00.000 --> 01:00:02.000\r\n<i>Ende</i> <00:00:01.000><c>gut</c>\r\n'
)
SRT = (
'1\n00:00:01,000 --> 00:00:03,500\nErste Zeile\nzweite Zeile\n\n'
'2\n00:00:03,500 --> 00:00:05,000\n<i>weiter.</i>\n\n'
'3\n00:00:40,000 --> 00:00:42,000\nNach der Pause\n'
'4\n00:00:42,000 --> 00:00:43,000\nohne Leerzeile davor\n'
)
TRANSCRIPT_JSON = {
'version': '1.0.0',
'segments': [
{'speaker': 'Anna', 'startTime': 0, 'endTime': 0.5, 'body': 'Hallo'},
{'speaker': 'Anna', 'startTime': 0.5, 'endTime': 1.2, 'body': 'zusammen.'},
{'speaker': 'Ben', 'startTime': 1.3, 'endTime': 2, 'body': 'Hi'},
{'speaker': '', 'startTime': 2.5, 'body': 'ohne Ende'},
{'startTime': 3, 'endTime': 4, 'body': 'und &amp; weiter'},
{'startTime': 'kaputt', 'body': 'weg'},
'kein Objekt',
],
}
class TranscriptParsingTests(TestCase):
def test_vtt(self):
from .transcripts import merge, parse
self.assertEqual(merge(parse(VTT.encode(), 'vtt')), [
{'start': 0.0, 'end': 4.5, 'speaker': 'Anna', 'text': 'Hallo und willkommen & so.'},
{'start': 4.5, 'end': 6.0, 'speaker': 'Ben', 'text': 'Danke!'},
{'start': 4.5, 'end': 6.0, 'speaker': 'Anna', 'text': 'Gern.'},
{'start': 3600.0, 'end': 3602.0, 'speaker': None, 'text': 'Ende gut'},
])
def test_srt(self):
from .transcripts import merge, parse
self.assertEqual(merge(parse(SRT.encode('cp1252'), 'srt')), [
{'start': 1.0, 'end': 5.0, 'speaker': None, 'text': 'Erste Zeile zweite Zeile weiter.'},
{'start': 40.0, 'end': 43.0, 'speaker': None, 'text': 'Nach der Pause ohne Leerzeile davor'},
])
def test_json(self):
from .transcripts import merge, parse
self.assertEqual(merge(parse(json.dumps(TRANSCRIPT_JSON).encode(), 'json')), [
{'start': 0.0, 'end': 1.2, 'speaker': 'Anna', 'text': 'Hallo zusammen.'},
{'start': 1.3, 'end': 2.0, 'speaker': 'Ben', 'text': 'Hi'},
{'start': 2.5, 'end': 4.0, 'speaker': None, 'text': 'ohne Ende und & weiter'},
])
def test_merging(self):
from .transcripts import merge
# One-second fragments without a sentence end: cut at 30 s.
words = [(float(i), i + 1.0, 'A', f'w{i}') for i in range(40)]
self.assertEqual([(s['start'], s['end']) for s in merge(words)], [(0.0, 30.0), (30.0, 40.0)])
# A sentence end closes a segment once it's 5 s long, not before.
frags = [(0.0, 1.0, None, 'Ja.'), (1.0, 6.0, None, 'Genau, so ist es.'), (6.0, 7.0, None, 'Weiter')]
self.assertEqual([s['text'] for s in merge(frags)], ['Ja. Genau, so ist es.', 'Weiter'])
def test_html_is_unsupported(self):
from .transcripts import Unsupported, parse
with self.assertRaises(Unsupported):
parse(b'<!DOCTYPE html><html><body><p>Hallo</p></body></html>', 'vtt')
def test_choice(self):
from .transcripts import candidates
ts = [
{'url': 'https://x.org/a.srt', 'type': 'application/x-subrip', 'language': 'de'},
{'url': 'https://x.org/a.html', 'type': 'text/html', 'language': 'de'},
{'url': 'https://x.org/en.json', 'type': 'application/json', 'language': 'en'},
{'url': 'https://x.org/a.vtt', 'type': 'text/vtt', 'language': 'de-DE'},
{'url': 'https://x.org/fr.json', 'type': 'application/json', 'language': 'fr'},
{'url': 'https://x.org/b.vtt', 'type': '', 'language': ''},
]
self.assertEqual([t['url'] for t in candidates(ts)], [
'https://x.org/a.vtt', 'https://x.org/b.vtt', 'https://x.org/a.srt',
'https://x.org/en.json', 'https://x.org/fr.json',
])
class TranscriptEndpointTests(TestCase):
VTT_T = {'url': 'https://example.org/a.vtt', 'type': 'text/vtt', 'language': 'de'}
JSON_T = {'url': 'https://example.org/a.json', 'type': 'application/json', 'language': 'de'}
HTML_T = {'url': 'https://example.org/a.html', 'type': 'text/html', 'language': 'de'}
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')
self.ep = PodcastEpisode.objects.create(
feed=self.feed, guid='a', title='A', audio_url='https://example.org/a.mp3',
chapters=[{'start': 0.0, 'title': 'Inline'}], # no chapter fetch in these tests
transcripts=[self.VTT_T, self.JSON_T, self.HTML_T],
)
self.requests = []
self.responses = {
self.VTT_T['url']: lambda: _FakeResponse(VTT.encode()),
self.JSON_T['url']: lambda: _FakeResponse(json.dumps(TRANSCRIPT_JSON).encode()),
}
def fake_get(self, url, **kwargs):
self.requests.append(url)
return self.responses[url]()
def get(self, ep=None, what='transcript/'):
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}/{what}')
def test_best_transcript_is_fetched_once(self):
r = self.get()
self.assertEqual(r.status_code, 200)
data = r.json()
self.assertEqual(data['source'], self.JSON_T)
self.assertEqual(data['segments'][0], {'start': 0.0, 'end': 1.2, 'speaker': 'Anna', 'text': 'Hallo zusammen.'})
self.assertEqual(self.get().json(), data)
self.assertEqual(self.requests, [self.JSON_T['url']])
self.assertTrue(self.get(what='').json()['has_transcript'])
def test_falls_back_to_the_next_one(self):
self.responses[self.JSON_T['url']] = lambda: _FakeResponse(status=404)
data = self.get().json()
self.assertEqual(data['source'], self.VTT_T)
self.assertEqual(data['segments'][1]['speaker'], 'Ben')
self.get()
self.assertEqual(self.requests, [self.JSON_T['url'], self.VTT_T['url']])
def test_temporary_failure_is_not_cached(self):
self.responses[self.JSON_T['url']] = lambda: _FakeResponse(status=503)
self.responses[self.VTT_T['url']] = lambda: _FakeResponse(status=503)
r = self.get()
self.assertEqual((r.status_code, r.json()), (502, {'error': 'unavailable'}))
# The better one failing for now: the other one is shown, but not kept.
self.responses[self.VTT_T['url']] = lambda: _FakeResponse(VTT.encode())
self.assertEqual(self.get().json()['source'], self.VTT_T)
self.responses[self.JSON_T['url']] = lambda: _FakeResponse(json.dumps(TRANSCRIPT_JSON).encode())
self.assertEqual(self.get().json()['source'], self.JSON_T)
self.get()
self.assertEqual(self.requests.count(self.JSON_T['url']), 3)
def test_html_only_is_unsupported(self):
self.ep.transcripts = [self.HTML_T]
self.ep.save()
r = self.get()
self.assertEqual((r.status_code, r.json()), (404, {'error': 'unsupported'}))
self.assertEqual(self.requests, [])
self.assertFalse(self.get(what='').json()['has_transcript'])
def test_html_served_under_another_type(self):
self.ep.transcripts = [self.VTT_T]
self.ep.save()
self.responses[self.VTT_T['url']] = lambda: _FakeResponse(b'<html><body>Hallo</body></html>')
self.assertTrue(self.get(what='').json()['has_transcript'])
r = self.get()
self.assertEqual((r.status_code, r.json()), (404, {'error': 'unsupported'}))
self.assertEqual(self.get().status_code, 404)
self.assertEqual(self.requests, [self.VTT_T['url']])
self.assertFalse(self.get(what='').json()['has_transcript'])
def test_no_transcript(self):
self.ep.transcripts = []
self.ep.save()
r = self.get()
self.assertEqual((r.status_code, r.json()), (404, {'error': 'no transcript'}))
self.assertFalse(self.get(what='').json()['has_transcript'])
def test_new_transcripts_in_the_feed_replace_the_cache(self):
self.ep.transcripts = [self.VTT_T]
self.ep.save()
self.assertEqual(self.get().json()['source'], self.VTT_T)
self.ep.transcripts = [self.VTT_T, self.JSON_T]
self.ep.save()
self.assertEqual(self.get().json()['source'], self.JSON_T)
self.assertEqual(self.requests, [self.VTT_T['url'], self.JSON_T['url']])
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',
transcripts=[self.VTT_T])
r = self.get(ep)
self.assertEqual((r.status_code, r.json()), (404, {'error': 'not found'}))
self.assertEqual(self.requests, [])
def test_login_required(self):
self.client.logout()
self.assertEqual(self.get().status_code, 401)
class AutoDownloadDefaultTests(TestCase):
"""The account-wide default that new subscriptions start with."""
def setUp(self):
self.user = User.objects.create_user(username='alice', password='pw12345678')
self.client.force_login(self.user)
def set_default(self, value):
return self.client.post('/podcasts/settings/', json.dumps({'auto_download_default': value}),
content_type='application/json')
def test_setting(self):
self.assertIs(self.client.get('/podcasts/settings/').json()['auto_download_default'], False)
self.assertIs(self.set_default(True).json()['auto_download_default'], True)
self.assertTrue(PodcastSettings.objects.get(user=self.user).auto_download_default)
self.assertIs(self.client.get('/podcasts/state/').json()['settings']['auto_download_default'], True)
for bad in (1, 'true', None):
with self.subTest(value=bad):
r = self.set_default(bad)
self.assertEqual(r.status_code, 400)
self.assertIn('auto_download_default', r.json()['error'])
self.assertTrue(PodcastSettings.objects.get(user=self.user).auto_download_default)
self.assertIs(self.set_default(False).json()['auto_download_default'], False)
def test_new_subscriptions_start_with_it(self):
from django.core.files.uploadedfile import SimpleUploadedFile
rss = _rss([('e1', 'Mon, 05 Jan 2026 10:00:00 GMT')])
def add(url):
with patch('podcasts.views._fetch_feed', lambda u, **kw: _ok(rss)):
self.client.post('/podcasts/feeds/add/', json.dumps({'rss_url': url}),
content_type='application/json')
def opml(url):
body = f'<opml version="2.0"><body><outline type="rss" text="F" xmlUrl="{url}"/></body></opml>'
self.client.post('/podcasts/feeds/import/',
{'file': SimpleUploadedFile('f.opml', body.encode(), 'text/x-opml')})
def gpodder(url):
auth = 'Basic ' + base64.b64encode(b'alice:pw12345678').decode()
r = self.client.post('/api/2/subscriptions/alice/phone.json', json.dumps({'add': [url]}),
content_type='application/json', HTTP_AUTHORIZATION=auth)
self.assertEqual(r.status_code, 200, r.content)
def auto_download(url):
return PodcastFeed.objects.get(user=self.user, rss_url=url).auto_download
add('https://example.org/a1')
opml('https://example.org/o1')
gpodder('https://example.org/g1')
self.set_default(True)
add('https://example.org/a2')
opml('https://example.org/o2')
gpodder('https://example.org/g2')
self.assertEqual(
[auto_download(f'https://example.org/{n}') for n in ('a1', 'o1', 'g1', 'a2', 'o2', 'g2')],
[False, False, False, True, True, True],
)
# Existing feeds keep theirs.
self.set_default(False)
opml('https://example.org/a2')
self.assertTrue(auto_download('https://example.org/a2'))
class FeedCountsTests(TestCase):
"""Per feed in feeds/ and state/: episodes, unplayed, new, last listened;
and what a podcast's own page needs (feeds/<id>/episodes/)."""
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',
link='https://example.org/', description='Ein Podcast über A.')
self.fb = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/b', title='Beta')
self.empty = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/c', title='Leer')
eps = [PodcastEpisode.objects.create(feed=self.fa, guid=f'a{i}', title=f'A{i}', pub_date=_at(1, i + 1),
audio_url=f'https://example.org/a{i}.mp3') for i in range(5)]
self.b1 = PodcastEpisode.objects.create(feed=self.fb, guid='b1', title='B1', audio_url='https://example.org/b1.mp3')
_progress(self.user, eps[0], _at(2, 1), played=True)
_progress(self.user, eps[1], _at(2, 3), position_seconds=40)
_progress(self.user, eps[2], _at(2, 5), dismissed=True)
_progress(self.user, eps[3], _at(2, 9), favorite=True) # a star isn't listening
self.eps = eps
# Bob's progress on alice's episode (can't happen) and his own feed don't count.
bob = User.objects.create_user(username='bob', password='pw12345678')
_progress(bob, eps[4], _at(3, 1), played=True)
fbob = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/a', title='Alpha')
xb = PodcastEpisode.objects.create(feed=fbob, guid='x', title='X', audio_url='https://example.org/x.mp3')
_progress(bob, xb, _at(3, 1), played=True)
def test_counts_in_feeds_and_state(self):
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']}
a, b, c = feeds[self.fa.id], feeds[self.fb.id], feeds[self.empty.id]
self.assertEqual((a['episode_count'], a['unplayed_count'], a['new_count']), (5, 4, 3))
self.assertEqual(a['last_played_at'], '2026-02-03T09:00:00+00:00')
self.assertEqual(a['link'], 'https://example.org/')
self.assertEqual((b['episode_count'], b['unplayed_count'], b['new_count'], b['last_played_at']),
(1, 1, 1, None))
self.assertEqual((c['episode_count'], c['unplayed_count'], c['new_count'], c['last_played_at']),
(0, 0, 0, None))
# The shownotes-sized description stays out of the state.
self.assertNotIn('description', a)
def test_new_count_matches_the_inbox(self):
inbox = self.client.get('/podcasts/state/').json()['inbox']
feeds = {f['id']: f for f in self.client.get('/podcasts/state/').json()['feeds']}
for feed in (self.fa, self.fb):
self.assertEqual(feeds[feed.id]['new_count'], sum(1 for e in inbox if e['feed__id'] == feed.id))
def test_two_queries_however_many_feeds(self):
from .views import _feeds_data
for i in range(10):
f = PodcastFeed.objects.create(user=self.user, rss_url=f'https://example.org/n{i}', title=f'N{i}')
e = PodcastEpisode.objects.create(feed=f, guid='e', title='E', audio_url='https://example.org/e.mp3')
_progress(self.user, e, _at(2, 2), played=True)
cache.clear()
with self.assertNumQueries(3): # feeds, progress counts, how often they come
feeds = _feeds_data(self.user)
self.assertEqual(len(feeds), 13)
with self.assertNumQueries(2): # that one is kept
_feeds_data(self.user)
self.assertEqual({f['unplayed_count'] for f in feeds if f['title'].startswith('N')}, {0})
def test_feed_page_has_favorites_and_info(self):
data = self.client.get(f'/podcasts/feeds/{self.fa.id}/episodes/').json()
favs = {e['id']: e['favorite'] for e in data['episodes']}
self.assertEqual(favs[self.eps[3].id], True)
self.assertEqual(favs[self.eps[0].id], False)
self.assertEqual(favs[self.eps[4].id], False) # bob's row isn't alice's
self.assertEqual({k: data['feed'][k] for k in ('description', 'link', 'rss_url', 'last_refreshed_at')}, {
'description': 'Ein Podcast über A.', 'link': 'https://example.org/', 'rss_url': 'https://example.org/a',
'last_refreshed_at': None,
})
class BulkFavoriteTests(TestCase):
"""POST /podcasts/episodes/favorite/: star or unstar several at once."""
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.a, self.b, self.c = (
PodcastEpisode.objects.create(feed=feed, guid=g, title=g, audio_url=f'https://example.org/{g}.mp3')
for g in 'abc'
)
bob = User.objects.create_user(username='bob', password='pw12345678')
fbob = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/feed', title='Feed')
self.bobs = PodcastEpisode.objects.create(feed=fbob, guid='x', title='X', audio_url='https://example.org/x.mp3')
def post(self, body):
return self.client.post('/podcasts/episodes/favorite/', json.dumps(body), content_type='application/json')
def test_star_and_unstar(self):
p = _progress(self.user, self.a, _at(2, 1), position_seconds=30)
r = self.post({'episode_ids': [self.a.id, self.b.id, self.bobs.id, 999999], 'favorite': True})
self.assertEqual(r.json(), {'ok': True, 'favorite': True, 'count': 2})
self.assertEqual(set(EpisodeProgress.objects.filter(user=self.user, favorite=True)
.values_list('episode_id', flat=True)), {self.a.id, self.b.id})
# The listening time of an existing row stays (history, statistics).
p.refresh_from_db()
self.assertEqual((p.position_seconds, p.updated_at), (30, _at(2, 1)))
self.assertFalse(EpisodeProgress.objects.filter(episode=self.bobs).exists())
# Again: nothing doubles.
self.post({'episode_ids': [self.a.id, self.b.id], 'favorite': True})
self.assertEqual(EpisodeProgress.objects.filter(user=self.user).count(), 2)
r = self.post({'episode_ids': [self.a.id, self.b.id, self.c.id], 'favorite': False})
self.assertEqual(r.json(), {'ok': True, 'favorite': False, 'count': 3})
self.assertFalse(EpisodeProgress.objects.filter(favorite=True).exists())
# Unstarring doesn't make rows for episodes that had none.
self.assertFalse(EpisodeProgress.objects.filter(episode=self.c).exists())
# The list shows it.
self.post({'episode_ids': [self.c.id], 'favorite': True})
favs = self.client.get('/podcasts/episodes/', {'filter': 'favorites'}).json()['episodes']
self.assertEqual([e['id'] for e in favs], [self.c.id])
def test_bad_requests(self):
for body in ({}, {'episode_ids': [self.a.id]}, {'episode_ids': [], 'favorite': True},
{'episode_ids': self.a.id, 'favorite': True}, {'episode_ids': ['x'], 'favorite': True},
{'episode_ids': [self.a.id], 'favorite': 'yes'}, [1, 2]):
with self.subTest(body=body):
self.assertEqual(self.post(body).status_code, 400)
self.assertEqual(self.client.get('/podcasts/episodes/favorite/').status_code, 405)
self.client.logout()
self.assertEqual(self.post({'episode_ids': [self.a.id], 'favorite': True}).status_code, 401)
class PlaybackSettingsTests(TestCase):
"""The account's playback settings: skip lengths, speed presets, what
skipping and deleting do, mobile data, times at the speed."""
def setUp(self):
self.user = User.objects.create_user(username='alice', password='pw12345678')
self.client.force_login(self.user)
def post(self, body):
return self.client.post('/podcasts/settings/', json.dumps(body), content_type='application/json')
def settings(self):
return {k: getattr(PodcastSettings.for_user(self.user), k) for k in PLAYBACK_DEFAULTS}
def test_defaults(self):
r = self.client.get('/podcasts/settings/').json()
self.assertEqual({k: r[k] for k in PLAYBACK_DEFAULTS}, PLAYBACK_DEFAULTS)
self.assertEqual(r['skip_choices'], [5, 10, 15, 20, 30, 45, 60])
self.assertEqual(r['speed_range'], [0.5, 3.5])
def test_change_all(self):
body = {'skip_back': 10, 'skip_forward': 30, 'smart_played': 30, 'speed_presets': [3, 1, 2.5, 2, 2.0],
'keep_skipped': False, 'delete_dequeue': False, 'mobile_downloads': False,
'mobile_streaming': 'ask', 'times_by_speed': True}
r = self.post(body)
self.assertEqual(r.status_code, 200, r.content)
expected = {**body, 'speed_presets': [1.0, 2.0, 2.5, 3.0]}
self.assertEqual({k: r.json()[k] for k in PLAYBACK_DEFAULTS}, expected)
self.assertEqual(self.settings(), expected)
self.assertEqual({k: self.client.get('/podcasts/state/').json()['settings'][k] for k in PLAYBACK_DEFAULTS},
expected)
# Only what is sent changes; null unsets again.
r = self.post({'mobile_streaming': 'never', 'speed_presets': None, 'skip_back': None})
self.assertEqual({k: r.json()[k] for k in ('mobile_streaming', 'speed_presets', 'skip_back', 'skip_forward')},
{'mobile_streaming': 'never', 'speed_presets': None, 'skip_back': None, 'skip_forward': 30})
def test_presets_up_to_three_and_a_half(self):
self.assertEqual(self.post({'speed_presets': [0.5, 3.5, 1.234]}).json()['speed_presets'], [0.5, 1.23, 3.5])
self.assertEqual(self.post({'speed_presets': [1.0] * 5}).json()['speed_presets'], [1.0])
def test_invalid_values_change_nothing(self):
self.post({'skip_back': 15})
bad = [
('skip_back', 7), ('skip_back', 0), ('skip_back', '10'), ('skip_back', 10.5), ('skip_back', True),
('skip_forward', 90), ('smart_played', -1), ('smart_played', 3601), ('smart_played', 'x'),
('speed_presets', []), ('speed_presets', [0.4]), ('speed_presets', [3.6]), ('speed_presets', ['1']),
('speed_presets', 1.5), ('speed_presets', [0.5 + 0.25 * i for i in range(13)]),
('keep_skipped', 1), ('delete_dequeue', None), ('mobile_downloads', 'false'), ('times_by_speed', 0),
('mobile_streaming', 'sometimes'), ('mobile_streaming', None), ('enqueue_position', 'middle'),
]
for name, value in bad:
with self.subTest(name=name, value=value):
r = self.post({'keep_skipped': False, name: value} if name != 'keep_skipped' else {name: value})
self.assertEqual(r.status_code, 400)
self.assertIn(name, r.json()['error'])
self.assertEqual(r.json()['field'], name)
self.assertEqual(self.settings(), {**PLAYBACK_DEFAULTS, 'skip_back': 15})
self.assertEqual(self.post([1]).status_code, 400)
self.assertEqual(self.post({'initial': [1]}).status_code, 400)
def test_initial_sets_only_what_is_unset(self):
"""An app hands over its own values once: they only fill what the account doesn't have yet."""
self.post({'skip_forward': 45})
r = self.post({'initial': {'skip_back': 10, 'skip_forward': 30, 'smart_played': 60,
'keep_skipped': False}}) # keep_skipped isn't one that starts unset
self.assertEqual(r.status_code, 200, r.content)
self.assertEqual({k: r.json()[k] for k in ('skip_back', 'skip_forward', 'smart_played', 'keep_skipped')},
{'skip_back': 10, 'skip_forward': 45, 'smart_played': 60, 'keep_skipped': True})
# A second device comes too late: nothing changes.
r = self.post({'initial': {'skip_back': 5, 'smart_played': 0, 'speed_presets': [1, 2]}})
self.assertEqual({k: r.json()[k] for k in ('skip_back', 'smart_played', 'speed_presets')},
{'skip_back': 10, 'smart_played': 60, 'speed_presets': [1.0, 2.0]})
# Invalid initial values are refused like any other.
r = self.post({'initial': {'skip_back': 7}})
self.assertEqual((r.status_code, r.json()['field']), (400, 'skip_back'))
# Initial and a change at once: the change wins.
PodcastSettings.objects.filter(user=self.user).update(smart_played=None)
r = self.post({'smart_played': 120, 'initial': {'smart_played': 30}})
self.assertEqual(r.json()['smart_played'], 120)
def test_login(self):
self.client.logout()
self.assertEqual(self.post({'skip_back': 10}).status_code, 401)
self.assertEqual(self.client.get('/podcasts/settings/').status_code, 401)
class DiscardTests(TestCase):
"""POST /podcasts/episodes/discard/ ("Folge löschen"): out of the queue
and "Neu", the downloads go on every device (state/ "discarded"), but
not played: the place and the star stay."""
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')
base = datetime(2026, 1, 1, tzinfo=timezone.utc)
self.eps = [
PodcastEpisode.objects.create(feed=feed, guid=f'e{i}', title=f'E{i}', duration_seconds=3600,
audio_url=f'https://example.org/e{i}.mp3', pub_date=base + timedelta(days=i))
for i in range(5)
]
bob = User.objects.create_user(username='bob', password='pw12345678')
fbob = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/feed', title='Feed')
self.bob = bob
self.bobs = PodcastEpisode.objects.create(feed=fbob, guid='x', title='X', audio_url='https://example.org/x.mp3')
PodcastQueue.objects.create(user=bob, episode=self.bobs, position=0)
def post(self, body):
return self.client.post('/podcasts/episodes/discard/', json.dumps(body), content_type='application/json')
def queue(self):
return list(PodcastQueue.objects.filter(user=self.user).order_by('position', 'added_at', 'id')
.values_list('episode_id', flat=True))
def state(self):
return self.client.get('/podcasts/state/').json()
def test_out_of_queue_and_inbox_not_played(self):
e0, e1, e2, e3, e4 = self.eps
for i, e in enumerate((e0, e1, e2)):
PodcastQueue.objects.create(user=self.user, episode=e, position=i)
p1 = _progress(self.user, e1, _at(2, 1), position_seconds=754, favorite=True)
self.assertIn(e3.id, [x['id'] for x in self.state()['inbox']])
r = self.post({'episode_ids': [e1.id, e3.id]})
self.assertEqual(r.status_code, 200)
d = r.json()
self.assertEqual((d['ok'], d['count']), (True, 2))
# The rest of the queue keeps its order.
self.assertEqual(self.queue(), [e0.id, e2.id])
# The place, the star and the listening time stay; not played.
p1.refresh_from_db()
self.assertEqual((p1.position_seconds, p1.favorite, p1.played, p1.dismissed), (754, True, False, True))
self.assertEqual(p1.updated_at, _at(2, 1))
self.assertIsNotNone(p1.discarded_at)
# An episode without a row gets one: out of "Neu", not played, no place.
p3 = EpisodeProgress.objects.get(user=self.user, episode=e3)
self.assertEqual((p3.played, p3.dismissed, p3.position_seconds), (False, True, 0))
s = self.state()
inbox = [x['id'] for x in s['inbox']]
self.assertNotIn(e3.id, inbox)
self.assertIn(e4.id, inbox)
self.assertEqual([q['episode__id'] for q in s['queue']], [e0.id, e2.id])
# The state lists both for the other devices, with the time of the discard.
self.assertEqual({x['id'] for x in s['discarded']}, {e1.id, e3.id})
at = next(x['at'] for x in s['discarded'] if x['id'] == e1.id)
self.assertRegex(at, r'^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\.\d{3}Z$')
self.assertEqual(at, d['at'])
# Still unplayed in the episode lists, with its place.
unplayed = self.client.get('/podcasts/episodes/', {'filter': 'unplayed'}).json()['episodes']
row = next(x for x in unplayed if x['id'] == e1.id)
self.assertEqual((row['position_seconds'], row['played'], row['in_queue']), (754, False, False))
def test_queued_again_works_and_clears_it(self):
e0 = self.eps[0]
PodcastQueue.objects.create(user=self.user, episode=e0, position=0)
self.post({'episode_ids': [e0.id]})
self.assertEqual(self.queue(), [])
r = self.client.post('/podcasts/queue/add/', json.dumps({'episode_ids': [e0.id]}), content_type='application/json')
self.assertEqual(r.json()['added'], [e0.id])
self.assertEqual(self.queue(), [e0.id])
self.assertIsNone(EpisodeProgress.objects.get(user=self.user, episode=e0).discarded_at)
self.assertEqual(self.state()['discarded'], [])
# Playing it saves its place as ever.
self.client.post('/podcasts/progress/save/', json.dumps({'episode_id': e0.id, 'position_seconds': 99}),
content_type='application/json')
self.assertEqual(EpisodeProgress.objects.get(user=self.user, episode=e0).position_seconds, 99)
def test_kept_queue_order_stays_sorted(self):
PodcastSettings.objects.update_or_create(user=self.user, defaults={'keep_sorted': 'old'})
for e in self.eps:
self.client.post('/podcasts/queue/add/', json.dumps({'episode_ids': [e.id]}), content_type='application/json')
self.assertEqual(self.queue(), [e.id for e in self.eps])
self.post({'episode_ids': [self.eps[2].id]})
self.assertEqual(self.queue(), [self.eps[i].id for i in (0, 1, 3, 4)])
def test_old_discards_leave_the_state(self):
e0, e1 = self.eps[:2]
self.post({'episode_ids': [e0.id, e1.id]})
EpisodeProgress.objects.filter(user=self.user, episode=e0).update(
discarded_at=datetime.now(timezone.utc) - timedelta(days=31))
self.assertEqual([x['id'] for x in self.state()['discarded']], [e1.id])
def test_other_users_episodes_are_left_alone(self):
r = self.post({'episode_ids': [self.bobs.id, 999999]})
self.assertEqual(r.status_code, 200)
self.assertEqual(r.json()['count'], 0)
self.assertTrue(PodcastQueue.objects.filter(user=self.bob, episode=self.bobs).exists())
self.assertFalse(EpisodeProgress.objects.filter(episode=self.bobs).exists())
self.assertEqual(self.state()['discarded'], [])
def test_bad_requests(self):
for body in ({}, {'episode_ids': []}, {'episode_ids': self.eps[0].id}, {'episode_ids': ['x']},
{'episode_ids': [True]}, [1, 2]):
with self.subTest(body=body):
self.assertEqual(self.post(body).status_code, 400)
self.assertEqual(self.client.get('/podcasts/episodes/discard/').status_code, 405)
self.client.logout()
self.assertEqual(self.post({'episode_ids': [self.eps[0].id]}).status_code, 401)
# ---------------------------------------------------------------------------
# Skip silence in the browser: pauses found on the server (silence.py)
# ---------------------------------------------------------------------------
class SilenceTests(TestCase):
"""The rules are diora-android's PauseCutter: what Android cuts, the browser runs through."""
@staticmethod
def speech(seconds, db=-22.0):
return [db + (0.5 if i % 2 else 0.0) for i in range(int(seconds * 100))]
@staticmethod
def quiet(seconds, db=-75.0):
return [db] * int(seconds * 100)
def test_a_pause_keeps_a_fifth_half_before_half_after(self):
from .silence import cuts
# 2 s of speech (the noise floor needs some history), a 0.5 s pause, speech.
levels = self.speech(12) + self.quiet(0.5) + self.speech(2)
self.assertEqual(cuts(levels), [[12.05, 12.45]]) # keeps 100 ms: 50 + 50
def test_short_gaps_between_words_stay(self):
from .silence import cuts
self.assertEqual(cuts(self.speech(12) + self.quiet(0.08) + self.speech(2)), [])
def test_a_long_pause_keeps_at_most_a_quarter_second(self):
from .silence import cuts
self.assertEqual(cuts(self.speech(12) + self.quiet(3) + self.speech(2)), [[12.125, 14.875]])
def test_a_short_pause_keeps_at_least_60_ms(self):
from .silence import cuts
self.assertEqual(cuts(self.speech(12) + self.quiet(0.1) + self.speech(2)), [[12.03, 12.07]])
def test_a_quiet_second_voice_is_not_a_pause(self):
from .silence import cuts
# A talk with real pauses now and then (that's where the noise floor shows),
# then a voice well below the speaker but far above the floor: someone
# asking from the back. Only the real pauses are cut.
talk = (self.speech(1.8) + self.quiet(0.2)) * 6
levels = talk + self.quiet(2, db=-48.0) + self.speech(2)
found = cuts(levels)
self.assertEqual(len(found), 6)
self.assertTrue(all(end <= 12.0 for _, end in found))
def test_a_pause_running_to_the_end_stays(self):
from .silence import cuts
self.assertEqual(cuts(self.speech(12) + self.quiet(2)), [])
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.ep = PodcastEpisode.objects.create(feed=feed, guid='e', title='E', audio_url='https://example.org/e.mp3')
def test_first_request_starts_the_search_then_answers_with_the_pauses(self):
with patch('podcasts.silence.ensure') as ensure:
r = self.client.get(f'/podcasts/episodes/{self.ep.id}/silences/')
self.assertEqual(r.json(), {'ready': False})
ensure.assert_called_once_with(self.ep.id)
with patch('podcasts.silence.analyze', return_value=[[1.5, 2.4]]):
from .silence import _work
_work(self.ep.id)
r = self.client.get(f'/podcasts/episodes/{self.ep.id}/silences/')
self.assertEqual(r.json(), {'ready': True, 'v': 2, 'spans': [[1.5, 2.4]]})
def test_a_result_of_the_old_rules_is_found_anew(self):
PodcastEpisode.objects.filter(pk=self.ep.id).update(silences={'spans': [[1.0, 2.0]]})
with patch('podcasts.silence.ensure') as ensure:
r = self.client.get(f'/podcasts/episodes/{self.ep.id}/silences/')
self.assertEqual(r.json(), {'ready': False})
ensure.assert_called_once_with(self.ep.id)
def test_a_failure_is_remembered_not_retried(self):
from .silence import _work
with patch('podcasts.silence.analyze', side_effect=RuntimeError('HTTP error 404 Not Found')):
_work(self.ep.id)
with patch('podcasts.silence.ensure') as ensure:
r = self.client.get(f'/podcasts/episodes/{self.ep.id}/silences/')
ensure.assert_not_called()
self.assertEqual(r.json(), {'ready': True, 'v': 2, 'spans': [], 'error': 'HTTP error 404 Not Found'})
def test_only_own_episodes(self):
bob = User.objects.create_user(username='bob', password='pw12345678')
self.client.force_login(bob)
self.assertEqual(self.client.get(f'/podcasts/episodes/{self.ep.id}/silences/').status_code, 404)
self.client.logout()
self.assertEqual(self.client.get(f'/podcasts/episodes/{self.ep.id}/silences/').status_code, 401)
Reference in a new issue View git blame Copy permalink