mrwnslz/diora-web
1
0
Fork 0
Code Issues 4 Pull requests Projects Releases Packages Wiki Activity Actions
diora-web/podcasts/tests.py
mrwnslz fd0a103fb2
Some checks failed
Build and push Docker image / build (push) Successful in 16s
Details
Test / test (push) Failing after 28s
Details
Podcasts: state/ ohne Shownotes in der Inbox 
Seit die Shownotes content:encoded nehmen (bis 50 KB pro Folge), würde
die eine Anfrage der Apps mit 200 Inbox-Folgen sehr groß. state/ lässt
sie weg; die Apps holen sie pro Folge über episodes/<id>/. inbox/ für
das Web liefert sie weiter.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-02 08:01:34 +02:00

537 lines
26 KiB
Python
Raw Blame History

import json
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
import feedparser
from django.contrib.auth.models import User
from django.test import TestCase
from .models import EpisodeProgress, PodcastEpisode, PodcastFeed, PodcastQueue, PodcastSettings
from .views import _refresh_feed
_parse = feedparser.parse
def _rss(items):
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"/></item>'
for guid, date in items
)
return f'<?xml version="1.0"?><rss version="2.0"><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.create(user=self.user, episode=self.eps[2], 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 AutoQueueTests(TestCase):
"""New episodes of an auto_queue feed land where the user's setting says."""
def setUp(self):
self.user = User.objects.create_user(username='alice', password='pw12345678')
self.feed = PodcastFeed.objects.create(
user=self.user, rss_url='https://example.org/feed', title='Feed', auto_queue=True,
last_refreshed_at=datetime(2026, 1, 1, tzinfo=timezone.utc), # not the first fetch
)
other = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/o', title='O')
self.old = [
PodcastEpisode.objects.create(feed=other, guid=f'o{i}', title=f'O{i}', audio_url=f'https://example.org/o{i}.mp3')
for i in range(3)
]
for i, ep in enumerate(self.old):
PodcastQueue.objects.create(user=self.user, episode=ep, position=i)
self.rss = _rss([
('new1', 'Mon, 05 Jan 2026 10:00:00 GMT'),
('new2', 'Tue, 06 Jan 2026 10:00:00 GMT'),
])
def refresh(self):
with patch('podcasts.views._fetch_feed', lambda url: _parse(self.rss)):
_refresh_feed(self.feed)
order = PodcastQueue.objects.filter(user=self.user).order_by('position')
return [q.episode.guid for q in order]
def test_front_is_the_default(self):
self.assertEqual(self.refresh(), ['new2', 'new1', 'o0', 'o1', 'o2'])
def test_back_appends_oldest_first(self):
PodcastSettings.objects.create(user=self.user, auto_queue_position='back')
self.assertEqual(self.refresh(), ['o0', 'o1', 'o2', 'new1', 'new2'])
def test_after_current(self):
PodcastSettings.objects.create(user=self.user, auto_queue_position='after_current')
EpisodeProgress.objects.create(user=self.user, episode=self.old[1], position_seconds=12)
self.assertEqual(self.refresh(), ['o0', 'o1', 'new2', 'new1', 'o2'])
def test_feed_without_auto_queue_queues_nothing(self):
self.feed.auto_queue = False
self.feed.save()
self.assertEqual(self.refresh(), ['o0', 'o1', 'o2'])
class FirstFetchTests(TestCase):
"""A new subscription doesn't flood the inbox with its back catalogue."""
def setUp(self):
self.user = User.objects.create_user(username='alice', password='pw12345678')
self.client.force_login(self.user)
self.rss = _rss([(f'e{i}', f'Mon, {i + 1:02d} Jan 2026 10:00:00 GMT') for i in range(5)])
def subscribe(self):
with patch('podcasts.views._fetch_feed', lambda url: _parse(self.rss)):
return self.client.post('/podcasts/feeds/add/', json.dumps({'rss_url': 'https://example.org/f'}),
content_type='application/json')
def test_only_the_newest_is_new(self):
self.assertEqual(self.subscribe().json()['new_episodes'], 5)
inbox = self.client.get('/podcasts/inbox/').json()['episodes']
self.assertEqual([e['title'] for e in inbox], ['e4'])
# The others are still listed for the podcast, not marked played.
feed = PodcastFeed.objects.get(user=self.user)
eps = self.client.get(f'/podcasts/feeds/{feed.id}/episodes/').json()['episodes']
self.assertEqual(len(eps), 5)
self.assertFalse(any(e['played'] for e in eps))
def test_later_fetches_keep_every_new_episode(self):
self.subscribe()
feed = PodcastFeed.objects.get(user=self.user)
self.rss = _rss([(f'e{i}', f'Mon, {i + 1:02d} Jan 2026 10:00:00 GMT') for i in range(8)])
with patch('podcasts.views._fetch_feed', lambda url: _parse(self.rss)):
_refresh_feed(feed)
inbox = self.client.get('/podcasts/inbox/').json()['episodes']
self.assertEqual([e['title'] for e in inbox], ['e7', 'e6', 'e5', 'e4'])
class RefreshAllTests(TestCase):
"""One request refreshes all feeds on the server; one request returns the state."""
def setUp(self):
self.user = User.objects.create_user(username='alice', password='pw12345678')
self.client.force_login(self.user)
self.a = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/a', title='A')
self.b = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/b', title='B')
bob = User.objects.create_user(username='bob', password='pw12345678')
self.bobs = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/a', title='A')
self.fetched = []
def fake_fetch(self, url):
self.fetched.append(url)
if url.endswith('/b'):
raise ValueError('broken feed')
return _parse(_rss([(url[-1] + '1', 'Mon, 05 Jan 2026 10:00:00 GMT')]))
def test_refresh_all_reports_counts_and_skips_recent(self):
with patch('podcasts.views._fetch_feed', self.fake_fetch):
r = self.client.post('/podcasts/feeds/refresh-all/', '{}', content_type='application/json')
self.assertEqual(r.json(), {'ok': True, 'refreshed': 1, 'failed': 1, 'new_episodes': 1})
self.assertEqual(sorted(self.fetched), ['https://example.org/a', 'https://example.org/b'])
# A was just refreshed: a second call within max_age only retries B.
self.fetched.clear()
with patch('podcasts.views._fetch_feed', self.fake_fetch):
self.client.post('/podcasts/feeds/refresh-all/', '{}', content_type='application/json')
self.assertEqual(self.fetched, ['https://example.org/b'])
def test_a_shared_feed_url_is_fetched_once(self):
from .views import _refresh_feeds
with patch('podcasts.views._fetch_feed', self.fake_fetch):
refreshed, failed, new = _refresh_feeds([self.a, self.bobs])
self.assertEqual(self.fetched, ['https://example.org/a'])
self.assertEqual((refreshed, failed, new), (2, 0, 2))
def test_state_in_one_response(self):
ep = PodcastEpisode.objects.create(feed=self.a, guid='x', title='X', audio_url='https://example.org/x.mp3')
PodcastQueue.objects.create(user=self.user, episode=ep, position=0)
r = self.client.get('/podcasts/state/')
data = r.json()
self.assertEqual({f['title'] for f in data['feeds']}, {'A', 'B'})
self.assertEqual([q['episode__id'] for q in data['queue']], [ep.id])
self.assertEqual([e['id'] for e in data['inbox']], [ep.id])
self.assertTrue(data['inbox'][0]['in_queue'])
# Shownotes stay out of the one-request state (they can be large).
self.assertNotIn('description', data['inbox'][0])
self.assertEqual(data['settings'], {'enqueue_position': 'back', 'auto_queue_position': 'front'})
# ---------------------------------------------------------------------------
# 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: _parse(rss)):
_refresh_feed(self.feed)
def test_feedparser_keys(self):
import podcasts.views # noqa: F401 -- installs the feedparser extension
rss = _rich_rss([('a', PSC + TRANSCRIPTS + '<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.assertEqual(data['feed'], {'id': self.feed.id, 'title': 'Feed',
'artwork_url': 'https://example.org/art.jpg'})
self.assertEqual((data['position_seconds'], data['played'], data['in_queue']), (42, False, False))
self.assertEqual(data['pub_date'], '2026-01-05T10:00:00+00:00')
self.assertEqual(data['chapters'], [
{'start': 0.0, 'title': 'Erstes'},
{'start': 95.5, 'title': 'Zweites', 'url': 'https://x.org/b', 'img': 'https://x.org/b.jpg'},
{'start': 120.0, 'title': 'Link'},
])
def test_chapters_are_fetched_once(self):
self.get()
self.get()
self.assertEqual(self.requests, ['https://x.org/c.json'])
self.ep.refresh_from_db()
self.assertEqual(len(self.ep.chapters), 3)
self.assertIsNotNone(self.ep.chapters_fetched_at)
def test_inline_chapters_need_no_fetch(self):
self.ep.chapters = [{'start': 0.0, 'title': 'Inline'}]
self.ep.save()
self.assertEqual(self.get().json()['chapters'], [{'start': 0.0, 'title': 'Inline'}])
self.assertEqual(self.requests, [])
def test_redirect_is_followed_and_rechecked(self):
def fake_get(url, **kwargs):
self.requests.append(url)
if url.endswith('c.json'):
return _FakeResponse(status=302, headers={'Location': '/real.json'})
return _FakeResponse(json.dumps(CHAPTERS_JSON).encode())
with patch('podcasts.chapters.assert_safe_url') as check, patch('podcasts.chapters.requests.get', fake_get):
r = self.client.get(f'/podcasts/episodes/{self.ep.id}/')
self.assertEqual(len(r.json()['chapters']), 3)
self.assertEqual([c.args[0] for c in check.call_args_list], ['https://x.org/c.json', 'https://x.org/real.json'])
def test_private_chapters_url_is_refused_and_remembered(self):
self.ep.chapters_url = 'http://127.0.0.1/c.json'
self.ep.save()
with patch('podcasts.chapters.requests.get', self.fake_get):
r = self.client.get(f'/podcasts/episodes/{self.ep.id}/')
self.assertEqual(r.json()['chapters'], [])
self.assertEqual(self.requests, [])
self.ep.refresh_from_db()
self.assertIsNotNone(self.ep.chapters_fetched_at)
def test_oversized_or_broken_chapters(self):
from . import chapters
with patch('podcasts.chapters.assert_safe_url'), patch(
'podcasts.chapters.requests.get',
lambda url, **kw: _FakeResponse(b'x' * (chapters.CHAPTERS_MAX_BYTES + 1)),
):
with self.assertRaises(chapters.ChaptersInvalid):
chapters.fetch('https://x.org/big.json')
with patch('podcasts.chapters.assert_safe_url'), patch(
'podcasts.chapters.requests.get', lambda url, **kw: _FakeResponse(b'{"chapters": 3}'),
):
with self.assertRaises(chapters.ChaptersInvalid):
chapters.fetch('https://x.org/odd.json')
def test_server_error_is_retried_later(self):
with patch('podcasts.chapters.assert_safe_url'), patch(
'podcasts.chapters.requests.get', lambda url, **kw: _FakeResponse(status=503),
):
r = self.client.get(f'/podcasts/episodes/{self.ep.id}/')
self.assertEqual(r.json()['chapters'], [])
self.ep.refresh_from_db()
self.assertIsNone(self.ep.chapters_fetched_at)
def test_other_users_episode_is_not_found(self):
bob = User.objects.create_user(username='bob', password='pw12345678')
feed = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/b', title='B')
ep = PodcastEpisode.objects.create(feed=feed, guid='x', title='X', audio_url='https://example.org/x.mp3',
chapters_url='https://x.org/c.json')
self.assertEqual(self.get(ep).status_code, 404)
self.assertEqual(self.requests, [])
def test_login_required(self):
self.client.logout()
self.assertEqual(self.get().status_code, 401)
Reference in a new issue View git blame Copy permalink