Cron 按日历重复触发任务

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

assert os.geteuid() == 0 and Path('/.dockerenv').exists(), 'use a disposable root Linux container'
assert subprocess.run(['pgrep', '-x', 'cron'], capture_output=True).returncode == 1
assert subprocess.run(['crontab', '-l'], capture_output=True).returncode != 0

with TemporaryDirectory(prefix='cron-overlap-') as directory:
    root = Path(directory)
    events_path = root / 'events.jsonl'
    stop = root / 'stop'
    worker = root / 'worker.py'
    worker.write_text('''from pathlib import Path
import fcntl,json,os,sys,time
root=Path(__file__).parent
kind=sys.argv[1]
def record(event):
    data=json.dumps({'event':event,'kind':kind,'pid':os.getpid(),'monotonic':time.monotonic()})+'\\n'
    fd=os.open(root/'events.jsonl',os.O_WRONLY|os.O_CREAT|os.O_APPEND,0o600)
    try: os.write(fd,data.encode())
    finally: os.close(fd)
lock=None
if kind=='locked':
    lock=open(root/'job.lock','a')
    try: fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
    except BlockingIOError:
        record('skip')
        raise SystemExit(0)
record('start')
try:
    until=time.monotonic()+70
    while time.monotonic()<until and not (root/'stop').exists(): time.sleep(0.1)
finally:
    record('end')
    if lock: lock.close()
''')
    table = ''.join(f'* * * * * /usr/bin/python3 {worker} {kind} >>{root}/worker.log 2>&1\n'
                    for kind in ('plain', 'locked'))
    subprocess.run(['crontab', '-'], input=table, text=True, check=True)
    daemon = subprocess.Popen(['cron', '-f'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    def read_events():
        if not events_path.exists():
            return []
        return [json.loads(line) for line in events_path.read_text().splitlines()]
    try:
        deadline = time.monotonic() + 140
        while time.monotonic() < deadline:
            assert daemon.poll() is None, 'cron exited'
            events = read_events()
            plain = [e for e in events if (e['kind'], e['event']) == ('plain', 'start')]
            locked = [e for e in events if (e['kind'], e['event']) == ('locked', 'start')]
            skipped = [e for e in events if (e['kind'], e['event']) == ('locked', 'skip')]
            if len(plain) >= 2 and skipped:
                break
            time.sleep(0.2)
        else:
            raise AssertionError('two real cron ticks were not observed')
        assert len(plain) == 2 and len(locked) == 1 and len(skipped) == 1
        assert plain[1]['monotonic'] - plain[0]['monotonic'] < 70
        assert not any(e['event'] == 'end' and e['pid'] == plain[0]['pid'] for e in events)
        print(json.dumps({'actual_cron_starts': len(plain), 'overlap_observed': True,
                          'seconds_between_starts': round(plain[1]['monotonic']-plain[0]['monotonic'], 3),
                          'locked_starts': len(locked), 'locked_skips': len(skipped)}), flush=True)
    finally:
        daemon.terminate()
        stop.touch()
        daemon.communicate(timeout=5)
        subprocess.run(['crontab', '-r'], check=True)
        deadline = time.monotonic() + 5
        while time.monotonic() < deadline:
            events = read_events()
            started = {e['pid'] for e in events if e['event'] == 'start'}
            ended = {e['pid'] for e in events if e['event'] == 'end'}
            if started <= ended:
                break
            time.sleep(0.1)
        else:
            raise AssertionError('test workers did not finish')