#!/usr/bin/env python3 """Observe stop/disable with disposable user units; never touch existing jobs.""" import argparse import json import os import platform import signal import subprocess import sys import tempfile import time import uuid from datetime import datetime, timezone from pathlib import Path def main(): parser = argparse.ArgumentParser() parser.add_argument('--output', required=True) output = Path(parser.parse_args().output) if output.exists(): raise SystemExit('Existing evidence will not be overwritten.') runtime = Path(os.environ.get('XDG_RUNTIME_DIR', f'/run/user/{os.getuid()}')) if not (runtime / 'bus').exists(): raise SystemExit('A running systemd user manager and user bus are required.') prefix = 'punchblog-timer-lab-' + uuid.uuid4().hex[:12] service, timer = prefix + '.service', prefix + '.timer' unit_dir = runtime / 'systemd/user' unit_dir.mkdir(parents=True, exist_ok=True) commands, observations = [], [] def ctl(*args, check=True): result = subprocess.run(['systemctl', '--user', *args], capture_output=True, text=True, timeout=10) commands.append({'arguments': list(args), 'exit_code': result.returncode, 'stdout': result.stdout.strip(), 'stderr': result.stderr.strip()}) if check and result.returncode: raise RuntimeError(f'systemctl {args}: {result.stderr}') return result def state(unit): raw = ctl('show', unit, '-p', 'ActiveState', '-p', 'SubState', '-p', 'UnitFileState', '-p', 'MainPID', '-p', 'ConditionResult').stdout return dict(line.split('=', 1) for line in raw.splitlines() if '=' in line) def interrupt(_signal, _frame): raise SystemExit('Interrupted; cleaning up experiment units.') signal.signal(signal.SIGTERM, interrupt) signal.signal(signal.SIGINT, interrupt) with tempfile.TemporaryDirectory(prefix=prefix + '-') as directory: root = Path(directory) counter, marker, worker = root / 'runs.txt', root / 'completed', root / 'worker.py' worker.write_text('import os, sys, time\n' 'with open(sys.argv[1], "a") as out:\n' ' out.write(str(os.getpid()) + "\\n")\n' 'time.sleep(30)\n') service_text = f'''[Unit] Description=Punchblog disposable timer experiment ConditionPathExists=!{marker} [Service] Type=simple ExecStart={sys.executable} {worker} {counter} Restart=no RuntimeMaxSec=40 TimeoutStopSec=3 MemoryMax=64M CPUQuota=10% Nice=15 ''' timer_text = f'''[Unit] Description=Punchblog disposable timer experiment trigger [Timer] OnActiveSec=2s OnUnitInactiveSec=2s AccuracySec=100ms Unit={service} [Install] WantedBy=timers.target ''' files = [unit_dir / service, unit_dir / timer] def runs(): return len(counter.read_text().splitlines()) if counter.exists() else 0 def wait_for_runs(expected): deadline = time.monotonic() + 10 while runs() < expected and time.monotonic() < deadline: time.sleep(.1) assert runs() == expected, (expected, runs()) def observe(case): item = {'case': case, 'runs': runs(), 'service': state(service), 'timer': state(timer)} observations.append(item) return item try: for path, body in zip(files, (service_text, timer_text)): with path.open('x') as stream: stream.write(body) ctl('daemon-reload') ctl('enable', '--runtime', '--now', timer) wait_for_runs(1) observe('initial') ctl('stop', service) wait_for_runs(2) restarted = observe('stop_service_only') assert restarted['service']['ActiveState'] == 'active' ctl('disable', '--runtime', timer) disabled = observe('disable_timer_only') assert disabled['timer']['UnitFileState'] == 'disabled' assert disabled['timer']['ActiveState'] == 'active' ctl('stop', service) wait_for_runs(3) observe('disabled_timer_still_triggers') ctl('disable', '--runtime', '--now', timer) timer_stopped = observe('disable_now_timer') assert timer_stopped['timer']['ActiveState'] == 'inactive' assert timer_stopped['service']['ActiveState'] == 'active' ctl('stop', service) time.sleep(4) stopped = observe('stop_timer_and_service') assert stopped['runs'] == 3 assert stopped['service']['MainPID'] == '0' assert stopped['service']['ActiveState'] == 'inactive' marker.touch() ctl('start', service) guarded = observe('completion_guard') assert guarded['service']['ConditionResult'] == 'no' assert guarded['runs'] == 3 finally: ctl('disable', '--runtime', '--now', timer, check=False) ctl('stop', service, check=False) for path in files: path.unlink(missing_ok=True) ctl('daemon-reload', check=False) ctl('reset-failed', service, timer, check=False) cleanup = {'unit_files_removed': all(not p.exists() for p in files), 'timer_enable_link_removed': not (unit_dir / 'timers.target.wants' / timer).is_symlink(), 'worker_pids_gone': all(not Path('/proc/' + pid).exists() for pid in counter.read_text().splitlines())} assert all(cleanup.values()), cleanup labels = ['최초 타이머 시작', '서비스만 stop 후 다음 실행', '타이머만 disable 직후', 'disabled 타이머가 다시 실행', '타이머 disable --now 직후', '서비스도 stop 후 4초 관찰', '완료 파일 생성 후 수동 start'] record = { 'schema_version': 1, 'observed_at': datetime.now(timezone.utc).isoformat(), 'environment': {'os': platform.system(), 'python': platform.python_version(), 'systemd': subprocess.check_output(['systemctl', '--version'], text=True).splitlines()[0], 'scope': 'disposable --user units, runtime enablement only'}, 'method': '고유 이름의 사용자 서비스·타이머를 만들었다. 작업은 임시 파일에 PID 한 줄을 남긴 뒤 대기했다. 서비스의 Restart=no, 타이머의 OnUnitInactiveSec=2s 조건에서 stop과 disable을 비교했다. 종료 시 실험 유닛과 실행 프로세스를 정리했다.', 'limits': '재부팅, cron, 외부 watchdog, Persistent=true의 누락된 달력 일정, mask는 실행 실험하지 않았다. 중지 뒤 추가 실행 여부의 관찰 창은 4초이며 영구 중지를 증명하는 부하 시험은 아니다.', 'unit_name': prefix, 'units': {'service': service_text, 'timer': timer_text}, 'observations': observations, 'commands': commands, 'cleanup': cleanup, 'tables': {'states': {'headers': ['관찰 시점', '누적 실행', '서비스', '타이머', '타이머 등록'], 'rows': [[label, r['runs'], r['service']['ActiveState'], r['timer']['ActiveState'], r['timer']['UnitFileState']] for label, r in zip(labels, observations)]}}, } with output.open('x', encoding='utf-8') as stream: json.dump(record, stream, ensure_ascii=False, indent=2) stream.write('\n') print(json.dumps({'result': 'passed', 'observations': len(observations), 'cleanup': cleanup})) if __name__ == '__main__': main()