Spaces:
Paused
Paused
| """Raw request limits apply before JSON validation and across body chunks.""" | |
| import asyncio | |
| import unittest | |
| from unittest.mock import patch | |
| from fastapi.testclient import TestClient | |
| import app as server | |
| from request_limits import RequestSizeLimitMiddleware | |
| class RequestLimitTests(unittest.TestCase): | |
| def test_oversized_json_is_rejected_before_inference(self): | |
| with patch.object(server.engine, "analyze_locked") as analyze: | |
| for path in ["/api/analyze", "/api/jobs"]: | |
| response = TestClient(server.app).post(path, json={ | |
| "task": "pii", "text": "x" * (12 * 1024 * 1024), | |
| }) | |
| self.assertEqual(response.status_code, 413) | |
| self.assertEqual(response.json()["detail"], "Request is too large. Use fewer or smaller media files, or shorten the text.") | |
| self.assertEqual(response.headers["Cache-Control"], "no-store, max-age=0") | |
| analyze.assert_not_called() | |
| def test_chunked_body_cannot_bypass_the_limit(self): | |
| async def exercise(): | |
| messages = iter([ | |
| {"type": "http.request", "body": b"12345", "more_body": True}, | |
| {"type": "http.request", "body": b"67890", "more_body": True}, | |
| ]) | |
| sent = [] | |
| async def receive(): | |
| return next(messages) | |
| async def send(message): | |
| sent.append(message) | |
| async def downstream(scope, receive, send): | |
| self.fail("An oversized request must not reach the framework") | |
| await RequestSizeLimitMiddleware(downstream, max_bytes=8)( | |
| {"type": "http", "path": "/api/analyze"}, receive, send, | |
| ) | |
| self.assertEqual(sent[0]["status"], 413) | |
| asyncio.run(exercise()) | |
| if __name__ == "__main__": | |
| unittest.main() | |