import datetime,hashlib,importlib.util,json,os,pathlib,subprocess,sys,time sys.dont_write_bytecode=True r=pathlib.Path(__file__).resolve().parent;p=r.parent spec=importlib.util.spec_from_file_location('row276_driver',p/'driver.py');d=importlib.util.module_from_spec(spec);spec.loader.exec_module(d) start=time.monotonic();results=dict(grant='sage-2964',steps=[]) (r/'start.txt').write_text(datetime.datetime.now(datetime.timezone.utc).isoformat()+'\n');print((r/'HEADER.json').read_text(),flush=True) def step(name,cmd,red=False,seconds=1200): print('START '+name,flush=True);v=d.run(cmd,r/(name+'.log'),seconds=seconds);v.update(name=name,expected_red=red);results['steps'].append(v);results['wall_seconds']=round(time.monotonic()-start,3);(r/'results.json').write_text(json.dumps(results,indent=2)+'\n') if v['timeout'] or (v['code']==0 if red else v['code']!=0):raise RuntimeError('unexpected result: stop and classify '+name) if red and 'legacy supervisor block exists' not in (r/(name+'.log')).read_text():raise RuntimeError('wrong RED reason: stop and classify') print('END '+name,flush=True) for f in ['packages/cli/tests/case1-accept.test.mjs','packages/cli/tests/case-sudo-fixture.mjs']: step('syntax-'+str(len(results['steps'])),['timeout','-k','10','60','node','--check',f],seconds=90) target=d.REPO/'scripts/case1-accept.sh';original=target.read_bytes();status=subprocess.check_output(['git','status','--porcelain','--untracked-files=no'],cwd=d.REPO) try: target.write_bytes(subprocess.check_output(['git','show','48e450ce8:scripts/case1-accept.sh'],cwd=d.REPO)) step('intended-red',['node','--test','--test-reporter=tap','--test-name-pattern','row276 talk legacy inner timeout','packages/cli/tests/case1-accept.test.mjs'],red=True) finally: target.write_bytes(original);assert target.read_bytes()==original;assert subprocess.check_output(['git','status','--porcelain','--untracked-files=no'],cwd=d.REPO)==status step('focused-green',['node','--test','--test-reporter=tap','--test-name-pattern','row276 talk legacy inner timeout','packages/cli/tests/case1-accept.test.mjs']) scope=json.loads((p/'GATE-SCOPE.json').read_text()) for i,f in enumerate(scope['whole_files']):step('whole-'+str(i+1),['node','--test','--test-reporter=tap',f]) step('changes-controller',[sys.executable,str(p/'driver.py'),'--grant','sage-2964','--out',str(r/'changes')],seconds=900) limits=json.loads((p/'SHELL-LIMITS.json').read_text())['scripts'] for i,f in enumerate(scope['shells']): limit=limits[f]['limit_seconds'];step('shell-'+str(i+1),['timeout','-k','10',str(limit),'bash',f],seconds=limit+30) (r/'end.txt').write_text(datetime.datetime.now(datetime.timezone.utc).isoformat()+'\n');print('GATE COMPLETE',flush=True)