systemd timer 安排服务启动，不保证任务成功

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

name='xbcatm-timer-'+uuid.uuid4().hex[:8];service=name+'.service';timer=name+'.timer'
paths={unit:Path('/run/systemd/system')/unit for unit in (service,timer)}
assert all(not p.exists() for p in paths.values())
def run(*args):
    result=subprocess.run(args,text=True,capture_output=True,timeout=10)
    assert result.returncode==0,(args,result.stderr)
    return result.stdout.strip()
def value(unit,key):return run('systemctl','show',unit,'-p',key,'--value')
def wait_for(check):
    deadline=time.monotonic()+5
    while not check():
        assert time.monotonic()<deadline,'transition timed out'
        time.sleep(0.02)
with TemporaryDirectory(prefix=name+'-',dir='/run') as directory:
    root=Path(directory);worker=root/'worker.py'
    worker.write_text(r'''from pathlib import Path
import json,sys,time
root=Path(__file__).parent;number=len(list(root.glob('start-*.json')))+1
def record(kind):
    temp=root/'record.tmp';temp.write_text(json.dumps(time.monotonic()))
    temp.replace(root/(kind+'-'+str(number)+'.json'))
record('start')
if (root/'fail').exists():sys.exit(7)
while not (root/('release-'+str(number))).exists():time.sleep(0.01)
record('end')
''')
    paths[service].write_text(f'[Service]\nType=oneshot\nTimeoutStartSec=10s\nExecStart=/usr/bin/python3 {worker}\n')
    definition='[Timer]\nOnActiveSec=100ms\nOnUnitInactiveSec=10s\nAccuracySec=1ms\n'
    paths[timer].write_text(definition);(root/'fail').touch()
    try:
        run('systemd-analyze','verify',*[str(p) for p in paths.values()])
        run('systemctl','daemon-reload');run('systemctl','start',timer)
        wait_for(lambda:value(service,'ActiveState')=='failed')
        assert value(service,'ExecMainStatus')=='7' and value(timer,'ActiveState')=='active'
        run('systemctl','stop',timer)
        print('timer active while its first job failed with exit status 7')
        (root/'fail').unlink();run('systemctl','reset-failed',service)
        paths[timer].write_text(definition.replace('OnUnitInactiveSec=10s','OnUnitInactiveSec=250ms'))
        run('systemctl','daemon-reload');run('systemctl','start',timer)
        wait_for(lambda:(root/'start-2.json').exists())
        time.sleep(0.65)
        assert len(list(root.glob('start-*.json')))==2
        (root/'release-2').touch();wait_for(lambda:(root/'start-3.json').exists())
        gap=json.loads((root/'start-3.json').read_text())-json.loads((root/'end-2.json').read_text())
        assert gap>=0.25,gap
        print('no overlapping launch; next job starts at least 250ms after prior deactivation')
        pid=value(service,'MainPID');assert pid!='0'
        run('systemctl','stop',timer)
        assert value(service,'MainPID')==pid and value(service,'ActiveState')=='activating'
        print('stopping timer leaves its already running oneshot alive')
        (root/'release-3').touch();wait_for(lambda:value(service,'ActiveState')=='inactive')
        time.sleep(0.4)
        assert len(list(root.glob('start-*.json')))==3 and value(service,'Result')=='success'
        print('released job completes; stopped timer schedules no further run')
    finally:
        run('systemctl','stop',timer,service)
        for unit,path in paths.items():
            if value(unit,'ActiveState')=='failed':run('systemctl','reset-failed',unit)
            path.unlink()
        run('systemctl','daemon-reload')