summaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/asgi/tests.py32
-rw-r--r--tests/requests_tests/tests.py38
2 files changed, 68 insertions, 2 deletions
diff --git a/tests/asgi/tests.py b/tests/asgi/tests.py
index f77bd997a4..1ff6892078 100644
--- a/tests/asgi/tests.py
+++ b/tests/asgi/tests.py
@@ -13,6 +13,7 @@ from asgiref.testing import ApplicationCommunicator
from django.contrib.staticfiles.handlers import ASGIStaticFilesHandler
from django.core.asgi import get_asgi_application
from django.core.exceptions import RequestDataTooBig
+from django.core.files.uploadedfile import InMemoryUploadedFile
from django.core.handlers.asgi import ASGIHandler, ASGIRequest
from django.core.signals import request_finished, request_started
from django.db import close_old_connections
@@ -804,8 +805,7 @@ class ASGITest(SimpleTestCase):
self.assertEqual(request.COOKIES, {"a": "abc", "b": "def", "c": "ghi"})
-class DataUploadMaxMemorySizeASGITests(SimpleTestCase):
-
+class MaxMemorySizeASGITests(SimpleTestCase):
def make_request(
self,
body,
@@ -923,6 +923,34 @@ class DataUploadMaxMemorySizeASGITests(SimpleTestCase):
self.addCleanup(uploaded.close)
self.assertEqual(uploaded.read(), file_content)
+ def test_multipart_file_upload_limited_by_file_upload_max(self):
+ boundary = "testboundary"
+ file_content = b"x" * 100
+ body = (
+ (
+ f"--{boundary}\r\n"
+ f'Content-Disposition: form-data; name="file"; filename="test.txt"\r\n'
+ f"Content-Type: application/octet-stream\r\n"
+ f"\r\n"
+ ).encode()
+ + file_content
+ + f"\r\n--{boundary}--\r\n".encode()
+ )
+ # Provide an understated content-length.
+ request = self.make_request(
+ body,
+ content_type=f"multipart/form-data; boundary={boundary}".encode(),
+ content_length=9,
+ )
+ with self.settings(FILE_UPLOAD_MAX_MEMORY_SIZE=10):
+ files = request.FILES
+ self.assertEqual(len(files), 1)
+ uploaded = files["file"]
+ # The file is not loaded into memory.
+ self.assertNotIsInstance(uploaded, InMemoryUploadedFile)
+ self.addCleanup(uploaded.close)
+ self.assertEqual(uploaded.read(), file_content)
+
async def test_read_body_buffers_all_chunks(self):
# read_body() consumes all chunks regardless of
# DATA_UPLOAD_MAX_MEMORY_SIZE; the limit is enforced later when
diff --git a/tests/requests_tests/tests.py b/tests/requests_tests/tests.py
index c25dd913cf..cf24ddd326 100644
--- a/tests/requests_tests/tests.py
+++ b/tests/requests_tests/tests.py
@@ -1265,6 +1265,44 @@ class RequestsTests(SimpleTestCase):
request.multipart_parser_class = MultiPartParser
+class MemoryFileUploadHandlerTests(SimpleTestCase):
+ def test_handle_raw_input_wsgi_request_within_limit_activated(self):
+
+ class WSGIRequest:
+ def __init__(self, body):
+ self._stream = LimitedStream(BytesIO(body), len(body))
+
+ handler = MemoryFileUploadHandler()
+ with self.settings(FILE_UPLOAD_MAX_MEMORY_SIZE=10):
+ handler.handle_raw_input(WSGIRequest(b"x" * 5), {}, 5, None)
+ self.assertIs(handler.activated, True)
+
+ def test_handle_raw_input_wsgi_request_exceeds_limit_deactivated(self):
+
+ class WSGIRequest:
+ def __init__(self, body):
+ self._stream = LimitedStream(BytesIO(body), len(body))
+
+ handler = MemoryFileUploadHandler()
+ with self.settings(FILE_UPLOAD_MAX_MEMORY_SIZE=10):
+ handler.handle_raw_input(WSGIRequest(b"x" * 15), {}, 15, None)
+ self.assertIs(handler.activated, False)
+
+ def test_handle_raw_input_seekable_within_limit_activated(self):
+ handler = MemoryFileUploadHandler()
+ with self.settings(FILE_UPLOAD_MAX_MEMORY_SIZE=10):
+ # content_length param is understated (0) but actual size is 10.
+ handler.handle_raw_input(BytesIO(b"x" * 10), {}, 0, None)
+ self.assertIs(handler.activated, True)
+
+ def test_handle_raw_input_seekable_exceeds_limit_deactivated(self):
+ handler = MemoryFileUploadHandler()
+ with self.settings(FILE_UPLOAD_MAX_MEMORY_SIZE=10):
+ # content_length param is understated (0) but actual size is 15.
+ handler.handle_raw_input(BytesIO(b"x" * 15), {}, 0, None)
+ self.assertIs(handler.activated, False)
+
+
class HostValidationTests(SimpleTestCase):
poisoned_hosts = [
"example.com@evil.tld",