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