feat: burn-after-read and expiry cleaner
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -199,3 +199,75 @@ def test_banned_ip_cannot_create(client):
|
||||
|
||||
r = client.post("/api/items", json={"body": "x"})
|
||||
assert r.status_code == 403
|
||||
|
||||
|
||||
def test_burn_after_read(client):
|
||||
r = client.post(
|
||||
"/api/items",
|
||||
json={"body": "once", "burn_after_read": True, "ttl": "never"},
|
||||
)
|
||||
assert r.status_code == 200
|
||||
slug = r.json()["slug"]
|
||||
assert client.get(f"/api/items/{slug}").status_code == 200
|
||||
assert client.get(f"/api/items/{slug}").status_code == 404
|
||||
|
||||
|
||||
def test_cleaner_removes_expired(client):
|
||||
import app.db as db_module
|
||||
from app.models import Item
|
||||
from app.services import cleaner
|
||||
from sqlalchemy import select
|
||||
|
||||
r = client.post("/api/items", json={"body": "old", "ttl": "never"})
|
||||
assert r.status_code == 200
|
||||
slug = r.json()["slug"]
|
||||
|
||||
db = db_module.SessionLocal()
|
||||
try:
|
||||
item = db.scalar(select(Item).where(Item.slug == slug))
|
||||
assert item is not None
|
||||
item.expires_at = datetime.utcnow() - timedelta(hours=1)
|
||||
db.commit()
|
||||
|
||||
n = cleaner.run_once(db)
|
||||
assert n >= 1
|
||||
assert db.scalar(select(Item).where(Item.slug == slug)) is None
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
assert client.get(f"/api/items/{slug}").status_code == 404
|
||||
|
||||
|
||||
def test_cleaner_removes_burned_leftover_files(client, monkeypatch):
|
||||
import app.db as db_module
|
||||
import app.api.public as public_api
|
||||
from app.models import Item
|
||||
from app.services import cleaner
|
||||
from app.services import items as items_service
|
||||
from sqlalchemy import select
|
||||
|
||||
monkeypatch.setattr(public_api, "_delete_file_path", lambda *_a, **_k: None)
|
||||
|
||||
files = {"file": ("gone.txt", b"bytes", "text/plain")}
|
||||
r = client.post(
|
||||
"/api/items/upload",
|
||||
files=files,
|
||||
data={"is_public": "true", "ttl": "never", "burn_after_read": "true"},
|
||||
)
|
||||
assert r.status_code == 200
|
||||
slug = r.json()["slug"]
|
||||
assert client.get(f"/api/items/{slug}/file").status_code == 200
|
||||
|
||||
db = db_module.SessionLocal()
|
||||
try:
|
||||
item = db.scalar(select(Item).where(Item.slug == slug))
|
||||
assert item is not None
|
||||
path = items_service.resolve_file_path(item)
|
||||
assert path is not None and path.is_file()
|
||||
assert item.burned is True
|
||||
|
||||
cleaner.run_once(db)
|
||||
assert not path.exists()
|
||||
assert db.scalar(select(Item).where(Item.slug == slug)) is None
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
Reference in New Issue
Block a user