1import hashlib
2import json
3from pathlib import Path
4import tempfile
5import unittest
6from unittest.mock import patch
7
8from verify_offline_confirmation import verify
9
10
11class ConfirmationOracle(unittest.TestCase):
12 def test_native_images_must_match_the_confirmation_and_preserved_native_prefix(self):
13 with tempfile.TemporaryDirectory() as folder:
14 root = Path(folder)
15 cold = root / 'cold'
16 (root / 'rust/confirmations').mkdir(parents=True)
17 (root / 'n0').mkdir()
18 (cold / 'results/123-456').mkdir(parents=True)
19 (root / 'run.json').write_text(json.dumps(dict(rust_writers=1, rust_readers=0, stress_clients=1, stress_operations=1)))
20 rows = [dict(event='ready', pid=123), dict(event='remote_attempt', state='Unknown', revision='revision', space='space'),
21 dict(event='remote_confirm', capture='123-456.one', revisions={'space': ['revision']}, text='Concurrent edits: [w0:0]')]
22 (root / 'rust/w0.jsonl').write_text('\n'.join(map(json.dumps, rows)))
23 (root / 'n0/stress-events.jsonl').write_text(json.dumps(dict(operation=0, token=' [n0:0]', before='Native 0:')))
24 snapshot = root / 'rust/confirmations/123-456.one'
25 snapshot.write_bytes(b'captured snapshot')
26 sha = hashlib.sha256(snapshot.read_bytes()).hexdigest()
27 (cold / 'run.json').write_text(json.dumps(dict(inputs={'123-456.one': sha})))
28 result = dict(name='123-456', error=None, pages=1, source_sha256=sha, seconds=1)
29 (cold / 'results.json').write_text(json.dumps(result))
30 (cold / 'teardown.json').write_text(json.dumps(dict(absent=True)))
31 page = cold / 'results/123-456/page-0.xml'
32 def xml(text, native):
33 return f'<one:Page xmlns:one="http://schemas.microsoft.com/office/onenote/2010/onenote"><one:Outline><one:OEChildren><one:OE><one:T>{text}</one:T></one:OE><one:OE><one:T>{native}</one:T></one:OE></one:OEChildren></one:Outline></one:Page>'
34 page.write_text(xml('Concurrent edits: [w0:0]', 'Native 0:'))
35 with patch('verify_offline_confirmation.publication_links') as ledger:
36 self.assertEqual(verify(root, cold)['validated_paragraphs'], 2)
37 self.assertTrue(ledger.called)
38 config = json.loads((root / 'run.json').read_text())
39 config['stress_operations'] = 2
40 (root / 'run.json').write_text(json.dumps(config))
41 with self.assertRaisesRegex(AssertionError, 'Missing native acknowledgements'):
42 verify(root, cold)
43 self.assertFalse(verify(root, cold, partial=True)['complete_workload'])
44 config['stress_operations'] = 1
45 (root / 'run.json').write_text(json.dumps(config))
46 for text, native in [('Concurrent edits:', 'Native 0:'), ('Concurrent edits: [w0:0]', 'Native 0: [n0:1]')]:
47 page.write_text(xml(text, native))
48 with self.assertRaises(AssertionError): verify(root, cold)
49 page.write_text(xml('Concurrent edits: [w0:0]', 'Native 0: [n0:0]'))
50 self.assertEqual(verify(root, cold)['native_images'], 1)
51 snapshot.write_bytes(b'changed')
52 with self.assertRaises(AssertionError): verify(root, cold)
53 snapshot.write_bytes(b'captured snapshot')
54 config['document_operations'] = True
55 (root / 'run.json').write_text(json.dumps(config))
56 document = dict(text='x', runs=[dict(text='x', bold=True, size=18, color=0x563412)])
57 rows[1]['document_changes'] = {'target': document}
58 rows[2]['started_us'] = 200
59 observed = dict(event='read', finished_us=100, text=rows[2]['text'], documents={'target': document})
60 rows.insert(2, observed)
61 (root / 'rust/w0.jsonl').write_text('\n'.join(map(json.dumps, rows)))
62 markup = '<one:OE><one:T><![CDATA[<span style="font-weight:bold;font-size:18pt;color:#123456">x</span>]]></one:T></one:OE>'
63 content = xml('Concurrent edits: [w0:0]', 'Native 0:').replace('</one:OEChildren>', markup + '</one:OEChildren>')
64 page.write_text(content)
65 with patch('offline_document_history.document_history', return_value={'target': {'states': {'format': {'attempt': {'receipt_revision': 'revision'}}}}}) as documents:
66 result = verify(root, cold)
67 self.assertEqual(result['validated_paragraphs'], 3)
68 self.assertEqual(result['native_intended_format_checks'], 3)
69 documents.assert_called_once_with({'w0': rows}, 1)
70 rows[3]['revisions'] = {'space': ['current']}
71 rows[3]['current_revisions'] = {'space': 'current'}
72 documents.return_value['target']['states']['format']['attempt']['receipt_revision'] = 'current'
73 (root / 'rust/w0.jsonl').write_text('\n'.join(map(json.dumps, rows)))
74 self.assertEqual(verify(root, cold)['confirmed_revision'], 'current')
75 rows[3]['current_revisions'] = {'space': 'unrelated'}
76 (root / 'rust/w0.jsonl').write_text('\n'.join(map(json.dumps, rows)))
77 with self.assertRaisesRegex(AssertionError, 'current effect-confirmation'): verify(root, cold)
78 rows[3]['current_revisions'] = {'space': 'current'}
79 (root / 'rust/w0.jsonl').write_text('\n'.join(map(json.dumps, rows)))
80 page.write_text(content.replace('18pt', '19pt'))
81 with self.assertRaisesRegex(AssertionError, 'font size'): verify(root, cold)
82 page.write_text(content)
83 observed['documents'] = {}
84 (root / 'rust/w0.jsonl').write_text('\n'.join(map(json.dumps, rows)))
85 with self.assertRaisesRegex(AssertionError, 'omitted'): verify(root, cold)
86
87
88if __name__ == '__main__': unittest.main()