27 lines
1.7 KiB
Python
27 lines
1.7 KiB
Python
"""Execute the real reservation block without starting a compositor or camera."""
|
|
from pathlib import Path
|
|
import json
|
|
import signal
|
|
import subprocess
|
|
import tempfile
|
|
|
|
runner=Path(__file__).with_name('run.py').resolve()
|
|
source=runner.read_text().split("scales = ")[0]
|
|
# CONTRACT: only the launch tail is replaced; production reservation executes.
|
|
root=Path(tempfile.mkdtemp(prefix='fractional-budget-test-'))
|
|
script=root/'reservation.py'
|
|
script.write_text('__file__='+repr(str(runner))+'\n'+source+"\nprint('RESERVED',flush=True)\nsys.stdin.readline()\nraise SystemExit(7)\n")
|
|
def launch(lab,pairs):
|
|
return subprocess.Popen(['python',str(script),str(lab),'private_mosaic','G1','--camera-pairs',str(pairs)],stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)
|
|
lab=root/'lab';lab.mkdir()
|
|
p=launch(lab,1);assert p.stdout.readline().strip()=='RESERVED'
|
|
q=launch(lab,1);out,err=q.communicate();assert q.returncode!=0 and 'BlockingIOError' in err
|
|
p.send_signal(signal.SIGINT);p.communicate(timeout=5)
|
|
assert json.loads((lab/'fractional-camera-budget.json').read_text())['consumed']==1
|
|
p=launch(lab,1);assert p.stdout.readline().strip()=='RESERVED';p.communicate();assert p.returncode==7
|
|
p=launch(lab,1);out,err=p.communicate();assert p.returncode==2 and 'exhausted' in err
|
|
p=launch(lab,0);assert p.stdout.readline().strip()=='RESERVED';p.communicate()
|
|
assert json.loads((lab/'fractional-camera-budget.json').read_text())['consumed']==2
|
|
old=root/'old';(old/'runs/control-fractional-G1-old').mkdir(parents=True)
|
|
p=launch(old,0);out,err=p.communicate();assert p.returncode==2 and 'silent budget reset' in err
|
|
print('PASS reserve/lock/interruption/failure/exhaustion/zero/missing-ledger; no compositor launched:',root)
|