36 lines
2.3 KiB
Python
36 lines
2.3 KiB
Python
"""Verify the completed isolated GPU outputs without submitting or touching user projects."""
|
|
from pathlib import Path
|
|
import hashlib,json,sqlite3,subprocess,urllib.request
|
|
|
|
root=Path(__file__).resolve().parent.parent
|
|
test=root/'integration-work/hd-test'
|
|
api='http://127.0.0.1:18879/api/v1'
|
|
with sqlite3.connect(test/'huobao.sqlite3') as db:
|
|
rows=db.execute('SELECT id,status,local_path,created_at,updated_at FROM sys_task WHERE id IN (86,87) ORDER BY id').fetchall()
|
|
if any(row[1]!='completed' for row in rows):
|
|
print(json.dumps({'ready':False,'tasks':[[r[0],r[1]] for r in rows]}))
|
|
raise SystemExit(2)
|
|
fixture=json.loads((test/'fixture.json').read_text())
|
|
original=db.execute('SELECT video_url FROM storyboards WHERE id=?',(fixture['storyboard'],)).fetchone()[0]
|
|
assert original==fixture['original'], 'HD must not replace the original automatically'
|
|
|
|
report={'passed':True,'original_sha256':hashlib.sha256((test/original).read_bytes()).hexdigest(),'outputs':[]}
|
|
assert report['original_sha256']=='16ea40eaca50ef6d1572123aa6252ca2615c288ece2f7f6bfcb2326526b6ec14'
|
|
for row,expected in zip(rows,[(2560,1440),(3840,2160)]):
|
|
raw=subprocess.check_output([str(root/'HuobaoDrama/resources/bin/ffprobe.exe'),'-v','error','-show_streams','-show_format','-of','json',str(test/row[2])],creationflags=0x08000000)
|
|
info=json.loads(raw); stream=next(s for s in info['streams'] if s['codec_type']=='video')
|
|
assert (stream['width'],stream['height'])==expected, (row[0],stream['width'],stream['height'])
|
|
report['outputs'].append({'task':row[0],'file':str(test/row[2]),'width':stream['width'],'height':stream['height'],'duration':info['format']['duration'],'fps':stream['r_frame_rate'],'created':row[3],'completed':row[4]})
|
|
|
|
def select(body):
|
|
req=urllib.request.Request(f"{api}/hd/storyboards/{fixture['storyboard']}/select",data=json.dumps(body).encode(),headers={'Content-Type':'application/json'})
|
|
with urllib.request.urlopen(req,timeout=30) as response: return json.load(response)['data']['video_url']
|
|
|
|
try:
|
|
assert select({'task_id':87})==rows[1][2]
|
|
finally:
|
|
assert select({'original':True})==fixture['original']
|
|
report['version_switch_and_restore']=True
|
|
(root/'integration-work/real-hd-verification.json').write_text(json.dumps(report,indent=2),encoding='utf-8')
|
|
print(json.dumps(report,ensure_ascii=False))
|