107 lines
5.6 KiB
Python
107 lines
5.6 KiB
Python
"""The Podcast-Radio (docs/podcasts.md, podcasts/radio.py)."""
|
|
import json
|
|
|
|
from django.contrib.auth.models import User
|
|
from django.core.cache import cache
|
|
from django.test import TestCase
|
|
|
|
from .models import EpisodeProgress, PlayerState, PodcastEpisode, PodcastFeed, PodcastQueue, PodcastRadioHeard
|
|
from .tests import _at, _progress
|
|
|
|
|
|
class RadioTests(TestCase):
|
|
def setUp(self):
|
|
cache.clear()
|
|
self.user = User.objects.create_user(username='alice', password='pw12345678')
|
|
self.client.force_login(self.user)
|
|
self.fa = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/a', title='Alpha')
|
|
self.fb = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/b', title='Beta')
|
|
self.fc = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/c', title='Abruf',
|
|
new_episodes=PodcastFeed.NEW_IGNORE)
|
|
self.book = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/h', title='Buch',
|
|
kind=PodcastFeed.AUDIOBOOK)
|
|
|
|
def ep(self, feed, guid, day):
|
|
return PodcastEpisode.objects.create(feed=feed, guid=guid, title=guid, pub_date=_at(1, day),
|
|
audio_url=f'https://example.org/{guid}.mp3')
|
|
|
|
def radio(self, **params):
|
|
r = self.client.get('/podcasts/radio/', params)
|
|
self.assertEqual(r.status_code, 200)
|
|
return [e['title'] for e in r.json()['episodes']]
|
|
|
|
def heard(self, episode, **body):
|
|
r = self.client.post('/podcasts/radio/heard/', json.dumps({'episode_id': episode.id, **body}),
|
|
content_type='application/json')
|
|
self.assertEqual(r.status_code, 200)
|
|
return [e['title'] for e in r.json()['episodes']]
|
|
|
|
def test_newest_untouched_strom_episodes_first(self):
|
|
self.ep(self.fa, 'a1', 1)
|
|
self.ep(self.fb, 'b2', 2)
|
|
played = self.ep(self.fa, 'a3', 3)
|
|
started = self.ep(self.fb, 'b4', 4)
|
|
self.ep(self.fc, 'c5', 5) # Auf Abruf: not in the Strom
|
|
self.ep(self.book, 'h6', 6) # an audiobook's chapter
|
|
_progress(self.user, played, _at(2, 1), played=True)
|
|
_progress(self.user, started, _at(2, 1), position_seconds=30)
|
|
self.assertEqual(self.radio(), ['b2', 'a1'])
|
|
|
|
def test_one_podcast_is_mixed_with_the_others(self):
|
|
for d in (1, 2, 3):
|
|
self.ep(self.fb, f'b{d}', d)
|
|
for d in (7, 8, 9):
|
|
self.ep(self.fa, f'a{d}', d)
|
|
self.assertEqual(self.radio(limit=6), ['a9', 'b3', 'a8', 'b2', 'a7', 'b1'])
|
|
# After an episode of Alpha: Beta first, and that episode not again.
|
|
a9 = PodcastEpisode.objects.get(guid='a9')
|
|
self.assertEqual(self.radio(after=a9.id, limit=3), ['b3', 'a8', 'b2'])
|
|
# One the device is done with (its /heard/ still on the way) doesn't count for the mixing.
|
|
a8 = PodcastEpisode.objects.get(guid='a8')
|
|
self.assertEqual(self.radio(after=a9.id, skip=str(a8.id), limit=3), ['b3', 'a7', 'b2'])
|
|
# Only one podcast left: its episodes in a row.
|
|
PodcastEpisode.objects.filter(feed=self.fb).delete()
|
|
self.assertEqual(self.radio(after=a9.id), ['a8', 'a7'])
|
|
|
|
def test_heard_leaves_no_trace_but_the_radio_moves_on(self):
|
|
a1 = self.ep(self.fa, 'a1', 1)
|
|
b2 = self.ep(self.fb, 'b2', 2)
|
|
self.assertEqual(self.heard(b2), ['a1'])
|
|
self.assertEqual(self.radio(), ['a1'])
|
|
self.assertFalse(EpisodeProgress.objects.exists())
|
|
self.assertFalse(PodcastQueue.objects.exists())
|
|
# The Strom and the history don't know about it.
|
|
r = self.client.get('/podcasts/episodes/', {'filter': 'history'}).json()
|
|
self.assertEqual(r['total'], 0)
|
|
r = self.client.get('/podcasts/episodes/', {'filter': 'stream,unplayed'}).json()
|
|
self.assertEqual(r['total'], 2)
|
|
self.heard(b2) # twice is fine
|
|
self.assertEqual(PodcastRadioHeard.objects.count(), 1)
|
|
self.heard(a1)
|
|
self.assertEqual(self.radio(), [])
|
|
|
|
def test_others_episodes_and_bad_requests(self):
|
|
bob = User.objects.create_user(username='bob', password='pw12345678')
|
|
fbob = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/a', title='Alpha')
|
|
theirs = self.ep(fbob, 'x', 1)
|
|
r = self.client.post('/podcasts/radio/heard/', json.dumps({'episode_id': theirs.id}), content_type='application/json')
|
|
self.assertEqual(r.status_code, 404)
|
|
r = self.client.post('/podcasts/radio/heard/', '{}', content_type='application/json')
|
|
self.assertEqual(r.status_code, 400)
|
|
self.assertEqual(self.client.get('/podcasts/radio/', {'after': 'x'}).status_code, 400)
|
|
self.assertEqual(self.radio(), [])
|
|
self.client.logout()
|
|
self.assertEqual(self.client.get('/podcasts/radio/').status_code, 401)
|
|
|
|
def test_player_state_says_radio(self):
|
|
a1 = self.ep(self.fa, 'a1', 1)
|
|
body = {'device_id': 'phone', 'device_name': 'Handy', 'kind': 'podcast', 'episode_id': a1.id,
|
|
'playing': True, 'position_ms': 1000, 'start': True, 'podcast_radio': True}
|
|
r = self.client.post('/podcasts/player/report/', json.dumps(body), content_type='application/json')
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertTrue(r.json()['podcast_radio'])
|
|
self.assertTrue(PlayerState.objects.get(user=self.user).podcast_radio)
|
|
body['podcast_radio'] = False
|
|
r = self.client.post('/podcasts/player/report/', json.dumps(body), content_type='application/json')
|
|
self.assertFalse(r.json()['podcast_radio'])
|
|
self.assertFalse(EpisodeProgress.objects.exists())
|