EnvironmentFile 在 systemd 执行进程时供应取值

from pathlib import Path
from tempfile import TemporaryDirectory
import json,subprocess,time

unit='xbcatm-environment-lab.service';unit_path=Path('/run/systemd/system')/unit
assert not unit_path.exists()
def ctl(*args,success=True):
    result=subprocess.run(['systemctl',*args],capture_output=True,text=True,timeout=10)
    assert (result.returncode==0)==success,(args,result.stderr)
    return result.stdout.strip()
def value(key):return ctl('show',unit,'-p',key,'--value')
def wait_for(check):
    deadline=time.monotonic()+5
    while not check():
        assert time.monotonic()<deadline,'state transition timed out'
        time.sleep(0.02)
with TemporaryDirectory(prefix='xbcatm-environment-',dir='/run') as directory:
    root=Path(directory);first=root/'required.env';second=root/'optional.env';worker=root/'worker.py'
    command=f'$(touch {root}/unexpected)'
    initial=f'MODE=first\nLITERAL=$HOME\nQUOTED="two words"\nCOMMAND={command}\n'
    first.write_text(initial);first.chmod(0o600);second.write_text('MODE=second\n')
    worker.write_text("""from pathlib import Path
import json,os,sys,time
root=Path(__file__).parent;name=sys.argv[1]
data={key:os.environ.get(key) for key in ('MODE','LITERAL','QUOTED','COMMAND')}
data['pid']=os.getpid()
tmp=root/(name+'.tmp');tmp.write_text(json.dumps(data));tmp.replace(root/(name+'.json'))
if name=='main': time.sleep(60)
""")
    unit_path.write_text(f'''[Service]
Type=exec
Environment=MODE=unit-default
EnvironmentFile={first}
EnvironmentFile=-{second}
ExecStart=/usr/bin/python3 {worker} main
ExecReload=/usr/bin/python3 {worker} reload
''')
    def read(name):return json.loads((root/(name+'.json')).read_text())
    try:
        subprocess.run(['systemd-analyze','verify',str(unit_path)],check=True)
        ctl('daemon-reload');ctl('start',unit)
        wait_for(lambda:(root/'main.json').exists())
        current=read('main');assert current['MODE']=='second'
        assert current['LITERAL']=='$HOME' and current['QUOTED']=='two words'
        assert current['COMMAND']==command and not (root/'unexpected').exists()
        assert 'MODE=unit-default' in value('Environment')
        print('ordered environment files override unit value; quoting works; shell text stays literal')
        second.write_text('MODE=third\n')
        ctl('daemon-reload');assert value('MainPID')==str(current['pid'])
        live=Path(f'/proc/{current["pid"]}/environ').read_bytes().split(b'\0')
        assert b'MODE=second' in live
        ctl('reload',unit)
        assert read('reload')['MODE']=='third' and read('main')==current
        assert value('MainPID')==str(current['pid'])
        live=Path(f'/proc/{current["pid"]}/environ').read_bytes().split(b'\0')
        assert b'MODE=second' in live
        print('reload helper receives new file value; existing main process retains old environment')
        second.write_text('MODE=fourth\n');ctl('restart',unit)
        wait_for(lambda:read('main')['pid']!=current['pid'])
        assert read('main')['MODE']=='fourth'
        print('restart reads changed environment file without another daemon-reload')
        first.unlink()
        subprocess.run(['systemd-analyze','verify',str(unit_path)],check=True)
        ctl('restart',unit,success=False)
        assert value('ActiveState')=='failed' and value('MainPID')=='0'
        print('missing required environment file prevents start')
        first.write_text(initial);first.chmod(0o600);second.unlink()
        ctl('reset-failed',unit);ctl('start',unit)
        wait_for(lambda:read('main')['MODE']=='first')
        print('missing optional file is ignored; required-file value remains')
    finally:
        ctl('stop',unit)
        if value('ActiveState')=='failed':ctl('reset-failed',unit)
        unit_path.unlink();ctl('daemon-reload')