diff options
Diffstat (limited to 'test/agent_session.py')
| -rw-r--r-- | test/agent_session.py | 171 |
1 files changed, 171 insertions, 0 deletions
diff --git a/test/agent_session.py b/test/agent_session.py new file mode 100644 index 00000000..b920b006 --- /dev/null +++ b/test/agent_session.py @@ -0,0 +1,171 @@ +#!/usr/bin/env python3 +import argparse +import hashlib +import json +from pathlib import Path +import shlex +import signal +import sys +import tempfile +import time +import uuid + +from fs import session + + +def screen_text(client): + screen = client.screen() + cols = screen['cols'] + return '\n'.join(''.join(cell[0] for cell in screen['cells'][at:at + cols]) + for at in range(0, len(screen['cells']), cols)) + + +def expired(signum, frame): + raise TimeoutError('session deadline exceeded') + + +def wait_ready(client, ready): + previous = None + while True: + text = screen_text(client) + if ready in text and text == previous: + return text + previous = text + time.sleep(.05) + + +def run(args, command, status): + temporary = tempfile.TemporaryDirectory(prefix='pardes-agent-session-') + try: + root = Path(temporary.name) + config = root / 'config' / 'pardes' + config.mkdir(parents=True) + (config / 'init').write_text('Shell /bin/sh\n') + script = root / 'command.sh' + script.write_text('exec ' + shlex.join(command) + '\n') + old_handler = signal.signal(signal.SIGALRM, expired) + signal.setitimer(signal.ITIMER_REAL, args.timeout) + try: + context = session(str(args.binary.resolve()), root, 'agent-session', tty=True, + socket_name='agent-session', launch=['--tty', '--9p=agent-session'], + inherited={'PARDES_TEST_GRID': '1'} if args.gui_grid else {}) + client, _ = context.__enter__() + try: + try: + while not client.read('/self/pane/1/body').strip(): + time.sleep(.01) + started = time.monotonic() + client.write('/self/pane/1/pty/data', shlex.join(['/bin/sh', str(script)]).encode() + b'\r') + status['phase'] = 'readiness' + wait_ready(client, args.ready) + if args.prepare_key is not None: + status['phase'] = 'preparation' + client.write('/self/pane/1/pty/data', args.prepare_key) + wait_ready(client, args.ready) + ready_ms = (time.monotonic() - started) * 1000 + status.update(phase='history', required_body_rows=args.min_rows) + while True: + started = time.monotonic() + body = client.read('/self/pane/1/body') + body_read_ms = (time.monotonic() - started) * 1000 + rows = body.count(b'\n') + int(bool(body) and not body.endswith(b'\n')) + status.update(body_bytes=len(body), body_rows=rows) + if rows >= args.min_rows: + break + time.sleep(.05) + status['phase'] = 'settling' + before = wait_ready(client, args.ready) + status['expected_visible_text_sha256'] = hashlib.sha256(before.encode()).hexdigest() + probe = 'p' + uuid.uuid4().hex[:12] + if probe in before: + raise AssertionError('probe is already visible') + status['phase'] = 'probe-visible' + started = time.monotonic() + client.write('/self/pane/1/pty/data', probe.encode()) + while probe not in screen_text(client).replace('\n', ''): + time.sleep(.01) + write_ms = (time.monotonic() - started) * 1000 + status['phase'] = 'probe-cleared' + started = time.monotonic() + client.write('/self/pane/1/pty/data', b'\x15') + while True: + observed = screen_text(client) + status['observed_visible_text_sha256'] = hashlib.sha256(observed.encode()).hexdigest() + if observed == before: + break + time.sleep(.01) + status['metrics'] = { + 'ready_ms': ready_ms, 'body_read_ms': body_read_ms, + 'ninep_write_to_observation_ms': write_ms, + 'ninep_clear_to_observation_ms': (time.monotonic() - started) * 1000, + 'visible_text_sha256': hashlib.sha256(before.encode()).hexdigest(), + 'body_bytes': len(body), 'body_rows': rows, + 'body_sha256': hashlib.sha256(body).hexdigest(), + } + finally: + previous_phase = status['phase'] + status['phase'] = 'interrupt-cleanup' + signal.setitimer(signal.ITIMER_REAL, 0) + client.socket.settimeout(.25) + for _ in range(2): + try: + client.write('/self/pane/1/pty/data', b'\x03') + except (OSError, EOFError): + pass + time.sleep(.05) + status['phase'] = previous_phase + finally: + previous_phase = status['phase'] + status['phase'] = 'session-cleanup' + context.__exit__(*sys.exc_info()) + status['phase'] = previous_phase + finally: + signal.setitimer(signal.ITIMER_REAL, 0) + signal.signal(signal.SIGALRM, old_handler) + finally: + previous_phase = status['phase'] + status['phase'] = 'files-cleanup' + temporary.cleanup() + status['phase'] = previous_phase + return status.pop('metrics') + + +def main(argv): + parser = argparse.ArgumentParser(description='Check an unsubmitted input probe through an owned Pardes 9P session.') + parser.add_argument('binary', type=Path) + parser.add_argument('--ready', required=True, help='visible application readiness text') + parser.add_argument('--timeout', type=float, default=60, help='interaction deadline in seconds, at most 600') + parser.add_argument('--min-rows', type=int, default=0) + parser.add_argument('--prepare-key', help='one hex control byte before history/probe checks; never CR or LF') + parser.add_argument('--gui-grid', action='store_true', help='use the GUI build headless grid host') + if '--' not in argv: + parser.error('supply the interactive command after --') + split = argv.index('--') + args = parser.parse_args(argv[:split]) + command = argv[split + 1:] + if not command or not command[0] or not args.ready.strip(): + parser.error('a command and nonempty --ready text are required') + if not 0 < args.timeout <= 600 or args.min_rows < 0: + parser.error('--timeout must be in (0, 600] and --min-rows must be nonnegative') + if args.prepare_key is not None: + try: + args.prepare_key = bytes.fromhex(args.prepare_key) + except ValueError: + parser.error('--prepare-key must be one hex control byte') + if len(args.prepare_key) != 1 or not 0 < args.prepare_key[0] < 32 or args.prepare_key[0] in (10, 13): + parser.error('--prepare-key must be one control byte other than CR or LF') + status = {'phase': 'startup'} + try: + metrics = run(args, command, status) + except (OSError, EOFError, AssertionError, ValueError) as error: + status['error'] = type(error).__name__ + if isinstance(error, OSError): + status['errno'] = error.errno + print(json.dumps(status), file=sys.stderr) + return 1 + print(json.dumps(metrics)) + return 0 + + +if __name__ == '__main__': + sys.exit(main(sys.argv[1:])) |