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_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)