1#!/usr/bin/env python3
2"""Edit a notebook on the owned Samba share through the application's session, relaunch,
3then cold-open the result in a disposable OneNote clone and compare its text."""
4import argparse
5import html
6import json
7import os
8import re
9import shutil
10import subprocess
11import sys
12import tarfile
13import xml.etree.ElementTree as ET
14from pathlib import Path
15
16ROOT = Path(__file__).resolve().parent.parent
17sys.path.insert(0, str(ROOT / 'tools/w7'))
18sys.path.insert(0, str(ROOT / 'tools'))
19import linux_vm # noqa: E402
20
21FIXTURE = ROOT / 'corpus/outline-edit/before/notebook'
22NS = {'one': 'http://schemas.microsoft.com/office/onenote/2010/onenote'}
23
24
25def native_texts(read):
26 pages = {}
27 for path in sorted(read.glob('page-*.xml')):
28 root = ET.parse(path).getroot()
29 texts = []
30 for element in root.iter('{%s}T' % NS['one']):
31 texts.append(html.unescape(re.sub(r'<[^>]*>', '', element.text or '')))
32 pages[root.get('name')] = texts
33 return pages
34
35
36def compare(expected, native):
37 """The client's edits carry its launch label; other text is compared by the corpus lanes."""
38 mismatches = []
39 for title, texts in expected.items():
40 body = [text for text in native[title] if text != title]
41 wanted = [text for text in texts if 'launch' in text]
42 assert wanted, title
43 for text in wanted:
44 if text not in body:
45 mismatches.append({'page': title, 'missing': text, 'native': body})
46 return mismatches
47
48
49def main():
50 parser = argparse.ArgumentParser(description=__doc__)
51 parser.add_argument('output', type=Path)
52 parser.add_argument('--linux', help='Disposable Linux VM owned by this run')
53 parser.add_argument('--expected-pages', type=int, default=15)
54 parser.add_argument('--verify-only', action='store_true',
55 help='Compare an existing capture in OUTPUT without the lab.')
56 args = parser.parse_args()
57 output = args.output.resolve()
58 if not args.verify_only and not args.linux:
59 parser.error('--linux names the disposable Linux VM this run owns')
60 if args.verify_only:
61 expected = {}
62 for launch in ('first', 'second'):
63 for line in (output / f'{launch}.log').read_text().splitlines():
64 if line.startswith('{') and '"published"' in line:
65 event = json.loads(line)
66 expected[event['title']] = event['texts']
67 native = native_texts(output / 'result/cold/read')
68 mismatches = compare(expected, native)
69 (output / 'verification.json').write_text(json.dumps(
70 {'expected': expected, 'native': {title: native[title] for title in expected}, 'mismatches': mismatches},
71 indent=2, ensure_ascii=False))
72 assert not mismatches, mismatches
73 print(f'Session acceptance verified: {len(expected)} pages', flush=True)
74 return
75 output.mkdir(parents=True)
76 server = args.linux
77 mount = output / 'mount'
78 mount.mkdir()
79 client = ROOT / 'target/debug/examples/session_client'
80 subprocess.run(['cargo', 'build', '-p', 'notebook', '--example', 'session_client'], cwd=ROOT, check=True)
81 record = {'server': server, 'fixture': str(FIXTURE.relative_to(ROOT)), 'launches': []}
82
83 def unmount():
84 for options in ([], ['-f']):
85 if not os.path.ismount(mount):
86 return
87 subprocess.run(['/sbin/umount', *options, str(mount)], capture_output=True, text=True, timeout=60)
88 if os.path.ismount(mount):
89 raise RuntimeError('The owned SMB mount remains attached')
90
91 try:
92 if not linux_vm.instance_path(server).exists():
93 linux_vm.create_instance(server)
94 linux_vm.launch(server)
95 linux_vm.wait_instance(server, 600)
96 config = linux_vm.load_instance(server)
97 (output / 'linux.json').write_text(json.dumps(config, indent=2))
98 linux_vm.run_ssh(server, 'mkdir /srv/agent/m4-session').check_returncode()
99 with tarfile.open(output / 'input.tar', 'w', dereference=True) as archive:
100 for name in ('synthetic.one', 'Open Notebook.onetoc2'):
101 archive.add(FIXTURE / name, arcname=name)
102 with (output / 'input.tar').open('rb') as stream:
103 subprocess.run(linux_vm.ssh_argv(server, 'tar xf - -C /srv/agent/m4-session'), stdin=stream, check=True)
104 linux_vm.run_ssh(server, 'chmod u+w /srv/agent/m4-session/*').check_returncode()
105 subprocess.run(['/sbin/mount_smbfs', '-N', f'//guest@127.0.0.1:{config["samba_port"]}/agent', mount],
106 check=True, stdin=subprocess.DEVNULL)
107 section = mount / 'm4-session/synthetic.one'
108 assert section.read_bytes() == (FIXTURE / 'synthetic.one').read_bytes()
109 cache = output / 'cache'
110 expected = {}
111 for launch in ('First launch', 'Second launch'):
112 result = subprocess.run([client, section, cache, launch], capture_output=True, text=True, timeout=300)
113 (output / f'{launch.split()[0].lower()}.log').write_text(result.stdout + result.stderr)
114 result.check_returncode()
115 events = [json.loads(line) for line in result.stdout.splitlines() if line.startswith('{')]
116 assert events[0]['event'] == 'opened' and events[0]['pending'] == 0, events[0]
117 published = [e for e in events if e['event'] == 'published']
118 assert published and all(e['revision'] for e in published), published
119 for e in published:
120 assert e['stored'] == e['texts'], e
121 expected[e['title']] = e['texts']
122 record['launches'].append({'label': launch, 'published': published})
123 unmount()
124 result = output / 'result'
125 (result / 'notebook').mkdir(parents=True)
126 with (output / 'result.tar').open('wb') as stream:
127 subprocess.run(linux_vm.ssh_argv(server, 'tar cf - -C /srv/agent/m4-session .'), stdout=stream, check=True)
128 with tarfile.open(output / 'result.tar') as archive:
129 archive.extractall(result / 'notebook', filter='data')
130 subprocess.run([sys.executable, ROOT / 'tools/native_runner.py', result / 'notebook', result / 'cold',
131 '--expected-pages', str(args.expected_pages), '--collect-notebook'], check=True)
132 native = native_texts(result / 'cold/read')
133 mismatches = compare(expected, native)
134 record['native'] = {title: native[title] for title in expected}
135 record['mismatches'] = mismatches
136 (output / 'record.json').write_text(json.dumps(record, indent=2, ensure_ascii=False))
137 assert not mismatches, mismatches
138 print(f'Session acceptance passed: {len(expected)} pages edited twice, reopened natively', flush=True)
139 finally:
140 unmount()
141 if linux_vm.instance_path(server).exists():
142 if linux_vm.running(server):
143 linux_vm.shutdown(server, 60)
144 linux_vm.delete_instance(server)
145 shutil.rmtree(mount, ignore_errors=True)
146
147
148if __name__ == '__main__':
149 main()