mrwnslz
1cb6584395
Eine App mit alter Bücherliste (Desktop während des langsamen Syncs) hat beim Öffnen fünf MEW-Bände mit ihrem alten Stand überschrieben – Titel, Autor, Ordner weg. base_iv sagt dem Server, von welchem Stand eine Änderung ausgeht. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
292 lines
15 KiB
Python
292 lines
15 KiB
Python
"""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)
|
|
|
|
|
|
class StaleMetaTests(TestCase):
|
|
"""A meta write that starts from an older meta (base_iv) is refused with the current one."""
|
|
|
|
def setUp(self):
|
|
from django.contrib.auth.models import User
|
|
from books.models import EBook
|
|
self.user = User.objects.create_user('metauser', password='pw-meta-123')
|
|
self.client.force_login(self.user)
|
|
self.book = EBook.objects.create(user=self.user, meta_ct='m1', meta_iv='aa', data_ct='ZA==', data_iv='i')
|
|
|
|
def post(self, body):
|
|
import json
|
|
return self.client.post(f'/books/{self.book.id}/meta/', json.dumps(body), content_type='application/json')
|
|
|
|
def test_current_base_is_taken(self):
|
|
self.assertEqual(self.post({'meta_ct': 'm2', 'meta_iv': 'bb', 'base_iv': 'aa'}).status_code, 200)
|
|
self.book.refresh_from_db()
|
|
self.assertEqual((self.book.meta_ct, self.book.meta_iv), ('m2', 'bb'))
|
|
|
|
def test_old_base_is_refused_with_the_current_meta(self):
|
|
self.post({'meta_ct': 'm2', 'meta_iv': 'bb', 'base_iv': 'aa'})
|
|
r = self.post({'meta_ct': 'old', 'meta_iv': 'cc', 'base_iv': 'aa'})
|
|
self.assertEqual(r.status_code, 409)
|
|
self.assertEqual((r.json()['meta_ct'], r.json()['meta_iv']), ('m2', 'bb'))
|
|
self.book.refresh_from_db()
|
|
self.assertEqual(self.book.meta_ct, 'm2')
|
|
|
|
def test_without_base_as_before(self):
|
|
self.assertEqual(self.post({'meta_ct': 'm3', 'meta_iv': 'dd'}).status_code, 200)
|