21 lines
1.3 KiB
Python
21 lines
1.3 KiB
Python
import ast,asyncio,json,pathlib,re,typing,gzip
|
|
p=pathlib.Path(__file__).resolve().parent/'results/medium-slot1-priority2-draft-cuda1/result.json.gz'
|
|
d=json.loads(gzip.decompress(p.read_bytes()));content=d['decode'][1]['choices'][0]['message']['content']
|
|
source=re.search(r'```python\s*\n(.*?)```',content,re.S).group(1)
|
|
# Reviewed above: only asyncio task orchestration and exception handling, no I/O.
|
|
node=next(n for n in ast.parse(source).body if isinstance(n,ast.AsyncFunctionDef) and n.name=='first_success')
|
|
scope={'asyncio':asyncio,'Any':typing.Any,'List':typing.List,'Awaitable':typing.Awaitable}
|
|
exec(compile(ast.Module(body=[node],type_ignores=[]),'<reviewed-dflash-answer>','exec'),scope)
|
|
async def check(all_fail):
|
|
async def task(delay,fail):
|
|
await asyncio.sleep(delay)
|
|
if fail:raise ValueError('synthetic failure')
|
|
return 7
|
|
try:
|
|
r=await asyncio.wait_for(scope['first_success']([task(.001,True),task(.005,all_fail)]),timeout=1)
|
|
return {'returned':r,'pass':not all_fail and r==7}
|
|
except Exception as e:
|
|
return {'error':type(e).__name__+': '+str(e),'pass':all_fail and type(e).__name__=='ExceptionGroup'}
|
|
out={'scope':'Two checks of manually reviewed decode answer, not broad quality evaluation','fast_failure_then_success':asyncio.run(check(False)),'all_fail':asyncio.run(check(True))}
|
|
print(json.dumps(out,indent=2))
|