diora-web/podcasts/tests.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 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,
        )
        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 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'])
        self.assertEqual(data['settings'], {'enqueue_position': 'back', 'auto_queue_position': 'front'})