| """使用内存 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() |
|
|
