35 lines
1.9 KiB
Python
35 lines
1.9 KiB
Python
"""Run inside the API container. Creates only synthetic retention fixtures."""
|
|
from datetime import datetime, timedelta, timezone
|
|
from uuid import uuid4
|
|
from botocore.exceptions import ClientError
|
|
from .adapters import LocalS3Storage
|
|
from .db import connect
|
|
from .worker import cleanup
|
|
|
|
def run():
|
|
storage=LocalS3Storage()
|
|
ids=[]
|
|
for expired in (True,False):
|
|
uid=uuid4();ids.append(uid);key=f'originals/{uid}'
|
|
upload=storage.begin(key)
|
|
result=storage.client.upload_part(Bucket=storage.bucket,Key=key,UploadId=upload,PartNumber=1,Body=b'LOCAL RETENTION TEST')
|
|
storage.complete(key,upload,[{'PartNumber':1,'ETag':result['ETag']}])
|
|
expiry=datetime.now(timezone.utc)+timedelta(days=-1 if expired else 1)
|
|
with connect() as c:
|
|
c.execute('INSERT INTO dtf_local.uploads(id,owner,name,size,object_key,multipart_id,complete,expires_at) VALUES(%s,%s,%s,%s,%s,%s,true,%s)',
|
|
(uid,uuid4(),'LOCAL-RETENTION-TEST.txt',len(b'LOCAL RETENTION TEST'),key,upload,expiry))
|
|
cleanup()
|
|
with connect() as c:
|
|
assert c.execute('SELECT purged_at FROM dtf_local.uploads WHERE id=%s',(ids[0],)).fetchone()['purged_at']
|
|
assert not c.execute('SELECT purged_at FROM dtf_local.uploads WHERE id=%s',(ids[1],)).fetchone()['purged_at']
|
|
try:storage.size(f'originals/{ids[0]}');raise AssertionError('Expired bytes still exist')
|
|
except ClientError as exc:assert exc.response['ResponseMetadata']['HTTPStatusCode']==404
|
|
assert storage.size(f'originals/{ids[1]}')==len(b'LOCAL RETENTION TEST')
|
|
# Clean only the other fixture just created above, leaving metadata intact.
|
|
with connect() as c:
|
|
c.execute("UPDATE dtf_local.uploads SET expires_at=now()-interval '1 second' WHERE id=%s",(ids[1],))
|
|
cleanup()
|
|
print('PASS: expired bytes removed, unexpired bytes preserved, upload metadata retained. Synthetic fixture bytes cleaned up.')
|
|
|
|
if __name__=='__main__':run()
|