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 = ('Das Buch' 'neuKapitel 11Mon, 05 Jan 2026 10:00:00 GMT' '') 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 = ('Buch' 'aErstes KapitelWed, 14 Oct 2026 21:45:54 GMT' '') 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 class AudiobookFolderTests(TestCase): def test_a_book_has_a_folder(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', kind='audiobook') r = self.client.post(f'/podcasts/feeds/{feed.pk}/settings/', json.dumps({'folder': ' Marx und Engels '}), content_type='application/json') self.assertEqual(r.json()['folder'], 'Marx und Engels') self.assertEqual(self.client.get('/podcasts/audiobooks/').json()['books'][0]['folder'], 'Marx und Engels')