62 lines
4.3 KiB
Python
62 lines
4.3 KiB
Python
from pathlib import Path
|
|
import hashlib, json, ctypes
|
|
OUT=Path(__file__).resolve().parent
|
|
REST=OUT/'rest-strict'
|
|
def data(path): return json.loads(path.read_text(encoding='utf-8-sig'))
|
|
def sha(path): return hashlib.sha256(path.read_bytes()).hexdigest().upper()
|
|
results=data(OUT/'runner-status.json')
|
|
assert len(results)==3 and all(x['exit_code']==0 for x in results)
|
|
assert data(REST/'result.json')['requests']==25
|
|
post=data(OUT/'postflight.json')
|
|
assert post['all_inputs_unchanged']
|
|
payload=data(OUT/'coverage-probe.json')
|
|
checks=[]
|
|
for a,b in [('02-before-model','05-candidate-no-commit-model'),('03-before-bundle','06-candidate-no-commit-bundle'),('08-incomplete-model','16-rejection-model'),('09-incomplete-bundle','17-rejection-bundle'),('19-repaired-model','22-restart-model'),('20-repaired-bundle','23-restart-bundle'),('21-repaired-validation','24-restart-validation')]:
|
|
pa=REST/(a+'.response.txt');pb=REST/(b+'.response.txt')
|
|
assert pa.read_bytes()==pb.read_bytes(),(a,b)
|
|
checks.append({'first':a,'second':b,'raw_response_sha256':sha(pa),'exact_bytes_equal':True})
|
|
if 'model' in a:
|
|
assert data(REST/(a+'.meta.json'))['etag']==data(REST/(b+'.meta.json'))['etag']
|
|
for label,fixture in [('09-incomplete-bundle','incomplete'),('20-repaired-bundle','repaired'),('23-restart-bundle','repaired')]:
|
|
assert data(REST/(label+'.response.txt'))==payload[fixture]
|
|
dest=OUT/('readback-'+label)
|
|
dest.mkdir()
|
|
for name,value in payload[fixture]['files'].items():
|
|
(dest/name).write_bytes(value.encode('utf-8'))
|
|
for file in [REST/'server-1.stop.json',REST/'server-2.stop.json',OUT/'schema-stop.json']:
|
|
assert data(file)['has_exited']
|
|
for file in [REST/'server-1.launch.json',REST/'server-2.launch.json',OUT/'schema-launch.json']:
|
|
assert data(file)['PATH']==data(OUT/'preflight.json')['server_PATH']
|
|
assert data(file)['cwd']==data(OUT/'preflight.json')['cwd']
|
|
kernel=ctypes.WinDLL('kernel32',use_last_error=True)
|
|
kernel.OpenProcess.argtypes=[ctypes.c_ulong,ctypes.c_int,ctypes.c_ulong]
|
|
kernel.OpenProcess.restype=ctypes.c_void_p
|
|
kernel.GetExitCodeProcess.argtypes=[ctypes.c_void_p,ctypes.POINTER(ctypes.c_ulong)]
|
|
kernel.QueryFullProcessImageNameW.argtypes=[ctypes.c_void_p,ctypes.c_ulong,ctypes.c_wchar_p,ctypes.POINTER(ctypes.c_ulong)]
|
|
kernel.CloseHandle.argtypes=[ctypes.c_void_p]
|
|
processes=[]
|
|
for file in [OUT/'rest/server-1.stop.json',OUT/'rest/server-2.stop.json',REST/'server-1.stop.json',REST/'server-2.stop.json',OUT/'schema-stop.json']:
|
|
pid=data(file)['pid'];handle=kernel.OpenProcess(0x1000,0,pid)
|
|
entry={'pid':pid,'probe_stop_receipt':str(file.relative_to(OUT))}
|
|
if handle:
|
|
try:
|
|
code=ctypes.c_ulong();assert kernel.GetExitCodeProcess(handle,ctypes.byref(code))
|
|
name=ctypes.create_unicode_buffer(32768);size=ctypes.c_ulong(32768)
|
|
assert kernel.QueryFullProcessImageNameW(handle,0,name,ctypes.byref(size))
|
|
entry.update({'current_exit_code':code.value,'current_image':name.value,'pid_reused':not name.value.lower().endswith('arcadia-server.exe')})
|
|
assert code.value!=259 or entry['pid_reused'],entry
|
|
finally:kernel.CloseHandle(handle)
|
|
else:
|
|
entry.update({'not_openable':True,'win32_error':ctypes.get_last_error()})
|
|
assert entry['win32_error']==87,entry
|
|
processes.append(entry)
|
|
(OUT/'process-readback.json').write_text(json.dumps(processes,indent=2),encoding='utf-8')
|
|
http=data(OUT/'schema-http.json')
|
|
spec=json.loads(http[0]['response'])
|
|
assert spec==data(OUT.parent.parent/'docs/openapi.json')
|
|
save={'status':'passed','raw_snapshot_pairs':checks,'source_bundle_exact_fixture_equality':['incomplete','repaired','restart-repaired'],'schema_http_requests_recorded':len(http),'schemas':len(spec['components']['schemas']),'server_processes_stopped':[data(p) for p in [REST/'server-1.stop.json',REST/'server-2.stop.json',OUT/'schema-stop.json']],'input_hashes_unchanged':post['inputs']}
|
|
(OUT/'readback.json').write_text(json.dumps(save,indent=2),encoding='utf-8')
|
|
files=[p for p in OUT.rglob('*') if p.is_file() and p.name!='evidence-sha256.json']
|
|
(OUT/'evidence-sha256.json').write_text(json.dumps({str(p.relative_to(OUT)):sha(p) for p in sorted(files)},indent=2),encoding='utf-8')
|
|
print(json.dumps({'result':'passed','exact_raw_response_pairs':len(checks),'manifest_files':len(files),'readback_sha256':sha(OUT/'readback.json')},indent=2))
|