116 lines
5.7 KiB
Python
116 lines
5.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise the daemon's capture arguments on an OWNED, private tmux server.
|
|
|
|
No backend process, default tmux socket, user configuration, network or installed
|
|
session is used. This is a tmux protocol regression, not installed acceptance.
|
|
"""
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import resource
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
HANDLER = ROOT / 'core/archipelago/src/api/handler/terminal.rs'
|
|
|
|
|
|
def bounded_child():
|
|
resource.setrlimit(resource.RLIMIT_AS, (64 * 1024 * 1024,) * 2)
|
|
resource.setrlimit(resource.RLIMIT_CPU, (5, 5))
|
|
resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
|
|
|
|
|
|
def capture_arguments():
|
|
source = HANDLER.read_text()
|
|
match = re.search(r'Command::new\("tmux"\)\.args\(\[("capture-pane".*?)\]\)', source)
|
|
if not match:
|
|
raise AssertionError('Update the protocol fixture for changed production capture construction')
|
|
tokens = [token.strip() for token in match.group(1).split(',')]
|
|
return ['fixture' if token == '&record.tmux' else token.strip('"') for token in tokens]
|
|
|
|
|
|
@unittest.skipUnless(shutil.which('tmux'), 'tmux required: protocol exit NOT qualified')
|
|
class TerminalScrollback(unittest.TestCase):
|
|
def test_history_is_bounded_and_survives_snapshot_replacement(self):
|
|
with tempfile.TemporaryDirectory(prefix='archy-terminal-history-') as directory:
|
|
socket = str(Path(directory) / 'owned.sock')
|
|
env = {'PATH': '/usr/bin:/bin', 'HOME': directory, 'TERM': 'xterm',
|
|
'LC_ALL': 'C.UTF-8', 'TMUX_TMPDIR': directory}
|
|
base = [shutil.which('tmux'), '-S', socket, '-f', '/dev/null']
|
|
owned_pids = []
|
|
|
|
def tmux(*args, check=True):
|
|
return subprocess.run(base + list(args), env=env, cwd=directory,
|
|
capture_output=True, text=True, timeout=3,
|
|
check=check, preexec_fn=bounded_child)
|
|
|
|
def wait_for(marker):
|
|
deadline = time.monotonic() + 3
|
|
while time.monotonic() < deadline:
|
|
text = tmux('capture-pane', '-p', '-t', 'fixture').stdout
|
|
if marker in text:
|
|
return
|
|
time.sleep(0.02)
|
|
self.fail(f'synthetic output did not reach {marker}')
|
|
|
|
# The child waits for one synthetic line before generating overflow.
|
|
program = ("import sys,time; "
|
|
"[print('HISTORY-%04d'%i) for i in range(100)]; "
|
|
"sys.stdout.flush(); sys.stdin.readline(); "
|
|
"[print('HISTORY-%04d'%i) for i in range(100,1200)]; "
|
|
"sys.stdout.flush(); time.sleep(10)")
|
|
try:
|
|
tmux('new-session', '-d', '-s', 'fixture', '-x', '80', '-y', '24',
|
|
'exec /usr/bin/python3 -u -c ' + shlex.quote(program))
|
|
for fmt in ('#{pane_pid}', '#{pid}'):
|
|
owned_pids.append(int(tmux('display-message', '-p', '-t', 'fixture', fmt).stdout))
|
|
wait_for('HISTORY-0099')
|
|
visible = tmux('capture-pane', '-p', '-t', 'fixture').stdout
|
|
self.assertNotIn('HISTORY-0000', visible, 'fixture must exceed the visible pane')
|
|
self.assertIn('HISTORY-0000', tmux('capture-pane', '-p', '-S', '-', '-t', 'fixture').stdout)
|
|
args = capture_arguments()
|
|
snapshot = tmux(*args).stdout
|
|
print(f'capture={args!r}; first={snapshot.splitlines()[0]!r}; lines={len(snapshot.splitlines())}', flush=True)
|
|
self.assertIn('HISTORY-0000', snapshot,
|
|
'production snapshot discards history that tmux still retains')
|
|
self.assertIn('HISTORY-0099', snapshot)
|
|
self.assertTrue(snapshot.startswith('HISTORY-0000\n'), 'missing history must not be padded with blank rows')
|
|
self.assertEqual(snapshot, tmux(*args).stdout,
|
|
'replacement/reconnect must reconstruct the same bounded history')
|
|
tmux('send-keys', '-t', 'fixture', 'Enter')
|
|
wait_for('HISTORY-1199')
|
|
bounded = tmux(*args).stdout
|
|
self.assertNotIn('HISTORY-0000', bounded, 'old history must be bounded')
|
|
self.assertIn('HISTORY-1199', bounded)
|
|
self.assertLessEqual(len(bounded.splitlines()), 1024,
|
|
'capture exceeds 1000 history lines plus 24 visible rows')
|
|
finally:
|
|
tmux('kill-server', check=False)
|
|
deadline = time.monotonic() + 3
|
|
while time.monotonic() < deadline:
|
|
# Zombies have exited and no longer hold a session/socket.
|
|
live = []
|
|
for pid in owned_pids:
|
|
try:
|
|
state = Path(f'/proc/{pid}/stat').read_text().split(') ')[1][0]
|
|
except FileNotFoundError:
|
|
continue
|
|
if state != 'Z':
|
|
live.append(pid)
|
|
if not live:
|
|
break
|
|
time.sleep(0.02)
|
|
self.assertFalse(live if owned_pids else [], 'owned tmux/fixture process survived cleanup')
|
|
self.assertNotEqual(tmux('list-sessions', check=False).returncode, 0,
|
|
'owned server still accepts connections')
|
|
print('CLEANUP: owned server and fixture exited; private socket accepts no sessions', flush=True)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|