"""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'])