#!/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()