Match parsed form fields in request expectations

doc/howto.rst

@@ -55,6 +55,26 @@ including an empty value. A ``werkzeug.datastructures.MultiDict`` matches all
 values, including empty ones.
 
 
+Matching form data
+------------------
+
+Use ``data_form`` to match parsed form fields without depending on their encoded
+order or spelling:
+
+.. literalinclude :: ../tests/examples/test_howto_form_data.py
+   :language: python
+
+A dictionary matches the first value for each field. Use a
+``werkzeug.datastructures.MultiDict`` to match all values of repeated fields.
+Extra fields do not match, and an empty value is distinct from a missing field.
+
+Werkzeug parses form fields when the request's ``Content-Type`` header is
+``application/x-www-form-urlencoded`` or ``multipart/form-data``.
+``data_form={}`` matches requests with no parsed form fields; use ``headers``
+if a particular content type is required. ``data_form`` is mutually exclusive
+with ``data`` and ``json``. The raw body remains available to response handlers.
+
+
 URI matching
 ------------
 

pytest_httpserver/blocking_httpserver.py

@@ -80,6 +80,7 @@ def assert_request(
         header_value_matcher: HeaderValueMatcher | None = None,
         json: Any = UNDEFINED,
         timeout: int = 30,
+        data_form: Mapping[str, str] | None = None,
     ) -> BlockingRequestHandler:
         """
         Wait for an incoming request and check whether it matches according to the given parameters.
@@ -109,10 +110,13 @@ def assert_request(
             is loaded as json. If load fails, this matcher will be failed also. *Content-Type* is not checked.
             If that's desired, add it to the headers parameter.
         :param timeout: waiting time in seconds for an incoming request.
+        :param data_form: expected form fields, parsed by werkzeug using the request's *Content-Type*.
+            A mapping matches the first value for each field; a ``MultiDict`` matches all values.
+            Mutually exclusive with `data` and `json`.
 
         :return: Created and registered :py:class:`BlockingRequestHandler`.
 
-        Parameters `json` and `data` are mutually exclusive.
+        Parameters `json`, `data`, and `data_form` are mutually exclusive.
         """
 
         matcher = self.create_matcher(
@@ -124,6 +128,7 @@ def assert_request(
             query_string=query_string,
             header_value_matcher=header_value_matcher,
             json=json,
+            **({"data_form": data_form} if data_form is not None else {}),
         )
 
         try:

pytest_httpserver/httpserver.py

@@ -304,6 +304,7 @@ class RequestMatcherKwargs(TypedDict, total=False):
     query_string: QueryMatcher | str | bytes | Mapping[str, str] | None
     header_value_matcher: HVMATCHER_T | None
     json: Any
+    data_form: Mapping[str, str] | None
 
 
 class RequestMatcher:
@@ -333,6 +334,9 @@ class RequestMatcher:
     :param json: a python object (eg. a dict) whose value will be compared to the request body after it
         is loaded as json. If load fails, this matcher will be failed also. *Content-Type* is not checked.
         If that's desired, add it to the headers parameter.
+    :param data_form: expected form fields, parsed by werkzeug using the request's *Content-Type*.
+        A mapping matches the first value for each field; a ``MultiDict`` matches all values.
+        Mutually exclusive with `data` and `json`.
     """
 
     def __init__(
@@ -345,15 +349,19 @@ def __init__(
         query_string: QueryMatcher | str | bytes | Mapping[str, str] | None = None,
         header_value_matcher: HVMATCHER_T | None = None,
         json: Any = UNDEFINED,
+        data_form: Mapping[str, str] | None = None,
     ) -> None:
         if json is not UNDEFINED and data is not None:
             raise ValueError("data and json parameters are mutually exclusive")
+        if data_form is not None and (data is not None or json is not UNDEFINED):
+            raise ValueError("data_form is mutually exclusive with data and json")
 
         self.uri = uri
         self.method = method
         self.query_string = query_string
         self.query_matcher = _create_query_matcher(self.query_string)
         self.json = json
+        self.data_form = data_form
 
         self.headers: Mapping[str, str] = {}
         if headers is not None:
@@ -379,9 +387,11 @@ def __repr__(self) -> str:
         retval = "<{} ".format(class_name)
         retval += (
             "uri={uri!r} method={method!r} query_string={query_string!r} "
-            "headers={headers!r} data={data!r} json={json!r}>"
+            "headers={headers!r} data={data!r} json={json!r}"
         ).format_map(self.__dict__)
-        return retval
+        if self.data_form is not None:
+            retval += f" data_form={self.data_form!r}"
+        return retval + ">"
 
     def match_data(self, request: Request) -> bool:
         """
@@ -440,6 +450,17 @@ def match_json(self, request: Request) -> bool:
 
         return bool(json_received == self.json)
 
+    def match_data_form(self, request: Request) -> bool:
+        """Match parsed form fields, or accept any form when no expectation is set."""
+        if self.data_form is None:
+            return True
+
+        # Cache the raw body before form parsing so response handlers can still read it.
+        request.get_data()
+        if isinstance(self.data_form, MultiDict):
+            return bool(request.form == self.data_form)
+        return request.form.to_dict() == dict(self.data_form)
+
     def difference(self, request: Request) -> list[tuple[str, str, str | URIPattern]]:
         """
         Calculates the difference between the matcher and the request.
@@ -478,6 +499,8 @@ def difference(self, request: Request) -> list[tuple[str, str, str | URIPattern]
 
         if not self.match_json(request):
             retval.append(("json", request.data, self.json))
+        if not self.match_data_form(request):
+            retval.append(("data_form", request.form, self.data_form))
         return retval
 
     def match(self, request: Request) -> bool:
@@ -603,9 +626,11 @@ def __repr__(self) -> str:
         retval = (
             f"<{class_name} uri={self.matcher.uri!r} method={self.matcher.method!r} "
             f"query_string={self.matcher.query_string!r} headers={self.matcher.headers!r} data={self.matcher.data!r} "
-            f"json={self.matcher.json!r}>"
+            f"json={self.matcher.json!r}"
         )
-        return retval
+        if self.matcher.data_form is not None:
+            retval += f" data_form={self.matcher.data_form!r}"
+        return retval + ">"
 
 
 class RequestHandlerList(list[RequestHandler]):
@@ -1087,6 +1112,7 @@ def expect_request(
         header_value_matcher: HVMATCHER_T | None = None,
         handler_type: HandlerType = HandlerType.PERMANENT,
         json: Any = UNDEFINED,
+        data_form: Mapping[str, str] | None = None,
     ) -> RequestHandler:
         """
         Create and register a request handler.
@@ -1134,10 +1160,13 @@ def expect_request(
         :param json: a python object (eg. a dict) whose value will be compared to the request body after it
             is loaded as json. If load fails, this matcher will be failed also. *Content-Type* is not checked.
             If that's desired, add it to the headers parameter.
+        :param data_form: expected form fields, parsed by werkzeug using the request's *Content-Type*.
+            A mapping matches the first value for each field; a ``MultiDict`` matches all values.
+            Mutually exclusive with `data` and `json`.
 
         :return: Created and register :py:class:`RequestHandler`.
 
-        Parameters `json` and `data` are mutually exclusive.
+        Parameters `json`, `data`, and `data_form` are mutually exclusive.
         """
 
         matcher = self.create_matcher(
@@ -1149,6 +1178,7 @@ def expect_request(
             query_string=query_string,
             header_value_matcher=header_value_matcher,
             json=json,
+            **({"data_form": data_form} if data_form is not None else {}),
         )
         request_handler = RequestHandler(matcher)
         if handler_type == HandlerType.PERMANENT:
@@ -1169,6 +1199,7 @@ def expect_oneshot_request(
         query_string: QueryMatcher | str | bytes | Mapping[str, str] | None = None,
         header_value_matcher: HVMATCHER_T | None = None,
         json: Any = UNDEFINED,
+        data_form: Mapping[str, str] | None = None,
     ) -> RequestHandler:
         """
         Create and register a oneshot request handler.
@@ -1196,10 +1227,13 @@ def expect_oneshot_request(
         :param json: a python object (eg. a dict) whose value will be compared to the request body after it
             is loaded as json. If load fails, this matcher will be failed also. *Content-Type* is not checked.
             If that's desired, add it to the headers parameter.
+        :param data_form: expected form fields, parsed by werkzeug using the request's *Content-Type*.
+            A mapping matches the first value for each field; a ``MultiDict`` matches all values.
+            Mutually exclusive with `data` and `json`.
 
         :return: Created and register :py:class:`RequestHandler`.
 
-        Parameters `json` and `data` are mutually exclusive.
+        Parameters `json`, `data`, and `data_form` are mutually exclusive.
         """
 
         return self.expect_request(
@@ -1212,6 +1246,7 @@ def expect_oneshot_request(
             header_value_matcher=header_value_matcher,
             handler_type=HandlerType.ONESHOT,
             json=json,
+            **({"data_form": data_form} if data_form is not None else {}),
         )
 
     def expect_ordered_request(
@@ -1224,6 +1259,7 @@ def expect_ordered_request(
         query_string: QueryMatcher | str | bytes | Mapping[str, str] | None = None,
         header_value_matcher: HVMATCHER_T | None = None,
         json: Any = UNDEFINED,
+        data_form: Mapping[str, str] | None = None,
     ) -> RequestHandler:
         """
         Create and register a ordered request handler.
@@ -1251,10 +1287,13 @@ def expect_ordered_request(
         :param json: a python object (eg. a dict) whose value will be compared to the request body after it
             is loaded as json. If load fails, this matcher will be failed also. *Content-Type* is not checked.
             If that's desired, add it to the headers parameter.
+        :param data_form: expected form fields, parsed by werkzeug using the request's *Content-Type*.
+            A mapping matches the first value for each field; a ``MultiDict`` matches all values.
+            Mutually exclusive with `data` and `json`.
 
         :return: Created and register :py:class:`RequestHandler`.
 
-        Parameters `json` and `data` are mutually exclusive.
+        Parameters `json`, `data`, and `data_form` are mutually exclusive.
         """
 
         return self.expect_request(
@@ -1267,6 +1306,7 @@ def expect_ordered_request(
             header_value_matcher=header_value_matcher,
             handler_type=HandlerType.ORDERED,
             json=json,
+            **({"data_form": data_form} if data_form is not None else {}),
         )
 
     def format_matchers(self) -> str:

releasenotes/notes/form-matching-b314582ecf709ad2.yaml

@@ -0,0 +1,8 @@
+features:
+  - |
+    Add the ``data_form`` keyword argument to request expectations and
+    ``BlockingHTTPServer.assert_request()`` for matching parsed form fields.
+    A mapping matches each field's first value, while a Werkzeug ``MultiDict``
+    matches all repeated values. Form expectations are mutually exclusive with
+    ``data`` and ``json`` expectations, and preserve the raw request body for
+    response handlers.

tests/examples/test_howto_form_data.py

@@ -0,0 +1,9 @@
+import requests
+
+
+def test_form_data(httpserver):
+    httpserver.expect_request("/form", method="POST", data_form={"name": "Jane Doe", "flag": ""}).respond_with_data(
+        "ok"
+    )
+    response = requests.post(httpserver.url_for("/form"), data={"flag": "", "name": "Jane Doe"})
+    assert response.text == "ok"

tests/test_bake.py

@@ -20,6 +20,17 @@ def test_bake_with_headers(httpserver: HTTPServer) -> None:
     assert response.json() == {"result": "ok"}
 
 
+def test_bake_with_form(httpserver: HTTPServer) -> None:
+    server = httpserver.bake(method="POST", data_form={"foo": "bar"})
+    server.expect_request("/form").respond_with_data("ok")
+
+    response = requests.post(server.url_for("/form"), data={"foo": "bar"})
+
+    assert response.status_code == 200
+    assert response.text == "ok"
+    httpserver.check()
+
+
 @pytest.mark.parametrize(
     ("bake_chain", "expect_kwargs", "request_method"),
     [

tests/test_blocking_httpserver.py

@@ -68,6 +68,19 @@ def test_behave_workflow(httpserver: BlockingHTTPServer):
         then_the_response_is_got_from(server_connection, response)
 
 
+def test_form_request(httpserver: BlockingHTTPServer):
+    request = dict(method="POST", url=httpserver.url_for("/form"), data={"foo": "bar"})
+
+    with when_a_request_is_being_sent_to_the_server(request) as server_connection:
+        handler = httpserver.assert_request("/form", method="POST", data_form={"foo": "bar"})
+        handler.respond_with_data("ok")
+
+        response = server_connection.get(timeout=9)
+        assert response.status_code == 200
+        assert response.text == "ok"
+        httpserver.check()
+
+
 def test_raises_assertion_error_when_request_does_not_match(httpserver: BlockingHTTPServer):
     request = dict(
         method="GET",

tests/test_form_matcher.py

@@ -0,0 +1,69 @@
+import pytest
+import requests
+from werkzeug import Request
+from werkzeug import Response
+from werkzeug.datastructures import MultiDict
+
+from pytest_httpserver import HTTPServer
+from pytest_httpserver.httpserver import RequestMatcher
+
+
+@pytest.mark.parametrize("expect_method", ["expect_request", "expect_oneshot_request", "expect_ordered_request"])
+def test_form_matcher_request(httpserver: HTTPServer, expect_method: str):
+    form = MultiDict([("tag", "first"), ("tag", "second"), ("name", "José + %"), ("flag", "")])
+    expect = getattr(httpserver, expect_method)
+    expect("/form", method="POST", data_form=form).respond_with_data("ok")
+
+    response = requests.post(httpserver.url_for("/form"), data=list(form.items(multi=True)))
+
+    assert response.status_code == 200
+    assert response.text == "ok"
+    httpserver.check()
+
+
+@pytest.mark.parametrize(
+    ("expected", "body", "matches"),
+    [
+        ({"foo": "bar"}, "foo=bar&foo=other", True),
+        ({"foo": "other"}, "foo=bar&foo=other", False),
+        ({"foo": "bar"}, "foo=bar&extra=value", False),
+        ({"flag": ""}, "flag=", True),
+        ({"flag": ""}, "", False),
+        ({}, "", True),
+        (MultiDict([("tag", "first"), ("tag", "second")]), "tag=first&tag=second", True),
+        (MultiDict([("tag", "first"), ("tag", "second")]), "tag=first", False),
+    ],
+)
+def test_form_matcher_values(expected, body: str, matches: bool):  # noqa: FBT001
+    request = Request.from_values(method="POST", data=body, content_type="application/x-www-form-urlencoded")
+    matcher = RequestMatcher("/", data_form=expected)
+
+    assert matcher.match(request) is matches
+    if not matches:
+        assert matcher.difference(request) == [("data_form", request.form, expected)]
+        assert "data_form=" in repr(matcher)
+
+
+def test_form_matcher_needs_form_content_type():
+    request = Request.from_values(method="POST", data="foo=bar", content_type="text/plain")
+    assert not RequestMatcher("/", data_form={"foo": "bar"}).match(request)
+
+
+def test_form_matcher_preserves_request_body(httpserver: HTTPServer):
+    def handler(request: Request) -> Response:
+        assert request.form["foo"] == "bar"
+        return Response(request.get_data())
+
+    httpserver.expect_request("/form", data_form={"foo": "bar"}).respond_with_handler(handler)
+
+    response = requests.post(httpserver.url_for("/form"), data={"foo": "bar"})
+
+    assert response.status_code == 200
+    assert response.content == b"foo=bar"
+    httpserver.check()
+
+
+@pytest.mark.parametrize("other_body", [{"data": "foo=bar"}, {"json": {}}])
+def test_form_matcher_body_parameters_mutually_exclusive(httpserver: HTTPServer, other_body):
+    with pytest.raises(ValueError, match="mutually exclusive"):
+        httpserver.expect_request("/form", data_form={}, **other_body)

tests/test_release.py

@@ -219,6 +219,7 @@ def test_sdist_contents(build: Build, version: str):
             "examples",
             "test_bake.py",
             "test_blocking_httpserver.py",
+            "test_form_matcher.py",
             "test_handler_errors.py",
             "test_headers.py",
             "test_hooks.py",