diff options
Diffstat (limited to 'prototypes/connections/check_tui.py')
| -rw-r--r-- | prototypes/connections/check_tui.py | 352 |
1 files changed, 352 insertions, 0 deletions
diff --git a/prototypes/connections/check_tui.py b/prototypes/connections/check_tui.py new file mode 100644 index 0000000..5328d02 --- /dev/null +++ b/prototypes/connections/check_tui.py @@ -0,0 +1,352 @@ +#!/usr/bin/env python3 +"""Exercise the real terminal editor through a PTY and the same document CLI.""" +import argparse +import fcntl +import hashlib +import json +import locale +import os +from pathlib import Path +import pty +import select +import signal +import struct +import subprocess +import tempfile +import termios +import time + +HERE = Path(__file__).resolve().parent + + +def command(*args): + return subprocess.run([str(arg) for arg in args], capture_output=True, + timeout=20, check=True) + + +def utf8_locale(): + original = locale.setlocale(locale.LC_CTYPE) + candidates = [os.environ.get('LC_ALL'), os.environ.get('LC_CTYPE'), + os.environ.get('LANG'), 'C.UTF-8', 'en_US.UTF-8'] + try: + for candidate in candidates: + if not candidate: + continue + try: + locale.setlocale(locale.LC_CTYPE, candidate) + except locale.Error: + continue + if locale.nl_langinfo(locale.CODESET).upper().replace('-', '') == 'UTF8': + return candidate + finally: + locale.setlocale(locale.LC_CTYPE, original) + raise AssertionError('no UTF-8 locale available for terminal test') + + +def memory_kib(pid, field, portable=False): + if portable: + return None + for line in Path(f'/proc/{pid}/status').read_text().splitlines(): + if line.startswith(field): + return int(line.split()[1]) + raise AssertionError(field) + + +class Terminal: + def __init__(self, binary, program, store, key): + self.master, self.slave = pty.openpty() + fcntl.ioctl(self.slave, termios.TIOCSWINSZ, struct.pack('HHHH', 24, 80, 0, 0)) + self.original = termios.tcgetattr(self.slave) + language = utf8_locale() + self.process = subprocess.Popen([str(binary), + *(['--program', str(program)] if program else []), + '--store', str(store), key], + stdin=self.slave, stdout=self.slave, stderr=self.slave, + env={**os.environ, 'LANG': language, + 'LC_ALL': language}) + self.wait_for(f'Pure | {key}'.encode()) + + def wait_for(self, marker, timeout=8): + seen = b'' + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + ready, _, _ = select.select([self.master], [], [], 0.1) + if ready: + part = os.read(self.master, 65536) + seen = (seen + part)[-65536:] + if marker in seen: + return seen + if self.process.poll() is not None: + break + raise AssertionError((marker, self.process.poll(), seen[-2000:])) + + def type(self, data): + os.write(self.master, data) + + def resize(self, rows, columns): + fcntl.ioctl(self.slave, termios.TIOCSWINSZ, + struct.pack('HHHH', rows, columns, 0, 0)) + os.kill(self.process.pid, signal.SIGWINCH) + + def close(self, discard=False): + if self.process.poll() is None: + self.type(b'\x11') + if discard: + self.wait_for(b'Unsaved draft') + self.type(b'\x11') + self.process.wait(timeout=8) + assert self.process.returncode == 0, self.process.returncode + restored = termios.tcgetattr(self.slave) + assert restored == self.original, (restored, self.original) + os.close(self.master) + os.close(self.slave) + + def signal_close(self): + os.kill(self.process.pid, signal.SIGTERM) + self.process.wait(timeout=8) + assert self.process.returncode == 0, self.process.returncode + assert termios.tcgetattr(self.slave) == self.original + os.close(self.master) + os.close(self.slave) + + +def shown(editor, store, key): + result = command(*editor, '--store', store, 'show', key) + return result.stdout, json.loads(result.stderr)['revision'] + + +def wait_stored(editor, store, key, expected, revision): + deadline = time.monotonic() + 8 + while time.monotonic() < deadline: + try: + body, found_revision = shown(editor, store, key) + if body == expected and found_revision == str(revision): + return + except subprocess.CalledProcessError: + pass + time.sleep(0.05) + raise AssertionError((key, expected[:100], shown(editor, store, key))) + + +def check(build, rss_limit_kib, source, program_path, portable=False): + binary = (build / ('pure-edit-tui' if source else 'pure-edit-tui-fixed')).resolve() + program = (program_path or HERE / 'examples/editor.pure').resolve() if source else None + editor = [str((build / ('pure-edit' if source else 'pure-edit-fixed')).resolve())] + if source: + editor += ['--program', str(program)] + with tempfile.TemporaryDirectory(prefix='pure-tui-') as name: + root = Path(name) + store = root / 'store' + if source: + invalid = root / 'invalid.pure' + invalid.write_text('not a pure program\n') + for candidate, marker in ((invalid, b'Program:'), + (HERE / 'examples/storage.pure', b'Program needs')): + denied = subprocess.run([str(binary), '--program', str(candidate), + '--store', str(store), 'note'], + capture_output=True, timeout=10) + assert denied.returncode == 2 and marker in denied.stderr + assert not store.exists() + terminal = Terminal(binary, program, store, 'note') + try: + terminal.type(b'hello Pure\x1b[D\x1b[D!\x13') + wait_stored(editor, store, 'note', b'hello Pu!re', 1) + if source: + assert command(*editor, '--store', store, 'show', 'note').stdout == b'hello Pu!re' + terminal.type(b'\x0fsecond\r') + terminal.wait_for(b'Pure | second') + terminal.type('안녕'.encode() + b'\x1b[D' + '하'.encode() + b'\x13') + terminal.wait_for('안하녕'.encode()) + wait_stored(editor, store, 'second', '안하녕'.encode(), 1) + terminal.resize(32, 100) + terminal.wait_for(b'Pure | second') + terminal.type(b'\x0fnote\r') + terminal.wait_for(b'Pure | note') + terminal.type(b'\x0c') + listing = terminal.wait_for(b'Pure | documents') + assert b'note' in listing and b'second' in listing, listing[-1000:] + terminal.type(b'\x1b[B\r') + terminal.wait_for(b'Pure | second') + terminal.type(b'\x0c\x1b[A\r') + terminal.wait_for(b'Pure | note') + terminal.type(b'\x0c') + terminal.wait_for(b'Pure | documents') + (store / 'obj-second').unlink() + terminal.type(b'\x1b[B\r') + terminal.wait_for(b'Document disappeared; refresh the list') + terminal.type(b'\x1b') + terminal.wait_for(b'Pure | note') + terminal.type(b'!') + terminal.type(b'\x0c') + terminal.wait_for(b'Save or reload the draft before browsing') + external = root / 'external' + external.write_bytes(b' remote') + command(*editor, '--store', store, 'edit', 'note', '1', '11', '0', external) + terminal.type(b'\x13') + terminal.wait_for(b'Version conflict') + assert shown(editor, store, 'note') == (b'hello Pu!re remote', '2') + small_peak = memory_kib(terminal.process.pid, 'VmHWM:', portable) + finally: + terminal.close(discard=True) + + multiline = b'first\n\tsecond\nthird' + insert = root / 'multiline-insert' + insert.write_bytes(multiline) + command(*editor, '--store', store, 'edit', 'multiline', '0', '0', '0', insert) + terminal = Terminal(binary, program, store, 'multiline') + try: + terminal.resize(25, 80) + terminal.wait_for(b'third') + terminal.type(b'!\x13') + wait_stored(editor, store, 'multiline', multiline + b'!', 2) + finally: + terminal.close() + terminal = Terminal(binary, program, store, 'multiline') + terminal.resize(25, 80) + terminal.wait_for(b'third!') + terminal.close() + + rows = [f'line-{index:02}'.encode() for index in range(60)] + insert.write_bytes(b'\n'.join(rows)) + command(*editor, '--store', store, 'edit', 'rows', '0', '0', '0', insert) + terminal = Terminal(binary, program, store, 'rows') + try: + terminal.type(b'\x1b[A' * 3 + b'X\x13') + rows[56] += b'X' + wait_stored(editor, store, 'rows', b'\n'.join(rows), 2) + terminal.wait_for(b'line-56X') + finally: + terminal.close() + + large = b'Q' * 1_048_576 + insert = root / 'large-insert' + insert.write_bytes(large) + command(*editor, '--store', store, 'edit', 'large', '0', '0', '0', insert) + terminal = Terminal(binary, program, store, 'large') + try: + terminal.type(b'Z') + terminal.wait_for(b'Z') + terminal.type(b'\x13') + wait_stored(editor, store, 'large', large + b'Z', 2) + large_peak = memory_kib(terminal.process.pid, 'VmHWM:', portable) + finally: + terminal.close() + if rss_limit_kib: + assert large_peak <= rss_limit_kib, (large_peak, rss_limit_kib) + + unicode_body = '안녕'.encode() * 174_762 + insert.write_bytes(unicode_body) + command(*editor, '--store', store, 'edit', 'unicode', '0', '0', '0', insert) + terminal = Terminal(binary, program, store, 'unicode') + try: + terminal.type(b'K') + terminal.wait_for(b'K') + terminal.type(b'\x1b[HX') + terminal.wait_for(b'X' + '안녕'.encode()) + terminal.type(b'\x13') + wait_stored(editor, store, 'unicode', b'X' + unicode_body + b'K', 2) + finally: + terminal.close() + + listed = root / 'listed-store' + insert.write_bytes(b'X') + for index in range(1024): + command(*editor, '--store', listed, 'edit', f'doc{index:04}', '0', '0', '0', insert) + terminal = Terminal(binary, program, listed, 'doc0000') + try: + terminal.type(b'\x0c') + terminal.wait_for(b'Pure | documents | 1024 keys') + terminal.resize(8, 12) + terminal.wait_for(b'\033[8;1H') + terminal.resize(24, 80) + terminal.wait_for(b'Pure | documents | 1024 keys') + terminal.type(b'\x1b[F\r') + terminal.wait_for(b'Pure | doc1023') + assert shown(editor, listed, 'doc1023') == (b'X', '1') + terminal.type(b'\x0c') + terminal.wait_for(b'Pure | documents | 1024 keys') + terminal.type(b'doc0919') + terminal.wait_for(b'Find: doc0919_') + terminal.type(b'\r') + terminal.wait_for(b'Pure | doc0919') + assert shown(editor, listed, 'doc0919') == (b'X', '1') + terminal.type(b'\x0czz') + terminal.wait_for(b'No matching key') + terminal.type(b'\r') + terminal.wait_for(b'No matching key') + terminal.type(b'\x7f\x7fdoc1023\r') + terminal.wait_for(b'Pure | doc1023') + list_peak = memory_kib(terminal.process.pid, 'VmHWM:', portable) + finally: + terminal.close() + if rss_limit_kib: + assert list_peak <= rss_limit_kib, (list_peak, rss_limit_kib) + + if source: + alternate = root / 'alternate.pure' + definition = program.read_text() + fields = ' start: u32le;\n remove: u32le;' + assert fields in definition + alternate.write_text(definition.replace(fields, + ' remove: u32le;\n start: u32le;')) + alternate_store = root / 'alternate-store' + terminal = Terminal(binary, alternate, alternate_store, 'different_layout') + try: + terminal.type(b'abcdef\x13') + wait_stored(editor, alternate_store, 'different_layout', b'abcdef', 1) + terminal.type(b'\x1b[D\x1b[D\x1b[D!\x13') + wait_stored(editor, alternate_store, 'different_layout', b'abc!def', 2) + finally: + terminal.close() + + terminal = Terminal(binary, program, store, 'note') + terminal.signal_close() + return {'profile': 'native-source-document-terminal-editor' if source else + 'native-fixed-document-terminal-editor', + 'checks': (['source parse and incompatible definitions fail before store creation', + 'source CLI and TUI read the same document', + 'changed source field layout drives TUI input encoding'] if source else []) + + ['cursor insertion and save through terminal', 'open another key', + 'bounded document browser selects keys and protects unsaved draft', + 'UTF-8 codepoint movement, visible redraw and edit', 'resize redraw', + 'external revision conflict preserves draft', + 'stored multiline and tabbed document survives edit and reopen', + 'cached viewport preserves vertical movement across many rows', + 'one MiB document viewport, edit and fresh CLI restore', + 'one MiB UTF-8 viewport and edit before cached anchor', + '1,024 real documents listed, searched, resized and opened' + + (' within RSS gate' if not portable else ''), + 'terminal mode restored after key and signal exit'], + 'small_peak_rss_kib': small_peak, + 'one_mib_peak_rss_kib': large_peak, + 'one_thousand_key_list_peak_rss_kib': list_peak, + 'one_mib_peak_rss_limit_kib': rss_limit_kib, + 'portable': portable, + 'terminal_locale': utf8_locale(), + 'binary_sha256': hashlib.sha256(binary.read_bytes()).hexdigest()} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--build', type=Path, default=HERE / 'build') + parser.add_argument('--rss-limit-kib', type=int, default=4096, + help='0 skips the release RSS gate for sanitizer builds') + parser.add_argument('--source', action='store_true', help='check the source-loaded TUI') + parser.add_argument('--portable', action='store_true', + help='run all PTY behavior without Linux /proc RSS measurement') + parser.add_argument('--program', type=Path, help='definition for --source (including relocated bundles)') + parser.add_argument('--output', type=Path) + args = parser.parse_args() + if args.output and args.output.exists(): + parser.error('evidence file already exists') + result = check(args.build, 0 if args.portable else args.rss_limit_kib, + args.source, args.program, args.portable) + body = json.dumps(result, indent=2) + '\n' + if args.output: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(body) + print(body, end='') + + +if __name__ == '__main__': + main() |