"""End-to-end continuity tests against an isolated SQLite DB and a loopback fake provider."""
import base64,json,os,pathlib,re,sqlite3,subprocess,threading,time,urllib.request,urllib.error
from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer

root=pathlib.Path(__file__).resolve().parent.parent
data=root/'integration-work'/('continuity-test-'+time.strftime('%Y%m%d-%H%M%S'));data.mkdir()
stage=root/'integration-work/stage-4.0.11/app'
dbfile=data/'huobao.sqlite3';posts=[];chat_posts=[];process=None
(data/'static/videos').mkdir(parents=True)
subprocess.run([str(stage/'resources/bin/ffmpeg.exe'),'-hide_banner','-loglevel','error','-f','lavfi','-i','color=c=blue:s=320x192:d=2','-f','lavfi','-i','anullsrc=r=32000:cl=stereo','-shortest','-c:v','libx264','-c:a','aac',str(data/'static/videos/source.mp4')],check=True)
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)
# Replace only the local-provider capability check in this private test copy.
needle='u.hostname === "192.168.110.111" && ["18792", "18794"].includes(u.port)'
assert source.count(needle)==1
source=source.replace(needle,'u.hostname === "127.0.0.1" && u.port === "18931"')
(data/'backend.mjs').write_text(source,'utf-8')
class Mock(BaseHTTPRequestHandler):
 def log_message(self,*args):pass
 def do_POST(self):
  body=json.loads(self.rfile.read(int(self.headers['Content-Length'])))
  if self.path.endswith('/chat/completions'):
   if 'tools' not in body:
    translated='Maintain continuity. Do not read these visual instructions aloud. '+ ' '.join(re.findall(r'__HB_KEEP_\d+__',body['messages'][-1]['content']))
    raw=json.dumps({'choices':[{'message':{'content':json.dumps({'translation':translated})}}]}).encode()
    self.send_response(200);self.send_header('Content-Type','application/json');self.end_headers();self.wfile.write(raw);return
   chat_posts.append(body)
   if len(chat_posts)==1:
    tool=next(t['function']['name'] for t in body['tools'] if 'storyboard' in t['function']['name'].lower() and ('context' in t['function']['name'].lower()))
    message={'role':'assistant','content':None,'tool_calls':[{'id':'call_fixture_context','type':'function','function':{'name':tool,'arguments':'{}'}}]};reason='tool_calls'
   else:message={'role':'assistant','content':'检查完成'};reason='stop'
   raw=json.dumps({'id':'fixture-chat','object':'chat.completion','created':1,'model':'fixture-text','choices':[{'index':0,'message':message,'finish_reason':reason}],'usage':{'prompt_tokens':20,'completion_tokens':20,'total_tokens':40}}).encode()
   self.send_response(200);self.send_header('Content-Type','application/json');self.end_headers();self.wfile.write(raw);return
  posts.append(body)
  raw=json.dumps({'task_id':str(len(posts)),'status':'succeeded','video_url':'http://127.0.0.1:18931/video.mp4','duration':2}).encode()
  self.send_response(200);self.send_header('Content-Type','application/json');self.end_headers();self.wfile.write(raw)
 def do_GET(self):
  if self.path!='/video.mp4':self.send_error(404);return
  self.send_response(200);self.send_header('Content-Type','video/mp4');self.end_headers();self.wfile.write((data/'static/videos/source.mp4').read_bytes())
server=ThreadingHTTPServer(('127.0.0.1',18931),Mock);threading.Thread(target=server.serve_forever,daemon=True).start()
env={**os.environ,'ELECTRON_RUN_AS_NODE':'1','PORT':'18930','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'),'FFMPEG_BIN':str(stage/'resources/bin/ffmpeg.exe'),'FFPROBE_BIN':str(stage/'resources/bin/ffprobe.exe')}
log=(data/'backend.log').open('w',encoding='utf-8')
def api(method,path,body=None,expect=200):
 req=urllib.request.Request('http://127.0.0.1:18930/api/v1'+path,data=None if body is None else json.dumps(body).encode(),method=method,headers={'Content-Type':'application/json'})
 try:
  with urllib.request.urlopen(req,timeout=40) as response:code=response.status;payload=json.load(response)
 except urllib.error.HTTPError as e:code=e.code;payload=json.load(e)
 assert code==expect,(method,path,code,payload)
 return payload.get('data',payload)
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,'Backend exited; inspect '+str(data/'backend.log')
  try:api('GET','/health');return
  except (OSError,AssertionError):time.sleep(.1)
 raise AssertionError('Backend not ready')
def stop():
 process.terminate();process.wait(timeout=10)
def sql(statement,args=()):
 with sqlite3.connect(dbfile) as db:
  db.row_factory=sqlite3.Row
  return [dict(r) for r in db.execute(statement,args).fetchall()]
def wait_task(id,stage=None,status=None):
 for _ in range(300):
  row=sql('SELECT * FROM sys_task WHERE id=?',(id,))[0]
  if (stage is None or row['stage']==stage) and (status is None or row['status']==status):return row
  time.sleep(.1)
 raise AssertionError(row)
def settings(id,mode,source=None,expect=200,**extra):return api('PUT',f'/storyboards/{id}/continuity',{'continuity_mode':mode,'continuity_source_id':source,**extra},expect)
def generate(id,**extra):return api('POST','/tasks',{'type':'video','storyboard_id':id,'config_id':1,'model':'MiniMax-H3','prompt':'@图片1人物向右迈步，无新增台词。','duration':4,'gpu':'gpu3',**extra},201)['id']
try:
 start()
 ts='2026-11-12T00:00:00Z'
 with sqlite3.connect(dbfile) as db:
  for id in [1,2]:db.execute('INSERT INTO dramas(id,title,created_at,updated_at) VALUES(?,?,?,?)',(id,'isolated',ts,ts))
  for id,drama,num,deleted in [(1,1,1,None),(2,1,2,None),(3,2,1,None),(4,1,0,ts)]:db.execute('INSERT INTO episodes(id,drama_id,episode_number,title,content,created_at,updated_at,deleted_at) VALUES(?,?,?,?,?,?,?,?)',(id,drama,num,'test','前集结尾',ts,ts,deleted))
  db.execute('INSERT INTO ai_service_configs(id,service_type,provider,name,base_url,api_key,model,is_active,created_at,updated_at) VALUES(1,?,?,?,?,?,?,1,?,?)',('video','minimax','fixture','http://127.0.0.1:18931','test-only-key',json.dumps(['MiniMax-H3']),ts,ts))
  db.execute('INSERT INTO ai_service_configs(id,service_type,provider,name,base_url,api_key,model,is_active,created_at,updated_at) VALUES(2,?,?,?,?,?,?,1,?,?)',('text','openai','fixture-text','http://127.0.0.1:18931/v1','test-only-key',json.dumps(['fixture-text']),ts,ts))
  for id,ep,num,scene in [(1,1,1,1),(2,1,2,1),(3,2,1,1),(4,2,2,1),(5,2,3,2),(6,3,1,1),(7,4,1,1),(8,2,4,1),(9,2,5,1)]:
   db.execute('INSERT INTO storyboards(id,episode_id,storyboard_number,scene_id,description,duration,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?)',(id,ep,num,scene,'动作描述',4,ts,ts))
 # Dependency validation across episodes, scenes, and projects.
 for scene_id in [1,2]:sql("INSERT INTO scenes(id,drama_id,location,time,prompt,lighting,created_at,updated_at) VALUES(?,1,'Room','night','Fixed stone walls and north window','Moonlight',?,?)",(scene_id,ts,ts))
 assert settings(3,'context')['source']['id']==2
 api('POST','/agent/storyboard_breaker/chat',{'drama_id':1,'episode_id':2,'config_id':2,'message':'先读取分镜上下文，核对上一集结尾，不修改任何内容。'})
 assert len(chat_posts)==2
 contexts=[m for m in chat_posts[-1]['messages'] if m.get('role')=='tool']
 text=json.dumps(contexts,ensure_ascii=False)
 assert 'previous_episode' in text and '前集结尾' in text and 'ending_storyboards' in text
 assert '跨镜头与跨集衔接' in json.dumps(chat_posts[0],ensure_ascii=False)
 settings(3,'first_frame',4,expect=400);settings(3,'first_frame',6,expect=400);settings(3,'first_frame',3,expect=400)
 settings(3,'first_frame',7,expect=400);settings(5,'first_frame',4,expect=400)
 settings(1,'context',expect=400)
 settings(3,'first_frame',2,continuity_start='右手持剑，朝向右侧',continuity_end='向右跨步结束')
 api('POST','/tasks',{'type':'video','storyboard_id':3,'config_id':1,'model':'MiniMax-H3-VDN-H3','prompt':'test'},400)
 task=generate(3);wait_task(task,stage='waiting_predecessor');assert not posts
 api('POST','/tasks',{'type':'video','storyboard_id':3,'config_id':1,'prompt':'duplicate'},400)
 # Independent shots still run while this shot waits.
 independent=generate(5);wait_task(independent,status='completed');assert len(posts)==1
 # Pending task survives restart, without resubmission.
 stop();start();wait_task(task,stage='waiting_predecessor');assert len(posts)==1
 sql('UPDATE storyboards SET video_url=? WHERE id=2',('static/videos/source.mp4',))
 result=wait_task(task,status='completed');assert len(posts)==2
 sent=posts[-1];images=[x for x in sent['content'] if x['type']=='image_url']
 assert len(images)==1 and images[0]['role']=='first_frame' and images[0]['image_url']['url'].startswith('data:image/')
 assert sent['gpu']=='gpu3' and '@图片1' not in sent['content'][0]['text'] and 'Do not read' in sent['content'][0]['text']
 params=json.loads(result['params']);assert params['continuity']['sourceId']==2 and params['continuity']['frameUrl']
 assert api('GET','/storyboards/3/continuity')['stale'] is False
 # Changing previous video flags the existing downstream output without replacing it.
 sql('UPDATE storyboards SET video_url=? WHERE id=2',('static/videos/changed.mp4',))
 assert api('GET','/storyboards/3/continuity')['stale'] is True
 assert sql('SELECT video_url FROM storyboards WHERE id=3')[0]['video_url']==result['local_path']
 # Failure blocks dependents; cancellation is local and never submits media.
 settings(4,'first_frame',3)
 sql('INSERT INTO sys_task(type,storyboard_id,status,error_code,created_at,updated_at) VALUES(?,?,?,?,?,?)',('video',3,'failed','generation_failed',ts,ts))
 blocked=generate(4);wait_task(blocked,stage='blocked_predecessor');count=len(posts)
 api('POST',f'/tasks/{blocked}/cancel-waiting',{});wait_task(blocked,status='failed');assert len(posts)==count
 # Replacing a waiting task affects only that task and persists the new GPU.
 settings(8,'first_frame',3);old=generate(8);wait_task(old,stage='blocked_predecessor')
 new=generate(8,replace_task_id=old,force_new=True,gpu='gpu2');wait_task(new,stage='blocked_predecessor')
 assert sql('SELECT error_code FROM sys_task WHERE id=?',(old,))[0]['error_code']=='user_cancelled'
 assert json.loads(sql('SELECT params FROM sys_task WHERE id=?',(new,))[0]['params'])['gpu']=='gpu2'
 api('POST',f'/tasks/{new}/cancel-waiting',{})
 assert len(posts)==count
 # A deleted source can be detached without losing this shot or its output.
 sql('UPDATE storyboards SET deleted_at=? WHERE id=3',(ts,))
 detached=settings(8,'none',3)
 assert detached['mode']=='none' and detached['source'] is None
 print(json.dumps({'passed':True,'fixture':str(data),'provider_submissions':len(posts),'agent_received_previous_episode':True,'waiting_restart':True,'cross_episode':True,'dependency_validation':True,'real_tail_frame_extracted':True,'single_first_frame':True,'gpu_preserved':True,'stale_warning':True,'cancel_and_replace_waiting':True}))
 (data/'result.json').write_text(json.dumps({'passed':True,'provider_submissions':len(posts)}))
 (root/'integration-work/continuity-test-current.txt').write_text(str(data))
finally:
 if process and process.poll() is None:stop()
 server.shutdown();log.close()