26 lines
1.8 KiB
Python
26 lines
1.8 KiB
Python
import ast,asyncio,json,pathlib,types
|
|
root=pathlib.Path(json.loads(pathlib.Path('/home/ps/huobao-gpu-selection/task-tracking-current.json').read_text())['directory'])
|
|
class Client:
|
|
def __init__(self,**kw):pass
|
|
async def __aenter__(self):return self
|
|
async def __aexit__(self,*args):pass
|
|
async def get(self,*args,**kw):
|
|
return types.SimpleNamespace(raise_for_status=lambda:None,json=lambda:queue)
|
|
httpx=types.SimpleNamespace(AsyncClient=Client);Any=object
|
|
provider_headers=lambda p:{}
|
|
decode_job=lambda jid:(types.SimpleNamespace(base_url='http://worker'),'target')
|
|
async def get_job(jid):return results.pop(0) if len(results)>1 else results[0]
|
|
for variant in ('general','vdn'):
|
|
tree=ast.parse((root/(variant+'.after.py')).read_text());node=next(n for n in tree.body if isinstance(n,ast.AsyncFunctionDef) and n.name=='huobao_job_status');node.decorator_list=[]
|
|
exec(compile(ast.Module(body=[node],type_ignores=[]),variant,'exec'),globals())
|
|
async def check():
|
|
global queue,results
|
|
results=[{'status':'running'}];queue={'queue_pending':[[1,'before'],[2,'target']]}
|
|
r=await huobao_job_status('job');assert r['worker_state']=='queued' and r['queue_position']==2
|
|
queue={'queue_running':[[1,'target']]};assert (await huobao_job_status('job'))['worker_state']=='running'
|
|
queue={};assert (await huobao_job_status('job'))['worker_state']=='missing'
|
|
results=[{'status':'running'},{'status':'succeeded','outputs':[{}]}];assert (await huobao_job_status('job'))['status']=='succeeded'
|
|
results=[{'status':'running','pollDeferred':True}];assert (await huobao_job_status('job'))['worker_state']=='checking'
|
|
results=[{'status':'failed'}];assert (await huobao_job_status('job'))['status']=='failed'
|
|
asyncio.run(check());print('PASS',variant,'queue/running/missing/finish race/quality-check/real failure')
|