# retoor <retoor@molodetz.nl>
import io
import uuid
import pytest
import requests
from PIL import Image
from tests.conftest import BASE_URL
from devplacepy.database import get_table, refresh_snapshot, set_setting
JSON_manage = {"Accept": "application/json"}
@pytest.fixture(scope="module", autouse=True)
def _manage_settings(app_server):
for key, value in {
"rate_limit_per_minute": "1000000",
"registration_open": "1",
"maintenance_mode": "0",
"max_upload_size_mb": "10",
"allowed_file_types": "",
}.items():
set_setting(key, value)
yield
def _db_user(name):
refresh_snapshot()
return get_table("users").find_one(username=name)
def _user_manage(prefix="mng"):
s = requests.Session()
name = f"{prefix}_{uuid.uuid4().hex[:10]}"
s.post(
f"{BASE_URL}/auth/signup",
data={
"username": name,
"email": f"{name}@test.dev",
"password": "secret123",
"confirm_password": "secret123",
},
allow_redirects=True,
)
return s, name
def _png_bytes():
buf = io.BytesIO()
Image.new("RGB", (4, 4), (0, 128, 255)).save(buf, "PNG")
return buf.getvalue()
def _upload(s, name="a.png", content=None, content_type="image/png"):
r = s.post(
f"{BASE_URL}/uploads/upload",
files={"file": (name, content or _png_bytes(), content_type)},
)
assert r.status_code == 201, r.text
return r.json()["uid"]
def _admin_session(seeded_db):
key = _db_user("alice_test")["api_key"]
s = requests.Session()
s.headers.update({"X-API-KEY": key})
return s
def test_list_requires_login(app_server):
r = requests.get(f"{BASE_URL}/uploads", headers=JSON_manage, allow_redirects=False)
assert r.status_code == 401
def test_list_own_attachments_newest_first(app_server):
s, _ = _user_manage()
first = _upload(s, "first.png")
second = _upload(s, "second.txt", b"hello world", "text/plain")
r = s.get(f"{BASE_URL}/uploads", headers=JSON_manage)
assert r.status_code == 200, r.text
body = r.json()
uids = [a["uid"] for a in body["attachments"]]
assert first in uids and second in uids
assert uids.index(second) < uids.index(first)
assert body["total"] >= 2
assert body["pagination"]["page"] == 1
def test_list_is_isolated_per_user(app_server):
alice, _ = _user_manage()
uid = _upload(alice, "alice.png")
bob, _ = _user_manage()
r = bob.get(f"{BASE_URL}/uploads", headers=JSON_manage)
assert r.status_code == 200
assert uid not in [a["uid"] for a in r.json()["attachments"]]
def test_list_linked_filter(app_server):
s, _ = _user_manage()
orphan = _upload(s, "orphan.png")
attached = _upload(s, "used.png")
post = s.post(
f"{BASE_URL}/posts/create",
headers=JSON_manage,
data={
"title": f"manage {uuid.uuid4().hex[:6]}",
"content": "a post that carries an attachment",
"topic": "devlog",
"attachment_uids": attached,
},
)
assert post.status_code in (200, 201), post.text
linked = s.get(f"{BASE_URL}/uploads", headers=JSON_manage, params={"linked": "true"})
linked_uids = [a["uid"] for a in linked.json()["attachments"]]
assert attached in linked_uids
assert orphan not in linked_uids
unlinked = s.get(
f"{BASE_URL}/uploads", headers=JSON_manage, params={"linked": "false"}
)
unlinked_uids = [a["uid"] for a in unlinked.json()["attachments"]]
assert orphan in unlinked_uids
assert attached not in unlinked_uids
attached_item = next(
a for a in linked.json()["attachments"] if a["uid"] == attached
)
assert attached_item["linked"] is True
assert attached_item["target_type"] == "post"
assert attached_item["target_url"]
def test_get_one_attachment(app_server):
s, _ = _user_manage()
uid = _upload(s, "single.png")
r = s.get(f"{BASE_URL}/uploads/{uid}", headers=JSON_manage)
assert r.status_code == 200, r.text
body = r.json()
assert body["uid"] == uid
assert body["original_filename"] == "single.png"
assert body["is_image"] is True
def test_get_missing_attachment_404(app_server):
s, _ = _user_manage()
r = s.get(f"{BASE_URL}/uploads/{uuid.uuid4().hex}", headers=JSON_manage)
assert r.status_code == 404
def test_get_other_users_attachment_forbidden(app_server):
alice, _ = _user_manage()
uid = _upload(alice, "private.png")
bob, _ = _user_manage()
r = bob.get(f"{BASE_URL}/uploads/{uid}", headers=JSON_manage)
assert r.status_code == 403
def test_admin_can_get_any_attachment(seeded_db):
member, _ = _user_manage()
uid = _upload(member, "member.png")
admin = _admin_session(seeded_db)
r = admin.get(f"{BASE_URL}/uploads/{uid}", headers=JSON_manage)
assert r.status_code == 200
assert r.json()["uid"] == uid
def test_rename_attachment(app_server):
s, _ = _user_manage()
uid = _upload(s, "before.png")
r = s.patch(
f"{BASE_URL}/uploads/{uid}", headers=JSON_manage, data={"filename": "after"}
)
assert r.status_code == 200, r.text
assert r.json()["original_filename"] == "after.png"
check = s.get(f"{BASE_URL}/uploads/{uid}", headers=JSON_manage)
assert check.json()["original_filename"] == "after.png"
def test_rename_preserves_extension(app_server):
s, _ = _user_manage()
uid = _upload(s, "safe.png")
r = s.patch(
f"{BASE_URL}/uploads/{uid}", headers=JSON_manage, data={"filename": "evil.html"}
)
assert r.status_code == 200, r.text
assert r.json()["original_filename"] == "evil.png"
def test_rename_other_users_attachment_forbidden(app_server):
alice, _ = _user_manage()
uid = _upload(alice, "keep.png")
bob, _ = _user_manage()
r = bob.patch(
f"{BASE_URL}/uploads/{uid}", headers=JSON_manage, data={"filename": "hijack"}
)
assert r.status_code == 403
def test_rename_missing_attachment_404(app_server):
s, _ = _user_manage()
r = s.patch(
f"{BASE_URL}/uploads/{uuid.uuid4().hex}",
headers=JSON_manage,
data={"filename": "x"},
)
assert r.status_code == 404
def test_rename_recorded(seeded_db):
s, _ = _user_manage()
uid = _upload(s, "audited.png")
s.patch(
f"{BASE_URL}/uploads/{uid}", headers=JSON_manage, data={"filename": "renamed"}
)
admin = _admin_session(seeded_db)
log = admin.get(
f"{BASE_URL}/admin/audit-log",
headers=JSON_manage,
params={"event_key": "attachment.rename"},
)
assert log.status_code == 200
assert any(e.get("target_uid") == uid for e in log.json()["entries"])
Reference in New Issue View Git Blame Copy Permalink