1import copy
2from contextlib import contextmanager
3import json
4from pathlib import Path
5import subprocess
6import tempfile
7import unittest
8from unittest.mock import patch
9import native_stress
10from native_stress import edit_history, native_history, verify_capture
11from native_collaboration import reachable_page_text
12from document_model import DEFAULT_CONTEXT, EXPORTER
13
14
15class NativeHistoryTests(unittest.TestCase):
16 def test_conflict_oracle_excludes_unreferenced_history_and_spaces(self):
17 root = Path(__file__).resolve().parent.parent
18 with tempfile.TemporaryDirectory() as folder:
19 output = Path(folder) / 'model'
20 subprocess.run([EXPORTER,
21 root / 'corpus/collaboration/round-01/offline/notebook/synthetic.one', output], check=True)
22 model = json.loads((output / 'document.json').read_text())
23 retained = reachable_page_text(model)
24 self.assertEqual(list(retained.values()), [['Rust edited during the native outage.'],
25 ['Native edited while disconnected.']])
26 main, conflict = retained
27 current = model['spaces'][main]
28 revision = current['revisions'][current['contexts'][DEFAULT_CONTEXT]]
29 current['revisions']['unreferenced'] = copy.deepcopy(revision)
30 for node in current['revisions']['unreferenced']['nodes'].values():
31 if node['kind']['type'] == 'RichText': node['kind']['text'] = 'History is not recovery'
32 self.assertEqual(reachable_page_text(model), retained)
33 revision['nodes'][revision['roots']['1']]['spaces'] = []
34 self.assertIn(conflict, model['spaces'])
35 self.assertEqual(reachable_page_text(model), {main: retained[main]})
36 page, = revision['nodes'][revision['roots']['1']]['content']
37 revision['nodes'][page]['children'] *= 2
38 self.assertEqual(reachable_page_text(model), {main: retained[main] * 2})
39
40 def setUp(self):
41 self.logs = {}
42 text = 'Concurrent edits:'
43 for i in range(2):
44 token = f' [w{i}:0]'
45 at = len(text.encode('utf-16-le')) // 2
46 self.logs[f'w{i}'] = [{'event': 'ready'},
47 {'event': 'intent', 'attempt': 1, 'operation': 0, 'token': token,
48 'replacement': token, 'before': text, 'range': [at, at], 'source_transaction': 1},
49 {'event': 'commit', 'attempt': 1, 'operation': 0, 'token': token,
50 'source_transaction': 1, 'started_us': i * 20, 'finished_us': i * 20 + 10},
51 {'event': 'done'}]
52 text += token
53 self.logs['r0'] = [{'event': 'ready'}, {'event': 'read', 'text': text,
54 'started_us': 40, 'finished_us': 50}, {'event': 'done'}]
55
56 def test_incomplete_capture_requires_explicit_retention_mode(self):
57 self.logs['w0'].pop()
58 with self.assertRaisesRegex(AssertionError, 'Incomplete client log'):
59 edit_history(self.logs, 2)
60 self.assertEqual(len(edit_history(self.logs, 2, partial=True)[0]), 2)
61 self.logs['w1'][2]['operation'] = 1
62 with self.assertRaisesRegex(AssertionError, 'acknowledgements'):
63 edit_history(self.logs, 2, partial=True)
64 self.logs['w1'][2]['operation'] = 0
65 self.logs['w1'].pop(2)
66 with self.assertRaisesRegex(AssertionError, 'reader observed'):
67 edit_history(self.logs, 2, partial=True)
68
69 def test_counter_renumbering_preserves_one_content_history(self):
70 commits, text = edit_history(self.logs, 1)
71 self.assertEqual(len(commits), 2)
72 self.assertEqual(text, self.logs['r0'][1]['text'])
73
74 def test_branches_unexplained_edits_and_incorrect_observations_fail(self):
75 for field, value in [('before', 'Concurrent edits:'), ('before', 'unknown'),
76 ('range', [0, 0]), ('replacement', 'wrong')]:
77 logs = copy.deepcopy(self.logs)
78 logs['w1'][1][field] = value
79 with self.assertRaises(AssertionError): edit_history(logs, 1)
80 for field, value in [('text', 'Concurrent edits:'), ('text', 'unknown'), ('finished_us', 1)]:
81 logs = copy.deepcopy(self.logs)
82 logs['r0'][1][field] = value
83 with self.assertRaises(AssertionError): edit_history(logs, 1)
84
85 def test_content_branch_and_disconnected_history_are_detected(self):
86 for before, message in [('Concurrent edits:', 'branched'), ('Concurrent edits: unexplained', 'complete content history')]:
87 logs = copy.deepcopy(self.logs)
88 logs['w1'][1]['before'] = before
89 offset = len(before.encode('utf-16-le')) // 2
90 logs['w1'][1]['range'] = [offset, offset]
91 with self.assertRaisesRegex(AssertionError, message): edit_history(logs, 1)
92
93 def test_replacements_chain_by_content_after_native_counter_renumbering(self):
94 expected = 'Concurrent edits:'
95 for i in range(2):
96 events = self.logs[f'w{i}']
97 intent = events[1]
98 intent['before'] = expected
99 intent['range'] = [len('Concurrent edits:'), len(expected.encode('utf-16-le')) // 2]
100 intent['replacement'] = ' café 🦀' + intent['token']
101 expected = 'Concurrent edits:' + intent['replacement']
102 self.logs['r0'][1]['text'] = expected
103 commits, text = edit_history(self.logs, 1, edit=True)
104 self.assertEqual((len(commits), text), (2, expected))
105 self.logs['w1'][1]['range'][0] = len('Concurrent edits: café ') + 1
106 with self.assertRaises(UnicodeDecodeError): edit_history(self.logs, 1, edit=True)
107
108 def test_missing_acknowledgement_is_not_silently_dropped(self):
109 self.logs['w1'].pop(2)
110 with self.assertRaises(AssertionError): edit_history(self.logs, 1)
111
112 def test_native_replacement_history_rejects_lost_edits_and_split_surrogates(self):
113 events = [{'operation': 0, 'token': ' [n0:0]', 'before': 'Native 0:',
114 'range': [9, 9], 'replacement': ' café 🦀 [n0:0]'},
115 {'operation': 1, 'token': ' [n0:1]', 'before': 'Native 0: café 🦀 [n0:0]',
116 'range': [10, 17], 'replacement': ' café 🦀 [n0:1]'}]
117 self.assertEqual(native_history(events, 0, 2, True), 'Native 0: café 🦀 [n0:1] [n0:0]')
118 for field, value in [('before', 'Native 0:'), ('range', [15, 16]),
119 ('operation', 0), ('replacement', 'unrecorded')]:
120 changed = copy.deepcopy(events)
121 changed[1][field] = value
122 with self.assertRaises((AssertionError, UnicodeDecodeError)):
123 native_history(changed, 0, 2, True)
124 with self.assertRaises(AssertionError): native_history(events[:-1], 0, 2, True)
125
126 def test_cold_capture_rejects_extra_missing_and_misformatted_content(self):
127 with tempfile.TemporaryDirectory() as folder:
128 root = Path(folder)
129 (root / 'rust').mkdir()
130 (root / 'n0').mkdir()
131 (root / 'run.json').write_text(json.dumps({'stress_operations': 1, 'edit': True,
132 'rust_writers': 2, 'rust_readers': 1, 'stress_clients': 1}))
133 text = 'Concurrent edits:'
134 for i in range(2):
135 events = self.logs[f'w{i}']
136 events[1].update(before=text, range=[17, len(text.encode('utf-16-le')) // 2],
137 replacement=' café 🦀' + events[1]['token'])
138 text = 'Concurrent edits:' + events[1]['replacement']
139 self.logs['r0'][1]['text'] = text
140 for actor, events in self.logs.items():
141 (root / 'rust' / f'{actor}.jsonl').write_text('\n'.join(map(json.dumps, events)))
142 event = {'operation': 0, 'token': ' [n0:0]', 'before': 'Native 0:',
143 'range': [9, 9], 'replacement': ' café 🦀 [n0:0]', 'bold': True, 'italic': False}
144 (root / 'n0/stress-events.jsonl').write_text(json.dumps(event), encoding='utf-8-sig')
145 native = '<OE style="font-weight:bold"><T>Native 0: café 🦀 [n0:0]</T></OE>'
146 body = f'<OE><T>{text}</T></OE>{native}'
147 page = root / 'page-000.xml'
148 page.write_text(f'<Page xmlns="http://schemas.microsoft.com/office/onenote/2010/onenote"><Outline>{body}</Outline></Page>')
149 result = verify_capture(root, root)
150 self.assertEqual((result['rust_intents'], result['native_intents'], result['exact_paragraphs']), (2, 1, 2))
151 self.assertGreater(result['native_intended_format_checks'], 0)
152 for changed in (body + native, body.replace(native, ''), body.replace('bold', 'normal')):
153 page.write_text(f'<Page xmlns="http://schemas.microsoft.com/office/onenote/2010/onenote"><Outline>{changed}</Outline></Page>')
154 with self.assertRaises(AssertionError): verify_capture(root, root)
155
156
157class FailureArtifacts(unittest.TestCase):
158 def test_failed_clients_preserve_native_logs_without_masking_the_failure(self):
159 @contextmanager
160 def failed_clients(*args, **kwargs):
161 yield {}
162 raise RuntimeError('Rust client exited')
163
164 with tempfile.TemporaryDirectory() as directory:
165 output = Path(directory)
166 (output / 'run.json').write_text(json.dumps({'maintenance': False}))
167 shared = output / 'shared'
168 shared.mkdir()
169 clients = [{'name': f'n{i}', 'folder': output / f'n{i}'} for i in range(2)]
170 for client in clients: client['folder'].mkdir()
171
172 def capture(remote, local, name):
173 self.assertTrue(remote.endswith('\\outbox\\7\\events.jsonl'))
174 if name == 'n0': raise OSError('Native client unavailable')
175 local.write_text('{"operation":0}\n')
176 return {'error': None}
177
178 with patch.object(native_stress, 'running_clients', failed_clients), \
179 patch.object(native_stress.windows, 'do_health', return_value={'utc_us': 0}), \
180 patch.object(native_stress.windows, 'do_cmd', return_value={'stdout': 'ready editing'}), \
181 patch.object(native_stress.windows, 'do_get', side_effect=capture):
182 with self.assertRaisesRegex(RuntimeError, 'Rust client exited'):
183 native_stress.exercise(output, shared, clients, lambda *a, **kw: 7,
184 lambda *a: None, lambda *a: None, lambda *a: None, 40, 1, 4, 4)
185 self.assertEqual(json.loads((output / 'n0/stress-capture.json').read_text())['error'], 'Native client unavailable')
186 self.assertEqual((output / 'n1/stress-events.jsonl').read_text(), '{"operation":0}\n')