devplacepy/tests/api/uploads/manage.py

# 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",
            "birth_date": "1990-01-01",
            "accept_terms": "1",
        },
        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"])