import pytest from fastapi import HTTPException from app.images import MAX_IMAGES_PER_RECORD, inspect_image, sniff from conftest import jpeg_bytes, png_bytes, record_payload def upload(client, headers, data=None, name="shot.png", content_type="image/png", thumbnail=True): files = {"file": (name, data if data is not None else png_bytes(), content_type)} if thumbnail: files["thumbnail"] = ("thumb.png", png_bytes(8, 6), "image/png") return client.post("/api/uploads/images", headers=headers, files=files) def test_headers_identify_supported_formats(): assert sniff(png_bytes(40, 25)) == ("image/png", 40, 25) assert sniff(jpeg_bytes(640, 480)) == ("image/jpeg", 640, 480) @pytest.mark.parametrize("blob", [ b'', b"

not an image

", b"GIF89a\x01\x00\x01\x00\x00\x00\x00", # GIF 不在白名单内 png_bytes()[:12], # 截断文件 b"\xff\xd8\xff" + b"\x00" * 40, # 伪造 JPEG 头但没有 SOF ]) def test_non_image_payloads_are_rejected(blob): with pytest.raises(HTTPException) as error: inspect_image(blob) assert error.value.status_code in (400, 413, 415) def test_oversized_payload_is_rejected(): with pytest.raises(HTTPException) as error: inspect_image(png_bytes(), limit=10) assert error.value.status_code == 413 def test_absurd_dimensions_are_rejected(): with pytest.raises(HTTPException) as error: inspect_image(jpeg_bytes(30000, 30000)) assert error.value.status_code == 413 def test_upload_requires_authentication(client): client.cookies.clear() assert client.post("/api/uploads/images", files={"file": ("a.png", png_bytes(), "image/png")}).status_code == 401 def test_upload_requires_csrf_token(admin): client, _ = admin assert client.post("/api/uploads/images", files={"file": ("a.png", png_bytes(), "image/png")}).status_code == 403 def test_upload_returns_metadata_and_token(admin): client, headers = admin body = upload(client, headers, png_bytes(64, 48)).json() assert body["ok"] and body["data"]["width"] == 64 and body["data"]["height"] == 48 assert body["data"]["content_type"] == "image/png" and body["data"]["token"] def test_disguised_html_upload_is_rejected(admin): client, headers = admin response = upload(client, headers, b"", name="evil.png") assert response.status_code == 415 def test_pending_image_is_private_until_attached(admin): client, headers = admin image_id = upload(client, headers).json()["data"]["id"] assert client.get(f"/api/images/{image_id}").status_code == 200 client.cookies.clear() assert client.get(f"/api/images/{image_id}").status_code == 404 def test_attached_image_is_public_and_cacheable(admin): client, headers = admin token = upload(client, headers, png_bytes(50, 40)).json()["data"]["token"] created = client.post("/api/records", headers=headers, json=record_payload("LAS-PUB-001", image_tokens=[token])) assert created.status_code == 200, created.text record_id = created.json()["data"]["id"] images = client.get(f"/api/public/records/{record_id}").json()["data"]["images"] assert len(images) == 1 client.cookies.clear() response = client.get(images[0]["url"]) assert response.status_code == 200 assert response.headers["content-type"] == "image/png" assert response.headers["x-content-type-options"] == "nosniff" assert "immutable" in response.headers["cache-control"] assert client.get(response.url, headers={"If-None-Match": response.headers["etag"]}).status_code == 304 def test_record_detail_never_leaks_binary_payload(admin): client, headers = admin token = upload(client, headers).json()["data"]["token"] record_id = client.post("/api/records", headers=headers, json=record_payload("LAS-BLOB-001", image_tokens=[token])).json()["data"]["id"] data = client.get(f"/api/records/{record_id}", headers=headers).json()["data"] assert "data" not in data and "thumbnail" not in data assert data["images"][0]["url"].startswith("/api/images/") def test_more_than_six_images_are_refused(admin): client, headers = admin tokens = [upload(client, headers).json()["data"]["token"] for _ in range(MAX_IMAGES_PER_RECORD + 1)] response = client.post("/api/records", headers=headers, json=record_payload("LAS-MANY-001", image_tokens=tokens)) assert response.status_code == 422 assert "6" in response.json()["message"] def test_unknown_or_reused_token_is_refused(admin): client, headers = admin token = upload(client, headers).json()["data"]["token"] assert client.post("/api/records", headers=headers, json=record_payload("LAS-TOK-001", image_tokens=[token])).status_code == 200 reused = client.post("/api/records", headers=headers, json=record_payload("LAS-TOK-002", image_tokens=[token])) assert reused.status_code == 404 assert client.post("/api/records", headers=headers, json=record_payload("LAS-TOK-003", image_tokens=["nope"])).status_code == 404 def test_editing_drops_images_left_out_of_the_keep_list(admin): client, headers = admin tokens = [upload(client, headers).json()["data"]["token"] for _ in range(2)] record_id = client.post("/api/records", headers=headers, json=record_payload("LAS-EDIT-001", image_tokens=tokens)).json()["data"]["id"] images = client.get(f"/api/records/{record_id}", headers=headers).json()["data"]["images"] assert len(images) == 2 kept = images[1]["id"] updated = client.put(f"/api/records/{record_id}", headers=headers, json=record_payload("LAS-EDIT-001", image_ids=[kept])) assert updated.status_code == 200, updated.text remaining = client.get(f"/api/records/{record_id}", headers=headers).json()["data"]["images"] assert [item["id"] for item in remaining] == [kept] assert client.get(f"/api/images/{images[0]['id']}").status_code == 404 def test_images_of_another_record_cannot_be_claimed(admin): client, headers = admin token = upload(client, headers).json()["data"]["token"] first = client.post("/api/records", headers=headers, json=record_payload("LAS-OWN-001", image_tokens=[token])).json()["data"]["id"] image_id = client.get(f"/api/records/{first}", headers=headers).json()["data"]["images"][0]["id"] second = client.post("/api/records", headers=headers, json=record_payload("LAS-OWN-002")).json()["data"]["id"] response = client.put(f"/api/records/{second}", headers=headers, json=record_payload("LAS-OWN-002", image_ids=[image_id])) assert response.status_code == 404 def test_deleting_a_record_removes_its_images(admin): client, headers = admin token = upload(client, headers).json()["data"]["token"] record_id = client.post("/api/records", headers=headers, json=record_payload("LAS-DEL-001", image_tokens=[token])).json()["data"]["id"] image_id = client.get(f"/api/records/{record_id}", headers=headers).json()["data"]["images"][0]["id"] assert client.delete(f"/api/records/{record_id}", headers=headers).status_code == 200 assert client.get(f"/api/images/{image_id}").status_code == 404 def test_expired_pending_images_are_purged(admin): from datetime import datetime, timedelta from app.db import SessionLocal from app.models import RecordImage client, headers = admin stale_token = upload(client, headers).json()["data"]["token"] with SessionLocal() as db: stale = db.query(RecordImage).filter_by(token=stale_token).one() stale.created_at = datetime.now().astimezone() - timedelta(hours=5) db.commit() assert upload(client, headers).status_code == 200 # 任何一次新上传都会顺手清理孤儿 with SessionLocal() as db: # 按 token 判断：SQLite 会复用被删除行的 rowid，按 id 判断会误判 assert db.query(RecordImage).filter_by(token=stale_token).count() == 0 def test_viewer_role_cannot_upload(admin): client, headers = admin created = client.post("/api/users", headers=headers, json={"username": "readonly1", "display_name": "只读", "password": "ViewerPass123", "role": "viewer"}) assert created.status_code in (200, 409) client.cookies.clear() client.post("/api/auth/login", data={"username": "readonly1", "password": "ViewerPass123"}) import hashlib import hmac from app.config import settings from app.main import SESSION_COOKIE csrf = hmac.new(settings.secret.encode(), client.cookies.get(SESSION_COOKIE).encode(), hashlib.sha256).hexdigest() assert upload(client, {"X-CSRF-Token": csrf}).status_code == 403 def test_filename_injection_characters_are_stripped(admin): client, headers = admin files = {"file": ('x" onload="alert(1)".png', png_bytes(), "image/png")} body = client.post("/api/uploads/images", headers=headers, files=files).json() stored = body["data"]["original_name"] assert '"' not in stored and "<" not in stored and ">" not in stored and "'" not in stored def test_payload_appended_after_image_end_marker_is_discarded(admin): client, headers = admin smuggled = png_bytes() + b"" * 20 body = client.post("/api/uploads/images", headers=headers, files={"file": ("ok.png", smuggled, "image/png")}).json() assert body["data"]["size_bytes"] == len(png_bytes()) token = body["data"]["token"] record_id = client.post("/api/records", headers=headers, json=record_payload("LAS-TRIM-001", image_tokens=[token])).json()["data"]["id"] url = client.get(f"/api/records/{record_id}", headers=headers).json()["data"]["images"][0]["url"] assert b"