按文档读取不断变化的进程 procfs 视图

from pathlib import Path
from tempfile import TemporaryDirectory
import json,os,subprocess,sys

worker=r'''
from pathlib import Path
import ctypes,json,os,sys
assert ctypes.CDLL(None).prctl(15,b'worker ) sample',0,0,0)==0
os.environ['XB_PROC_TEST']='changed-after-start'
path=Path(sys.argv[1]);path.write_bytes(b'fixture-data')
with path.open('rb') as handle:
    info=os.fstat(handle.fileno());path.unlink()
    print(json.dumps({'fd':handle.fileno(),'device':info.st_dev,'inode':info.st_ino,'current_environment':os.environ['XB_PROC_TEST']}),flush=True)
    assert sys.stdin.readline()=='done\n'
'''
with TemporaryDirectory(prefix='xbcatm-proc-') as directory:
    path=Path(directory)/'data'
    process=subprocess.Popen([sys.executable,'-c',worker,str(path),'argument with spaces'],stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True,env={'PATH':'/usr/bin:/bin','XB_PROC_TEST':'fixture-value'})
    try:
        metadata=json.loads(process.stdout.readline());root=Path('/proc')/str(process.pid)
        status_path=root/'status';status_text=status_path.read_text()
        assert status_path.stat().st_size==0 and len(status_text)>0
        status=dict(line.split(':',1) for line in status_text.splitlines() if ':' in line)
        assert int(status['Pid'])==process.pid and int(status['PPid'])==os.getpid()
        raw=(root/'stat').read_text();pid_text,rest=raw.split(' (',1);comm,fields=rest.rsplit(') ',1)
        fields=fields.split();assert int(pid_text)==process.pid and comm=='worker ) sample'
        assert int(fields[1])==os.getpid() and int(fields[19])>0
        assert raw.split()[2] not in ('R','S','D','T','Z')
        print('proc files can have size zero but readable data; stat comm can contain spaces and a closing parenthesis')
        args=(root/'cmdline').read_bytes().rstrip(b'\0').split(b'\0')
        environment=(root/'environ').read_bytes().rstrip(b'\0').split(b'\0')
        assert args[-1]==b'argument with spaces' and b'XB_PROC_TEST=fixture-value' in environment
        assert metadata['current_environment']=='changed-after-start'
        assert os.readlink('/proc/self')==str(os.getpid())
        print('cmdline and environ use NUL separators; environ retains the initial value here; self resolves for the reader')
        entry=root/'fd'/str(metadata['fd']);target=os.readlink(entry);actual=entry.stat()
        assert target==str(path)+' (deleted)' and not path.exists()
        assert (actual.st_dev,actual.st_ino)==(metadata['device'],metadata['inode'])
        assert entry.read_bytes()==b'fixture-data'
        print('a deleted pathname remains inspectable through an open descriptor subject to access checks')
        output,error=process.communicate('done\n',timeout=5)
        assert process.returncode==0,error
        assert not root.exists()
        print('after the controlled child exits and is reaped, its proc directory disappears')
    finally:
        if process.poll() is None:process.kill();process.wait()