194 lines
11 KiB
Python
194 lines
11 KiB
Python
"""The two lists, the subscription levels, the fetch interval and the Strom (docs/podcasts.md)."""
|
|
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
from unittest.mock import patch
|
|
|
|
from django.contrib.auth.models import User
|
|
from django.test import TestCase
|
|
from django.utils import timezone as dj_tz
|
|
|
|
from .models import EpisodeProgress, PodcastEpisode, PodcastFeed, PodcastLater, PodcastQueue, PodcastSettings
|
|
from .tests import _ok, _rss
|
|
from .views import _refresh_feed, refresh_due, refresh_hours
|
|
|
|
|
|
class Base(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(username='lists', password='pw12345678')
|
|
self.client.force_login(self.user)
|
|
self.feed = PodcastFeed.objects.create(
|
|
user=self.user, rss_url='https://example.org/feed', title='Feed',
|
|
last_refreshed_at=datetime(2026, 1, 1, tzinfo=timezone.utc))
|
|
self.eps = [PodcastEpisode.objects.create(feed=self.feed, guid=f'e{i}', title=f'E{i}', duration_seconds=600,
|
|
audio_url=f'https://example.org/e{i}.mp3',
|
|
pub_date=datetime(2026, 1, 1 + i, tzinfo=timezone.utc))
|
|
for i in range(4)]
|
|
|
|
def post(self, url, body=None):
|
|
return self.client.post(url, json.dumps(body or {}), content_type='application/json')
|
|
|
|
def state(self):
|
|
return self.client.get('/podcasts/state/').json()
|
|
|
|
def queue(self):
|
|
return list(PodcastQueue.objects.filter(user=self.user).order_by('position').values_list('episode_id', flat=True))
|
|
|
|
def later(self):
|
|
return set(PodcastLater.objects.filter(user=self.user).values_list('episode_id', flat=True))
|
|
|
|
|
|
class SwitchTests(Base):
|
|
def test_switching_on_makes_the_queue_the_merkliste(self):
|
|
for i, ep in enumerate(self.eps[:3]):
|
|
PodcastQueue.objects.create(user=self.user, episode=ep, position=i)
|
|
r = self.post('/podcasts/lists/', {'on': True}).json()
|
|
self.assertEqual((r['lists_v2'], r['queue'], r['later']), (True, 0, 3))
|
|
self.assertEqual(self.later(), {e.id for e in self.eps[:3]})
|
|
self.assertEqual(self.queue(), [])
|
|
st = self.state()
|
|
self.assertTrue(st['lists_v2'])
|
|
self.assertEqual([i['episode__id'] for i in st['later']], [self.eps[2].id, self.eps[1].id, self.eps[0].id])
|
|
# And back: the Merkliste joins the queue, oldest first.
|
|
self.post('/podcasts/lists/', {'on': False})
|
|
self.assertEqual(self.queue(), [e.id for e in self.eps[:3]])
|
|
self.assertEqual(self.later(), set())
|
|
|
|
def test_switching_twice_changes_nothing(self):
|
|
PodcastQueue.objects.create(user=self.user, episode=self.eps[0], position=0)
|
|
self.post('/podcasts/lists/', {'on': True})
|
|
PodcastQueue.objects.create(user=self.user, episode=self.eps[1], position=0)
|
|
self.post('/podcasts/lists/', {'on': True})
|
|
self.assertEqual(self.queue(), [self.eps[1].id])
|
|
self.assertEqual(self.later(), {self.eps[0].id})
|
|
|
|
|
|
class MerklisteTests(Base):
|
|
def test_add_remove_and_both_lists(self):
|
|
r = self.post('/podcasts/later/add/', {'episode_ids': [self.eps[0].id, self.eps[1].id, 999999]}).json()
|
|
self.assertEqual(r['added'], [self.eps[0].id, self.eps[1].id])
|
|
self.post('/podcasts/queue/add/', {'episode_ids': [self.eps[0].id], 'where': 'front'})
|
|
rows = {i['episode__id']: i for i in self.state()['later']}
|
|
self.assertTrue(rows[self.eps[0].id]['in_queue'])
|
|
self.assertFalse(rows[self.eps[1].id]['in_queue'])
|
|
self.post('/podcasts/later/remove/', {'episode_ids': [self.eps[1].id]})
|
|
self.assertEqual(self.later(), {self.eps[0].id})
|
|
self.assertEqual(self.queue(), [self.eps[0].id], 'the Wiedergabeliste is left alone')
|
|
|
|
def test_heard_leaves_the_merkliste(self):
|
|
self.post('/podcasts/later/add/', {'episode_ids': [e.id for e in self.eps[:3]]})
|
|
self.post('/podcasts/progress/save/', {'episode_id': self.eps[0].id, 'position_seconds': 590})
|
|
self.post('/podcasts/progress/mark-played/', {'episode_id': self.eps[1].id})
|
|
self.post('/podcasts/progress/save/', {'episode_id': self.eps[2].id, 'position_seconds': 100})
|
|
self.assertEqual(self.later(), {self.eps[2].id}, 'started stays, heard goes')
|
|
|
|
def test_audiobooks_go_on_neither_list(self):
|
|
book = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/book', title='Buch',
|
|
kind=PodcastFeed.AUDIOBOOK)
|
|
ch = PodcastEpisode.objects.create(feed=book, guid='c1', title='Kapitel 1', audio_url='https://example.org/c1.mp3')
|
|
self.post('/podcasts/queue/add/', {'episode_ids': [ch.id, self.eps[0].id]})
|
|
self.post('/podcasts/later/add/', {'episode_ids': [ch.id]})
|
|
self.assertEqual((self.queue(), self.later()), ([self.eps[0].id], set()))
|
|
|
|
def test_discard_leaves_both_lists(self):
|
|
self.post('/podcasts/later/add/', {'episode_ids': [self.eps[0].id]})
|
|
self.post('/podcasts/queue/add/', {'episode_ids': [self.eps[0].id]})
|
|
self.post('/podcasts/episodes/discard/', {'episode_ids': [self.eps[0].id]})
|
|
self.assertEqual((self.later(), self.queue()), (set(), []))
|
|
|
|
def test_episode_lists_tell_and_filter(self):
|
|
self.post('/podcasts/later/add/', {'episode_ids': [self.eps[3].id]})
|
|
r = self.client.get('/podcasts/episodes/?filter=later').json()
|
|
self.assertEqual([e['id'] for e in r['episodes']], [self.eps[3].id])
|
|
self.assertTrue(r['episodes'][0]['in_later'])
|
|
self.assertIn('discovered_at', r['episodes'][0])
|
|
|
|
|
|
class LevelTests(Base):
|
|
"""New episodes by level: Stamm into the Merkliste (or the Wiedergabeliste), Strom
|
|
nowhere, and the old one queue without lists_v2."""
|
|
|
|
RSS = _rss([('n1', 'Mon, 05 Jan 2026 10:00:00 GMT'), ('n2', 'Tue, 06 Jan 2026 10:00:00 GMT')])
|
|
|
|
def refresh(self, level, v2=True):
|
|
PodcastSettings.objects.update_or_create(user=self.user, defaults={'lists_v2': v2})
|
|
self.feed.new_episodes = level
|
|
self.feed.save()
|
|
with patch('podcasts.views._fetch_feed', lambda url, **kw: _ok(self.RSS)):
|
|
_refresh_feed(self.feed)
|
|
guid = dict(PodcastEpisode.objects.filter(feed=self.feed).values_list('id', 'guid'))
|
|
return ([guid[i] for i in self.queue()], sorted(guid[i] for i in self.later()))
|
|
|
|
def test_stamm_goes_to_the_merkliste_marked_auto(self):
|
|
self.assertEqual(self.refresh('queue'), ([], ['n1', 'n2']))
|
|
self.assertTrue(all(PodcastLater.objects.filter(user=self.user).values_list('auto', flat=True)))
|
|
|
|
def test_stamm_play_goes_to_the_wiedergabeliste(self):
|
|
self.assertEqual(self.refresh('play'), (['n2', 'n1'], []))
|
|
self.assertTrue(PodcastFeed.objects.get(pk=self.feed.pk).auto_queue)
|
|
|
|
def test_strom_and_auf_abruf_go_nowhere(self):
|
|
self.assertEqual(self.refresh('inbox'), ([], []))
|
|
|
|
def test_without_lists_v2_stamm_queues_as_before(self):
|
|
self.assertEqual(self.refresh('queue', v2=False), (['n2', 'n1'], []))
|
|
|
|
def test_the_strom_leaves_out_auf_abruf(self):
|
|
other = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/o', title='O', new_episodes='ignore')
|
|
PodcastEpisode.objects.create(feed=other, guid='x', title='X', audio_url='https://example.org/x.mp3')
|
|
r = self.client.get('/podcasts/episodes/?filter=stream&limit=50').json()
|
|
self.assertNotIn('X', [e['title'] for e in r['episodes']])
|
|
self.assertEqual(r['total'], 4)
|
|
|
|
|
|
class IntervalTests(Base):
|
|
def test_fixed_and_by_rhythm(self):
|
|
self.assertEqual(refresh_hours({'refresh_interval': 'daily'}, 0.5), 24)
|
|
self.assertEqual(refresh_hours({'refresh_interval': 'weekly'}, 0.5), 168)
|
|
self.assertEqual(refresh_hours({'refresh_interval': ''}, 0.1), 0.25)
|
|
self.assertEqual(refresh_hours({'refresh_interval': ''}, 7), 7)
|
|
self.assertEqual(refresh_hours({'refresh_interval': ''}, 90), 24)
|
|
self.assertEqual(refresh_hours({'refresh_interval': ''}, None), 1)
|
|
|
|
def test_due(self):
|
|
now = dj_tz.now()
|
|
daily = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/d', title='D',
|
|
refresh_interval='daily', last_refreshed_at=now - timedelta(hours=5))
|
|
stale = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/s', title='S',
|
|
refresh_interval='daily', last_refreshed_at=now - timedelta(hours=25))
|
|
new = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/n', title='N')
|
|
due = refresh_due(list(PodcastFeed.objects.filter(pk__in=[daily.pk, stale.pk, new.pk])), now)
|
|
self.assertEqual(set(due), {stale.pk, new.pk})
|
|
|
|
def test_favourite_subscriptions(self):
|
|
r = self.post(f'/podcasts/feeds/{self.feed.pk}/settings/', {'favorite': True}).json()
|
|
self.assertTrue(r['favorite'])
|
|
other = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/o', title='O')
|
|
PodcastEpisode.objects.create(feed=other, guid='o', title='O1', audio_url='https://example.org/o.mp3')
|
|
got = self.client.get('/podcasts/episodes/?filter=fave_feeds&limit=50').json()
|
|
self.assertEqual({e['feed__id'] for e in got['episodes']}, {self.feed.pk})
|
|
self.assertTrue([f for f in self.state()['feeds'] if f['id'] == self.feed.pk][0]['favorite'])
|
|
self.assertEqual(PodcastLater.objects.filter(user=self.user).count(), 0, 'a favourite queues nothing')
|
|
|
|
def test_setting_it(self):
|
|
r = self.post(f'/podcasts/feeds/{self.feed.pk}/settings/', {'refresh_interval': 'weekly'}).json()
|
|
self.assertEqual(r['refresh_interval'], 'weekly')
|
|
self.assertEqual(self.post(f'/podcasts/feeds/{self.feed.pk}/settings/', {'refresh_interval': 'often'}).status_code, 400)
|
|
|
|
|
|
class StromTests(Base):
|
|
def test_heard_in_twelve_months_per_feed(self):
|
|
EpisodeProgress.objects.create(user=self.user, episode=self.eps[0], played=True)
|
|
EpisodeProgress.objects.create(user=self.user, episode=self.eps[1], position_seconds=120)
|
|
EpisodeProgress.objects.create(user=self.user, episode=self.eps[2], position_seconds=10) # just a taste
|
|
old = EpisodeProgress.objects.create(user=self.user, episode=self.eps[3], played=True)
|
|
EpisodeProgress.objects.filter(pk=old.pk).update(updated_at=dj_tz.now() - timedelta(days=400))
|
|
feed = [f for f in self.state()['feeds'] if f['id'] == self.feed.id][0]
|
|
self.assertEqual(feed['heard_365'], 2)
|
|
|
|
def test_seen(self):
|
|
self.assertIsNone(self.state()['stream_seen_at'])
|
|
r = self.post('/podcasts/stream/seen/').json()
|
|
self.assertIsNone(r['before'])
|
|
r2 = self.post('/podcasts/stream/seen/').json()
|
|
self.assertEqual(r2['before'], r['now'])
|
|
self.assertEqual(self.state()['stream_seen_at'], r2['now'])
|