AI Flow

#!/usr/bin/env python3
import tempfile,pathlib,json,subprocess,os,hashlib
root=pathlib.Path(__file__).resolve().parents[1];helper=root/'tools/aiflow-widget.py'
with tempfile.TemporaryDirectory() as td:
 home=pathlib.Path(td);run=home/'run';run.mkdir();env=dict(os.environ,HOME=td);env['PATH']=str(home/'bin')+':'+env['PATH'];(home/'bin').mkdir();(home/'bin/adom-aiflow-widget').symlink_to(helper)
 def call(*args):return subprocess.check_output(['python3',str(helper),*args,'--run',str(run)],env=env,text=True)
 r={'currentStep':'placement','flow':{'steps':[{'name':'components'},{'name':'placement'}]},'steps':{'components':{'visits':[{'at':'2026-01-01T00:00:00Z','end':'2026-01-01T00:01:00Z'}]},'placement':{'visits':[{'at':'2026-01-01T00:01:00Z','end':None}]}}}
 (run/'run.json').write_text(json.dumps(r));ledger=run/'run.jsonl';ledger.write_text('');before=hashlib.sha256((run/'run.json').read_bytes()).hexdigest()
 assert 'disabled' in call('event','--label','ignored');assert not (run/'widget-state.json').exists()
 call('enable');d=json.loads(call('status'));assert d['progress']==50 and d['step']=='placement'
 call('event','--step','components','--label','5 of 40 reviewed','--progress','12.5');d=json.loads(call('status'));assert d['progress']==12.5 and d['step']=='components'
 old=(run/'widget-state.json').read_bytes();call('disable');call('event','--label','must not write');assert (run/'widget-state.json').read_bytes()==old;assert json.loads(call('status'))['thumbs']==[]
 assert hashlib.sha256((run/'run.json').read_bytes()).hexdigest()==before and ledger.read_text()==''
 print('PASS: opt-out prevents writes; enable selects run; planned-step progress; milestone override; ledger and run untouched')
# Expanded previews reuse the source image; opting out still prevents all new cache work.
with tempfile.TemporaryDirectory() as td:
 home=pathlib.Path(td);run=home/'run';run.mkdir();env=dict(os.environ,HOME=td)
 (run/'run.json').write_text(json.dumps({'currentStep':'components'}))
 def call(*args):return subprocess.run(['python3',str(helper),*args,'--run',str(run)],env=env,text=True,capture_output=True)
 src=run/'image.png';subprocess.run(['ffmpeg','-y','-v','error','-f','lavfi','-i','color=c=teal:s=800x600','-frames:v','1',str(src)],check=True)
 assert call('event','--file',str(src)).returncode==0 and not (run/'widget-previews').exists()
 assert call('enable').returncode==0
 result=call('event','--file',str(src),'--label','Generated test image');assert result.returncode==0,result.stderr
 name=json.loads(result.stdout)['thumb'];assert (run/'widget-thumbs'/name).exists() and (run/'widget-previews'/name).exists()
 state=json.loads(call('status').stdout);assert state['thumbs'][-1]['preview']=='/preview-image/'+name
 outside=home/'outside.png';outside.write_bytes(src.read_bytes());assert call('event','--file',str(outside)).returncode!=0
 assert call('disable').returncode==0
 assert json.loads(call('status').stdout)['thumbs']==[]
 print('PASS: large cached preview, named URL, outside-run rejection and disabled cache avoidance')