| 1 | import copy |
| 2 | from pathlib import Path |
| 3 | import json |
| 4 | import subprocess |
| 5 | import tempfile |
| 6 | import unittest |
| 7 | from random_edit_campaign import EDITOR, ROOT, export, verify |
| 8 | from document_model import ordered_pages, view |
| 9 | |
| 10 | |
| 11 | class RandomEditOracleTests(unittest.TestCase): |
| 12 | def test_a_random_replacement_cannot_be_a_no_op(self): |
| 13 | with tempfile.TemporaryDirectory() as temporary: |
| 14 | source = Path(temporary) / 'source.one' |
| 15 | subprocess.run([ROOT / 'target/debug/examples/create_section', source, ' café ', 'No-op regression'], check=True) |
| 16 | before = export(source) |
| 17 | for mode in ['create', 'in-place']: |
| 18 | destination = Path(temporary) / (mode + '.one') |
| 19 | if mode == 'create': |
| 20 | args = [EDITOR, source, destination, '301'] |
| 21 | else: |
| 22 | destination.write_bytes(source.read_bytes()) |
| 23 | args = [EDITOR, destination, '--in-place', '301'] |
| 24 | record = json.loads(subprocess.check_output(args)) |
| 25 | self.assertEqual(record['range'], [0, 0]) |
| 26 | self.assertNotEqual(source.read_bytes(), destination.read_bytes()) |
| 27 | verify(before, export(destination), record) |
| 28 | |
| 29 | def test_native_fixture_splice_and_unrelated_damage(self): |
| 30 | source = ROOT / 'corpus/native/20260905-05/snapshots/06-attachment/notebook/synthetic.one' |
| 31 | with tempfile.TemporaryDirectory() as temporary: |
| 32 | destination = Path(temporary) / 'edited.one' |
| 33 | record = json.loads(subprocess.check_output([EDITOR, source, destination, '42'])) |
| 34 | before, after = export(source), export(destination) |
| 35 | verify(before, after, record) |
| 36 | sid, oid = record['space'], record['object'] |
| 37 | rid = after[0]['spaces'][sid]['contexts']['{00000000-0000-0000-0000-000000000000},0'] |
| 38 | for kind in ['format', 'metadata', 'payload', 'structure', 'ancestor-time', 'unrelated-time', |
| 39 | 'revision-time', 'revision-author', 'revision-extra', 'missing-attribution', |
| 40 | 'author-name', 'author-initials', 'paragraph-author', 'original-author']: |
| 41 | damaged = copy.deepcopy(after) |
| 42 | if kind == 'format': damaged[1][sid][rid][oid][0]['format']['bold'] = not damaged[1][sid][rid][oid][0]['format']['bold'] |
| 43 | if kind == 'metadata': damaged[0]['spaces'][sid]['revisions'][rid]['nodes'][oid]['created'] = 0 |
| 44 | if kind == 'payload': damaged[2].append('another payload') |
| 45 | if kind == 'structure': damaged[0]['spaces'][sid]['revisions'][rid]['nodes'][oid]['children'].append('another object') |
| 46 | if kind == 'ancestor-time': damaged[0]['spaces'][sid]['revisions'][rid]['nodes'][record['page']]['modified'] = 0 |
| 47 | nodes = damaged[0]['spaces'][sid]['revisions'][rid]['nodes'] |
| 48 | metadata = nodes[after[0]['spaces'][sid]['revisions'][rid]['roots']['4']] |
| 49 | author_id = metadata['latest_author'] |
| 50 | paragraph = next(key for key, node in nodes.items() |
| 51 | if node['kind']['type'] == 'Paragraph' and oid in node['content']) |
| 52 | if kind == 'revision-time': metadata['kind']['modified_filetime'] = 0 |
| 53 | if kind == 'revision-author': metadata['latest_author'] = paragraph |
| 54 | if kind == 'revision-extra': metadata['extra'][0].append({'id': 123, 'value': {'Bytes': '00'}}) |
| 55 | if kind == 'missing-attribution': del damaged[0]['spaces'][sid]['revisions'][rid]['roots']['4'] |
| 56 | if kind == 'author-name': nodes[author_id]['kind']['name'] = 'Another author' |
| 57 | if kind == 'author-initials': nodes[author_id]['extra'][0][0]['value']['Bytes'] = '0000' |
| 58 | if kind == 'paragraph-author': nodes[paragraph]['latest_author'] = nodes[paragraph]['original_author'] |
| 59 | if kind == 'original-author': nodes[paragraph]['original_author'] = author_id |
| 60 | if kind == 'unrelated-time': |
| 61 | old_rid = before[0]['spaces'][sid]['contexts']['{00000000-0000-0000-0000-000000000000},0'] |
| 62 | old = before[0]['spaces'][sid]['revisions'][old_rid]['nodes'] |
| 63 | nodes = damaged[0]['spaces'][sid]['revisions'][rid]['nodes'] |
| 64 | key = next(key for key, node in nodes.items() if node['modified'] is not None and node['modified'] == old[key]['modified']) |
| 65 | nodes[key]['modified'] = 0 |
| 66 | with self.assertRaises(AssertionError, msg=kind): verify(before, damaged, record) |
| 67 | |
| 68 | def test_repeated_edits_reuse_attribution_and_preserve_existing_author_data(self): |
| 69 | source = ROOT / 'corpus/native/20260905-05/snapshots/06-attachment/notebook/synthetic.one' |
| 70 | with tempfile.TemporaryDirectory() as temporary: |
| 71 | destination = Path(temporary) / 'edited.one' |
| 72 | record = json.loads(subprocess.check_output([EDITOR, source, destination, '42'])) |
| 73 | before = export(destination) |
| 74 | _, previous = view(before[0], record['space']) |
| 75 | author_id = previous['nodes'][previous['roots']['4']]['latest_author'] |
| 76 | record = json.loads(subprocess.check_output([EDITOR, destination, '--in-place', '43', record['page']])) |
| 77 | after = export(destination) |
| 78 | verify(before, after, record) |
| 79 | _, current = view(after[0], record['space']) |
| 80 | self.assertEqual(current['roots']['4'], previous['roots']['4']) |
| 81 | self.assertEqual(current['nodes'][current['roots']['4']]['latest_author'], author_id) |
| 82 | self.assertEqual(current['nodes'][author_id], previous['nodes'][author_id]) |
| 83 | damaged = copy.deepcopy(after) |
| 84 | _, damaged_revision = view(damaged[0], record['space']) |
| 85 | damaged_revision['nodes'][author_id]['extra'][0][0]['value']['Bytes'] = '0000' |
| 86 | with self.assertRaises(AssertionError): |
| 87 | verify(before, damaged, record) |
| 88 | |
| 89 | def test_seeded_edits_work_on_native_legacy_text(self): |
| 90 | source = ROOT / 'corpus/native/20260905-05/snapshots/02-text/notebook/synthetic.one' |
| 91 | with tempfile.TemporaryDirectory() as temporary: |
| 92 | for seed in range(42, 74): |
| 93 | destination = Path(temporary) / f'{seed}.one' |
| 94 | result = subprocess.run([EDITOR, source, destination, str(seed)], capture_output=True, text=True) |
| 95 | self.assertEqual(result.returncode, 0, (seed, result.stderr)) |
| 96 | self.assertTrue(destination.is_file()) |
| 97 | |
| 98 | def test_title_cache_changes_are_required_and_bounded(self): |
| 99 | source = ROOT / 'corpus/m6/native-structure-01/notebook/synthetic.one' |
| 100 | before = export(source) |
| 101 | sid, _, revision, page = next(p for p in ordered_pages(before[0]) if any( |
| 102 | n['kind']['type'] == 'RichText' and any(f['id'] == 0x88001cb4 for f in n['extra'][0]) |
| 103 | for n in p[2]['nodes'].values())) |
| 104 | with tempfile.TemporaryDirectory() as temporary: |
| 105 | destination = Path(temporary) / 'edited.one' |
| 106 | record = json.loads(subprocess.check_output([EDITOR, source, destination, '98', page])) |
| 107 | self.assertTrue(any(f['id'] == 0x88001cb4 for f in revision['nodes'][record['object']]['extra'][0])) |
| 108 | after = export(destination) |
| 109 | verify(before, after, record) |
| 110 | rid = after[0]['spaces'][sid]['contexts']['{00000000-0000-0000-0000-000000000000},0'] |
| 111 | damaged = copy.deepcopy(after) |
| 112 | damaged[0]['spaces'][sid]['revisions'][rid]['nodes'][revision['roots']['2']]['kind']['title'] = revision['nodes'][revision['roots']['2']]['kind']['title'] |
| 113 | with self.assertRaises(AssertionError): verify(before, damaged, record) |
| 114 | |
| 115 | |
| 116 | if __name__ == '__main__': |
| 117 | unittest.main() |