1import copy
2import hashlib
3import json
4from pathlib import Path
5import tempfile
6import unittest
7from unittest.mock import patch
8from offline_publication_crash import state, TOKEN, confirmation_only
9from verify_offline_recovery import verify
10
11
12class RecoveryOracleTests(unittest.TestCase):
13 def test_confirmation_cannot_hide_revision_or_content_writes(self):
14 source = bytes(1024)
15 changed = bytearray(source)
16 changed[212:252] = bytes(range(40))
17 events = [{'event':'write', 'offset':212, 'bytes':40}]
18 confirmation_only(events, source, changed)
19 for offset in (96, 211, 252, 1023):
20 bad = changed.copy()
21 bad[offset] ^= 1
22 with self.assertRaises(AssertionError): confirmation_only(events, source, bad)
23 for event in ({'event':'write','offset':96,'bytes':1}, {'event':'phase','name':'publish-before'}):
24 with self.assertRaises(AssertionError): confirmation_only([*events,event], source, changed)
25
26 def test_text_acknowledgements_and_revision_identity_must_agree(self):
27 original = 'Concurrent edits: 🦀'
28 at = len(original.encode('utf-16-le')) // 2
29 row = {'event': 'state', 'status': 'uncertain', 'revision': 'new', 'remote_revision': 'old',
30 'local_text': original+TOKEN, 'remote_text': original,
31 'pending': [{'id': 1, 'replacement': TOKEN, 'range': [at, at]}]}
32 self.assertEqual(state([row], original), row)
33 for field, value in [('revision', None), ('revision', 'old'), ('local_text', original),
34 ('remote_text', original+TOKEN+TOKEN), ('status', 'missing'), ('pending', [])]:
35 changed = copy.deepcopy(row)
36 changed[field] = value
37 with self.subTest(field=field):
38 with self.assertRaises(AssertionError): state([changed], original)
39 for field, value in [('id', 2), ('replacement', 'other'), ('range', [at-1, at-1])]:
40 changed = copy.deepcopy(row)
41 changed['pending'][0][field] = value
42 with self.assertRaises(AssertionError): state([changed], original)
43 row.update(status='published', remote_revision='new', remote_text=original+TOKEN, pending=[])
44 self.assertEqual(state([row], original), row)
45 for field, value in [('revision', 'old'), ('remote_text', original), ('pending', [{'id': 1}])]:
46 changed = copy.deepcopy(row)
47 changed[field] = value
48 with self.assertRaises(AssertionError): state([changed], original)
49
50 def test_native_inventory_hashes_errors_and_content_are_independently_checked(self):
51 with tempfile.TemporaryDirectory() as folder:
52 run, cold, source = [Path(folder)/name for name in ('run', 'cold', 'source.xml')]
53 (run/'images').mkdir(parents=True)
54 cold.mkdir()
55 data = b'owned fixture bytes'
56 digest = hashlib.sha256(data).hexdigest()
57 (run/'images'/f'{digest}.one').write_bytes(data)
58 expected = ['Concurrent edits:'+TOKEN, 'Other paragraph']
59 (run/'results.json').write_text(json.dumps([{'remote_image': digest, 'remote_text': expected[0]}]))
60 (cold/'run.json').write_text(json.dumps({'inputs': {digest+'.one': digest}}))
61 record = {'name': digest, 'source_sha256': digest, 'error': None, 'pages': 1, 'seconds': 1}
62 (cold/'results.json').write_text(json.dumps(record))
63 (cold/'teardown.json').write_text(json.dumps({'absent': True}))
64 (cold/'results'/digest).mkdir(parents=True)
65 (cold/'results'/digest/'page-0.xml').touch()
66 def content(path):
67 return ['Concurrent edits:', 'Other paragraph'] if path == source else expected
68 with patch('verify_offline_recovery.paragraphs', side_effect=content):
69 self.assertEqual(verify(run, cold, source)['exact_images'], 1)
70 for key, value in [('source_sha256', 'wrong'), ('error', 'failed'), ('pages', 0), ('name', 'wrong')]:
71 (cold/'results.json').write_text(json.dumps({**record, key:value}))
72 with self.subTest(key=key):
73 with self.assertRaises(AssertionError): verify(run, cold, source)
74 (cold/'results.json').write_text(json.dumps([record, record]))
75 with self.assertRaises(AssertionError): verify(run, cold, source)
76 (cold/'results.json').write_text(json.dumps(record))
77 expected[0] = 'Concurrent edits:'
78 with self.assertRaises(AssertionError): verify(run, cold, source)
79 expected[0] += TOKEN
80 (run/'images'/f'{digest}.one').write_bytes(b'changed')
81 with self.assertRaises(AssertionError): verify(run, cold, source)