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.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
Hallo Link'
''
'böseauch'
'
'
'
Hallo Link'
'böseauch'
'
Text bleibtoffen
Kapitel:
00:00 Intro
(12:34) - Thema eins
1:02:03 Zum Schluss
Hallo
01:02 Thema x
Hallo
01:02 Thema x
Heute mit dem Gast Bonusmaterial
') # 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"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', ' Hallo