35 lines
2.4 KiB
Python
35 lines
2.4 KiB
Python
"""Receipt API tests against temporary SQLite through ASGI, without a network listener."""
|
|
import asyncio,importlib.util,pathlib,sqlite3,tempfile
|
|
from contextlib import closing
|
|
import httpx
|
|
from fastapi import FastAPI,Header,HTTPException
|
|
source=pathlib.Path(__file__).resolve().parent.parent/'model-bridge/video_cleanup_api.py'
|
|
spec=importlib.util.spec_from_file_location('video_cleanup_api',source);module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
|
|
async def main():
|
|
with tempfile.TemporaryDirectory(prefix='huobao-cleanup-api-') as directory:
|
|
database=pathlib.Path(directory)/'jobs.sqlite3'
|
|
with closing(sqlite3.connect(database)) as db,db:
|
|
db.execute('create table jobs(id text primary key)');db.execute("insert into jobs values('known-video')")
|
|
app=FastAPI()
|
|
async def auth(authorization:str|None=Header(default=None)):
|
|
if authorization!='Bearer fixture-token':raise HTTPException(401,'Unauthorized')
|
|
module.install_video_cleanup(app,auth,database)
|
|
receipt={'version':1,'index':0,'sha256':'a'*64,'bytes':1234}
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app),base_url='http://fixture') as client:
|
|
url='/v2/video_generation/known-video/cleanup'
|
|
assert (await client.post(url,json=receipt)).status_code==401
|
|
client.headers['Authorization']='Bearer fixture-token'
|
|
assert (await client.post('/v2/video_generation/unknown/cleanup',json=receipt)).status_code==404
|
|
for patch in [{'filename':'../../anything.mp4'},{'index':True},{'bytes':0},{'sha256':'invalid'},{'version':2}]:
|
|
assert (await client.post(url,json={**receipt,**patch})).status_code==400
|
|
assert (await client.post(url,json=receipt)).json()['status']=='pending'
|
|
assert (await client.post(url,json=receipt)).json()['status']=='pending'
|
|
assert (await client.post(url,json={**receipt,'sha256':'b'*64})).status_code==409
|
|
with closing(sqlite3.connect(database)) as db,db:
|
|
assert db.execute('select count(*) from video_cleanup_acks').fetchone()[0]==1
|
|
db.execute("update video_cleanup_acks set status='cleaned' where id='known-video'")
|
|
result=(await client.post(url,json=receipt)).json()
|
|
assert result['status']=='cleaned' and result['sha256']==receipt['sha256'] and result['bytes']==1234
|
|
print('PASS: authenticated known-video-only receipts; invalid paths/types denied; retries are idempotent; conflicting receipts denied')
|
|
asyncio.run(main())
|