"""End-to-end isolated image->video lifecycle tests. Never calls live models or the live DB."""
import base64, hashlib, json, os, pathlib, re, sqlite3, struct, subprocess, threading, time, urllib.request, urllib.error, zlib
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from email.parser import BytesParser
from email.policy import default
root=pathlib.Path(__file__).resolve().parent.parent
data=root/'integration-work'/('composed-frame-test-'+time.strftime('%Y%m%d-%H%M%S'));data.mkdir()
stage=root/'integration-work/stage-4.0.11/app';dbfile=data/'huobao.sqlite3'
source=(root/'huobao-source/desktop/build/backend.mjs').read_text('utf-8')
source=source.replace('const require = __hbCreateRequire(import.meta.url)','const require = __hbCreateRequire('+json.dumps(str(stage/'resources/app.asar/build/backend.mjs'))+')',1)
source=re.sub(r'^import (\w+) from "(better-sqlite3|sharp)";',r'const \1 = require("\2");',source,flags=re.M)
source=re.sub(r'(\w+)\.hostname === "192\.168\.110\.111" && \["18792", "18794"\]\.includes\(\1.port\)',r'\1.hostname === "127.0.0.1" && \1.port === "18971"',source)
source=re.sub(r'&& !!taskRouting\(\{ provider: config\d*\.provider, taskId \}\)','&& !!taskId',source)
source=source.replace('intervalMs: 1e4','intervalMs: 100').replace('intervalMs: 5e3','intervalMs: 100')
(data/'backend.mjs').write_text(source,'utf-8')
def png(color):
 def chunk(tag,raw):return struct.pack('>I',len(raw))+tag+raw+struct.pack('>I',zlib.crc32(tag+raw)&0xffffffff)
 return b'\x89PNG\r\n\x1a\n'+chunk(b'IHDR',struct.pack('>IIBBBBB',32,18,8,2,0,0,0))+chunk(b'IDAT',zlib.compress((b'\0'+bytes(color)*32)*18))+chunk(b'IEND',b'')
events=[];image_jobs={};image_mode='complete';image_submit_release=threading.Event();image_submit_release.set();video_submit_release=threading.Event();video_submit_release.set()
class Mock(BaseHTTPRequestHandler):
 def log_message(self,*args):pass
 def reply(self,status,body):
  try:
   raw=json.dumps(body).encode();self.send_response(status);self.send_header('Content-Type','application/json');self.end_headers();self.wfile.write(raw)
  except (BrokenPipeError,ConnectionResetError):pass
 def do_POST(self):
  raw=self.rfile.read(int(self.headers.get('Content-Length','0')))
  if 'multipart/form-data' in self.headers.get('Content-Type',''):
   message=BytesParser(policy=default).parsebytes(('Content-Type: '+self.headers['Content-Type']+'\r\nMIME-Version: 1.0\r\n\r\n').encode()+raw)
   body={'images':[]}
   for part in message.iter_parts():
    name=part.get_param('name',header='content-disposition')
    if name=='image[]':body['images'].append(hashlib.sha256(part.get_payload(decode=True)).hexdigest())
    else:body[name]=part.get_payload(decode=True).decode()
  else:body=json.loads(raw or b'{}')
  events.append((self.path,body))
  if self.path.endswith('/images/edits'):
   jid='img-'+str(len(image_jobs)+1);image_jobs[jid]={'status':image_mode,'color':(len(image_jobs)*11%200,120,210)}
   image_submit_release.wait(25);self.reply(200,{'task_id':jid,'status':'processing'});return
  if '/images/task/' in self.path and self.path.endswith('/cancel'):
   jid=self.path.split('/')[-2];image_jobs[jid]['status']='cancelled';self.reply(200,{'task':{'id':jid,'status':'cancelled'}});return
  if self.path.endswith('/validate'):self.reply(200,{'valid':True});return
  if self.path.endswith('/cancel'):self.reply(200,{'task':{'id':self.path.split('/')[-2],'status':'cancelled'}});return
  if self.path.endswith('/video_generation'):
   video_submit_release.wait(25);self.reply(200,{'task_id':'vid-'+str(len(events))});return
  if self.path.endswith('/chat/completions'):
   content='An establishing shot. '+' '.join(re.findall(r'__HB_KEEP_\d+__',body['messages'][-1]['content']))
   self.reply(200,{'choices':[{'message':{'content':json.dumps({'translation':content})}}]});return
  self.reply(400,{'error':'unexpected fixture route'})
 def do_GET(self):
  if self.path.startswith('/image/'):
   raw=png(image_jobs[self.path.split('/')[-1]]['color']);self.send_response(200);self.send_header('Content-Type','image/png');self.end_headers();self.wfile.write(raw);return
  if '/images/task/' in self.path:
   jid=self.path.split('/')[-1];state=image_jobs[jid]['status']
   if state=='complete':self.reply(200,{'status':'completed','image_url':'http://127.0.0.1:18971/image/'+jid})
   elif state in ('fail','cancelled'):self.reply(200,{'status':'failed','cancelled':state=='cancelled','error':{'message':'Fixture image failure'}})
   else:self.reply(200,{'status':'processing','worker_state':'queued'})
   return
  self.reply(200,{'task':{'id':self.path.split('/')[-1],'status':'running','worker_state':'running'}})
server=ThreadingHTTPServer(('127.0.0.1',18971),Mock);threading.Thread(target=server.serve_forever,daemon=True).start()
env={**os.environ,'FFMPEG_BIN':str(stage/'resources/bin/ffmpeg.exe'),'FFPROBE_BIN':str(stage/'resources/bin/ffprobe.exe'),'ELECTRON_RUN_AS_NODE':'1','PORT':'18970','HUOBAO_DATA_DIR':str(data),'SQLITE_PATH':str(dbfile),'FRONTEND_DIST':str(root/'huobao-source/frontend/.output/public'),'WORKSPACE_PATH':str(stage/'resources/workspace-template')}
log=(data/'backend.log').open('w',encoding='utf-8');process=None
def api(method,path,body=None,expected=200):
 req=urllib.request.Request('http://127.0.0.1:18970/api/v1'+path,method=method,data=None if body is None else json.dumps(body).encode(),headers={'Content-Type':'application/json'})
 try:
  with urllib.request.urlopen(req,timeout=30) as response:status=response.status;value=json.load(response)
 except urllib.error.HTTPError as error:status=error.code;value=json.load(error)
 assert status==expected,(path,status,value)
 return value.get('data',value)
def sql(query,args=()):
 with sqlite3.connect(dbfile) as db:db.row_factory=sqlite3.Row;return [dict(r) for r in db.execute(query,args)]
def task(id):return sql('select * from sys_task where id=?',(id,))[0]
def plan(id):return json.loads(task(id)['params'])['firstFrameComposition']
def wait(fn,seconds=15):
 deadline=time.monotonic()+seconds
 while time.monotonic()<deadline:
  value=fn()
  if value:return value
  time.sleep(.1)
 raise AssertionError('Condition did not become true')
def start():
 global process
 process=subprocess.Popen([str(stage/'HuobaoDrama.exe'),str(data/'backend.mjs')],cwd=data,env=env,stdout=log,stderr=log,creationflags=subprocess.CREATE_NO_WINDOW)
 for _ in range(150):
  assert process.poll() is None
  try:api('GET','/health');return
  except OSError:time.sleep(.1)
 raise AssertionError('Backend did not start')
def stop():
 global process
 process.terminate();process.wait(10);process=None
def send(shot,**more):return api('POST','/tasks',{'type':'video','storyboard_id':shot,'drama_id':1,'config_id':1,'model':'LTX-2.3','frame_image_config_id':2,'frame_image_model':'qwen_image_2_1','prompt':'0-4s: @图片1404宿舍, @图片2Girl looks at @Other.','duration':4,'gpu':'gpu3','force_new':True,**more},201)['id']
def videos():return [b for p,b in events if p.endswith('/video_generation')]
def images():return [b for p,b in events if p.endswith('/images/edits')]
def finished_frame(id):
 wait(lambda:task(id)['task_id'] or task(id)['status']=='failed')
 assert task(id)['task_id'],task(id)
 return plan(id)
try:
 start();ts='2026-11-12T00:00:00Z'
 sql('insert into dramas(id,title,created_at,updated_at) values(1,?,?,?)',('Composition fixture',ts,ts))
 sql('insert into episodes(id,drama_id,episode_number,title,content,script_content,image_config_id,created_at,updated_at) values(1,1,1,?,?,?,?,?,?)',('Fixture','A girl in a dormitory.','A girl in a dormitory.',2,ts,ts))
 for id,service,provider,model,gpu in [(1,'video','localvideo','LTX-2.3','gpu3'),(2,'image','openai','qwen_image_2_1','gpu1'),(3,'text','openai','fixture-translator','auto')]:
  sql('insert into ai_service_configs(id,service_type,provider,name,base_url,api_key,model,settings,is_active,created_at,updated_at) values(?,?,?,?,?,?,?,?,1,?,?)',(id,service,provider,'Fixture '+service,'http://127.0.0.1:18971','fixture-only',json.dumps([model,'Wan2.2-I2V','MiniMax-H3'] if service=='video' else [model]),json.dumps({'gpu':gpu}),ts,ts))
 (data/'static/images').mkdir(parents=True)
 for i in range(16):(data/f'static/images/input{i}.png').write_bytes(png((i*13%255,15,70)))
 sql("insert into scenes(id,drama_id,location,time,prompt,lighting,image_url,created_at,updated_at) values(1,1,?,'night','Blue walls; east window; oak desk.','Warm east lamp.','static/images/input0.png',?,?)",('404宿舍',ts,ts))
 sql('insert into episode_scenes(episode_id,scene_id,created_at) values(1,1,?)',(ts,))
 for i in range(1,14):
  name='Girl' if i==1 else 'Other' if i==2 else 'Character'+str(i)
  sql('insert into characters(id,drama_id,name,image_url,created_at,updated_at) values(?,1,?,?,?,?)',(i,name,f'static/images/input{i}.png',ts,ts))
  sql('insert into episode_characters(episode_id,character_id,created_at) values(1,?,?)',(i,ts))
 sql("insert into props(id,drama_id,name,image_url,created_at,updated_at) values(1,1,'Lamp','static/images/input14.png',?,?)",(ts,ts))
 sql('insert into episode_props(episode_id,prop_id,created_at) values(1,1,?)',(ts,))
 for i in range(1,24):
  sql("insert into storyboards(id,episode_id,storyboard_number,scene_id,description,video_prompt,duration,created_at,updated_at) values(?,1,?,1,'Girl looks up in 404 dormitory.','0-4s: @404宿舍, @Girl looks at @Other.',4,?,?)",(i,i,ts,ts))
  for c in [1,2]:sql('insert into storyboard_characters(storyboard_id,character_id) values(?,?)',(i,c))
  sql('insert into storyboard_props(storyboard_id,prop_id) values(?,1)',(i,))
 sources={name:sql('select * from '+name+' order by id') for name in ['characters','scenes','props']}
 # Real multipart edits receive every bound asset, and videos wait for that output.
 image_mode='hold';one=send(1);wait(lambda:len(images())==1)
 assert not videos() and len(images()[0]['images'])==4 and images()[0]['gpu']=='gpu1'
 assert '404宿舍' in images()[0]['prompt'] and '单个' not in images()[0]['prompt']
 image_jobs['img-1']['status']='complete';first=finished_frame(one)
 assert first['frameUrl'] and sql('select first_frame_image from storyboards where id=1')[0]['first_frame_image']==first['frameUrl']
 v=videos()[-1];assert v['model']=='LTX-2.3' and v['gpu']=='gpu3'
 refs=[x for x in v['content'] if x['type']=='image_url'];assert len(refs)==1 and refs[0]['role']=='first_frame'
 assert '@图片' not in v['content'][0]['text'] and '@404宿舍' in v['content'][0]['text']
 assert base64.b64decode(refs[0]['image_url']['url'].split(',')[1])==(data/first['frameUrl']).read_bytes()
 # Same exact input reuses the finished frame across LTX/Wan video retries.
 sql("update sys_task set status='failed',error_code='generation_failed' where id=?",(one,));before=len(images())
 two=send(1,model='Wan2.2-I2V');reused=finished_frame(two)
 assert len(images())==before and reused['frameUrl']==first['frameUrl'] and reused['reusedFromTaskId']==one
 # Cancellation stops the child image and never sends its video.
 cancelled=send(2);wait(lambda:len(images())==before+1);wait(lambda:task(plan(cancelled)['childTaskIds'][0])['task_id'])
 before_v=len(videos());result=api('POST',f'/tasks/{cancelled}/cancel',{})
 assert result['outcome']=='cancelled';assert task(cancelled)['error_code']=='user_cancelled';assert len(videos())==before_v
 # All 15 original references (> route's old 9-image limit) reach Qwen over 3 rounds.
 for c in range(3,14):sql('insert into storyboard_characters(storyboard_id,character_id) values(3,?)',(c,))
 image_mode='complete';before=len(images());many=send(3,reference_image_urls=[f'static/images/input{i}.png' for i in range(15)])
 multi=finished_frame(many);assert len(multi['assets'])==15 and multi['rounds']==3 and len(multi['images'])==3
 assert [len(x['images']) for x in images()[before:]]==[6,6,5]
 originals=[]
 for child_id in multi['childTaskIds']:originals+=json.loads(task(child_id)['params'])['referenceAssets']
 assert originals==multi['assets'] and len(videos()[-1]['content'])==2
 # Missing references fail before replacement touches an existing running video.
 sql("update characters set image_url=null where id=2");before=len(events)
 api('POST','/tasks',{'type':'video','storyboard_id':1,'model':'LTX-2.3','config_id':1,'prompt':'test','replace_task_id':two,'force_new':True},400)
 assert len(events)==before and task(two)['status']=='processing'
 sql("update characters set image_url='static/images/input2.png' where id=2")
 # Image failure cannot emit a video or overwrite source assets.
 image_mode='fail';before_v=len(videos());failed=send(4)
 wait(lambda:task(failed)['status']=='failed');assert '首帧合成' in task(failed)['error_msg'] and len(videos())==before_v
 # Edits during generation keep the old output in history but prohibit stale video submission.
 image_mode='hold';stale=send(5);wait(lambda:plan(stale)['childTaskIds']);child=plan(stale)['childTaskIds'][0];wait(lambda:task(child)['task_id'])
 sql("update storyboards set description='Changed pose' where id=5");image_jobs[task(child)['task_id']]['status']='complete'
 wait(lambda:task(stale)['status']=='failed');assert '已变化' in task(stale)['error_msg'] and len(videos())==before_v
 assert task(child)['local_path'] and not sql('select first_frame_image from storyboards where id=5')[0]['first_frame_image']
 # Restart follows the SAME image task; no duplicate image or video generation.
 restart=send(6);wait(lambda:plan(restart)['childTaskIds']);child=plan(restart)['childTaskIds'][0];wait(lambda:task(child)['task_id'])
 jid=task(child)['task_id'];before=len(images());stop();start();assert task(child)['task_id']==jid
 image_jobs[jid]['status']='complete';finished_frame(restart);assert len(images())==before
 # Cancel while the image POST has not returned its ID. Wait for acknowledgement, then cancel exactly that job.
 image_submit_release.clear();image_mode='hold';pending=send(7);wait(lambda:len(images())==before+1)
 assert api('POST',f'/tasks/{pending}/cancel',{})['outcome']=='pending'
 image_submit_release.set();wait(lambda:task(pending)['error_code']=='user_cancelled')
 assert image_jobs['img-'+str(len(image_jobs))]['status']=='cancelled'
 # Previous-shot tail frame is an EXTRA reference, never a replacement for bound assets.
 (data/'static/videos').mkdir(exist_ok=True)
 subprocess.run([str(stage/'resources/bin/ffmpeg.exe'),'-hide_banner','-loglevel','error','-f','lavfi','-i','color=c=blue:s=320x192:d=1','-c:v','libx264',str(data/'static/videos/source.mp4')],check=True,creationflags=subprocess.CREATE_NO_WINDOW)
 sql("update storyboards set video_url='static/videos/source.mp4',continuity_end='At the east window' where id=20")
 api('PUT','/storyboards/21/continuity',{'continuity_mode':'first_frame','continuity_source_id':20,'continuity_start':'At the east window'})
 image_mode='complete';before=len(images());continued=send(21,model='Wan2.2-I2V');continued_plan=finished_frame(continued)
 assert continued_plan['continuityFrameUrl'] and len(images()[before]['images'])==5
 assert len(continued_plan['assets'])==4 and continued_plan['frameUrl']!=continued_plan['continuityFrameUrl']
 assert len([x for x in videos()[-1]['content'] if x['type']=='image_url'])==1
 # A crash after a video POST may have reached the provider: never submit a duplicate.
 before=len(videos());video_submit_release.clear();unknown=send(22);wait(lambda:len(videos())==before+1)
 assert task(unknown)['stage']=='submitting' and not task(unknown)['task_id'];stop();start()
 wait(lambda:task(unknown)['error_code']=='submission_unknown');video_submit_release.set();assert len(videos())==before+1
 # Explicit regeneration bypasses the valid frame cache, then supports normal cancellation.
 image_mode='hold';before=len(images());fresh=send(1,replace_task_id=two,regenerate_first_frame=True)
 wait(lambda:len(images())==before+1);assert task(two)['error_code']=='user_cancelled';wait(lambda:task(plan(fresh)['childTaskIds'][0])['task_id'])
 api('POST',f'/tasks/{fresh}/cancel',{})
 # Existing direct multi-reference H3 flow must not create an image task.
 image_mode='complete';before=len(images());h3=send(8,model='MiniMax-H3',reference_image_urls=['static/images/input0.png','static/images/input1.png'])
 wait(lambda:task(h3)['task_id']);assert len(images())==before and len([x for x in videos()[-1]['content'] if x['type']=='image_url'])==2
 # The default selected video config also accepts every binding when model is omitted.
 for c in range(3,14):sql('insert into storyboard_characters(storyboard_id,character_id) values(23,?)',(c,))
 default_task=send(23,model=None,reference_image_urls=[f'static/images/input{i}.png' for i in range(15)])
 assert len(finished_frame(default_task)['assets'])==15
 # No source artwork or their metadata was changed by composition.
 assert all(sql('select * from '+name+' order by id')==snapshot for name,snapshot in sources.items())
 assert sql('pragma integrity_check')[0]['integrity_check']=='ok'
 report={'passed':True,'fixture':str(data),'checks':['all bound references','image before video','one final frame sent','correct image and video GPUs','15 refs over 3 rounds','cache reuse','forced remake','missing-reference before cancellation','image failure isolation','stale-edit protection','restart no duplicate','pending submission cancellation','tail-frame plus all bound images','unknown video acknowledgement no resubmission','default model all refs','H3 unchanged','source artwork unchanged']}
 (data/'result.json').write_text(json.dumps(report,indent=2));print(json.dumps(report))
 if os.environ.get('COMPOSE_UI_TEST'):
  sql("update storyboards set title='Composition fixture shot' where id=23")
  sql("update sys_task set status='failed',error_code='generation_failed' where id=?",(default_task,))
  subprocess.run(['node',str(root/'integration-work/browser-test/verify_composed_frame_ui.cjs')],env={**os.environ,'COMPOSE_FIXTURE':str(data)},check=True)
finally:
 image_submit_release.set();video_submit_release.set()
 if process:stop()
 server.shutdown();log.close()