122 lines
4.0 KiB
Python
122 lines
4.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
文件上传测试。POST /api/upload 上传图片，DELETE /api/upload 删除。
|
||
验证：本地存储（OSS 未配置时）、响应格式、删除流程。
|
||
"""
|
||
import io
|
||
import pytest
|
||
import requests
|
||
|
||
|
||
# 最小有效 JPEG（1x1 像素，约 100 字节）
|
||
_MIN_JPEG = (
|
||
b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00"
|
||
b"\xff\xdb\x00C\x00\x08\x06\x06\x07\x06\x05\x08\x07\x07\x07\t\t\x08\n\x0c\x14\r\x0c\x0b\x0b\x0c\x19\x12\x13\x0f\x14\x1d\x1a\x1f\x1e\x1d\x1a\x1c\x1c"
|
||
b" $.' \",#\x1c\x1c(7),01444\x1f'9=82<.342\xff\xc0\x00\x0b\x08\x00\x01\x00\x01\x01\x01\x11\x00\xff\xc4\x00\x1f\x00\x00\x01\x05\x01\x01\x01\x01\x01\x01\x00\x00\x00\x00\x00\x00\x00\x00\x01\x02\x03\x04\x05\x06\x07\x08\t\n\x0b\xff\xda\x00\x08\x01\x01\x00\x00\x00?\x00\xfe\x02\x1f\xff\xd9"
|
||
)
|
||
|
||
|
||
def test_upload_post_success(base_url):
|
||
"""POST /api/upload 上传图片成功，返回 url、fileName、size、type"""
|
||
files = {"file": ("test_upload.jpg", io.BytesIO(_MIN_JPEG), "image/jpeg")}
|
||
data = {"folder": "test"}
|
||
r = requests.post(
|
||
f"{base_url}/api/upload",
|
||
files=files,
|
||
data=data,
|
||
timeout=10,
|
||
)
|
||
assert r.status_code == 200, f"期望 200，实际 {r.status_code}: {r.text}"
|
||
body = r.json()
|
||
assert body.get("success") is True, body
|
||
assert "url" in body, body
|
||
assert body["url"], "url 不应为空"
|
||
data_out = body.get("data", {})
|
||
assert "url" in data_out
|
||
assert "fileName" in data_out
|
||
assert "size" in data_out
|
||
assert data_out["size"] == len(_MIN_JPEG)
|
||
assert "type" in data_out
|
||
assert "image" in str(data_out.get("type", "")).lower()
|
||
|
||
|
||
def test_upload_post_with_admin_token(base_url, admin_token):
|
||
"""POST /api/upload 带管理端 token 也可上传（接口不强制鉴权）"""
|
||
if not admin_token:
|
||
pytest.skip("admin 登录失败")
|
||
files = {"file": ("avatar.jpg", io.BytesIO(_MIN_JPEG), "image/jpeg")}
|
||
# multipart 上传不设 Content-Type，让 requests 自动带 boundary
|
||
headers = {"Authorization": f"Bearer {admin_token}"}
|
||
r = requests.post(
|
||
f"{base_url}/api/upload",
|
||
files=files,
|
||
headers=headers,
|
||
timeout=10,
|
||
)
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert body.get("success") is True
|
||
assert body.get("url")
|
||
|
||
|
||
def test_upload_post_no_file(base_url):
|
||
"""POST /api/upload 无 file 返回 400"""
|
||
r = requests.post(
|
||
f"{base_url}/api/upload",
|
||
data={"folder": "test"},
|
||
timeout=10,
|
||
)
|
||
assert r.status_code == 400
|
||
body = r.json()
|
||
assert body.get("success") is False
|
||
assert "error" in body or "请选择" in body.get("error", "")
|
||
|
||
|
||
def test_upload_post_invalid_type(base_url):
|
||
"""POST /api/upload 非图片格式返回 400"""
|
||
files = {"file": ("test.txt", io.BytesIO(b"hello"), "text/plain")}
|
||
r = requests.post(
|
||
f"{base_url}/api/upload",
|
||
files=files,
|
||
timeout=10,
|
||
)
|
||
assert r.status_code == 400
|
||
body = r.json()
|
||
assert body.get("success") is False
|
||
|
||
|
||
def test_upload_delete_local(base_url):
|
||
"""DELETE /api/upload 删除本地文件：先上传再删除"""
|
||
# 1. 上传
|
||
files = {"file": ("del_test.jpg", io.BytesIO(_MIN_JPEG), "image/jpeg")}
|
||
r1 = requests.post(
|
||
f"{base_url}/api/upload",
|
||
files=files,
|
||
data={"folder": "test"},
|
||
timeout=10,
|
||
)
|
||
assert r1.status_code == 200
|
||
url = r1.json().get("url")
|
||
assert url, "上传应返回 url"
|
||
# path 支持 /uploads/xxx 或含 /uploads/ 的完整 URL
|
||
path = url
|
||
|
||
# 2. 删除
|
||
r2 = requests.delete(
|
||
f"{base_url}/api/upload",
|
||
params={"path": path},
|
||
timeout=10,
|
||
)
|
||
assert r2.status_code == 200
|
||
body = r2.json()
|
||
assert body.get("success") is True
|
||
assert "删除成功" in body.get("message", "")
|
||
|
||
|
||
def test_upload_delete_no_path(base_url):
|
||
"""DELETE /api/upload 无 path 返回 400"""
|
||
r = requests.delete(f"{base_url}/api/upload", timeout=10)
|
||
assert r.status_code == 400
|
||
body = r.json()
|
||
assert body.get("success") is False
|