Add attachment management CRUD to the /uploads API

Complete the read and update faces of the signed-in user's attachment management over the existing attachments table: - GET /uploads: paginated list of the user's own attachments, newest first, with an optional linked/orphaned filter - GET /uploads/{uid}: fetch one attachment (owner or admin) - PATCH /uploads/{uid}: rename the display filename, always preserving the original extension (owner or admin, audited as attachment.rename) Adds get_user_attachments/get_user_attachment data helpers, the rename_attachment operation, AttachmentRenameForm, the UploadItemOut and UploadsListOut schemas, the Devii tools list_attachments/get_attachment/ rename_attachment, expanded API reference documentation for the full lifecycle including delete, and api-tier tests.
2026-07-26 16:46:41 +02:00
parent ac04cf6817
commit 5774d83ece
12 changed files with 592 additions and 13 deletions
@@ -0,0 +1,227 @@
+# 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"])