diora-web/podcasts/test_audiobooks.py

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