mrwnslz/diora-web
1
0
Fork 0
Code Issues 4 Pull requests Projects Releases Packages Wiki Activity Actions
diora-web/podcasts/test_player.py
mrwnslz 3eb6a27ec3 Player-Sync: ein Player für alle Geräte (wie Spotify Connect) 
Server: PlayerState und PlayerDevice (Migration 0012), /podcasts/player/
(Stand), report/ (das Gerät, das zu spielen anfängt, wird das aktive und
bleibt es auch pausiert, solange es online ist), command/ (Fernbedienung,
transfer auf ein anderes Gerät) und stream/ (Server-Sent Events, zugleich
die Anwesenheit der Geräte). seq steigt in einem UPDATE und nie zurück.

Browser: podcasts-web.js meldet Abspielen, Pause, Springen, Folgen- und
Senderwechsel (alle 10 s beim Spielen), hört den Stream, hört auf, wenn ein
anderes Gerät startet, führt Befehle aus, zeigt das andere Gerät als remote
und übernimmt, was ein verschwundenes Gerät zuletzt hatte. Eine pausierte
Folge folgt einem neueren Stand von anderswo; im Vordergrund wird nach
15 s neu geholt. Podcast-Seite neu kopiert, sw v58.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-02 15:18:38 +02:00

365 lines
19 KiB
Python
Raw Blame History

import json
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
from django.contrib.auth.models import User
from django.test import TestCase
from django.utils import timezone as dj_timezone
from accounts.models import ApiToken
from . import views
from .models import PlayerDevice, PlayerState, PodcastEpisode, PodcastFeed, PodcastQueue
STATION = {'url': 'https://example.org/stream', 'name': 'Example FM', 'id': 7}
class PlayerSyncTests(TestCase):
"""The shared player (podcasts/views.py, "Player sync"): reports, commands,
presence and the stream."""
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://example.org/feed', title='Feed')
self.ep = PodcastEpisode.objects.create(
feed=self.feed, guid='e1', title='E1', audio_url='https://example.org/e1.mp3', duration_seconds=3600,
pub_date=datetime(2026, 1, 1, tzinfo=timezone.utc),
)
self.ep2 = PodcastEpisode.objects.create(
feed=self.feed, guid='e2', title='E2', audio_url='https://example.org/e2.mp3',
)
def post(self, path, body):
return self.client.post(f'/podcasts/player/{path}/', json.dumps(body), content_type='application/json')
def report(self, device, playing, kind='podcast', episode=None, pos=0, **extra):
body = {'device_id': device, 'device_name': device.upper(), 'kind': kind, 'position_ms': pos,
'playing': playing, **extra}
if kind == 'podcast':
body['episode_id'] = (episode or self.ep).id
elif kind == 'radio':
body['station'] = STATION
return self.post('report', body)
def command(self, device, action, value=None, to=None):
body = {'device_id': device, 'device_name': device.upper(), 'action': action}
if value is not None:
body['value'] = value
if to is not None:
body['to_device_id'] = to
return self.post('command', body)
def state(self):
return PlayerState.objects.get(user=self.user)
def go_offline(self, device, seconds=60):
PlayerDevice.objects.filter(user=self.user, device_id=device).update(
last_seen=dj_timezone.now() - timedelta(seconds=seconds))
def test_logged_out(self):
self.client.logout()
for r in (self.client.get('/podcasts/player/'), self.client.get('/podcasts/player/stream/'),
self.report('a', True), self.command('a', 'pause')):
self.assertEqual(r.status_code, 401)
self.assertEqual(r.json(), {'error': 'authentication required'})
def test_bearer_token(self):
self.client.logout()
token = ApiToken.objects.create(user=self.user)
r = self.client.get('/podcasts/player/', HTTP_AUTHORIZATION=f'Bearer {token.token}')
self.assertEqual(r.status_code, 200)
self.assertEqual(r.json()['kind'], '')
def test_empty_state(self):
d = self.client.get('/podcasts/player/').json()
self.assertEqual(d['seq'], 0)
self.assertEqual(d['kind'], '')
self.assertIsNone(d['episode'])
self.assertIsNone(d['station'])
self.assertIs(d['playing'], False)
self.assertIsNone(d['command'])
self.assertEqual(d['devices'], [])
self.assertIs(d['active_online'], False)
self.assertRegex(d['server_time'], r'^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\.\d{3}Z$')
def test_play_makes_the_device_active(self):
PodcastQueue.objects.create(user=self.user, episode=self.ep, position=0)
d = self.report('a', True, pos=12345, speed=1.5).json()
self.assertEqual(d['seq'], 2) # the report, and device a coming online
self.assertEqual((d['kind'], d['playing'], d['position_ms'], d['speed']), ('podcast', True, 12345, 1.5))
self.assertEqual((d['device_id'], d['device_name']), ('a', 'A'))
self.assertEqual(set(d['episode']), {'id', 'title', 'audio_url', 'duration_seconds', 'pub_date', 'artwork_url',
'feed__id', 'feed__title', 'feed__artwork_url', 'position_seconds',
'in_queue'})
self.assertEqual((d['episode']['id'], d['episode']['in_queue']), (self.ep.id, True))
self.assertRegex(d['updated_at'], r'Z$')
self.assertEqual(d['devices'], [{'id': 'a', 'name': 'A', 'online': True, 'active': True}])
self.assertIs(d['active_online'], True)
self.assertEqual(self.client.get('/podcasts/player/').json()['seq'], 2)
def test_play_on_another_device_takes_over(self):
self.report('a', True)
self.command('b', 'pause')
d = self.report('b', True, kind='radio', start=True).json()
self.assertEqual((d['device_id'], d['kind'], d['playing']), ('b', 'radio', True))
self.assertEqual(d['station'], STATION)
self.assertIsNone(d['episode'])
self.assertIsNone(d['command']) # it was for device a
self.assertEqual([(x['id'], x['active']) for x in d['devices']], [('a', False), ('b', True)])
def test_late_periodic_report_of_the_old_device_is_ignored(self):
# b takes over; a's 10 s "playing" report, already on its way, comes after.
self.report('a', True, pos=1000)
self.report('b', True, pos=1500, start=True)
seq = self.state().seq
d = self.report('a', True, pos=11000).json()
self.assertEqual((d['seq'], d['device_id'], d['position_ms'], d['playing']), (seq, 'b', 1500, True))
# The user starts playback on a again: then a is the active device.
d = self.report('a', True, pos=12000, start=True).json()
self.assertEqual((d['device_id'], d['position_ms']), ('a', 12000))
def test_playing_without_start_wins_only_without_an_online_active_device(self):
self.report('a', False, pos=1000) # a is the active device, paused
self.assertEqual(self.report('b', True, pos=7).json()['device_id'], 'a')
self.go_offline('a')
self.assertEqual(self.report('b', True, pos=7).json()['device_id'], 'b')
def test_seq_only_goes_up(self):
seqs = [self.client.get('/podcasts/player/').json()['seq']]
seqs.append(self.report('a', True, start=True).json()['seq'])
seqs.append(self.report('b', True).json()['seq']) # ignored, b came online
seqs.append(self.command('b', 'pause').json()['seq'])
seqs.append(self.report('b', False).json()['seq']) # ignored
seqs.append(self.report('a', False).json()['seq'])
self.go_offline('a')
views._check_online(self.user)
seqs.append(self.client.get('/podcasts/player/').json()['seq'])
self.assertEqual(seqs, sorted(seqs))
self.assertEqual(seqs, [0, 2, 3, 4, 4, 5, 6])
def test_active_device_updates_and_stays_active_while_paused(self):
self.report('a', True, pos=1000)
d = self.report('a', False, episode=self.ep2, pos=5000).json()
self.assertEqual((d['device_id'], d['playing'], d['episode']['id'], d['position_ms']), ('a', False, self.ep2.id, 5000))
# Paused, but online: device b's paused choice doesn't count.
seq = d['seq']
d = self.report('b', False, episode=self.ep, pos=99).json()
self.assertEqual((d['seq'], d['device_id'], d['episode']['id'], d['position_ms']), (seq + 1, 'a', self.ep2.id, 5000))
# (seq + 1: b came online; the report itself changed nothing.)
self.assertIs(d['active_online'], True)
self.assertEqual(self.state().device_id, 'a')
def test_paused_report_ignored_while_the_active_device_plays(self):
self.report('a', True, pos=1000)
self.report('b', True, start=True)
self.report('a', True, pos=2000, start=True) # a plays again: a is active
seq = self.state().seq
d = self.report('b', False, kind='').json()
self.assertEqual((d['seq'], d['device_id'], d['playing'], d['kind']), (seq, 'a', True, 'podcast'))
def test_active_device_offline_newest_report_wins(self):
self.report('a', True, pos=1000)
self.report('b', False, pos=1) # b online too
self.go_offline('a')
d = self.report('b', False, kind='radio').json()
self.assertEqual((d['device_id'], d['kind'], d['playing']), ('b', 'radio', False))
self.assertIs(d['active_online'], True)
self.assertEqual([(x['id'], x['online']) for x in d['devices']], [('b', True), ('a', False)])
def test_online_is_presence_not_the_last_report(self):
# Rule 4: playing with the last report a minute old, but the device is
# still online (its stream pings): still active.
self.report('a', True, pos=1000)
PlayerState.objects.filter(user=self.user).update(updated_at=dj_timezone.now() - timedelta(seconds=60))
d = self.report('b', False, kind='radio').json()
self.assertEqual(d['device_id'], 'a')
# 45 s without a sign of life: gone.
self.go_offline('a', 44)
self.assertEqual(self.report('b', False, kind='radio').json()['device_id'], 'a')
self.go_offline('a', 46)
self.assertEqual(self.report('b', False, kind='radio').json()['device_id'], 'b')
def test_no_active_device_newest_report_wins(self):
d = self.report('a', False, pos=4000).json()
self.assertEqual((d['device_id'], d['position_ms'], d['playing']), ('a', 4000, False))
def test_empty_kind_clears_the_item(self):
self.report('a', True, kind='radio')
d = self.report('a', True, kind='').json()
self.assertEqual((d['kind'], d['station'], d['episode'], d['playing']), ('', None, None, False))
def test_invalid_reports(self):
other = User.objects.create_user(username='bob', password='pw12345678')
feed = PodcastFeed.objects.create(user=other, rss_url='https://example.org/feed', title='Bob')
theirs = PodcastEpisode.objects.create(feed=feed, guid='x', title='X', audio_url='https://example.org/x.mp3')
r = self.report('a', True, episode=theirs)
self.assertEqual(r.status_code, 400)
self.assertEqual(r.json()['error'], 'unknown episode')
for body in ({'kind': 'podcast', 'playing': True}, # no device
{'device_id': 'a', 'kind': 'tv', 'playing': True},
{'device_id': 'a', 'kind': 'podcast', 'playing': True}, # no episode
{'device_id': 'a', 'kind': 'radio', 'playing': True, 'station': {'name': 'x'}},
{'device_id': 'a', 'kind': '', 'playing': 'yes'},
{'device_id': 'a', 'kind': '', 'playing': True, 'position_ms': 'x'}):
with self.subTest(body=body):
self.assertEqual(self.post('report', body).status_code, 400)
self.assertFalse(PlayerState.objects.filter(user=self.user, seq__gt=0).exists())
def test_command_is_stored_and_cleared_by_done_command(self):
self.report('a', True, pos=1000)
r = self.command('b', 'seek', 61000)
self.assertEqual(r.status_code, 200)
d = r.json()
cmd = d['command']
self.assertEqual((cmd['action'], cmd['value'], cmd['from_device_id'], cmd['from_device_name'], cmd['to_device_id']),
('seek', 61000, 'b', 'B', 'a'))
self.assertRegex(cmd['id'], r'^[0-9a-f]{32}$')
self.assertRegex(cmd['at'], r'Z$')
seq = d['seq']
# A report without it (the 10 s one) leaves it.
d = self.report('a', True, pos=2000).json()
self.assertEqual((d['seq'], d['command']), (seq + 1, cmd))
d = self.report('a', True, pos=61000, done_command='nope').json()
self.assertEqual(d['command'], cmd)
d = self.report('a', True, pos=61000, done_command=cmd['id']).json()
self.assertIsNone(d['command'])
self.assertEqual(d['seq'], seq + 3)
self.assertEqual(self.command('b', 'skip', -15).json()['command']['value'], -15)
def test_commands_to_a_paused_active_device(self):
# Device 1 plays, device 2 pauses it; "play" on device 2 resumes on device 1.
self.report('d1', True, pos=1000)
pause = self.command('d2', 'pause').json()['command']
self.report('d1', False, pos=3000, done_command=pause['id'])
d = self.command('d2', 'play').json()
self.assertEqual((d['command']['action'], d['command']['to_device_id']), ('play', 'd1'))
self.assertEqual((d['device_id'], d['playing'], d['position_ms']), ('d1', False, 3000))
self.assertIs(d['active_online'], True)
def test_command_needs_an_online_device(self):
r = self.command('b', 'pause')
self.assertEqual(r.status_code, 409) # nothing active
self.report('a', True)
self.go_offline('a')
self.assertEqual(self.command('b', 'pause').status_code, 409)
self.assertEqual(self.command('b', 'transfer', to='zzz').status_code, 409)
# Device a gone: device b's play wins (even without start).
d = self.report('b', True, pos=5).json()
self.assertEqual((d['device_id'], d['playing']), ('b', True))
def test_invalid_commands(self):
self.report('a', True)
for body in ({'device_id': 'b', 'action': 'louder'},
{'device_id': 'b', 'action': 'seek'},
{'device_id': 'b', 'action': 'skip', 'value': 'x'},
{'action': 'pause'}):
with self.subTest(body=body):
self.assertEqual(self.post('command', body).status_code, 400)
def test_transfer_to_another_device(self):
self.report('a', True, pos=1000)
self.report('c', False, pos=0) # c is online (paused report: ignored)
d = self.command('b', 'transfer', to='c').json()
cmd = d['command']
self.assertEqual((cmd['action'], cmd['to_device_id'], cmd['from_device_id']), ('transfer', 'c', 'b'))
self.assertEqual(d['device_id'], 'a')
# a keeps reporting meanwhile: the command stays for c.
self.assertEqual(self.report('a', True, pos=2000).json()['command'], cmd)
# c plays it from the place: c is the active device, the command is done.
d = self.report('c', True, pos=2100, done_command=cmd['id'], start=True).json()
self.assertEqual((d['device_id'], d['playing'], d['command']), ('c', True, None))
def test_transfer_done_in_the_report_that_takes_over(self):
self.report('a', True, pos=1000)
self.report('c', False)
cmd = self.command('b', 'transfer', to='c').json()['command']
d = self.report('c', True, pos=1100, start=True, done_command=cmd['id']).json()
self.assertEqual((d['device_id'], d['command']), ('c', None))
# a's periodic report, sent before it heard of the transfer: ignored.
self.assertEqual(self.report('a', True, pos=11000).json()['device_id'], 'c')
def test_transfer_that_could_not_play_is_done_too(self):
self.report('a', True)
self.report('c', False)
cmd = self.command('b', 'transfer', to='c').json()['command']
d = self.report('c', False, done_command=cmd['id']).json()
self.assertEqual((d['device_id'], d['command']), ('a', None))
def test_devices_coming_and_going_change_seq(self):
self.report('a', True)
seq = self.state().seq
views._check_online(self.user)
self.assertEqual(self.state().seq, seq) # nothing changed
self.report('a', True)
self.assertEqual(self.state().seq, seq + 1) # a report, no new device
self.go_offline('a')
views._check_online(self.user)
self.assertEqual(self.state().seq, seq + 2) # a went away
self.assertIs(self.client.get('/podcasts/player/').json()['active_online'], False)
def test_device_list(self):
now = dj_timezone.now()
for i in range(12):
PlayerDevice.objects.create(user=self.user, device_id=f'old{i}', device_name=f'Old {i}',
last_seen=now - timedelta(hours=i + 1))
PlayerDevice.objects.create(user=self.user, device_id='ancient', last_seen=now - timedelta(days=8))
self.report('z', True)
self.report('m', False)
devices = self.client.get('/podcasts/player/').json()['devices']
self.assertEqual(len(devices), 10)
self.assertEqual([d['id'] for d in devices[:3]], ['m', 'z', 'old0'])
self.assertNotIn('ancient', [d['id'] for d in devices])
def stream(self, since=None, **params):
if since is not None:
params['since'] = since
r = self.client.get('/podcasts/player/stream/', params)
self.assertEqual(r['Content-Type'], 'text/event-stream')
self.assertEqual(r['Cache-Control'], 'no-cache')
self.assertEqual(r['X-Accel-Buffering'], 'no')
events = []
for chunk in r.streaming_content:
text = chunk.decode()
if text.startswith('data: '):
events.append(json.loads(text[len('data: '):]))
else:
self.assertEqual(text, ': ping\n\n')
return events
@patch.object(views, 'STREAM_SECONDS', 0.05)
@patch.object(views, 'STREAM_POLL_SECONDS', 0.01)
def test_stream_sends_newer_state_and_ends(self):
self.report('a', True, pos=777)
seq = self.state().seq
events = self.stream(seq - 1)
self.assertEqual(len(events), 1)
self.assertEqual((events[0]['seq'], events[0]['position_ms'], events[0]['device_id']), (seq, 777, 'a'))
self.assertEqual(self.stream(seq), []) # nothing newer: pings only
@patch.object(views, 'STREAM_SECONDS', 0.05)
@patch.object(views, 'STREAM_POLL_SECONDS', 0.01)
def test_stream_sends_changes(self):
self.report('a', True)
seq = self.state().seq
sleep = views.time.sleep
changes = []
def change_once(seconds):
if not changes:
changes.append(1)
self.report('a', False, pos=42)
sleep(seconds)
with patch.object(views.time, 'sleep', change_once):
events = self.stream(seq)
self.assertEqual([(e['playing'], e['position_ms']) for e in events], [(False, 42)])
@patch.object(views, 'STREAM_SECONDS', 0.05)
@patch.object(views, 'STREAM_POLL_SECONDS', 0.01)
def test_stream_counts_as_presence(self):
events = self.stream(0, device_id='web1', device_name='Browser (Firefox/Linux)', device_kind='web')
self.assertEqual(len(events), 1) # web1 came online: seq 1
self.assertEqual(events[0]['devices'], [{'id': 'web1', 'name': 'Browser (Firefox/Linux)', 'online': True,
'active': False}])
self.assertEqual(PlayerDevice.objects.get(device_id='web1').kind, 'web')
Reference in a new issue View git blame Copy permalink