import json from unittest.mock import patch import feedparser from django.contrib.auth.models import User from django.test import TestCase from . import youtube from .models import PodcastEpisode, PodcastFeed from .views import FeedFetch, _load_chapters CHANNEL = 'UC5NOEUbkLheQcaaRldYW5GA' FEATURED = 'UC1_uAIS3r8Vu6JjXWvastJg' # another channel the page shows PAGE = ( f'' '' f'' f'' ) def _entry(vid, title, kind='watch?v=', when='2026-10-14T10:00:00+00:00', text='Hallo\nhttps://example.org/x\n\n00:00 Anfang'): link = f'https://www.youtube.com/{kind}{vid}' return ( f'yt:video:{vid}{vid}{CHANNEL}' f'{title}' f'Tages Schau{when}{when}' f'{title}' f'' f'' f'{text}' ) FEED_XML = ( '' '' f'yt:channel:{CHANNEL[2:]}' f'{CHANNEL[2:]}Tages Schau' f'' 'Tages Schau2010-01-01T00:00:00+00:00' + _entry('aaaaaaaaaaa', 'Video eins') + _entry('bbbbbbbbbbb', 'Ein Short', kind='shorts/') + _entry('ccccccccccc', 'Video zwei', when='2026-10-13T10:00:00+00:00') + '' ) class _Page: def __init__(self, text, status=200): self.status_code = status self.raw = self self._body = text.encode() def read(self, n, decode_content=True): return self._body[:n] def close(self): pass class ResolveTests(TestCase): def resolve(self, url, page=PAGE): seen = [] def get(u, **kw): seen.append(u) return _Page(page) with patch('podcasts.youtube.requests.get', get): return youtube.resolve(url), seen def test_a_channel_page_is_its_own_channel_not_one_it_features(self): page = (f'' f'') found, _ = self.resolve('https://www.youtube.com/@3blue1brown', page) self.assertEqual(found['feed'], f'{youtube.FEED}?channel_id={CHANNEL}') def test_a_handle_leads_to_its_channel_feed(self): found, seen = self.resolve('https://www.youtube.com/@tagesschau/videos') self.assertEqual(found, {'feed': f'{youtube.FEED}?channel_id={CHANNEL}', 'title': 'Tages & Schau', 'artwork': 'https://yt3.example/pic.jpg'}) self.assertEqual(seen, ['https://www.youtube.com/@tagesschau/videos']) def test_a_video_leads_to_its_channel_without_the_videos_name(self): page = (f'') found, seen = self.resolve('https://youtu.be/aaaaaaaaaaa', page) self.assertEqual(found['feed'], f'{youtube.FEED}?channel_id={CHANNEL}') self.assertEqual((found['title'], found['artwork']), ('', '')) self.assertEqual(seen, [youtube.WATCH + 'aaaaaaaaaaa']) def test_a_playlist_has_its_own_feed(self): found, _ = self.resolve('https://www.youtube.com/playlist?list=PLabcdefghijk') self.assertEqual(found['feed'], f'{youtube.FEED}?playlist_id=PLabcdefghijk') def test_other_youtube_pages_are_no_channel(self): with self.assertRaises(youtube.NotAChannel): youtube.resolve('https://www.youtube.com/feed/subscriptions') def test_a_page_without_a_channel_is_none(self): with self.assertRaises(youtube.NotAChannel): self.resolve('https://www.youtube.com/@nobody', 'consent') class YouTubeFeedTests(TestCase): def setUp(self): self.user = User.objects.create_user(username='alice', password='pw12345678') self.client.force_login(self.user) def subscribe(self, url): pages = [] def get(u, **kw): pages.append(u) return _Page(PAGE) fetched = [] def fetch(u, **kw): fetched.append(u) return FeedFetch(200, feedparser.parse(FEED_XML)) with patch('podcasts.youtube.requests.get', get), patch('podcasts.views._fetch_feed', fetch): r = self.client.post('/podcasts/feeds/add/', json.dumps({'rss_url': url, 'title': url}), content_type='application/json') return r, pages, fetched def test_subscribing_to_a_channel_reads_its_feed_never_a_video(self): r, pages, fetched = self.subscribe('https://www.youtube.com/@tagesschau') self.assertEqual(r.status_code, 200, r.content) feed = PodcastFeed.objects.get() self.assertEqual(feed.rss_url, f'{youtube.FEED}?channel_id={CHANNEL}') self.assertEqual(feed.title, 'Tages Schau') self.assertEqual(feed.artwork_url, 'https://yt3.example/pic.jpg') # kept: the feed has none self.assertEqual(pages, ['https://www.youtube.com/@tagesschau']) self.assertEqual(fetched, [feed.rss_url]) eps = {e.title: e for e in PodcastEpisode.objects.all()} self.assertEqual(set(eps), {'Video eins', 'Video zwei'}) # no Short one = eps['Video eins'] self.assertEqual(one.audio_url, 'https://www.youtube.com/watch?v=aaaaaaaaaaa') self.assertEqual(one.guid, 'yt:video:aaaaaaaaaaa') self.assertEqual(one.artwork_url, 'https://i1.ytimg.com/vi/aaaaaaaaaaa/hqdefault.jpg') self.assertEqual(one.description, '

Hallo
https://example.org/x

00:00 Anfang

') def test_the_same_channel_twice_is_one_subscription(self): self.subscribe('https://www.youtube.com/@tagesschau') r, _, _ = self.subscribe(f'https://www.youtube.com/channel/{CHANNEL}') self.assertFalse(r.json()['created']) self.assertEqual(PodcastFeed.objects.count(), 1) def test_a_feed_subscribed_directly_gets_the_channels_picture(self): r, pages, _ = self.subscribe(f'{youtube.FEED}?channel_id={CHANNEL}') self.assertEqual(r.status_code, 200) self.assertEqual(PodcastFeed.objects.get().artwork_url, 'https://yt3.example/pic.jpg') self.assertEqual(pages, [f'https://www.youtube.com/channel/{CHANNEL}']) def test_no_channel_there_is_an_error(self): with patch('podcasts.youtube.requests.get', lambda u, **kw: _Page('')): r = self.client.post('/podcasts/feeds/add/', json.dumps({'rss_url': 'https://www.youtube.com/@x'}), content_type='application/json') self.assertEqual(r.status_code, 400) self.assertFalse(PodcastFeed.objects.exists()) def test_search_finds_a_pasted_channel(self): with patch('podcasts.youtube.requests.get', lambda u, **kw: _Page(PAGE)): r = self.client.get('/podcasts/search/', {'q': 'https://www.youtube.com/@tagesschau'}) self.assertEqual(r.json()['results'][0]['rss_url'], f'{youtube.FEED}?channel_id={CHANNEL}') self.assertEqual(r.json()['results'][0]['title'], 'Tages & Schau') def test_the_server_never_fetches_a_videos_sound(self): self.subscribe('https://www.youtube.com/@tagesschau') ep = PodcastEpisode.objects.get(title='Video eins') with patch('podcasts.silence.ensure') as ensure: d = self.client.get(f'/podcasts/episodes/{ep.pk}/silences/').json() ensure.assert_not_called() self.assertEqual((d['ready'], d['spans'], d['error']), (True, [], 'youtube')) with patch('podcasts.chapters.fetch_id3') as id3: self.assertEqual(_load_chapters(ep), []) id3.assert_not_called() def test_the_players_duration_is_kept_when_the_feed_has_none(self): self.subscribe('https://www.youtube.com/@tagesschau') ep = PodcastEpisode.objects.get(title='Video eins') save = lambda body: self.client.post('/podcasts/progress/save/', json.dumps({'episode_id': ep.pk, **body}), content_type='application/json') save({'position_seconds': 10, 'duration_seconds': 'x'}) save({'position_seconds': 10, 'duration_seconds': 10 ** 9}) ep.refresh_from_db() self.assertEqual(ep.duration_seconds, 0) save({'position_seconds': 10, 'duration_seconds': 929.3}) save({'position_seconds': 20, 'duration_seconds': 5}) # known now: not changed again ep.refresh_from_db() self.assertEqual(ep.duration_seconds, 929) class MoveFeedTests(TestCase): """A channel fed through another server (podsync-like) moved to the channel itself.""" OLD = 'http://creamfresh.example:8001/feed/abc/rss.xml' def setUp(self): from .models import EpisodeProgress, PodcastQueue 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=self.OLD, title='Tages Schau', playback_speed=1.2, http_etag='"x"') mk = lambda vid, title, guid=None: PodcastEpisode.objects.create( feed=self.feed, guid=guid or f'old-{vid}', title=title, duration_seconds=600, audio_url=f'http://creamfresh.example:8001/feed/abc/audio/{vid}.mp3') self.newest = mk('aaaaaaaaaaa', 'Video eins') self.older = mk('zzzzzzzzzzz', 'Altes Video') # no longer in the channel's feed self.trailer = mk('UCs8mbJ-M142ZskR5VR0gBig', 'Kanal - Videos') EpisodeProgress.objects.create(user=self.user, episode=self.newest, position_seconds=0, played=True) EpisodeProgress.objects.create(user=self.user, episode=self.older, position_seconds=300, favorite=True) PodcastQueue.objects.create(user=self.user, episode=self.older, position=0) def move(self, url): def fetch(u, **kw): self.assertEqual(kw.get('etag', ''), '') # the old address's validators are gone return FeedFetch(200, feedparser.parse(FEED_XML)) with patch('podcasts.youtube.requests.get', lambda u, **kw: _Page(PAGE)), \ patch('podcasts.views._fetch_feed', fetch): return self.client.post(f'/podcasts/feeds/{self.feed.pk}/move/', json.dumps({'rss_url': url}), content_type='application/json') def test_the_subscription_keeps_everything_and_its_episodes_become_videos(self): from .models import EpisodeProgress, PodcastQueue r = self.move('https://www.youtube.com/@tagesschau') self.assertEqual(r.status_code, 200, r.content) self.assertEqual(r.json()['videos'], 2) self.feed.refresh_from_db() self.assertEqual(self.feed.rss_url, f'{youtube.FEED}?channel_id={CHANNEL}') self.assertEqual(self.feed.playback_speed, 1.2) self.assertEqual(self.feed.artwork_url, 'https://yt3.example/pic.jpg') self.newest.refresh_from_db() self.older.refresh_from_db() self.trailer.refresh_from_db() self.assertEqual((self.newest.guid, self.newest.audio_url), ('yt:video:aaaaaaaaaaa', youtube.WATCH + 'aaaaaaaaaaa')) self.assertEqual(self.older.audio_url, youtube.WATCH + 'zzzzzzzzzzz') self.assertTrue(self.trailer.audio_url.endswith('.mp3')) # not a video: left as it was # Video eins came again with the channel's feed: the same episode, not a second one. self.assertEqual(PodcastEpisode.objects.filter(title='Video eins').count(), 1) self.assertEqual(set(PodcastEpisode.objects.values_list('title', flat=True)), {'Video eins', 'Video zwei', 'Altes Video', 'Kanal - Videos'}) self.assertTrue(EpisodeProgress.objects.get(episode=self.newest).played) p = EpisodeProgress.objects.get(episode=self.older) self.assertEqual((p.position_seconds, p.favorite), (300, True)) self.assertTrue(PodcastQueue.objects.filter(episode=self.older).exists()) def test_an_address_subscribed_already_is_refused(self): PodcastFeed.objects.create(user=self.user, rss_url=f'{youtube.FEED}?channel_id={CHANNEL}', title='x') self.assertEqual(self.move('https://www.youtube.com/@tagesschau').status_code, 409) self.feed.refresh_from_db() self.assertEqual(self.feed.rss_url, self.OLD) def test_only_ones_own_feed(self): bob = User.objects.create_user(username='bob', password='pw12345678') self.client.force_login(bob) self.assertEqual(self.move('https://www.youtube.com/@tagesschau').status_code, 404)