Merge pull request #78 from a-gerhard/request_data

Explicitly store request data

.gitignore

@@ -6,6 +6,7 @@ junit-*.xml
 build
 dist
 *.egg-info
+/pytest_localserver/_version.py
 
 # project files
 .idea

AUTHORS

@@ -14,3 +14,4 @@ Hasan Ramezani <[email protected]>
 Felix Yan <[email protected]>
 Henri Hulski <[email protected]>
 Theodore Ni
+Alissa Gerhard <[email protected]>

README.rst

@@ -91,12 +91,13 @@ poking around in the code itself.
     * ``content`` - content of next response (str, bytes, or iterable of either)
     * ``headers`` - response headers (dict)
     * ``chunked`` - whether to chunk-encode the response (enumeration)
+    * ``store_request_data`` - whether to store request data for later use
 
     Once these attributes are set, all subsequent requests will be answered with
     these values until they are changed or the server is stopped. A more
     convenient way to change these is ::
 
-        httpserver.serve_content(content=None, code=200, headers=None, chunked=pytest_localserver.http.Chunked.NO)
+        httpserver.serve_content(content=None, code=200, headers=None, chunked=pytest_localserver.http.Chunked.NO, store_request_data=True)
 
     The ``chunked`` attribute or parameter can be set to
 
@@ -108,6 +109,11 @@ poking around in the code itself.
     If chunk encoding is applied, each str or bytes in ``content`` becomes one
     chunk in the response.
 
+    You can use ``store_request_data=False`` to disable loading the request data into
+    memory. This will make it impossible to check the request data using
+    ``httpserver.requests[index].data`` but may make sense when posting a larger amount of
+    data and you don't need to check this.
+
     The server address can be found in property
 
     * ``url``

pytest_localserver/http.py

@@ -89,12 +89,18 @@ def __init__(self, host="127.0.0.1", port=0, ssl_context=None):
         self.compress = None
         self.requests = []
         self.chunked = Chunked.NO
+        self.store_request_data = False
 
     def __call__(self, environ, start_response):
         """
         This is the WSGI application.
         """
         request = Request(environ)
+
+        if self.store_request_data:
+            # need to invoke this method to cache the data
+            request.get_data(cache=True)
+
         self.requests.append(request)
         if (
             request.content_type == "application/x-www-form-urlencoded"
@@ -129,7 +135,7 @@ def __call__(self, environ, start_response):
 
         return response(environ, start_response)
 
-    def serve_content(self, content, code=200, headers=None, chunked=Chunked.NO):
+    def serve_content(self, content, code=200, headers=None, chunked=Chunked.NO, store_request_data=True):
         """
         Serves string content (with specified HTTP error code) as response to
         all subsequent request.
@@ -138,6 +144,7 @@ def serve_content(self, content, code=200, headers=None, chunked=Chunked.NO):
         :param code: HTTP status code
         :param headers: HTTP headers to be returned
         :param chunked: whether to apply chunked transfer encoding to the content
+        :param store_request_data: whether to store data sent as request payload.
         """
         if not isinstance(content, (str, bytes, list, tuple)):
             # If content is an iterable which is not known to be a string,
@@ -153,6 +160,7 @@ def serve_content(self, content, code=200, headers=None, chunked=Chunked.NO):
         self.content = content
         self.code = code
         self.chunked = chunked
+        self.store_request_data = store_request_data
         if headers:
             self.headers = Headers(headers)
 

tests/test_http.py

@@ -5,6 +5,7 @@
 
 import pytest
 import requests
+from werkzeug.exceptions import ClientDisconnected
 
 from pytest_localserver import http
 from pytest_localserver import plugin
@@ -90,6 +91,29 @@ def test_HEAD_request(httpserver):
 #     assert resp.status_code == 200
 
 
+def test_POST_request_no_store_data(httpserver):
+    headers = {"Content-type": "text/plain"}
+    httpserver.serve_content("TEST!", store_request_data=False)
+    requests.post(httpserver.url, data=b"testdata", headers=headers)
+
+    request = httpserver.requests[-1]
+    request.input_stream.close()
+
+    with pytest.raises(ClientDisconnected):
+        request.data
+
+
+def test_POST_request_store_data(httpserver):
+    headers = {"Content-type": "text/plain"}
+    httpserver.serve_content("TEST!", store_request_data=True)
+    requests.post(httpserver.url, data=b"testdata", headers=headers)
+
+    request = httpserver.requests[-1]
+    request.input_stream.close()
+
+    assert httpserver.requests[-1].data == b"testdata"
+
+
 @pytest.mark.parametrize("chunked_flag", [http.Chunked.YES, http.Chunked.AUTO, http.Chunked.NO])
 def test_chunked_attribute_without_header(httpserver, chunked_flag):
     """
@@ -274,19 +298,23 @@ def test_GET_request_chunked_no_content_length(httpserver, chunked_flag):
 def test_httpserver_init_failure_no_stderr_during_cleanup(tmp_path):
     """
     Test that, when the server encounters an error during __init__, its cleanup
-    does not raise an AttributeError in its __del__ method, which would emit a 
+    does not raise an AttributeError in its __del__ method, which would emit a
     warning onto stderr.
     """
 
     script_path = tmp_path.joinpath("script.py")
 
-    script_path.write_text(textwrap.dedent("""
+    script_path.write_text(
+        textwrap.dedent(
+            """
         from pytest_localserver import http
         from unittest.mock import patch
 
         with patch("pytest_localserver.http.make_server", side_effect=RuntimeError("init failure")):
             server = http.ContentServer()
-    """))
+    """
+        )
+    )
 
     result = subprocess.run([sys.executable, str(script_path)], stderr=subprocess.PIPE)
 

tests/test_https.py

@@ -1,5 +1,5 @@
-import requests
 import pytest
+import requests
 
 from pytest_localserver import https
 from pytest_localserver import plugin
@@ -43,11 +43,13 @@ def test_HEAD_request(httpsserver):
     assert resp.status_code == 200
     assert resp.headers["Content-type"] == "text/plain"
 
+
 def test_client_does_not_trust_self_signed_certificate(httpsserver):
     httpsserver.serve_content("TEST!", headers={"Content-type": "text/plain"})
     with pytest.raises(requests.exceptions.SSLError, match="CERTIFICATE_VERIFY_FAILED"):
         requests.get(httpsserver.url, verify=True)
 
+
 def test_add_server_certificate_to_client_trust_chain(httpsserver):
     httpsserver.serve_content("TEST!", headers={"Content-type": "text/plain"})
     resp = requests.get(httpsserver.url, verify=httpsserver.certificate)