diora-web/podcasts/test_sponsor.py

"""Skipping the ad and the outro of a YouTube episode (sponsor.py)."""
from datetime import timedelta
from unittest.mock import patch

from django.contrib.auth.models import User
from django.test import TestCase
from django.utils import timezone

from . import sponsor
from .models import PodcastEpisode, PodcastFeed

VIDEO = 'https://www.youtube.com/watch?v=aqz-KE-bpKQ'
VID = 'aqz-KE-bpKQ'


def seg(start, end, category='sponsor', action='skip'):
    return {'UUID': f'{category}-{start}', 'category': category, 'actionType': action,
            'segment': [start, end], 'votes': 10, 'videoDuration': 600}


class SpanTests(TestCase):
    """What is taken out of SponsorBlock's answer."""

    def test_only_sponsor_and_outro(self):
        self.assertEqual(sponsor._spans([
            seg(10, 20), seg(30, 40, 'outro'),
            seg(50, 60, 'intro'), seg(70, 80, 'selfpromo'),
            seg(90, 100, 'interaction'), seg(110, 120, 'music_offtopic'),
        ]), [[10, 20], [30, 40]])

    def test_full_and_poi_segments_are_left_alone(self):
        # "full": the whole video is a sponsor — jumping would leave no sound at all.
        self.assertEqual(sponsor._spans([seg(0, 600, 'sponsor', 'full')]), [])

    def test_mute_counts_as_skipped(self):
        self.assertEqual(sponsor._spans([seg(10, 20, 'sponsor', 'mute')]), [[10, 20]])

    def test_touching_parts_become_one(self):
        self.assertEqual(sponsor._spans([seg(10, 20), seg(20, 30)]), [[10, 30]])
        self.assertEqual(sponsor._spans([seg(10, 20), seg(19.5, 30)]), [[10, 30]])

    def test_order_does_not_matter(self):
        self.assertEqual(sponsor._spans([seg(30, 40, 'outro'), seg(10, 20)]), [[10, 20], [30, 40]])

    def test_nonsense_is_ignored(self):
        self.assertEqual(sponsor._spans([
            seg(10, 20), 'nonsense', {}, {'category': 'sponsor'}, {'category': 'sponsor', 'segment': [1]},
            {'category': 'sponsor', 'segment': ['a', 'b']}, {'category': 'sponsor', 'segment': [20, 10]},
            {'category': 'sponsor', 'segment': [-5, 5]}, seg(30, 30.1),      # too short to be worth it
        ]), [[10, 20]])

    def test_nothing_at_all(self):
        self.assertEqual(sponsor._spans(None), [])
        self.assertEqual(sponsor._spans([]), [])

    def test_the_video_id_of_an_episode(self):
        self.assertEqual(sponsor.video_id_of(VIDEO), VID)
        self.assertEqual(sponsor.video_id_of(f'https://example.org/podcasts/{VID}.mp3'), VID)
        self.assertEqual(sponsor.video_id_of('https://example.org/folge.mp3'), '')


class FetchTests(TestCase):
    def setUp(self):
        self.answers = []

        def get(url, **kw):
            self.asked = (url, kw.get('params'), kw.get('timeout'))
            return self.answers.pop(0)

        self.get = get

    def _answer(self, status=200, body=None):
        class R:
            status_code = status

            def json(_self):
                return body
        r = R()
        r.status_code = status
        return r

    def test_a_video_nobody_marked_is_not_an_error(self):
        self.answers = [self._answer(404)]
        with patch('podcasts.sponsor.requests.get', self.get):
            self.assertEqual(sponsor.fetch(VID), [])
        url, params, timeout = self.asked
        self.assertEqual(url, sponsor.API)
        self.assertEqual(params['videoID'], VID)
        self.assertEqual(params['categories'], ['sponsor', 'outro'])
        self.assertEqual(timeout, sponsor.TIMEOUT)

    def test_what_the_api_answers_is_read(self):
        self.answers = [self._answer(200, [seg(60, 90, 'outro'), seg(10, 20)])]
        with patch('podcasts.sponsor.requests.get', self.get):
            self.assertEqual(sponsor.fetch(VID), [[10, 20], [60, 90]])

    def test_a_broken_answer_is_an_error(self):
        for answer in (self._answer(500), self._answer(200, {'error': 'nope'})):
            self.answers = [answer]
            with patch('podcasts.sponsor.requests.get', self.get):
                with self.assertRaises(RuntimeError):
                    sponsor.fetch(VID)

    def test_ask_remembers_an_error_instead_of_raising(self):
        def boom(url, **kw):
            raise OSError('no route to host')
        with patch('podcasts.sponsor.requests.get', boom):
            got = sponsor.ask(VID)
        self.assertEqual(got['spans'], [])
        self.assertEqual(got['v'], sponsor.VERSION)
        self.assertIn('no route', got['error'])
        self.assertTrue(got['at'])

    def test_a_stored_answer_is_used_until_it_is_old(self):
        now = timezone.now().isoformat()
        fresh = {'v': sponsor.VERSION, 'spans': [[1, 2]], 'at': now}
        self.assertTrue(sponsor.current(fresh))
        self.assertFalse(sponsor.current(None))
        self.assertFalse(sponsor.current({}))
        self.assertFalse(sponsor.current({'v': sponsor.VERSION - 1, 'spans': [], 'at': now}))
        self.assertFalse(sponsor.current({'v': sponsor.VERSION, 'spans': []}))   # no "at"

    def test_a_failure_is_asked_again_sooner_than_an_answer(self):
        def ago(seconds):
            return (timezone.now() - timedelta(seconds=seconds)).isoformat()
        old = {'v': sponsor.VERSION, 'spans': [[1, 2]], 'at': ago(sponsor.STALE + 60)}
        self.assertFalse(sponsor.current(old))
        self.assertTrue(sponsor.current(dict(old, at=ago(sponsor.STALE - 60))))
        failed = {'v': sponsor.VERSION, 'spans': [], 'error': 'HTTP 500', 'at': ago(sponsor.STALE_ERROR + 60)}
        self.assertFalse(sponsor.current(failed))


class EpisodeSponsorViewTests(TestCase):
    def setUp(self):
        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='https://www.youtube.com/feeds/videos.xml?channel_id=UCx',
                                              title='Kanal')
        self.video = PodcastEpisode.objects.create(
            feed=self.feed, guid=f'yt:video:{VID}', title='Eine Folge', audio_url=VIDEO,
            pub_date=timezone.now())
        self.other = PodcastEpisode.objects.create(
            feed=self.feed, guid='e1', title='Eine Folge', audio_url='https://example.org/e1.mp3',
            pub_date=timezone.now())
        self.answers = []

    def _get(self, url, **kw):
        return self.answers.pop(0)

    def _answer(self, body):
        class R:
            status_code = 200

            def json(_self):
                return body
        return R()

    def url(self, ep):
        return f'/podcasts/episodes/{ep.id}/sponsor/'

    def test_a_video_is_asked_once_and_the_answer_is_kept(self):
        self.answers = [self._answer([seg(10, 20), seg(60, 90, 'outro')])]
        with patch('podcasts.sponsor.requests.get', self._get):
            d = self.client.get(self.url(self.video)).json()
        self.assertEqual(d, {'ready': True, 'v': sponsor.VERSION, 'spans': [[10, 20], [60, 90]]})
        self.assertEqual(PodcastEpisode.objects.get(pk=self.video.pk).sponsor['spans'], [[10, 20], [60, 90]])
        # Kept: the second call doesn't ask again.
        with patch('podcasts.sponsor.requests.get', self._get):
            d = self.client.get(self.url(self.video)).json()
        self.assertEqual(d['spans'], [[10, 20], [60, 90]])

    def test_a_failed_ask_answers_empty_and_is_remembered(self):
        def boom(url, **kw):
            raise OSError('timed out')
        with patch('podcasts.sponsor.requests.get', boom):
            d = self.client.get(self.url(self.video)).json()
        self.assertEqual(d['spans'], [])
        self.assertEqual(d['error'], 'timed out')
        self.assertIn('error', PodcastEpisode.objects.get(pk=self.video.pk).sponsor)

    def test_an_episode_that_is_no_video_is_never_asked_for(self):
        with patch('podcasts.sponsor.requests.get', self._get):
            d = self.client.get(self.url(self.other)).json()
        self.assertEqual(d, {'ready': True, 'v': sponsor.VERSION, 'spans': []})
        self.assertIsNone(PodcastEpisode.objects.get(pk=self.other.pk).sponsor)

    def test_a_foreign_episode_is_not_found(self):
        other = User.objects.create_user(username='bob', password='pw12345678')
        feed = PodcastFeed.objects.create(user=other, rss_url='https://example.org/f', title='F')
        ep = PodcastEpisode.objects.create(feed=feed, guid='x', title='X', audio_url=VIDEO,
                                          pub_date=timezone.now())
        self.assertEqual(self.client.get(self.url(ep)).status_code, 404)

    def test_asking_without_a_login_is_refused(self):
        self.client.logout()
        self.assertEqual(self.client.get(self.url(self.video)).status_code, 401)