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