diora-web/books/tests.py

"""The reading position: the place last read wins (docs/progress-sync.md)."""
import json
from unittest.mock import patch

from django.contrib.auth.models import User
from django.test import TestCase

from accounts.models import ApiToken
from .models import EBook, EBookProgress

NOW = 1_760_270_000_000  # the server's clock in these tests (ms)
MIN = 60_000


class ProgressTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(username='alice', password='pw12345678')
        self.auth = {'HTTP_AUTHORIZATION': f'Bearer {ApiToken.objects.create(user=self.user).token}'}
        self.book = EBook.objects.create(user=self.user, meta_ct='m', meta_iv='i', data_ct='d', data_iv='i')
        self.now = NOW
        clock = patch('books.views._now_ms', side_effect=lambda: self.now)
        clock.start()
        self.addCleanup(clock.stop)

    def save(self, anchor, frac=None, read_at=None, sent_at=None, device='Handy', **extra):
        body = {'scroll_fraction': frac if frac is not None else 0.1, 'position_anchor': anchor, **extra}
        if read_at is not None:
            body.update(read_at=read_at, sent_at=sent_at if sent_at is not None else self.now, device=device)
        resp = self.client.post(f'/books/{self.book.id}/progress/', json.dumps(body),
                                content_type='application/json', **self.auth)
        self.assertEqual(resp.status_code, 200)
        return resp.json()

    def state(self):
        return EBookProgress.objects.get(book=self.book)

    # -- the new rule

    def test_newer_place_wins_even_if_it_is_earlier_in_the_book(self):
        self.save('300:0.5', read_at=NOW - 10 * MIN, device='Handy')
        r = self.save('12:0.0', read_at=NOW - 1 * MIN, device='diora Desktop')
        self.assertTrue(r['stored'])
        p = self.state()
        self.assertEqual((p.position_anchor, p.device, p.read_at), ('12:0.0', 'diora Desktop', NOW - MIN))
        self.assertEqual(p.furthest_anchor, '300:0.5')

    def test_an_older_write_arriving_late_does_not_win(self):
        # The phone read offline at 10:00 and syncs at 12:00; the desktop read at 11:00.
        self.save('40:0.0', read_at=NOW - 60 * MIN, device='diora Desktop')
        r = self.save('90:0.0', read_at=NOW - 120 * MIN, device='Handy')
        self.assertFalse(r['stored'])
        self.assertEqual(r['position_anchor'], '40:0.0')
        p = self.state()
        self.assertEqual((p.position_anchor, p.device), ('40:0.0', 'diora Desktop'))
        self.assertEqual(p.furthest_anchor, '90:0.0', 'the late write still counts as reached')

    def test_resaving_an_unchanged_place_keeps_its_time_and_cannot_win(self):
        self.save('50:0.0', read_at=NOW - 5 * MIN, device='Handy')
        self.save('70:0.0', read_at=NOW - 2 * MIN, device='diora Desktop')
        # The phone, idle since, saves its place again on going to the background.
        self.now += MIN
        r = self.save('50:0.0', read_at=NOW - 5 * MIN, device='Handy')
        self.assertFalse(r['stored'])
        self.assertEqual(self.state().position_anchor, '70:0.0')

    def test_the_device_clock_error_is_taken_out(self):
        # A tolino 3 hours behind: it read one minute ago by its clock, and sends now by its clock.
        behind = 3 * 60 * MIN
        self.save('10:0.0', read_at=NOW - 30 * MIN, device='Handy')
        r = self.save('20:0.0', read_at=NOW - behind - MIN, sent_at=NOW - behind, device='tolino')
        self.assertTrue(r['stored'])
        self.assertEqual(self.state().read_at, NOW - MIN)

    def test_nothing_wins_from_the_future(self):
        self.save('10:0.0', read_at=NOW + 24 * 60 * MIN, sent_at=NOW)
        self.assertEqual(self.state().read_at, NOW)
        self.now += MIN
        self.assertTrue(self.save('11:0.0', read_at=self.now, sent_at=self.now)['stored'])

    def test_equal_times_take_the_new_write(self):
        self.save('10:0.0', read_at=NOW - MIN)
        self.assertTrue(self.save('11:0.0', read_at=NOW - MIN)['stored'])

    def test_force_sets_the_place_now(self):
        self.save('300:0.0', read_at=NOW - MIN)
        self.assertTrue(self.save('0:0.0', frac=0.0, force=True)['stored'])
        p = self.state()
        self.assertEqual((p.position_anchor, p.read_at), ('0:0.0', NOW))
        self.assertEqual(p.furthest_anchor, '300:0.0')

    def test_furthest_only_grows(self):
        self.save('80:0.0', read_at=NOW - 3 * MIN)
        self.save('20:0.0', read_at=NOW - 2 * MIN)
        self.save('81:0.0', read_at=NOW - 9 * MIN)
        p = self.state()
        self.assertEqual((p.position_anchor, p.furthest_anchor), ('20:0.0', '81:0.0'))

    def test_pdfs_compare_by_fraction(self):
        self.save('', frac=0.7, read_at=NOW - 2 * MIN)
        self.save('', frac=0.3, read_at=NOW - MIN)
        p = self.state()
        self.assertEqual((p.scroll_fraction, p.furthest_fraction), (0.3, 0.7))

    def test_response_is_the_state_after_the_write(self):
        r = self.save('5:0.25', frac=0.05, read_at=NOW - MIN, device='Browser')
        self.assertEqual(r['server_now'], NOW)
        for key in ('scroll_fraction', 'position_anchor', 'read_at', 'device',
                    'furthest_fraction', 'furthest_anchor', 'last_read'):
            self.assertIn(key, r)
        self.assertEqual(r['read_at'], NOW - MIN)

    def test_implausible_times_count_as_an_old_app(self):
        self.save('10:0.0', read_at=NOW - MIN)
        # read_at=True or seconds instead of ms are not times: only forward, like an old app.
        self.assertFalse(self.save('5:0.0', read_at=True)['stored'])
        self.assertFalse(self.save('5:0.0', read_at=1_760_270_000)['stored'])

    # -- apps from before the change (no read_at)

    def test_old_apps_only_move_forward(self):
        self.save('50:0.0', read_at=NOW - MIN)
        self.assertFalse(self.save('20:0.0')['stored'])
        self.assertTrue(self.save('60:0.0')['stored'])
        self.assertEqual(self.state().read_at, NOW)

    def test_old_apps_resaving_the_same_place_change_nothing(self):
        self.save('50:0.0', read_at=NOW - 5 * MIN, device='Handy')
        self.now += MIN
        self.assertFalse(self.save('50:0.0')['stored'])
        p = self.state()
        self.assertEqual((p.read_at, p.device), (NOW - 5 * MIN, 'Handy'))

    def test_first_write_of_an_old_app_is_stored(self):
        self.assertTrue(self.save('0:0.0', frac=0.0)['stored'])

    def test_old_force_still_works(self):
        self.save('50:0.0', read_at=NOW - MIN)
        self.assertTrue(self.save('1:0.0', force=True)['stored'])

    # -- reading

    def test_book_list_and_snapshot_carry_the_state(self):
        self.save('7:0.5', frac=0.2, read_at=NOW - MIN, device='Browser')
        resp = self.client.get('/books/', **self.auth)
        b = resp.json()[0]
        self.assertEqual(resp['X-Server-Now'], str(NOW))
        self.assertEqual((b['position_anchor'], b['read_at'], b['device'], b['furthest_anchor']),
                         ('7:0.5', NOW - MIN, 'Browser', '7:0.5'))
        self.assertTrue(b['last_read'].startswith('2025-10-12T'))
        snap = self.client.get('/api/sync/?only=books', **self.auth).json()
        p = snap['book_progress'][0]
        self.assertEqual((p['book_id'], p['read_at'], p['device']), (self.book.id, NOW - MIN, 'Browser'))
        self.assertIn('server_now', snap)

    def test_rows_from_before_have_their_save_time_as_read_at(self):
        p = EBookProgress.objects.create(user=self.user, book=self.book, scroll_fraction=0.4, position_anchor='9:0.0')
        b = self.client.get('/books/', **self.auth).json()[0]
        self.assertEqual(b['read_at'], int(p.updated_at.timestamp() * 1000))

    def test_a_never_opened_book(self):
        b = self.client.get('/books/', **self.auth).json()[0]
        self.assertEqual((b['read_at'], b['last_read'], b['position_anchor']), (None, None, ''))