56 lines
3.9 KiB
Python
56 lines
3.9 KiB
Python
"""Exercise saved GPU settings through the staged app; local mock media only."""
|
|
import json,sqlite3,threading,time,urllib.request,shutil
|
|
from pathlib import Path
|
|
from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
|
|
from email.parser import BytesParser
|
|
from email import policy
|
|
root=Path(__file__).resolve().parent.parent;test=root/'integration-work/hd-test-gpu'
|
|
assert test.name=='hd-test-gpu'
|
|
base='http://127.0.0.1:18882/api/v1';calls=[]
|
|
(test/'static/images').mkdir(exist_ok=True)
|
|
shutil.copy2(root/'integration-work/pinned-images-gpu1.png',test/'static/images/reference.png')
|
|
def api(path,body=None,method=None):
|
|
r=urllib.request.Request(base+path,data=json.dumps(body).encode() if body is not None else None,headers={'Content-Type':'application/json'},method=method)
|
|
with urllib.request.urlopen(r,timeout=30) as response:return json.load(response)['data']
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self,*args):pass
|
|
def do_GET(self):
|
|
file={'/image.png':root/'integration-work/pinned-images-gpu1.png','/video.mp4':test/'static/videos/hd-test-original.mp4','/hd.mp4':root/'integration-work/hd-test-alignment/2K.mp4'}[self.path]
|
|
data=file.read_bytes();self.send_response(200);self.send_header('Content-Length',str(len(data)));self.end_headers();self.wfile.write(data)
|
|
def do_POST(self):
|
|
raw=self.rfile.read(int(self.headers['Content-Length']));ctype=self.headers.get('Content-Type','')
|
|
if ctype.startswith('multipart/'):
|
|
msg=BytesParser(policy=policy.default).parsebytes(('Content-Type: '+ctype+'\r\n\r\n').encode()+raw)
|
|
body={p.get_param('name',header='content-disposition'):p.get_payload(decode=True).decode() for p in msg.iter_parts() if not p.get_filename()}
|
|
else:body=json.loads(raw)
|
|
calls.append((self.path,body.get('gpu')))
|
|
name='/image.png' if '/images/' in self.path else '/hd.mp4' if body['model']=='MiniMax-H3-VDN-H3' else '/video.mp4'
|
|
url='http://127.0.0.1:18880'+name
|
|
payload={'data':[{'url':url}]} if name.endswith('.png') else {'video_url':url,'duration':4}
|
|
data=json.dumps(payload).encode();self.send_response(200);self.send_header('Content-Type','application/json');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 sqlite3.connect(test/'huobao.sqlite3') as db:
|
|
saved=db.execute('SELECT id,base_url,settings,is_active FROM ai_service_configs WHERE id IN (4,6,10,11)').fetchall()
|
|
db.execute('UPDATE ai_service_configs SET is_active=0 WHERE id=11')
|
|
def wait(tid):
|
|
for _ in range(80):
|
|
d=api('/tasks/'+str(tid)+'/diagnostics')
|
|
if d['status']!='processing':assert d['status']=='completed',d;return
|
|
time.sleep(.25)
|
|
raise AssertionError('mock task timed out')
|
|
try:
|
|
for gpu,refs in [('gpu1',[]),('gpu3',['static/images/reference.png'])]:
|
|
api('/ai-configs/6',{'base_url':'http://127.0.0.1:18880','gpu':gpu},'PUT')
|
|
t=api('/tasks',{'type':'image','config_id':6,'model':'qwen_image_2_1','prompt':'A teacup.','size':'512x512','reference_images':refs});wait(t['id'])
|
|
api('/ai-configs/4',{'base_url':'http://127.0.0.1:18880','gpu':'gpu2'},'PUT')
|
|
t=api('/tasks',{'type':'video','config_id':4,'model':'MiniMax-H3','prompt':'A quiet lake.','duration':4,'aspect_ratio':'16:9'});wait(t['id'])
|
|
api('/ai-configs/10',{'base_url':'http://127.0.0.1:18880','gpu':'gpu3'},'PUT')
|
|
t=api('/hd/storyboards/1',{'resolution':'2K'});wait(t['id'])
|
|
assert calls==[('/v1/images/generations','gpu1'),('/v1/images/edits','gpu3'),('/v2/video_generation','gpu2'),('/v2/video_generation','gpu3')],calls
|
|
(root/'integration-work/gpu-dispatch-test.json').write_text(json.dumps({'passed':True,'requests':calls},indent=2))
|
|
print('PASS: saved settings reach text-to-image, reference-image, MiniMax video and VDN HD requests')
|
|
finally:
|
|
server.shutdown()
|
|
with sqlite3.connect(test/'huobao.sqlite3') as db:
|
|
for id,url,settings,active in saved:db.execute('UPDATE ai_service_configs SET base_url=?,settings=?,is_active=? WHERE id=?',(url,settings,active,id))
|