mrwnslz
12b651b772
- Jedes Kapitel hat ✓/○ zum Umschalten. Ein bis zum Ende gehörtes, das wieder als nicht gehört gilt, beginnt von vorn (mark-played setzt die Stelle zurück). - Feeds, die Titel, Datum oder Nummer schon bekannter Folgen korrigieren, werden übernommen: davon hängt die Kapitelreihenfolge ab (Das Kapital, Erster Band, hatte Daten aus einer älteren Fassung des Feeds). Ein Aktualisieren von Hand und das Markieren als Hörbuch lesen den Feed dafür ganz. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
189 lines
12 KiB
Python
189 lines
12 KiB
Python
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 EpisodeBookmark, EpisodeProgress, PodcastEpisode, PodcastFeed, PodcastQueue
|
|
from .views import FeedFetch, _refresh_feed
|
|
|
|
T0 = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
|
|
|
|
|
class AudiobookTests(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/book', title='Das Buch',
|
|
author='A. Autor', last_refreshed_at=T0)
|
|
# Out of order on purpose: the same date, chapter numbers only in the titles.
|
|
self.eps = {}
|
|
for title, dur in [('Kapitel 10', 600), ('Kapitel 2', 300), ('Kapitel 1', 100), ('Vorwort', 0)]:
|
|
self.eps[title] = PodcastEpisode.objects.create(
|
|
feed=self.feed, guid=title, title=title, audio_url=f'https://example.org/{title}.mp3',
|
|
duration_seconds=dur, pub_date=T0 - (timedelta(days=1) if title == 'Vorwort' else timedelta(0)))
|
|
|
|
def post(self, url, body):
|
|
return self.client.post(url, json.dumps(body), content_type='application/json')
|
|
|
|
def mark(self):
|
|
r = self.post(f'/podcasts/feeds/{self.feed.pk}/settings/', {'kind': 'audiobook'})
|
|
self.assertEqual(r.status_code, 200, r.content)
|
|
self.assertEqual(r.json()['kind'], 'audiobook')
|
|
|
|
def test_marking_takes_its_episodes_out_of_neu_without_playing_them(self):
|
|
self.mark()
|
|
self.assertIsNone(self.client.get('/podcasts/audiobooks/').json()['books'][0]['last_played_at'])
|
|
self.assertEqual(EpisodeProgress.objects.filter(episode__feed=self.feed, dismissed=True).count(), 4)
|
|
self.assertFalse(EpisodeProgress.objects.filter(played=True).exists())
|
|
|
|
def test_an_unknown_kind_is_refused(self):
|
|
r = self.post(f'/podcasts/feeds/{self.feed.pk}/settings/', {'kind': 'radio'})
|
|
self.assertEqual(r.status_code, 400)
|
|
|
|
def test_chapters_in_order_and_one_progress_for_the_book(self):
|
|
self.mark()
|
|
EpisodeProgress.objects.filter(episode=self.eps['Vorwort']).update(played=True)
|
|
EpisodeProgress.objects.filter(episode=self.eps['Kapitel 1']).update(played=True)
|
|
EpisodeProgress.objects.filter(episode=self.eps['Kapitel 2']).update(position_seconds=120)
|
|
d = self.client.get(f'/podcasts/audiobooks/{self.feed.pk}/').json()
|
|
self.assertEqual([c['episode__title'] for c in d['chapters']], ['Vorwort', 'Kapitel 1', 'Kapitel 2', 'Kapitel 10'])
|
|
self.assertEqual([c['index'] for c in d['chapters']], [0, 1, 2, 3])
|
|
book = d['book']
|
|
self.assertEqual((book['chapters'], book['done'], book['total_seconds'], book['heard_seconds']), (4, 2, 1000, 220))
|
|
self.assertEqual(book['next'], {'episode_id': self.eps['Kapitel 2'].pk, 'index': 2, 'position_seconds': 120})
|
|
listed = self.client.get('/podcasts/audiobooks/').json()['books']
|
|
self.assertEqual([b['feed_id'] for b in listed], [self.feed.pk])
|
|
|
|
def test_it_goes_on_where_it_was_listened_to_last_not_at_a_skipped_chapter(self):
|
|
self.mark()
|
|
# Started at Kapitel 1 (the foreword skipped), finished it, now in Kapitel 2.
|
|
EpisodeProgress.objects.filter(episode=self.eps['Kapitel 1']).update(played=True, position_seconds=100, updated_at=T0 + timedelta(hours=1))
|
|
EpisodeProgress.objects.filter(episode=self.eps['Kapitel 2']).update(position_seconds=30, updated_at=T0 + timedelta(hours=2))
|
|
nxt = self.client.get(f'/podcasts/audiobooks/{self.feed.pk}/').json()['book']['next']
|
|
self.assertEqual(nxt['episode_id'], self.eps['Kapitel 2'].pk)
|
|
# Kapitel 2 finished as well (last): on with Kapitel 10, not back to the foreword.
|
|
EpisodeProgress.objects.filter(episode=self.eps['Kapitel 2']).update(played=True, updated_at=T0 + timedelta(hours=3))
|
|
nxt = self.client.get(f'/podcasts/audiobooks/{self.feed.pk}/').json()['book']['next']
|
|
self.assertEqual(nxt['episode_id'], self.eps['Kapitel 10'].pk)
|
|
# The end reached with the foreword still open: that one.
|
|
EpisodeProgress.objects.filter(episode=self.eps['Kapitel 10']).update(played=True, updated_at=T0 + timedelta(hours=4))
|
|
nxt = self.client.get(f'/podcasts/audiobooks/{self.feed.pk}/').json()['book']['next']
|
|
self.assertEqual(nxt['episode_id'], self.eps['Vorwort'].pk)
|
|
|
|
def test_a_podcast_is_no_audiobook(self):
|
|
self.assertEqual(self.client.get(f'/podcasts/audiobooks/{self.feed.pk}/').status_code, 404)
|
|
self.assertEqual(self.client.get('/podcasts/audiobooks/').json()['books'], [])
|
|
|
|
def test_a_finished_book_has_no_next(self):
|
|
self.mark()
|
|
EpisodeProgress.objects.filter(episode__feed=self.feed).update(played=True)
|
|
self.assertIsNone(self.client.get('/podcasts/audiobooks/').json()['books'][0]['next'])
|
|
|
|
def test_audiobooks_stay_out_of_folgen(self):
|
|
self.mark()
|
|
self.assertEqual(self.client.get('/podcasts/episodes/').json()['episodes'], [])
|
|
d = self.client.get('/podcasts/episodes/', {'feed_id': self.feed.pk}).json()
|
|
self.assertEqual(len(d['episodes']), 4) # in its own list they are
|
|
|
|
def test_new_chapters_never_go_to_neu_or_the_queue(self):
|
|
self.feed.new_episodes = PodcastFeed.NEW_QUEUE
|
|
self.feed.save()
|
|
self.mark()
|
|
rss = ('<?xml version="1.0"?><rss version="2.0"><channel><title>Das Buch</title>'
|
|
'<item><guid>neu</guid><title>Kapitel 11</title><pubDate>Mon, 05 Jan 2026 10:00:00 GMT</pubDate>'
|
|
'<enclosure url="https://example.org/11.mp3" type="audio/mpeg" length="1"/></item></channel></rss>')
|
|
self.feed.refresh_from_db()
|
|
with patch('podcasts.views._fetch_feed', lambda url, **kw: FeedFetch(200, feedparser.parse(rss))):
|
|
_refresh_feed(self.feed)
|
|
new = PodcastEpisode.objects.get(guid='neu')
|
|
self.assertFalse(PodcastQueue.objects.filter(episode=new).exists())
|
|
self.assertTrue(EpisodeProgress.objects.get(episode=new).dismissed)
|
|
|
|
def test_bookmarks(self):
|
|
self.mark()
|
|
ep = self.eps['Kapitel 2']
|
|
r = self.post('/podcasts/bookmarks/', {'episode_id': ep.pk, 'position_ms': 61500, 'note': ' Wichtig '})
|
|
self.assertEqual(r.status_code, 200, r.content)
|
|
bm = r.json()['bookmark']
|
|
self.assertEqual((bm['episode_id'], bm['position_ms'], bm['note']), (ep.pk, 61500, 'Wichtig'))
|
|
self.post('/podcasts/bookmarks/', {'episode_id': self.eps['Kapitel 1'].pk, 'position_ms': 5})
|
|
marks = self.client.get(f'/podcasts/audiobooks/{self.feed.pk}/').json()['bookmarks']
|
|
self.assertEqual([m['episode_id'] for m in marks], [self.eps['Kapitel 1'].pk, ep.pk]) # chapter order
|
|
r = self.post(f'/podcasts/bookmarks/{bm["id"]}/', {'note': 'Anders'})
|
|
self.assertEqual(r.json()['bookmark']['note'], 'Anders')
|
|
self.post(f'/podcasts/bookmarks/{bm["id"]}/delete/', {})
|
|
self.assertEqual(EpisodeBookmark.objects.count(), 1)
|
|
|
|
def test_bookmarks_are_ones_own(self):
|
|
self.mark()
|
|
bob = User.objects.create_user(username='bob', password='pw12345678')
|
|
mine = EpisodeBookmark.objects.create(user=self.user, episode=self.eps['Kapitel 1'], position_ms=1)
|
|
self.client.force_login(bob)
|
|
self.assertEqual(self.post('/podcasts/bookmarks/', {'episode_id': self.eps['Kapitel 1'].pk, 'position_ms': 1}).status_code, 404)
|
|
self.assertEqual(self.post(f'/podcasts/bookmarks/{mine.pk}/', {'note': 'x'}).status_code, 404)
|
|
self.post(f'/podcasts/bookmarks/{mine.pk}/delete/', {})
|
|
self.assertTrue(EpisodeBookmark.objects.filter(pk=mine.pk).exists())
|
|
self.assertEqual(self.client.get(f'/podcasts/audiobooks/{self.feed.pk}/').status_code, 404)
|
|
|
|
def test_bad_bookmarks_are_refused(self):
|
|
ep = self.eps['Kapitel 1']
|
|
self.assertEqual(self.post('/podcasts/bookmarks/', {'episode_id': ep.pk, 'position_ms': -1}).status_code, 400)
|
|
self.assertEqual(self.post('/podcasts/bookmarks/', {'episode_id': ep.pk, 'position_ms': 'x'}).status_code, 400)
|
|
self.assertEqual(self.post('/podcasts/bookmarks/', {'episode_id': ep.pk, 'position_ms': 1, 'note': 5}).status_code, 400)
|
|
|
|
|
|
class FeedCorrectionTests(TestCase):
|
|
"""A feed that corrects titles and dates of episodes it had already (an audiobook's
|
|
order depends on them): a refresh takes them over, a refresh by hand even when the
|
|
feed looks unchanged."""
|
|
|
|
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/book', title='Buch',
|
|
last_refreshed_at=T0, http_body_hash='same')
|
|
PodcastEpisode.objects.create(feed=self.feed, guid='a', title='010 - Erstes Kapitel', pub_date=T0,
|
|
audio_url='https://example.org/a.mp3')
|
|
|
|
RSS = ('<?xml version="1.0"?><rss version="2.0"><channel><title>Buch</title>'
|
|
'<item><guid>a</guid><title>Erstes Kapitel</title><pubDate>Wed, 14 Oct 2026 21:45:54 GMT</pubDate>'
|
|
'<enclosure url="https://example.org/a.mp3" type="audio/mpeg" length="1"/></item></channel></rss>')
|
|
|
|
def test_a_refresh_by_hand_takes_over_corrected_titles_and_dates(self):
|
|
seen = {}
|
|
|
|
def fetch(url, **kw):
|
|
seen.update(kw)
|
|
if kw.get('body_hash') == 'same': # what an unchanged feed gets: nothing parsed
|
|
return FeedFetch(304, None, body_hash='same')
|
|
return FeedFetch(200, feedparser.parse(self.RSS), body_hash='same')
|
|
with patch('podcasts.views._fetch_feed', fetch):
|
|
r = self.client.post('/podcasts/feeds/refresh/', json.dumps({'feed_id': self.feed.pk}),
|
|
content_type='application/json')
|
|
self.assertEqual(r.status_code, 200, r.content)
|
|
self.assertEqual(seen.get('body_hash', ''), '')
|
|
ep = PodcastEpisode.objects.get(guid='a')
|
|
self.assertEqual(ep.title, 'Erstes Kapitel')
|
|
self.assertEqual(ep.pub_date, datetime(2026, 10, 14, 21, 45, 54, tzinfo=timezone.utc))
|
|
self.assertEqual(PodcastEpisode.objects.count(), 1)
|
|
|
|
|
|
class UnplayedTests(TestCase):
|
|
def test_a_chapter_heard_to_the_end_starts_over_once_not_heard(self):
|
|
user = User.objects.create_user(username='alice', password='pw12345678')
|
|
self.client.force_login(user)
|
|
feed = PodcastFeed.objects.create(user=user, 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',
|
|
duration_seconds=600)
|
|
half = PodcastEpisode.objects.create(feed=feed, guid='y', title='y', audio_url='https://example.org/y.mp3',
|
|
duration_seconds=600)
|
|
EpisodeProgress.objects.create(user=user, episode=ep, position_seconds=598, played=True)
|
|
EpisodeProgress.objects.create(user=user, episode=half, position_seconds=200, played=True)
|
|
for e in (ep, half):
|
|
self.client.post('/podcasts/progress/mark-played/', json.dumps({'episode_id': e.pk, 'played': False}),
|
|
content_type='application/json')
|
|
self.assertEqual(EpisodeProgress.objects.get(episode=ep).position_seconds, 0)
|
|
self.assertEqual(EpisodeProgress.objects.get(episode=half).position_seconds, 200) # its place stays
|