# retoor 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"])