#!/usr/bin/env python3
"""Exercise Tty9p and its real launcher with a nonprivileged sudo stand-in.

This checks PTY routing and lifecycle, not kernel-mount compatibility (v9fs.py).
"""
import json
import os
from pathlib import Path
import signal
import shutil
import sys
import tempfile
import time

from fs import execute, new_pane, session


def until(probe, description):
    deadline = time.monotonic() + 8
    while time.monotonic() < deadline:
        result = probe()
        if result:
            return result
        time.sleep(.02)
    raise AssertionError(description)


def test(binary, helper):
    with tempfile.TemporaryDirectory(prefix='pardes-v9fs-terminal-') as directory:
        root = Path(directory)
        # Exercise shell quoting with a real executable path containing both
        # whitespace and an apostrophe, under bash and fish where available.
        quoted_helper = root / "helper's quoted path"
        shutil.copy2(helper, quoted_helper)
        helper = quoted_helper
        sudo = root / 'sudo'
        # A fresh PATH in the owned fixture selects this stand-in. It never
        # invokes real sudo, mounts anything, or handles a real password.
        sudo.write_text(f'#!{sys.executable}\n' + '''
import json, os, signal, sys, termios
from pathlib import Path
tty = os.open('/dev/tty', os.O_RDWR)
assert os.isatty(tty)
report = dict(argv=sys.argv[1:], pane=os.environ['PARDES_PANE'],
              socket=os.environ['PARDES_9P'], mount=os.environ['PARDES_MOUNT'],
              path=os.environ['PARDES_V9FS_PATH'], launcher=os.getppid(), pid=os.getpid())
parent_status = Path(f"/proc/{report['launcher']}/status").read_text().splitlines()
shell_pid = next(line.split()[1] for line in parent_status if line.startswith('PPid:'))
report['shell'] = os.readlink(f'/proc/{shell_pid}/exe')
Path(os.environ['V9FS_REPORT']).write_text(json.dumps(report))
before = termios.tcgetattr(tty)
hidden = termios.tcgetattr(tty)
hidden[3] &= ~termios.ECHO
termios.tcsetattr(tty, termios.TCSANOW, hidden)
os.write(tty, b'V9FS_AUTH_READY: ')
response = os.read(tty, 100)
termios.tcsetattr(tty, termios.TCSANOW, before)
os.write(tty, b'V9FS_INPUT_RECEIVED\\n')
os.write(tty, b'sudo stand-in: authentication refused\\n')
sys.exit(1)
''')
        sudo.chmod(0o700)
        shells = [shutil.which('bash') or '/bin/sh', shutil.which('fish') or '/bin/sh']
        for shell, interrupted in zip(shells, (False, True)):
            report_path = root / ('interrupted.json' if interrupted else 'normal.json')
            path = str(root) + ':' + os.environ.get('PATH', '/usr/bin:/bin')
            name = 'interrupted' if interrupted else 'normal'
            with session(str(binary), root, name, inherited={
                'PARDES_V9FS_HELPER': str(helper), 'PATH': path,
                'V9FS_REPORT': str(report_path), 'SHELL': shell,
            }) as (client, address):
                control = new_pane(client, b'')
                execute(client, control, 'Shell ' + shell)
                execute(client, control, 'Tty9p')
                until(report_path.exists, 'Tty9p did not reach the sudo stand-in')
                report = json.loads(report_path.read_text())
                pane = f"/pane/{report['pane']}"
                until(lambda: b'V9FS_AUTH_READY' in client.read(pane + '/body'), 'prompt not visible')
                assert report['socket'] == str(address)
                assert report['argv'][:2] == ['-E', '--'], report
                assert report['argv'][2] == str(helper), report
                assert report['argv'][3] == str(address), report
                assert report['argv'][4] == report['mount'], report
                assert report['argv'][5:8] == [str(os.getuid()), str(os.getgid()), '--'], report
                assert report['path'] == path
                assert report['shell'] == str(Path(shell).resolve()), report
                # The detached core remains responsive during authentication.
                assert client.read('/pane/1/body') == b'initial\n'
                target = Path(report['mount'])
                assert target.is_dir() and list(target.iterdir()) == []
                if interrupted:
                    os.kill(report['launcher'], signal.SIGTERM)
                else:
                    client.write(pane + '/pty/data', b'probe-response\r')
                    until(lambda: b'V9FS_INPUT_RECEIVED' in client.read(pane + '/body'), 'input did not reach new PTY')
                    assert b'probe-response' not in client.read(pane + '/body'), 'password input was echoed'
                until(lambda: not target.exists(), 'launcher left its temporary mountpoint')
                # Failure/cancellation returns to the original interactive
                # shell, so the pane is useful and its errors remain visible.
                client.write(pane + '/pty/data', b'printf "OUTER_READY:%s\\n" "$PARDES_PANE"\r')
                expected = ('OUTER_READY:' + report['pane']).encode()
                until(lambda: expected in client.read(pane + '/body'), 'original shell did not remain usable')
                assert client.read('/pane/1/body') == b'initial\n'
    print('Tty9p: shell-first startup, quoted paths, hidden password input, failure recovery and cleanup passed')


if __name__ == '__main__':
    test(Path(sys.argv[1]).resolve(), Path(sys.argv[2]).resolve())