肉末
274 lines
8.9 KiB
Python
274 lines
8.9 KiB
Python
from datetime import datetime, timedelta
|
|
|
|
|
|
def test_create_text_default_public_and_24h(client):
|
|
r = client.post("/api/items", json={"body": "hello **md**"})
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["slug"]
|
|
assert data["is_public"] is True
|
|
assert data["expires_at"] is not None
|
|
assert data["kind"] == "text"
|
|
assert data["title"] == "hello **md**"
|
|
assert data["body"] == "hello **md**"
|
|
assert data["burn_after_read"] is False
|
|
assert data["view_count"] == 0
|
|
assert data["created_at"]
|
|
|
|
|
|
def test_link_only_not_on_wall(client):
|
|
r = client.post("/api/items", json={"body": "secret", "is_public": False})
|
|
slug = r.json()["slug"]
|
|
wall = client.get("/api/items/wall").json()["items"]
|
|
assert all(i["slug"] != slug for i in wall)
|
|
assert client.get(f"/api/items/{slug}").status_code == 200
|
|
|
|
|
|
def test_ttl_options(client):
|
|
never = client.post("/api/items", json={"body": "n", "ttl": "never"}).json()
|
|
assert never["expires_at"] is None
|
|
|
|
one_h = client.post("/api/items", json={"body": "h", "ttl": "1h"}).json()
|
|
expires = datetime.fromisoformat(one_h["expires_at"])
|
|
delta = expires - datetime.fromisoformat(one_h["created_at"])
|
|
assert timedelta(minutes=50) < delta < timedelta(hours=2)
|
|
|
|
week = client.post("/api/items", json={"body": "w", "ttl": "7d"}).json()
|
|
expires_w = datetime.fromisoformat(week["expires_at"])
|
|
delta_w = expires_w - datetime.fromisoformat(week["created_at"])
|
|
assert timedelta(days=6) < delta_w < timedelta(days=8)
|
|
|
|
|
|
def test_wall_pagination_shape(client):
|
|
for i in range(3):
|
|
client.post("/api/items", json={"body": f"item {i}", "is_public": True})
|
|
|
|
r = client.get("/api/items/wall", params={"page": 1, "page_size": 2})
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert "items" in data
|
|
assert data["page"] == 1
|
|
assert data["page_size"] == 2
|
|
assert data["total"] >= 3
|
|
assert len(data["items"]) == 2
|
|
# newest first
|
|
titles = [i["title"] for i in data["items"]]
|
|
assert titles[0] == "item 2"
|
|
|
|
|
|
def test_get_missing_404(client):
|
|
assert client.get("/api/items/does-not-exist").status_code == 404
|
|
|
|
|
|
def test_title_from_first_line(client):
|
|
r = client.post("/api/items", json={"body": "First line\nSecond line"})
|
|
assert r.json()["title"] == "First line"
|
|
|
|
|
|
def test_explicit_title(client):
|
|
r = client.post("/api/items", json={"body": "body text", "title": "Custom"})
|
|
assert r.json()["title"] == "Custom"
|
|
|
|
|
|
def test_upload_small_file(client):
|
|
files = {"file": ("hi.txt", b"abc", "text/plain")}
|
|
data = {"is_public": "true", "ttl": "24h"}
|
|
r = client.post("/api/items/upload", files=files, data=data)
|
|
assert r.status_code == 200
|
|
slug = r.json()["slug"]
|
|
f = client.get(f"/api/items/{slug}/file")
|
|
assert f.status_code == 200
|
|
assert f.content == b"abc"
|
|
|
|
|
|
def test_reject_oversize(client, monkeypatch):
|
|
import app.config as config_module
|
|
|
|
monkeypatch.setattr(config_module.settings, "max_upload_mb", 1)
|
|
big = b"x" * (1 * 1024 * 1024 + 1)
|
|
files = {"file": ("big.txt", big, "text/plain")}
|
|
r = client.post("/api/items/upload", files=files, data={"is_public": "true", "ttl": "24h"})
|
|
assert r.status_code == 413
|
|
|
|
|
|
def test_reject_disallowed_extension(client):
|
|
files = {"file": ("malware.exe", b"mz", "application/octet-stream")}
|
|
r = client.post(
|
|
"/api/items/upload",
|
|
files=files,
|
|
data={"is_public": "true", "ttl": "24h"},
|
|
)
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_file_burn_after_download_not_metadata(client):
|
|
files = {"file": ("secret.txt", b"top-secret", "text/plain")}
|
|
data = {"is_public": "true", "ttl": "24h", "burn_after_read": "true"}
|
|
r = client.post("/api/items/upload", files=files, data=data)
|
|
assert r.status_code == 200
|
|
slug = r.json()["slug"]
|
|
|
|
meta = client.get(f"/api/items/{slug}")
|
|
assert meta.status_code == 200
|
|
assert meta.json()["file_name"] == "secret.txt"
|
|
assert meta.json()["burn_after_read"] is True
|
|
|
|
first = client.get(f"/api/items/{slug}/file")
|
|
assert first.status_code == 200
|
|
assert first.content == b"top-secret"
|
|
|
|
assert client.get(f"/api/items/{slug}/file").status_code == 404
|
|
assert client.get(f"/api/items/{slug}").status_code == 404
|
|
|
|
|
|
def test_svg_forced_attachment_disposition(client):
|
|
files = {"file": ("xss.svg", b"<svg xmlns='http://www.w3.org/2000/svg'></svg>", "image/svg+xml")}
|
|
r = client.post(
|
|
"/api/items/upload",
|
|
files=files,
|
|
data={"is_public": "true", "ttl": "24h"},
|
|
)
|
|
assert r.status_code == 200
|
|
slug = r.json()["slug"]
|
|
f = client.get(f"/api/items/{slug}/file")
|
|
assert f.status_code == 200
|
|
cd = f.headers.get("content-disposition", "")
|
|
assert "attachment" in cd
|
|
assert "inline" not in cd.split(";")[0]
|
|
|
|
|
|
def test_burn_claims_before_stream(client, monkeypatch):
|
|
"""burned=True is set before streaming, not only in the post-response BackgroundTask."""
|
|
import app.db as db_module
|
|
import app.api.public as public_api
|
|
from app.models import Item
|
|
from sqlalchemy import select
|
|
|
|
# Background only deletes bytes; claiming is done in the request handler.
|
|
monkeypatch.setattr(public_api, "_delete_file_path", lambda *_a, **_k: None)
|
|
|
|
files = {"file": ("once.txt", b"payload", "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"]
|
|
|
|
first = client.get(f"/api/items/{slug}/file")
|
|
assert first.status_code == 200
|
|
assert first.content == b"payload"
|
|
|
|
db = db_module.SessionLocal()
|
|
try:
|
|
item = db.scalar(select(Item).where(Item.slug == slug))
|
|
assert item is not None
|
|
assert item.burned is True
|
|
finally:
|
|
db.close()
|
|
|
|
assert client.get(f"/api/items/{slug}/file").status_code == 404
|
|
|
|
|
|
def test_rate_limit_third_request_in_same_second(client, monkeypatch):
|
|
import app.config as config_module
|
|
from app.services import rate_limit as rate_limit_module
|
|
|
|
monkeypatch.setattr(config_module.settings, "rate_limit_per_second", 2)
|
|
rate_limit_module.reset()
|
|
|
|
r1 = client.post("/api/items", json={"body": "one"})
|
|
r2 = client.post("/api/items", json={"body": "two"})
|
|
r3 = client.post("/api/items", json={"body": "three"})
|
|
assert r1.status_code == 200
|
|
assert r2.status_code == 200
|
|
assert r3.status_code == 429
|
|
|
|
|
|
def test_banned_ip_cannot_create(client):
|
|
import app.db as db_module
|
|
from app.models import BannedIP
|
|
|
|
db = db_module.SessionLocal()
|
|
try:
|
|
db.add(BannedIP(ip="testclient", reason="test"))
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
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()
|