diora-web/podcasts/test_youtube.py

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'<html><script>{{"channelId":"{FEATURED}"}}</script><head><meta property="og:title" content="Tages &amp; Schau">'
    '<meta property="og:image" content="https://yt3.example/pic.jpg">'
    f'<link rel="alternate" type="application/rss+xml" title="RSS" href="{youtube.FEED}?channel_id={CHANNEL}">'
    f'</head><body><script>var x = {{"externalId":"{CHANNEL}"}};</script></body></html>'
)


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'<entry><id>yt:video:{vid}</id><yt:videoId>{vid}</yt:videoId><yt:channelId>{CHANNEL}</yt:channelId>'
        f'<title>{title}</title><link rel="alternate" href="{link}"/>'
        f'<author><name>Tages Schau</name></author><published>{when}</published><updated>{when}</updated>'
        f'<media:group><media:title>{title}</media:title>'
        f'<media:content url="https://www.youtube.com/v/{vid}?version=3" type="application/x-shockwave-flash" width="640" height="390"/>'
        f'<media:thumbnail url="https://i1.ytimg.com/vi/{vid}/hqdefault.jpg" width="480" height="360"/>'
        f'<media:description>{text}</media:description></media:group></entry>'
    )


FEED_XML = (
    '<?xml version="1.0" encoding="UTF-8"?>'
    '<feed xmlns:yt="http://www.youtube.com/xml/schemas/2015" xmlns:media="http://search.yahoo.com/mrss/" '
    'xmlns="http://www.w3.org/2005/Atom">'
    f'<link rel="self" href="{youtube.FEED}?channel_id={CHANNEL}"/><id>yt:channel:{CHANNEL[2:]}</id>'
    f'<yt:channelId>{CHANNEL[2:]}</yt:channelId><title>Tages Schau</title>'
    f'<link rel="alternate" href="https://www.youtube.com/channel/{CHANNEL}"/>'
    '<author><name>Tages Schau</name></author><published>2010-01-01T00:00:00+00:00</published>'
    + _entry('aaaaaaaaaaa', 'Video eins')
    + _entry('bbbbbbbbbbb', 'Ein Short', kind='shorts/')
    + _entry('ccccccccccc', 'Video zwei', when='2026-10-13T10:00:00+00:00')
    + '</feed>'
)


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'<script>{{"channelId":"{FEATURED}"}}</script>'
                f'<link rel="canonical" href="https://www.youtube.com/channel/{CHANNEL}">')
        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'<meta property="og:title" content="Ein Video"><script>{{"channelId":"{FEATURED}"}};'
                f'{{"videoDetails":{{"videoId":"aaaaaaaaaaa","channelId":"{CHANNEL}"}}}}</script>')
        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_video_whose_details_hold_objects_still_names_its_channel(self):
        page = (f'<script>{{"channelId":"{FEATURED}"}};{{"videoDetails":{{"videoId":"aaaaaaaaaaa",'
                f'"thumbnail":{{"x":1}},"channelId":"{CHANNEL}","isOwnerViewing":false}}}}</script>')
        found, _ = self.resolve(youtube.WATCH + 'aaaaaaaaaaa', page)
        self.assertEqual(found['feed'], f'{youtube.FEED}?channel_id={CHANNEL}')

    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', '<html>consent</html>')


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,
                         '<p>Hallo<br><a href="https://example.org/x">https://example.org/x</a></p><p>00:00 Anfang</p>')

    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('<html></html>')):
            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)


class FlakyFeedTests(TestCase):
    """YouTube's channel feeds answer 404 now and then; the next request gets them."""

    FEED = 'https://www.youtube.com/feeds/videos.xml?channel_id=UCabcdefghijklmnopqrstuv'
    BODY = (b'<?xml version="1.0"?><feed xmlns="http://www.w3.org/2005/Atom"><title>Kanal</title>'
            b'<entry><id>yt:video:abcdefghijk</id><title>Folge</title></entry></feed>')

    def _responses(self, codes):
        from types import SimpleNamespace
        calls = []

        def get(url, **kw):
            code = codes[min(len(calls), len(codes) - 1)]
            calls.append(url)
            return SimpleNamespace(status_code=code, content=self.BODY if code == 200 else b'',
                                   headers={}, url=url, history=[])
        return get, calls

    def test_a_404_is_asked_again(self):
        from .views import _fetch_feed
        get, calls = self._responses([404, 404, 200])
        with patch('podcasts.views.requests.get', get), patch('podcasts.views.time.sleep', lambda s: None):
            fetched = _fetch_feed(self.FEED)
        self.assertEqual(fetched.status, 200)
        self.assertEqual(len(calls), 3)

    def test_a_feed_that_stays_away_is_an_error(self):
        from .views import _fetch_feed, FeedFetchError, YOUTUBE_FEED_TRIES
        get, calls = self._responses([404])
        with patch('podcasts.views.requests.get', get), patch('podcasts.views.time.sleep', lambda s: None):
            with self.assertRaises(FeedFetchError):
                _fetch_feed(self.FEED)
        self.assertEqual(len(calls), YOUTUBE_FEED_TRIES)

    def test_other_feeds_are_asked_once(self):
        from .views import _fetch_feed, FeedFetchError
        get, calls = self._responses([404, 200])
        with patch('podcasts.views.requests.get', get), patch('podcasts.views.time.sleep', lambda s: None):
            with self.assertRaises(FeedFetchError):
                _fetch_feed('https://example.org/feed.xml')
        self.assertEqual(len(calls), 1)