mrwnslz
a1e17081c5
Beim ersten Abruf eines Feeds (Abonnieren, OPML-Import, gpodder) kam der ganze Rückkatalog als „neu“ in die Inbox (bei Lage der Nation 200 Folgen). Jetzt zählt nur die neueste; die älteren sind ausgeblendet, nicht gehört, und stehen weiter in der Folgenliste des Podcasts. Auch automatisches Einreihen nimmt beim ersten Abruf nur diese eine. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
230 lines
11 KiB
Python
230 lines
11 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 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,
|
|
last_refreshed_at=datetime(2026, 1, 1, tzinfo=timezone.utc), # not the first fetch
|
|
)
|
|
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 FirstFetchTests(TestCase):
|
|
"""A new subscription doesn't flood the inbox with its back catalogue."""
|
|
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(username='alice', password='pw12345678')
|
|
self.client.force_login(self.user)
|
|
self.rss = _rss([(f'e{i}', f'Mon, {i + 1:02d} Jan 2026 10:00:00 GMT') for i in range(5)])
|
|
|
|
def subscribe(self):
|
|
with patch('podcasts.views._fetch_feed', lambda url: _parse(self.rss)):
|
|
return self.client.post('/podcasts/feeds/add/', json.dumps({'rss_url': 'https://example.org/f'}),
|
|
content_type='application/json')
|
|
|
|
def test_only_the_newest_is_new(self):
|
|
self.assertEqual(self.subscribe().json()['new_episodes'], 5)
|
|
inbox = self.client.get('/podcasts/inbox/').json()['episodes']
|
|
self.assertEqual([e['title'] for e in inbox], ['e4'])
|
|
# The others are still listed for the podcast, not marked played.
|
|
feed = PodcastFeed.objects.get(user=self.user)
|
|
eps = self.client.get(f'/podcasts/feeds/{feed.id}/episodes/').json()['episodes']
|
|
self.assertEqual(len(eps), 5)
|
|
self.assertFalse(any(e['played'] for e in eps))
|
|
|
|
def test_later_fetches_keep_every_new_episode(self):
|
|
self.subscribe()
|
|
feed = PodcastFeed.objects.get(user=self.user)
|
|
self.rss = _rss([(f'e{i}', f'Mon, {i + 1:02d} Jan 2026 10:00:00 GMT') for i in range(8)])
|
|
with patch('podcasts.views._fetch_feed', lambda url: _parse(self.rss)):
|
|
_refresh_feed(feed)
|
|
inbox = self.client.get('/podcasts/inbox/').json()['episodes']
|
|
self.assertEqual([e['title'] for e in inbox], ['e7', 'e6', 'e5', 'e4'])
|
|
|
|
|
|
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'})
|