124 lines
8.1 KiB
Python
124 lines
8.1 KiB
Python
"""Linux isolated fixtures for scoped video deletion and crash-safe receipts."""
|
|
import asyncio,contextlib,hashlib,importlib.util,io,json,os,pathlib,sqlite3,sys,tempfile,types
|
|
source=pathlib.Path(sys.argv[1])
|
|
spec=importlib.util.spec_from_file_location('cleaner',source);cleaner=importlib.util.module_from_spec(spec);spec.loader.exec_module(cleaner)
|
|
PID='11111111-1111-4111-8111-111111111111';OTHER='22222222-2222-4222-8222-222222222222'
|
|
video=b'fixture video data';digest=hashlib.sha256(video).hexdigest()
|
|
entry={'status':{'completed':True,'status_str':'success'},'outputs':{'1':{'gifs':[{'filename':'own.mp4','subfolder':'shots','type':'output'}]}}}
|
|
history={PID:entry,OTHER:{'outputs':{'2':{'gifs':[{'filename':'other.mp4','subfolder':'shots','type':'output'}]}}}}
|
|
queue={'queue_running':[],'queue_pending':[]};state='succeeded';output_kind='video';calls=[]
|
|
def execute(code,request):
|
|
stdin,stdout=sys.stdin,sys.stdout;capture=io.StringIO()
|
|
try:
|
|
sys.stdin=io.StringIO(json.dumps(request));sys.stdout=capture
|
|
try:exec(code,{'__name__':'fixture'})
|
|
except SystemExit:pass
|
|
finally:sys.stdin,sys.stdout=stdin,stdout
|
|
values=[line.split(':',1)[1] for line in capture.getvalue().splitlines() if line.startswith('HB_CLEANUP_RESULT:')]
|
|
assert values,capture.getvalue()
|
|
result=json.loads(values[-1])
|
|
if not result['ok']:raise RuntimeError(result.get('error'))
|
|
return result['result']
|
|
app=types.ModuleType('app')
|
|
provider=types.SimpleNamespace(id='default',base_url='http://worker')
|
|
app.decode_job=lambda _: (provider,PID)
|
|
app.provider_headers=lambda _: {'fixture':'only'}
|
|
app.providers=lambda: {'default':provider}
|
|
async def job_state(_):return provider,PID,history.get(PID,{}),[{'filename':'own.mp4','kind':output_kind}]
|
|
app.job_state=job_state
|
|
class Reply:
|
|
def __init__(self,value):self.value=value
|
|
def json(self):return self.value
|
|
def raise_for_status(self):pass
|
|
async def aiter_bytes(self):yield video
|
|
async def __aenter__(self):return self
|
|
async def __aexit__(self,*args):pass
|
|
class Client:
|
|
def __init__(self,**kwargs):pass
|
|
async def __aenter__(self):return self
|
|
async def __aexit__(self,*args):pass
|
|
async def get(self,url,**kwargs):
|
|
if '/v1/jobs/' in url:return Reply({'status':state})
|
|
if url.endswith('/queue'):return Reply(queue)
|
|
if url.endswith('/history'):return Reply(history)
|
|
return Reply({PID:history[PID]} if PID in history else {})
|
|
async def post(self,url,**kwargs):
|
|
assert url=='http://worker/history';body=kwargs['json'];assert set(body)=={'delete'}
|
|
assert body['delete'] in ([],[PID]);calls.append(body)
|
|
for key in body['delete']:history.pop(key,None)
|
|
return Reply({})
|
|
def stream(self,*args,**kwargs):return Reply(None)
|
|
httpx=types.ModuleType('httpx');httpx.AsyncClient=Client
|
|
sys.modules['app']=app;sys.modules['httpx']=httpx
|
|
with tempfile.TemporaryDirectory(prefix='huobao-cleanup-fixture-',dir=source.parent) as directory:
|
|
root=pathlib.Path(directory);out=root/'output';(out/'shots').mkdir(parents=True)
|
|
# All recursive test teardown remains within the newly created private fixture.
|
|
assert root.resolve().parent==source.parent.resolve()
|
|
code=cleaner.WORKER_CODE.replace("'/app/output'",repr(str(out)))
|
|
locator={'filename':'own.mp4','subfolder':'shots','type':'output'}
|
|
def worker(mode,loc=locator,**kw):return execute(code,{'mode':mode,'locator':loc,**kw})
|
|
def fail(fn):
|
|
try:fn()
|
|
except RuntimeError:return
|
|
raise AssertionError('Unsafe operation accepted')
|
|
own=out/'shots/own.mp4';other=out/'shots/other.mp4';own.write_bytes(video);other.write_bytes(video)
|
|
receipt=worker('inspect');assert receipt['sha256']==digest
|
|
worker('delete',expected=receipt);assert not own.exists() and other.exists()
|
|
assert worker('delete',expected=receipt)['missing']
|
|
own.write_bytes(video);receipt=worker('inspect');own.write_bytes(b'replaced')
|
|
fail(lambda:worker('delete',expected=receipt));assert own.exists()
|
|
for loc in [dict(locator,filename='../own.mp4'),dict(locator,subfolder='../output/shots'),dict(locator,type='input'),dict(locator,filename='image.png'),dict(locator,filename='own\x00.mp4')]:
|
|
fail(lambda:worker('delete',loc,expected=receipt))
|
|
outside=root/'protected.mp4';outside.write_bytes(video)
|
|
(out/'symlink.mp4').symlink_to(outside)
|
|
fail(lambda:worker('inspect',dict(locator,subfolder='',filename='symlink.mp4')))
|
|
(out/'linked').symlink_to(root,target_is_directory=True)
|
|
fail(lambda:worker('inspect',dict(locator,subfolder='linked',filename='protected.mp4')))
|
|
os.link(outside,out/'hard.mp4')
|
|
fail(lambda:worker('inspect',dict(locator,subfolder='',filename='hard.mp4')))
|
|
assert outside.read_bytes()==video
|
|
verify={'mode':'verify','id':'fixture','index':0,'sha256':digest,'bytes':len(video)}
|
|
manifest=execute(cleaner.NORMALIZER_CODE,verify)
|
|
fail(lambda:execute(cleaner.NORMALIZER_CODE,{**verify,'sha256':'a'*64}))
|
|
state='running';fail(lambda:execute(cleaner.NORMALIZER_CODE,verify));state='succeeded'
|
|
output_kind='image';fail(lambda:execute(cleaner.NORMALIZER_CODE,verify));output_kind='video'
|
|
guard={'mode':'guard','id':'fixture','manifest':manifest}
|
|
queue['queue_running']=[[1,PID]];fail(lambda:execute(cleaner.NORMALIZER_CODE,guard));queue['queue_running']=[]
|
|
queue['queue_pending']=[[1,OTHER,{'video':'shots/own.mp4'}]];fail(lambda:execute(cleaner.NORMALIZER_CODE,guard));queue['queue_pending']=[]
|
|
other_entry=history[OTHER];history[OTHER]=entry;fail(lambda:execute(cleaner.NORMALIZER_CODE,guard));history[OTHER]=other_entry
|
|
entry['changed']=True;fail(lambda:execute(cleaner.NORMALIZER_CODE,guard));del entry['changed']
|
|
execute(cleaner.NORMALIZER_CODE,{**guard,'mode':'remove_history'});assert PID not in history and OTHER in history
|
|
history[PID]=entry;own.write_bytes(video)
|
|
database=root/'jobs.sqlite3'
|
|
with contextlib.closing(sqlite3.connect(database)) as db,db:
|
|
db.row_factory=sqlite3.Row
|
|
db.execute('create table jobs(id text primary key)');db.execute("insert into jobs values('fixture')")
|
|
db.execute('create table video_cleanup_acks(id text primary key,output_index integer,sha256 text,bytes integer,status text,manifest text,attempts integer,next_attempt real,error text,created real,updated real)')
|
|
db.execute("insert into video_cleanup_acks values('fixture',0,?,?,'pending',null,0,0,null,0,0)",(digest,len(video)))
|
|
fault=[True];verifications=[]
|
|
def run(container,script,payload):
|
|
if script==cleaner.WORKER_CODE:return execute(code,payload)
|
|
if payload['mode']=='verify':verifications.append(True)
|
|
if payload['mode']=='remove_history' and fault[0]:fault[0]=False;raise RuntimeError('simulated crash after unlink')
|
|
return execute(script,payload)
|
|
config={'name':'fixture','normalizer':'fixture','workers':{'default':'worker'}}
|
|
row=lambda:db.execute("select * from video_cleanup_acks where id='fixture'").fetchone()
|
|
fail(lambda:cleaner.process_ack(db,config,row(),run))
|
|
assert row()['status']=='prepared' and not own.exists() and PID in history and other.exists()
|
|
cleaner.process_ack(db,config,row(),run)
|
|
assert row()['status']=='cleaned' and PID not in history and OTHER in history and len(verifications)==1
|
|
assert other.read_bytes()==video and outside.read_bytes()==video
|
|
# A restored, exact old snapshot is removed; active or altered records survive.
|
|
snapshot_hash=hashlib.sha256(json.dumps(entry,sort_keys=True,separators=(',',':'),ensure_ascii=False).encode()).hexdigest()
|
|
reconcile={'providers':{'default':[{'prompt':PID,'hashes':[snapshot_hash]}]}}
|
|
history[PID]=entry
|
|
queue['queue_running']=[[1,PID]]
|
|
assert execute(cleaner.CLEANED_HISTORY_CODE,reconcile)['deferred']==[PID] and PID in history
|
|
queue['queue_running']=[];entry['changed']=True
|
|
assert execute(cleaner.CLEANED_HISTORY_CODE,reconcile)['deferred']==[PID] and PID in history
|
|
del entry['changed']
|
|
assert execute(cleaner.CLEANED_HISTORY_CODE,reconcile)['removed']==[PID]
|
|
assert PID not in history and OTHER in history and other.read_bytes()==video
|
|
assert execute(cleaner.CLEANED_HISTORY_CODE,reconcile)['removed']==[]
|
|
print(json.dumps({'passed':True,'checks':['exact video only','idempotent missing file retry','changed inode/hash protection','path traversal denied','symlink/hardlink denied','download checksum proof','completed video only','running/queued/shared file protection','single history delete','crash recovery after unlink','other files and history unchanged']}))
|