99 lines
6.4 KiB
Python
99 lines
6.4 KiB
Python
"""Exercise the staged application against local media fixtures, never production data."""
|
|
import hashlib,json,sqlite3,subprocess,threading,time,urllib.request,urllib.error,os
|
|
from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
|
|
from pathlib import Path
|
|
ROOT=Path(__file__).resolve().parent.parent
|
|
TEST=ROOT/'integration-work'/os.environ.get('HUOBAO_TEST_FOLDER','hd-test')
|
|
assert TEST.resolve().is_relative_to(ROOT/'integration-work')
|
|
FIXTURE=json.loads((TEST/'fixture.json').read_text())
|
|
BASE='http://127.0.0.1:'+os.environ.get('HUOBAO_TEST_PORT','18879')+'/api/v1'
|
|
BIN=ROOT/'integration-work/stage-4.0.11/app/resources/bin'
|
|
def api(path,body=None):
|
|
req=urllib.request.Request(BASE+path,data=json.dumps(body).encode() if body is not None else None,headers={'Content-Type':'application/json'})
|
|
try:
|
|
with urllib.request.urlopen(req,timeout=30) as r: return json.load(r)['data']
|
|
except urllib.error.HTTPError as e: raise RuntimeError(e.read().decode())
|
|
def connect(): return sqlite3.connect(TEST/'huobao.sqlite3')
|
|
def probe(path):
|
|
meta=json.loads(subprocess.check_output([str(BIN/'ffprobe.exe'),'-v','error','-show_streams','-of','json',str(path)],creationflags=0x08000000))
|
|
s=next(s for s in meta['streams'] if s['codec_type']=='video')
|
|
return s['width'],s['height']
|
|
for label,size in [('2K','2560x1440'),('4K','3840x2176')]:
|
|
target=TEST/(label+'.mp4')
|
|
if not target.exists() or probe(target)!=tuple(map(int,size.split('x'))):
|
|
subprocess.run([str(BIN/'ffmpeg.exe'),'-y','-hide_banner','-loglevel','error','-f','lavfi','-i',f'color=c=0x456789:s={size}:r=24:d=1',
|
|
'-c:v','libx264','-preset','ultrafast','-pix_fmt','yuv420p',str(target)],check=True,creationflags=0x08000000)
|
|
calls=[]
|
|
bad=False
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self,*args):pass
|
|
def answer(self,data):
|
|
self.send_response(200);self.send_header('Content-Type','application/json');self.end_headers();self.wfile.write(json.dumps(data).encode())
|
|
def do_POST(self):
|
|
body=json.loads(self.rfile.read(int(self.headers['Content-Length'])))
|
|
calls.append(body)
|
|
self.answer({'status':'succeeded','video_url':'http://127.0.0.1:18880/'+('2K' if bad else body['resolution'])+'.mp4','duration':4})
|
|
def do_GET(self):
|
|
if self.path=='/v1/models': return self.answer({'data':[{'id':'MiniMax-H3-VDN-H3'}]})
|
|
name=self.path.lstrip('/')
|
|
if name not in ('2K.mp4','4K.mp4'): self.send_error(404);return
|
|
data=(TEST/name).read_bytes()
|
|
self.send_response(200);self.send_header('Content-Type','video/mp4');self.send_header('Content-Length',str(len(data)));self.end_headers();self.wfile.write(data)
|
|
server=ThreadingHTTPServer(('127.0.0.1',18880),Handler)
|
|
threading.Thread(target=server.serve_forever,daemon=True).start()
|
|
with connect() as db:
|
|
cfgid=db.execute("SELECT id FROM ai_service_configs WHERE model LIKE '%VDN-H3%' ORDER BY id DESC LIMIT 1").fetchone()[0]
|
|
db.execute("UPDATE ai_service_configs SET is_active=0 WHERE model LIKE '%VDN-H3%' AND id!=?",(cfgid,))
|
|
db.execute('UPDATE ai_service_configs SET base_url=?,is_active=1 WHERE id=?',('http://127.0.0.1:18880',cfgid))
|
|
assert db.execute("SELECT COUNT(*) FROM ai_service_configs WHERE is_active=1 AND model LIKE '%VDN-H3%' AND base_url!='http://127.0.0.1:18880'").fetchone()[0]==0
|
|
sid=FIXTURE['storyboard']; eid=FIXTURE['episode']; original=FIXTURE['original']
|
|
sha=hashlib.sha256((TEST/original).read_bytes()).hexdigest()
|
|
def wait_task(tid):
|
|
for _ in range(60):
|
|
data=api('/tasks/'+str(tid)+'/diagnostics')
|
|
if data['status']!='processing':return data
|
|
time.sleep(0.5)
|
|
raise AssertionError('task timed out')
|
|
try:
|
|
complete=[]
|
|
for resolution,size in [('2K',(2560,1440)),('4K',(3840,2160))]:
|
|
tid=api(f'/hd/storyboards/{sid}',{'resolution':resolution})['id']
|
|
result=wait_task(tid)
|
|
assert result['status']=='completed',result
|
|
with connect() as db:
|
|
assert db.execute('SELECT video_url FROM storyboards WHERE id=?',(sid,)).fetchone()[0]==original
|
|
video=db.execute('SELECT local_path FROM sys_task WHERE id=?',(tid,)).fetchone()[0]
|
|
assert probe(TEST/video)==size
|
|
complete.append((tid,video))
|
|
listing=api(f'/hd/episodes/{eid}')
|
|
shot=next(s for s in listing['items'] if s['id']==sid)
|
|
assert shot['original']==original and shot['current']==original
|
|
assert all(v['error'] is None for v in shot['versions'] if v['status']=='completed')
|
|
api(f'/hd/storyboards/{sid}/select',{'task_id':complete[1][0]})
|
|
with connect() as db: assert db.execute('SELECT video_url FROM storyboards WHERE id=?',(sid,)).fetchone()[0]==complete[1][1]
|
|
api(f'/hd/storyboards/{sid}/select',{'original':True})
|
|
bad=True
|
|
tid=api(f'/hd/storyboards/{sid}',{'resolution':'4K'})['id']
|
|
assert wait_task(tid)['status']=='failed'
|
|
try: api(f'/hd/storyboards/{sid}/select',{'task_id':tid})
|
|
except RuntimeError:pass
|
|
else:raise AssertionError('failed low-resolution output was selectable')
|
|
assert hashlib.sha256((TEST/original).read_bytes()).hexdigest()==sha
|
|
assert all(c['model']=='MiniMax-H3-VDN-H3' and c['content'][1]['role']=='reference_image' and c['content'][1]['image_url']['url'].startswith('data:') for c in calls)
|
|
with connect() as db:
|
|
second=db.execute('SELECT id FROM storyboards WHERE episode_id=? AND id!=? ORDER BY id LIMIT 1',(eid,sid)).fetchone()[0]
|
|
db.execute('UPDATE storyboards SET video_url=? WHERE id=?',(complete[1][1],second))
|
|
merge=api(f'/merge/episodes/{eid}/merge',{'storyboard_ids':[sid,second]})
|
|
merge_id=merge.get('merge_id') or merge.get('id')
|
|
for _ in range(90):
|
|
with connect() as db: row=db.execute('SELECT status,merged_url,error_msg FROM video_merges WHERE id=?',(merge_id,)).fetchone()
|
|
if row and row[0]!='processing':break
|
|
time.sleep(1)
|
|
assert row and row[0]=='completed',row
|
|
assert probe(TEST/row[1])==(3840,2160)
|
|
(TEST/'test-results.json').write_text(json.dumps({'passed':True,'checks':['2K delivery','4K delivery','original remains unchanged','select and restore export version','wrong output size rejected','mixed resolution export keeps 4K'],'complete':complete,'original_sha256':sha},indent=2))
|
|
print('PASS: 2K/4K delivery, preserved original, version switching, wrong-size rejection and mixed-size 4K export')
|
|
finally:
|
|
with connect() as db: db.execute('UPDATE ai_service_configs SET base_url=? WHERE id=?',('http://192.168.110.111:18794',cfgid))
|
|
server.shutdown()
|