作业控制在进程组之间交接终端前台

from pathlib import Path
import os,pty,re,select,signal,termios,time

pid,master=pty.fork()
if pid==0:
    attributes=termios.tcgetattr(0);attributes[3]&=~termios.ECHO;attributes[3]|=termios.ISIG|termios.ICANON
    attributes[6][termios.VINTR]=b'\x03';attributes[6][termios.VSUSP]=b'\x1a';attributes[6][termios.VEOF]=b'\x04'
    termios.tcsetattr(0,termios.TCSANOW,attributes)
    os.execve('/bin/bash',['bash','--noprofile','--norc','-i'],{'PATH':'/usr/bin:/bin','TERM':'dumb','LC_ALL':'C','PS1':'XB-PROMPT> ','HISTFILE':'/dev/null'})
known_groups={pid};pending=b''
def until_prompt():
    global pending
    deadline=time.monotonic()+5
    while b'XB-PROMPT> ' not in pending:
        remaining=deadline-time.monotonic();assert remaining>0,pending
        ready,_,_=select.select([master],[],[],remaining);assert ready,pending
        chunk=os.read(master,65536);assert chunk,'PTY closed before prompt';pending+=chunk
    output,pending=pending.split(b'XB-PROMPT> ',1)
    return output.decode(errors='replace')
def command(text):os.write(master,text.encode()+b'\n');return until_prompt()
def wait_for(predicate):
    deadline=time.monotonic()+5
    while not predicate():
        assert time.monotonic()<deadline,'terminal state did not change';time.sleep(0.01)
def state(process):
    try:return (Path('/proc')/str(process)/'stat').read_text().rsplit(') ',1)[1].split()[0]
    except FileNotFoundError:return None
try:
    until_prompt();assert os.tcgetpgrp(master)==pid
    os.write(master,b'sleep 60 | cat\n')
    wait_for(lambda:os.tcgetpgrp(master)!=pid)
    group=os.tcgetpgrp(master);known_groups.add(group)
    def pipeline_started():
        children=[]
        for process in Path('/proc').iterdir():
            if not process.name.isdecimal():continue
            try:
                fields=(process/'stat').read_text().rsplit(') ',1)[1].split()
                if int(fields[1])==pid:
                    children.append(((process/'comm').read_text().strip(),int(fields[2])))
            except (FileNotFoundError,ProcessLookupError):continue
        return sorted(children)==[('cat',group),('sleep',group)]
    wait_for(pipeline_started)
    os.write(master,b'\x1a');output=until_prompt()
    assert 'Stopped' in output and os.tcgetpgrp(master)==pid,output
    assert 'Stopped' in command('jobs -l')
    print('Ctrl-Z stops the foreground pipeline and returns the terminal to the interactive shell')
    command('bg %1');assert os.tcgetpgrp(master)==pid
    os.write(master,b'fg %1\n');wait_for(lambda:os.tcgetpgrp(master)==group)
    os.write(master,b'\x03');until_prompt()
    assert 'RESULT:130' in command("printf 'RESULT:%s\\n' \"$?\"")
    print('bg resumes without foreground ownership; fg restores ownership; Ctrl-C gives this pipeline status 130')
    output=command("cat & printf 'BG:%s\\n' \"$!\"")
    background=int(re.search(r'BG:(\d+)',output)[1]);known_groups.add(background)
    wait_for(lambda:state(background)=='T')
    assert os.tcgetpgrp(master)==pid
    assert 'Stopped (tty input)' in command('jobs -l')
    print('a background reader is stopped by terminal input rules, rather than receiving the shell keyboard stream')
    os.write(master,b'fg %+\n');wait_for(lambda:os.tcgetpgrp(master)==background)
    os.write(master,b'hello\n\x04');output=until_prompt();assert 'hello' in output,output
    assert 'RESULT:0' in command("printf 'RESULT:%s\\n' \"$?\"")
    print('foregrounding that reader lets it read; EOF ends it normally; all jobs are reaped')
    assert not command('jobs -p').strip()
    os.write(master,b'exit\n');waited,status=os.waitpid(pid,0);assert waited==pid and os.WIFEXITED(status)
    pid=None
finally:
    if pid is not None:
        for group in known_groups:
            try:os.killpg(group,signal.SIGKILL)
            except ProcessLookupError:pass
        os.waitpid(pid,0)
    os.close(master)