systemd 依赖要区分激活、顺序与就绪

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

provider='xbcatm-provider-lab.service';consumer='xbcatm-consumer-lab.service'
paths={name:Path('/run/systemd/system')/name for name in (provider,consumer)}
assert all(not p.exists() for p in paths.values())
def ctl(*args,success=True):
    result=subprocess.run(['systemctl',*args],text=True,capture_output=True,timeout=10)
    assert (result.returncode==0)==success,(args,result.stderr)
    return result.stdout.strip()
def value(name,key):return ctl('show',name,'-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-dependency-',dir='/run') as directory:
    root=Path(directory)
    (root/'provider.py').write_text(r"""from pathlib import Path
import os,socket,sys,time
root=Path(__file__).parent;mode=(root/'mode').read_text()
if mode=='fail': sys.exit(7)
(root/'entered').touch()
if mode in ('exec','notify'):
    while not (root/'gate').exists(): time.sleep(0.01)
(root/'ready').touch()
if mode=='notify':
    address=os.environ['NOTIFY_SOCKET']
    if address.startswith('@'): address='\0'+address[1:]
    with socket.socket(socket.AF_UNIX,socket.SOCK_DGRAM) as notification:
        notification.connect(address);notification.sendall(b'READY=1')
if mode in ('exec','notify'): time.sleep(60)
""")
    (root/'consumer.py').write_text("from pathlib import Path\nr=Path(__file__).parent\n(r/'observed').write_text(str((r/'ready').exists()))\n")
    def install(mode,requirement):
        (root/'mode').write_text(mode)
        kind='oneshot' if mode in ('oneshot','fail') else mode
        paths[provider].write_text(f'[Service]\nType={kind}\nExecStart=/usr/bin/python3 {root}/provider.py\nTimeoutStartSec=5s\n'+('RemainAfterExit=yes\n' if kind=='oneshot' else ''))
        paths[consumer].write_text(f'[Unit]\nAfter={provider}\n'+(f'{requirement}={provider}\n' if requirement else '')+f'[Service]\nType=oneshot\nExecStart=/usr/bin/python3 {root}/consumer.py\n')
        subprocess.run(['systemd-analyze','verify',*[str(p) for p in paths.values()]],check=True)
        ctl('daemon-reload')
    def reset():
        ctl('stop',consumer,provider)
        for name in paths:
            if value(name,'ActiveState')=='failed':ctl('reset-failed',name)
        for name in ('entered','gate','ready','observed'):(root/name).unlink(missing_ok=True)
    pending=None
    install('oneshot','')
    try:
        ctl('start',consumer)
        assert (root/'observed').read_text()=='False' and value(provider,'ActiveState')=='inactive'
        print('After alone: provider not pulled in; consumer ran before readiness')
        reset();install('fail','Wants');ctl('start',consumer)
        assert value(provider,'ActiveState')=='failed' and (root/'observed').read_text()=='False'
        print('Wants plus After: consumer runs despite provider failure')
        reset();install('fail','Requires');ctl('start',consumer,success=False)
        assert value(provider,'ActiveState')=='failed' and not (root/'observed').exists()
        print('Requires plus After: failed provider prevents consumer execution')
        reset();install('oneshot','Requires');ctl('start',consumer)
        assert (root/'observed').read_text()=='True'
        print('oneshot provider: consumer sees completed setup')
        reset();install('exec','Requires');ctl('start',consumer)
        assert (root/'observed').read_text()=='False' and value(provider,'ActiveState')=='active'
        (root/'gate').touch();wait_for(lambda:(root/'ready').exists())
        print('Type=exec provider: process started before application readiness')
        reset();install('notify','Requires')
        pending=subprocess.Popen(['systemctl','start',consumer],text=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
        wait_for(lambda:(root/'entered').exists())
        assert pending.poll() is None and not (root/'observed').exists()
        assert value(provider,'ActiveState')=='activating'
        (root/'gate').touch();out,err=pending.communicate(timeout=10)
        assert pending.returncode==0,err
        assert (root/'observed').read_text()=='True'
        print('Type=notify provider: dependent waits for actual READY=1')
    finally:
        reset()
        if pending is not None and pending.poll() is None:pending.communicate(timeout=5)
        for path in paths.values():path.unlink()
        ctl('daemon-reload')