diora-web/podcasts/test_player.py

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_sleep_timer_travels_with_the_state(self):
        self.report('phone', True, kind='radio', start=True, sleep={'left_ms': 900_000, 'episodes': 0})
        d = self.client.get('/podcasts/player/').json()
        self.assertEqual(d['sleep'], {'left_ms': 900_000, 'episodes': 0})
        # Episodes left; nonsense is none; an older client without it: none.
        self.report('phone', True, sleep={'left_ms': 0, 'episodes': 2})
        self.assertEqual((self.state().sleep_left_ms, self.state().sleep_episodes), (0, 2))
        self.report('phone', True, sleep={'left_ms': 'x', 'episodes': -4})
        self.assertEqual((self.state().sleep_left_ms, self.state().sleep_episodes), (0, 0))
        self.report('phone', True, sleep={'left_ms': 10 ** 12})
        self.assertEqual(self.state().sleep_left_ms, views.SLEEP_MAX_MINUTES * 60_000)
        self.report('phone', True)
        self.assertEqual(self.state().sleep_left_ms, 0)
        # Another device's ignored report doesn't change it.
        self.report('phone', True, sleep={'left_ms': 60_000})
        self.report('web', True, sleep={'left_ms': 5})
        self.assertEqual(self.state().sleep_left_ms, 60_000)

    def test_sleep_command(self):
        self.report('phone', True, kind='radio', start=True)
        r = self.command('web', 'sleep', 30)
        self.assertEqual(r.status_code, 200)
        self.assertEqual((r.json()['command']['action'], r.json()['command']['value']), ('sleep', 30))
        self.assertEqual(self.command('web', 'sleep', -1).json()['command']['value'], -1)
        self.assertEqual(self.command('web', 'sleep', 0).json()['command']['value'], 0)
        for bad in (None, 'x', 601, -100):
            self.assertEqual(self.command('web', 'sleep', bad).status_code, 400, bad)

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