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


class CoverAndPrefsTests(TestCase):
    """Covers and the library's settings travel encrypted (docs/library.md)."""

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

    def post(self, url, body, **auth):
        return self.client.post(url, json.dumps(body), content_type='application/json', **(auth or self.auth))

    def test_a_cover_is_kept_and_listed(self):
        url = f'/books/{self.book.id}/cover/'
        self.assertEqual(self.client.get(url, **self.auth).status_code, 404)
        self.assertFalse(self.client.get('/books/', **self.auth).json()[0]['has_cover'])
        self.assertEqual(self.post(url, {'cover_ct': 'Q0lQSEVS', 'cover_iv': 'a1b2c3d4e5f60718293a4b5c'}).status_code, 200)
        self.assertEqual(self.client.get(url, **self.auth).json(),
                         {'cover_ct': 'Q0lQSEVS', 'cover_iv': 'a1b2c3d4e5f60718293a4b5c'})
        self.assertTrue(self.client.get('/books/', **self.auth).json()[0]['has_cover'])

    def test_bad_or_huge_covers_are_refused(self):
        url = f'/books/{self.book.id}/cover/'
        self.assertEqual(self.post(url, {'cover_ct': 'x', 'cover_iv': 'not hex!'}).status_code, 400)
        self.assertEqual(self.post(url, {'cover_ct': 'x' * 300_000, 'cover_iv': 'a1b2c3d4'}).status_code, 400)

    def test_covers_are_ones_own(self):
        bob = User.objects.create_user(username='bob', password='pw12345678')
        bob_auth = {'HTTP_AUTHORIZATION': f'Bearer {ApiToken.objects.create(user=bob).token}'}
        url = f'/books/{self.book.id}/cover/'
        self.assertEqual(self.post(url, {'cover_ct': 'x', 'cover_iv': 'a1b2c3d4'}, **bob_auth).status_code, 404)
        self.assertEqual(self.client.get(url, **bob_auth).status_code, 404)

    def test_prefs_are_one_encrypted_blob(self):
        self.assertEqual(self.client.get('/books/prefs/', **self.auth).json(), {'prefs_ct': '', 'prefs_iv': ''})
        self.post('/books/prefs/', {'prefs_ct': 'AAA', 'prefs_iv': 'a1b2c3d4e5f6'})
        self.post('/books/prefs/', {'prefs_ct': 'BBB', 'prefs_iv': 'a1b2c3d4e5f7'})
        self.assertEqual(self.client.get('/books/prefs/', **self.auth).json(), {'prefs_ct': 'BBB', 'prefs_iv': 'a1b2c3d4e5f7'})


class BookRowsReadPastNoFileTests(TestCase):
    """The list must not read a column that stands after the file (data_ct) in a book's row:
    SQLite would walk through the whole file to get there (docs/database.md)."""

    def test_the_list_queries_use_covering_indexes(self):
        from django.contrib.auth.models import User
        from django.db import connection
        from django.test.utils import CaptureQueriesContext
        from books.models import EBook, book_rows
        user = User.objects.create_user('rows', password='x')
        EBook.objects.create(user=user, meta_ct='m', meta_iv='i', data_ct='d' * 100000, data_iv='j', cover_iv='c', cover_ct='cc')
        EBook.objects.create(user=user, meta_ct='m2', meta_iv='i2', data_ct='d', data_iv='j')
        with CaptureQueriesContext(connection) as q:
            rows = book_rows(user)
        self.assertEqual([r['meta_ct'] for r in rows], ['m', 'm2'])
        self.assertEqual([r['has_cover'] for r in rows], [True, False])
        late = ('uploaded_at', 'is_read', 'cover_iv', 'cover_ct', 'data_ct', 'data_iv')
        with connection.cursor() as cur:
            for query in q.captured_queries:
                sql = query['sql']
                if any(col in sql for col in late):
                    cur.execute('EXPLAIN QUERY PLAN ' + sql.replace('%s', '?'), [user.id] if '?' in sql or '%s' in sql else [])
                    plan = ' '.join(str(r) for r in cur.fetchall())
                    self.assertIn('COVERING INDEX', plan, sql)


class EBookFileTests(TestCase):
    """The file lives in EBookFile; upload, data, replace-data and rekey go there."""

    def setUp(self):
        from django.contrib.auth.models import User
        self.user = User.objects.create_user('filer', password='pw-filer-123')
        self.client.force_login(self.user)

    def test_upload_read_replace_rekey_delete(self):
        import base64, json
        from books.models import EBook, EBookFile
        data = base64.b64encode(b'x' * 100).decode()
        r = self.client.post('/books/upload/', json.dumps({'meta_ct': 'm', 'meta_iv': 'i', 'data_ct': data, 'data_iv': 'd'}),
                             content_type='application/json')
        pk = r.json()['id']
        self.assertEqual(EBookFile.objects.get(book_id=pk).data_ct, data)
        self.assertEqual(self.client.get(f'/books/{pk}/data/').json(), {'data_ct': data, 'data_iv': 'd'})
        self.client.post(f'/books/{pk}/replace-data/', json.dumps({'data_ct': 'bmV3', 'data_iv': 'd2'}), content_type='application/json')
        self.assertEqual(self.client.get(f'/books/{pk}/data/').json()['data_ct'], 'bmV3')
        self.client.post(f'/books/{pk}/rekey/', json.dumps({'meta_ct': 'm3', 'meta_iv': 'i3', 'data_ct': 'cmVr', 'data_iv': 'd3'}),
                         content_type='application/json')
        self.assertEqual(EBook.objects.get(pk=pk).meta_ct, 'm3')
        self.assertEqual(EBookFile.objects.get(book_id=pk).data_iv, 'd3')
        self.client.post(f'/books/{pk}/delete/')
        self.assertFalse(EBookFile.objects.filter(book_id=pk).exists())

    def test_someone_elses_file_is_not_found(self):
        from django.contrib.auth.models import User
        from books.models import EBook
        other = User.objects.create_user('other-filer', password='pw-other-123')
        b = EBook.objects.create(user=other, meta_ct='m', meta_iv='i', data_ct='ZA==', data_iv='i')
        self.assertEqual(self.client.get(f'/books/{b.id}/data/').status_code, 404)