1import json
2from pathlib import Path
3import tempfile
4from threading import Thread
5import unittest
6from unittest.mock import patch
7from urllib.error import HTTPError
8from urllib.parse import urlencode
9from urllib.request import Request, urlopen
10
11from document_model import view, walk
12import notebook_editor as editor
13from random_edit_campaign import export, verify
14
15
16class EditorTest(unittest.TestCase):
17 def setUp(self):
18 self.temporary = tempfile.TemporaryDirectory()
19 self.addCleanup(self.temporary.cleanup)
20 source = editor.ROOT / 'corpus/native/20260905-05/snapshots/06-attachment/notebook'
21 self.session = editor.Session(source, Path(self.temporary.name) / 'session')
22 self.server = editor.ThreadingHTTPServer(('127.0.0.1', 0), editor.Handler)
23 self.server.session = self.session
24 self.thread = Thread(target=self.server.serve_forever)
25 self.thread.start()
26 self.addCleanup(self.stop)
27 self.url = f'http://127.0.0.1:{self.server.server_port}'
28 self.file = self.session.output / 'notebook/synthetic.one'
29 self.before = export(self.file)
30 self.row = json.loads((self.session.output / 'g/0/report/pages.json').read_text())[0]
31 _, revision = view(self.before[0], self.row['space'])
32 self.oid, node = next((oid, node) for oid, node in walk(revision, self.row['object'])
33 if node['kind']['type'] == 'RichText' and node['kind']['text'].startswith('Fictitious'))
34 self.selection = {'generation': 0, 'page': self.row['report'], 'object': self.oid, 'run': 0, 'action': 'text'}
35 self.text = self.before[1][self.row['space']][self.row['revision']][self.oid][0]['text']
36
37 def stop(self):
38 self.server.shutdown()
39 self.thread.join()
40 self.server.server_close()
41
42 def request(self, path, data=None):
43 request = Request(self.url + path, data=json.dumps(data).encode() if data is not None else None,
44 headers={'Content-Type': 'application/json', 'X-OneNote-Diagnostic': '1'})
45 try:
46 response = urlopen(request)
47 except HTTPError as error:
48 response = error
49 with response:
50 return response.status, json.loads(response.read())
51
52 def test_unicode_stale_snapshot_and_unrelated_content(self):
53 status, checked = self.request('/api/run?' + urlencode(self.selection))
54 self.assertEqual((status, checked['ok'], checked['text']), (200, True, self.text))
55 replacement = self.text + ' café 🦀 e\u0301 <diagnostic>'
56 status, saved = self.request('/api/save', {**self.selection, 'replacement': replacement})
57 self.assertEqual((status, saved['state'], saved['ok']), (200, 'Committed', True))
58 intent, outcome = [json.loads(s) for s in (self.session.output / 'operations.jsonl').read_text().splitlines()]
59 edit = intent['edit']
60 verify(self.before, export(self.file), {'page': self.row['object'], 'space': edit['space'], 'object': self.oid,
61 'range': [edit['action']['start'], edit['action']['end']], 'replacement': replacement, 'run_start': edit['action']['start'],
62 'run_before': self.text, 'started_ms': intent['started_ms'], 'finished_ms': outcome['finished_ms']})
63 after = self.file.read_bytes()
64 fresh = self.session.output / 'fresh-report'
65 editor.generate(self.session.output / 'g/1/snapshot', fresh, editable=True)
66 cached = self.session.output / 'g/1/report'
67 self.assertEqual({p.relative_to(fresh): p.read_bytes() for p in fresh.rglob('*') if p.is_file()},
68 {p.relative_to(cached): p.read_bytes() for p in cached.rglob('*') if p.is_file()})
69 status, stale = self.request('/api/save', {**self.selection, 'replacement': 'another draft'})
70 self.assertEqual((status, stale['state'], stale['kind']), (409, 'NotCommitted', 'ResourceBusy'))
71 self.assertEqual(self.file.read_bytes(), after)
72 self.assertEqual((self.session.output / 'g/0/snapshot/synthetic.one').read_bytes(),
73 (editor.ROOT / 'corpus/native/20260905-05/snapshots/06-attachment/notebook/synthetic.one').read_bytes())
74 self.assertEqual((self.session.output / 'g/1/snapshot/synthetic.one').stat().st_ino,
75 (self.session.output / 'g/2/snapshot/synthetic.one').stat().st_ino)
76 with urlopen(self.url + saved['location']) as response:
77 html = response.read().decode()
78 self.assertIn('&lt;diagnostic&gt;', html)
79 self.assertNotIn('<diagnostic>', html)
80
81 def test_invalid_edits_and_selection_never_write(self):
82 before = self.file.read_bytes()
83 for changes in [{'replacement': 'first\nsecond'}, {'replacement': '\ud800'},
84 {'run': -1}, {'generation': True}, {'object': 'missing'}, {'extra': 'field'}]:
85 status, result = self.request('/api/save', {**self.selection, 'replacement': 'changed', **changes})
86 self.assertEqual((status, result['state']), (422, 'NotCommitted'))
87 self.assertEqual(self.file.read_bytes(), before)
88
89 def test_unknown_response_does_not_replay_a_real_commit(self):
90 original = editor.bridge
91 calls = []
92 def uncertain(mode, *args):
93 result = original(mode, *args)
94 if mode == 'commit':
95 calls.append(result)
96 self.assertTrue(result['ok'])
97 return {'ok': False, 'state': 'Unknown', 'error': 'Simulated lost outcome'}
98 return result
99 with patch.object(editor, 'bridge', side_effect=uncertain):
100 status, result = self.request('/api/save', {**self.selection, 'replacement': self.text + ' once'})
101 self.assertEqual((status, result['state'], len(calls)), (503, 'Unknown', 1))
102 after = export(self.file)
103 rid, _ = view(after[0], self.row['space'])
104 self.assertEqual(after[1][self.row['space']][rid][self.oid][0]['text'], self.text + ' once')
105 status, stale = self.request('/api/save', {**self.selection, 'replacement': self.text + ' twice'})
106 self.assertEqual((status, stale['kind']), (409, 'ResourceBusy'))
107
108 def test_refresh_failure_preserves_committed_outcome(self):
109 with patch.object(self.session, 'snapshot', side_effect=OSError('Report storage unavailable')):
110 status, result = self.request('/api/save', {**self.selection, 'replacement': self.text + ' saved'})
111 self.assertEqual((status, result['state'], result['ok']), (503, 'Committed', False))
112 self.assertIn('Report storage', result['report_error'])
113 after = export(self.file)
114 rid, _ = view(after[0], self.row['space'])
115 self.assertEqual(after[1][self.row['space']][rid][self.oid][0]['text'], self.text + ' saved')
116 with urlopen(self.url + '/latest?' + urlencode(self.selection)) as response:
117 self.assertIn(' saved', response.read().decode())
118
119 def test_document_actions_preserve_placement_unicode_and_unselected_formatting(self):
120 status, page = self.request('/api/page?' + urlencode(self.selection))
121 self.assertEqual(status, 200)
122 outline = next(target for target in page['targets'] if target['label'].startswith('Outline'))
123 base = {'generation': 0, 'page': self.row['report'], 'object': outline['object'], 'action': 'paragraph',
124 'before': outline['children'][0]['object'], 'text': 'A🦀 café\rsecond line', 'author': 'Diagnostic test'}
125 status, saved = self.request('/api/save', base)
126 self.assertEqual((status, saved['state']), (200, 'Committed'))
127 after = export(self.file)
128 _, revision = view(after[0], self.row['space'])
129 children = revision['nodes'][outline['object']]['children']
130 self.assertEqual(children[1], base['before'])
131 oid, node = next((key, node) for key, node in walk(revision, children[0]) if node['kind']['type'] == 'RichText')
132 self.assertEqual(node['kind']['text'], base['text'])
133 self.assertEqual(self.before[2], after[2])
134 attrs = [{'Bold': True}, {'Italic': True}, {'Underline': True}, {'Strike': True},
135 {'Superscript': True}, {'Subscript': False}, {'Font': 'Arial'}, {'FontSize': 20.5},
136 {'Color': [18, 52, 86]}, {'Highlight': [255, 255, 0]}]
137 selected = {'generation': 1, 'page': saved['location'].split('/')[-1], 'object': oid,
138 'run': 0, 'action': 'format', 'start': 1, 'end': 3, 'attributes': attrs}
139 status, formatted = self.request('/api/save', selected)
140 self.assertEqual((status, formatted['state']), (200, 'Committed'))
141 after_format = export(self.file)
142 rid, _ = view(after_format[0], self.row['space'])
143 runs = after_format[1][self.row['space']][rid][oid]
144 self.assertEqual([run['text'] for run in runs], ['A', '🦀', ' café\rsecond line'])
145 original_rid, _ = view(after[0], self.row['space'])
146 original_format = after[1][self.row['space']][original_rid][oid][0]['format']
147 self.assertEqual(runs[0]['format'], original_format)
148 self.assertEqual(runs[2]['format'], original_format)
149 for key, value in {'bold': True, 'italic': True, 'underline': True, 'strike': True,
150 'superscript': True, 'subscript': False, 'font': 'Arial', 'font_size': 20.5,
151 'color': 0x563412, 'highlight': 0xffff}.items():
152 self.assertEqual(runs[1]['format'][key], value, key)
153 status, cleared = self.request('/api/save', {**selected, 'generation': 2, 'run': 1, 'start': 0, 'end': 2,
154 'attributes': [{'Subscript': True}, {'Color': None}, {'Highlight': None}]})
155 self.assertEqual(status, 200)
156 cleared_model = export(self.file)
157 rid, _ = view(cleared_model[0], self.row['space'])
158 fmt = cleared_model[1][self.row['space']][rid][oid][1]['format']
159 self.assertEqual((fmt['superscript'], fmt['subscript'], fmt['color'], fmt['highlight']), (False, True, 0xff000000, 0xff000000))
160 status, added = self.request('/api/save', {'generation': 3, 'page': selected['page'], 'action': 'outline',
161 'object': self.row['object'], 'x': 216.5, 'y': 360,
162 'text': 'New outline 🦀', 'author': 'Diagnostic test'})
163 self.assertEqual(status, 200)
164 _, revision = view(export(self.file)[0], self.row['space'])
165 new_outline = revision['nodes'][revision['nodes'][self.row['object']]['children'][-1]]
166 self.assertEqual(new_outline['kind']['type'], 'Outline')
167 self.assertEqual((new_outline['layout']['x'], new_outline['layout']['y']), (216.5, 360))
168 committed = self.file.read_bytes()
169 for request in (base, selected):
170 status, stale = self.request('/api/save', request)
171 self.assertEqual((status, stale['state'], stale['kind']), (409, 'NotCommitted', 'ResourceBusy'))
172 self.assertEqual(self.file.read_bytes(), committed)
173
174 def test_document_rejections_do_not_publish(self):
175 before = self.file.read_bytes()
176 common = {'generation': 0, 'page': self.row['report'], 'object': self.oid}
177 requests = [
178 {**common, 'action': 'format', 'run': 0, 'start': 0, 'end': 1, 'attributes': []},
179 {**common, 'action': 'format', 'run': 0, 'start': True, 'end': 1, 'attributes': [{'Bold': True}]},
180 {**common, 'action': 'format', 'run': 0, 'start': 0, 'end': 99999, 'attributes': [{'Bold': True}]},
181 {**common, 'action': 'format', 'run': 0, 'start': 0, 'end': 1, 'attributes': [{'FontSize': 144}]},
182 {**common, 'action': 'format', 'run': 0, 'start': 0, 'end': 1, 'attributes': [{'Bold': True}, {'Bold': False}]},
183 {**common, 'action': 'outline', 'x': 1, 'y': 1, 'text': 'No page parent', 'author': 'test'},
184 {**common, 'action': 'paragraph', 'before': None, 'text': 'No paragraph parent', 'author': 'test'},
185 ]
186 for request in requests:
187 status, result = self.request('/api/save', request)
188 self.assertEqual((status, result['state']), (422, 'NotCommitted'))
189 self.assertEqual(self.file.read_bytes(), before)
190
191 def test_uncertain_insertion_has_one_publication_and_a_stale_retry(self):
192 request = {'generation': 0, 'page': self.row['report'], 'object': self.row['object'],
193 'action': 'outline', 'x': 144, 'y': 288, 'text': 'Only once 🦀', 'author': 'test'}
194 original = editor.bridge
195 def uncertain(mode, *args):
196 result = original(mode, *args)
197 return {'ok': False, 'state': 'Unknown'} if mode == 'commit' and result['ok'] else result
198 with patch.object(editor, 'bridge', side_effect=uncertain):
199 status, result = self.request('/api/save', request)
200 self.assertEqual((status, result['state']), (503, 'Unknown'))
201 status, result = self.request('/api/save', request)
202 self.assertEqual((status, result['kind']), (409, 'ResourceBusy'))
203 _, revision = view(export(self.file)[0], self.row['space'])
204 self.assertEqual(sum(n['kind'].get('text') == 'Only once 🦀' for _, n in walk(revision, self.row['object'])), 1)
205
206 def test_generated_fields_are_read_only(self):
207 source = editor.ROOT / 'corpus/m6/native-structure-01/notebook'
208 session = editor.Session(source, Path(self.temporary.name) / 'fields')
209 self.server.session = session
210 pages = json.loads((session.output / 'g/0/report/pages.json').read_text())
211 document = export(source / 'synthetic.one')[0]
212 for row in pages:
213 _, revision = view(document, row['space'])
214 fields = [oid for oid, node in walk(revision, row['object'])
215 if node['kind']['type'] == 'RichText' and node['kind']['boilerplate']]
216 if fields: break
217 self.assertTrue(fields)
218 status, result = self.request('/api/run?' + urlencode({'generation': 0, 'page': row['report'], 'object': fields[0], 'run': 0}))
219 self.assertEqual((status, result['ok']), (422, False))
220
221 def test_templates_keep_reader_only_controls(self):
222 session = editor.Session(editor.ROOT / 'corpus/m6/native-template-controls-01/notebook',
223 Path(self.temporary.name) / 'template')
224 self.server.session = session
225 pages = json.loads((session.output / 'g/0/report/pages.json').read_text())
226 protected = [row for row in pages if row['category'] == 'Default page template']
227 self.assertTrue(protected)
228 for row in protected:
229 status, result = self.request('/api/run?' + urlencode({'generation': 0, 'page': row['report'], 'object': row['object'], 'run': 0}))
230 self.assertEqual(status, 422)
231 self.assertIn('active page', result['error'])
232 html = (session.output / 'g/0/report' / row['report']).read_text()
233 self.assertNotIn('src="/editor.js"', html)
234
235
236if __name__ == '__main__':
237 unittest.main()