83 lines
7.3 KiB
Python
83 lines
7.3 KiB
Python
"""Exercise deletion against a disposable desktop backend; never touch live data."""
|
|
import contextlib, hashlib, json, os, pathlib, re, sqlite3, subprocess, time, urllib.request, urllib.error
|
|
ROOT = pathlib.Path(__file__).resolve().parent.parent
|
|
OUT = ROOT/'integration-work'/('storyboard-delete-test-'+time.strftime('%Y%m%d-%H%M%S')); OUT.mkdir()
|
|
STAGE = ROOT/'integration-work/stage-4.0.11/app'; DB = OUT/'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)
|
|
(OUT/'backend.mjs').write_text(source, 'utf-8')
|
|
env = {**os.environ, 'ELECTRON_RUN_AS_NODE':'1','PORT':'18986','HUOBAO_DATA_DIR':str(OUT),'SQLITE_PATH':str(DB),
|
|
'FRONTEND_DIST':str(ROOT/'huobao-source/frontend/.output/public'), 'WORKSPACE_PATH':str(STAGE/'resources/workspace-template')}
|
|
def sql(query,args=()):
|
|
with contextlib.closing(sqlite3.connect(DB)) as c,c:
|
|
c.row_factory=sqlite3.Row
|
|
return [dict(r) for r in c.execute(query,args)]
|
|
def api(method,path,body=None,code=200):
|
|
req=urllib.request.Request('http://127.0.0.1:18986/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=10) as response: actual=response.status; value=json.load(response)
|
|
except urllib.error.HTTPError as error: actual=error.code; value=json.load(error)
|
|
assert actual==code,(method,path,actual,value)
|
|
return value.get('data',value)
|
|
def remove(ids,code=200): return api('POST','/storyboards/batch-delete',{'episode_id':1,'storyboard_ids':ids},code)
|
|
def live():return api('GET','/episodes/1/storyboards')
|
|
def row(id):return sql('select * from storyboards where id=?',(id,))[0]
|
|
log=(OUT/'backend.log').open('w',encoding='utf-8')
|
|
process=subprocess.Popen([str(STAGE/'HuobaoDrama.exe'),str(OUT/'backend.mjs')],cwd=OUT,env=env,stdout=log,stderr=log,creationflags=subprocess.CREATE_NO_WINDOW)
|
|
try:
|
|
for _ in range(100):
|
|
assert process.poll() is None,'Backend exited'
|
|
try:api('GET','/health');break
|
|
except OSError:time.sleep(.15)
|
|
ts='2026-11-13T00:00:00Z'
|
|
sql("insert into app_settings(key,value,updated_at) values('tours_seen',?,?) on conflict(key) do update set value=excluded.value",(json.dumps(['episode']),ts))
|
|
sql('insert into dramas(id,title,created_at,updated_at) values(1,?,?,?)',('Deletion fixture',ts,ts))
|
|
for id in (1,2):sql('insert into episodes(id,drama_id,episode_number,title,content,script_content,created_at,updated_at) values(?,1,?,?,?,?,?,?)',(id,id,'fixture','剧本','剧本',ts,ts))
|
|
sql('insert into characters(id,drama_id,name,created_at,updated_at) values(1,1,?,?,?)',('角色',ts,ts))
|
|
sql("insert into scenes(id,drama_id,location,time,prompt,created_at,updated_at) values(1,1,'房间','night','room',?,?)",(ts,ts))
|
|
sql("insert into props(id,drama_id,name,created_at,updated_at) values(1,1,'道具',?,?)",(ts,ts))
|
|
for table,key in [('episode_characters','character_id'),('episode_scenes','scene_id'),('episode_props','prop_id')]:
|
|
sql(f'insert into {table}(episode_id,{key},created_at) values(1,1,?)',(ts,))
|
|
for id in list(range(1,11))+[20]:
|
|
sql('insert into storyboards(id,episode_id,storyboard_number,title,description,duration,scene_id,created_at,updated_at) values(?,?,?,?,?,9,1,?,?)',(id,2 if id==20 else 1,id,'镜头 '+str(id),'分镜画面 '+str(id),ts,ts))
|
|
sql('insert into storyboard_characters values(1,1)'); sql('insert into storyboard_props values(1,1)')
|
|
media=OUT/'static/videos/preserved.mp4';media.parent.mkdir(parents=True);media.write_bytes(b'fixture media bytes kept intact')
|
|
original_hash=hashlib.sha256(media.read_bytes()).hexdigest()
|
|
sql("update storyboards set video_url='static/videos/preserved.mp4',status='completed' where id=1")
|
|
def task(id,sb,status,params=None):
|
|
sql("insert into sys_task(id,type,storyboard_id,drama_id,provider,model,prompt,status,params,local_path,created_at,updated_at) values(?,'video',?,1,'minimax','MiniMax-H3','fixture',?,?,?, ?,?)",(id,sb,status,json.dumps(params or {}),'static/videos/preserved.mp4' if id==1 else None,ts,ts))
|
|
task(1,1,'completed');task(2,2,'processing');task(90,3,'processing',{'continuity':{'sourceId':4}})
|
|
baseline={id:row(id) for id in (2,3,4,20)}
|
|
remove([2,20],400);remove([2,3],400);remove([4],400);remove([],400);remove(['2'],400)
|
|
assert all(row(id)==baseline[id] for id in baseline)
|
|
sql("update sys_task set status='completed' where id=90")
|
|
assert api('DELETE','/storyboards/1')['count']==1
|
|
assert row(1)['deleted_at'] and row(1)['video_url']=='static/videos/preserved.mp4'
|
|
assert sql('select * from storyboard_characters where storyboard_id=1') and sql('select * from storyboard_props where storyboard_id=1')
|
|
assert all(r['id']!=1 for r in live())
|
|
assert all(r['id']!=1 for r in api('GET','/hd/episodes/1')['items'])
|
|
assert api('GET','/episodes/1/pipeline-status')['steps']['extract_storyboards']['count']==9
|
|
merged=api('POST','/merge/episodes/1/merge',{'storyboard_ids':[1]},400)
|
|
assert '没有可拼接的视频' in merged['message']
|
|
history=api('GET','/episodes/1/generation-tasks')['tasks'];old=next(t for t in history if t['id']==1)
|
|
assert old['storyboard_deleted'] and old['storyboard_number']==1 and old['local_path']=='static/videos/preserved.mp4'
|
|
api('PUT','/storyboards/1',{'description':'should not save'},400)
|
|
api('POST','/tasks',{'type':'image','storyboard_id':1,'prompt':'no model call'},400)
|
|
api('POST','/tasks',{'type':'video','storyboard_id':1,'prompt':'no model call'},400)
|
|
assert remove([3,4])['count']==2 and remove([3,4])['count']==0
|
|
assert row(2)==baseline[2] and row(20)==baseline[20]
|
|
ui=subprocess.run(['node',str(ROOT/'integration-work/browser-test/verify_storyboard_delete_ui.cjs')],cwd=ROOT/'integration-work/browser-test',env={**os.environ,'STORYBOARD_FIXTURE':str(OUT)},capture_output=True,text=True,encoding='utf-8',timeout=75)
|
|
print(ui.stdout,flush=True);assert ui.returncode==0,ui.stderr
|
|
assert row(5)['deleted_at'] and row(6)['deleted_at'] and row(7)['deleted_at'] and not row(2)['deleted_at']
|
|
sql("update sys_task set status='cancelled' where id=2")
|
|
remove([r['id'] for r in live()]);assert live()==[]
|
|
assert api('GET','/episodes/1/pipeline-status')['steps']['extract_storyboards']['count']==0
|
|
assert hashlib.sha256(media.read_bytes()).hexdigest()==original_hash
|
|
assert len(sql('select * from characters'))==1 and len(sql('select * from scenes'))==1 and len(sql('select * from props'))==1
|
|
assert len(sql('select * from sys_task'))==3
|
|
report={'passed':True,'checks':['single/batch soft delete','batch all-or-none','active and dependent task protection','cross-episode validation','deleted shot update and generation rejected','pipeline, HD and merge filtering','task history and references retained','UI confirmation/cancel/batch/reload','delete all','media bytes unchanged']}
|
|
(OUT/'result.json').write_text(json.dumps(report,indent=2),'utf-8');print(json.dumps(report),flush=True)
|
|
finally:
|
|
process.terminate();process.wait(10);log.close()
|