File size: 5,288 Bytes
d590bfb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | """使用内存 TestClient 验证表单解析，不启动服务器、不导入模型/数据库。"""
import json
import unittest
from unittest.mock import patch
from fastapi import APIRouter, FastAPI, File, Form, HTTPException, Query, UploadFile
from fastapi.testclient import TestClient
from upload_routes import FaceAnalysisFormRoute
class UploadFormLimitsTest(unittest.TestCase):
def setUp(self) -> None:
self.received_files = []
app = FastAPI()
router = APIRouter(prefix="/facescore", route_class=FaceAnalysisFormRoute)
@router.post("/analyze")
@router.post("/app/analyze")
async def analyze(
images: str = Form(None), nickname: str = Form(None),
file: UploadFile = File(None), files: list[UploadFile] = File(None),
model: str = Query("hybrid"), fail: bool = Query(False),
) -> dict:
uploads = ([file] if file is not None else []) + (files or [])
self.received_files.extend(uploads)
if fail:
raise HTTPException(status_code=400, detail="business error")
return {
"images": images, "nickname": nickname, "model": model,
"uploads": [{"name": item.filename, "size": len(await item.read())} for item in uploads],
}
@router.post("/other")
async def other(images: str = Form(...)) -> dict:
return {"size": len(images)}
app.include_router(router)
self.client = TestClient(app)
self.large_images = json.dumps(["A" * (2 * 1024 * 1024)])
def test_two_analysis_routes_accept_large_multipart_base64(self) -> None:
for path in ("/facescore/analyze", "/facescore/app/analyze"):
with self.subTest(path=path):
response = self.client.post(path, params={"model": "deepface"}, files={
"images": (None, self.large_images), "nickname": (None, "测试昵称"),
})
self.assertEqual(200, response.status_code, response.text)
self.assertEqual(self.large_images, response.json()["images"])
self.assertEqual("测试昵称", response.json()["nickname"])
self.assertEqual("deepface", response.json()["model"])
def test_urlencoded_base64_is_preserved(self) -> None:
response = self.client.post("/facescore/analyze", data={"images": self.large_images})
self.assertEqual(200, response.status_code, response.text)
self.assertEqual(self.large_images, response.json()["images"])
def test_other_routes_retain_default_limit(self) -> None:
response = self.client.post("/facescore/other", files={"images": (None, self.large_images)})
self.assertEqual(400, response.status_code)
self.assertIn("maximum size", response.json()["detail"])
def test_configured_limit_rejects_larger_fields_with_413(self) -> None:
with patch("upload_routes.FACE_SCORE_MAX_FORM_FIELD_MB", 1):
for multipart in (True, False):
with self.subTest(multipart=multipart):
kwargs = {"files": {"images": (None, self.large_images)}} if multipart else {
"data": {"images": self.large_images}}
response = self.client.post("/facescore/analyze", **kwargs)
self.assertEqual(413, response.status_code, response.text)
self.assertIn("1MB", response.json()["detail"])
def test_field_at_limit_is_accepted(self) -> None:
with patch("upload_routes.FACE_SCORE_MAX_FORM_FIELD_MB", 1):
response = self.client.post("/facescore/analyze", files={"images": (None, "A" * (1024 * 1024))})
self.assertEqual(200, response.status_code, response.text)
def test_file_uploads_still_work_and_are_closed(self) -> None:
for field, count in (("file", 1), ("files", 2)):
with self.subTest(field=field), patch("upload_routes.FACE_SCORE_MAX_FORM_FIELD_MB", 1):
response = self.client.post("/facescore/analyze", files=[
(field, (f"photo{i}.jpg", b"x" * (2 * 1024 * 1024), "image/jpeg")) for i in range(count)
])
self.assertEqual(200, response.status_code, response.text)
self.assertEqual(count, len(response.json()["uploads"]))
self.assertTrue(all(item["size"] == 2 * 1024 * 1024 for item in response.json()["uploads"]))
self.assertTrue(all(item.file.closed for item in self.received_files))
def test_business_error_is_preserved_and_upload_closed(self) -> None:
response = self.client.post("/facescore/analyze?fail=true", files={"file": ("photo.jpg", b"image")})
self.assertEqual(400, response.status_code)
self.assertEqual("business error", response.json()["detail"])
self.assertTrue(self.received_files[0].file.closed)
def test_invalid_multipart_is_still_rejected(self) -> None:
response = self.client.post("/facescore/analyze", content=b"invalid", headers={"content-type": "multipart/form-data"})
self.assertEqual(400, response.status_code)
self.assertIn("boundary", response.json()["detail"])
if __name__ == "__main__":
unittest.main()
|