Fix matching request data when files are provided (#252)

Author: lundberg

docs/api.md

@@ -309,9 +309,9 @@ respx.post("https://example.org/", content__contains="bar")
 ```
 
 ### Data
-Matches request *form data*, using [eq](#eq) as default lookup.
+Matches request *form data*, excluding files, using [eq](#eq) as default lookup.
 > Key: `data`  
-> Lookups: [eq](#eq)
+> Lookups: [eq](#eq), [contains](#contains)
 ``` python
 respx.post("https://example.org/", data={"foo": "bar"})
 ```

respx/patterns.py

@@ -25,6 +25,8 @@
 
 import httpx
 
+from respx.utils import MultiItems, decode_data
+
 from .types import (
     URL as RawURL,
     CookieTypes,
@@ -536,14 +538,16 @@ def hash(self, value: Union[str, List, Dict]) -> str:
         return jsonlib.dumps(value, sort_keys=True)
 
 
-class Data(ContentMixin, Pattern):
-    lookups = (Lookup.EQUAL,)
+class Data(MultiItemsMixin, Pattern):
+    lookups = (Lookup.EQUAL, Lookup.CONTAINS)
     key = "data"
-    value: bytes
+    value: MultiItems
+
+    def clean(self, value: Dict) -> MultiItems:
+        return MultiItems(value)
 
-    def clean(self, value: Dict) -> bytes:
-        request = httpx.Request("POST", "/", data=value)
-        data = request.read()
+    def parse(self, request: httpx.Request) -> Any:
+        data, _ = decode_data(request)
         return data
 
 

respx/utils.py

@@ -0,0 +1,73 @@
+import email
+from email.message import Message
+from typing import List, Tuple, cast
+from urllib.parse import parse_qsl
+
+import httpx
+
+
+class MultiItems(dict):
+    def get_list(self, key: str) -> List[str]:
+        try:
+            return [self[key]]
+        except KeyError:  # pragma: no cover
+            return []
+
+    def multi_items(self) -> List[Tuple[str, str]]:
+        return list(self.items())
+
+
+def _parse_multipart_form_data(
+    content: bytes, *, content_type: str, encoding: str
+) -> Tuple[MultiItems, MultiItems]:
+    form_data = b"\r\n".join(
+        (
+            b"MIME-Version: 1.0",
+            b"Content-Type: " + content_type.encode(encoding),
+            b"\r\n" + content,
+        )
+    )
+    data = MultiItems()
+    files = MultiItems()
+    for payload in email.message_from_bytes(form_data).get_payload():
+        payload = cast(Message, payload)
+        name = payload.get_param("name", header="Content-Disposition")
+        filename = payload.get_filename()
+        content_type = payload.get_content_type()
+        value = payload.get_payload(decode=True)
+        assert isinstance(value, bytes)
+        if content_type.startswith("text/") and filename is None:
+            # Text field
+            data[name] = value.decode(payload.get_content_charset() or "utf-8")
+        else:
+            # File field
+            files[name] = filename, value
+
+    return data, files
+
+
+def _parse_urlencoded_data(content: bytes, *, encoding: str) -> MultiItems:
+    return MultiItems(
+        (key, value)
+        for key, value in parse_qsl(content.decode(encoding), keep_blank_values=True)
+    )
+
+
+def decode_data(request: httpx.Request) -> Tuple[MultiItems, MultiItems]:
+    content = request.read()
+    content_type = request.headers.get("Content-Type", "")
+
+    if content_type.startswith("multipart/form-data"):
+        data, files = _parse_multipart_form_data(
+            content,
+            content_type=content_type,
+            encoding=request.headers.encoding,
+        )
+    else:
+        data = _parse_urlencoded_data(
+            content,
+            encoding=request.headers.encoding,
+        )
+        files = MultiItems()
+
+    return data, files

tests/test_api.py

@@ -263,6 +263,15 @@ def test_json_post_body():
         assert get_route.called
 
 
+def test_data_post_body():
+    with respx.mock:
+        url = "https://foo.bar/"
+        route = respx.post(url, data={"foo": "bar"}) % 201
+        response = httpx.post(url, data={"foo": "bar"}, files={"file": b"..."})
+        assert response.status_code == 201
+        assert route.called
+
+
 async def test_raising_content(client):
     async with MockRouter() as respx_mock:
         url = "https://foo.bar/"

tests/test_patterns.py

@@ -323,14 +323,69 @@ def test_content_pattern(lookup, content, expected):
 
 
 @pytest.mark.parametrize(
-    ("lookup", "data", "expected"),
+    ("lookup", "data", "request_data", "expected"),
     [
-        (Lookup.EQUAL, {"foo": "bar", "ham": "spam"}, True),
+        (
+            Lookup.EQUAL,
+            {"foo": "bar", "ham": "spam"},
+            None,
+            True,
+        ),
+        (
+            Lookup.EQUAL,
+            {"foo": "bar", "ham": "spam"},
+            {"ham": "spam", "foo": "bar"},
+            True,
+        ),
+        (
+            Lookup.EQUAL,
+            {"uni": "äpple", "mixed": "Gehäusegröße"},
+            None,
+            True,
+        ),
+        (
+            Lookup.EQUAL,
+            {"blank_value": ""},
+            None,
+            True,
+        ),
+        (
+            Lookup.EQUAL,
+            {"x": "a"},
+            {"x": "b"},
+            False,
+        ),
+        (
+            Lookup.EQUAL,
+            {"foo": "bar"},
+            {"foo": "bar", "ham": "spam"},
+            False,
+        ),
+        (
+            Lookup.CONTAINS,
+            {"foo": "bar"},
+            {"foo": "bar", "ham": "spam"},
+            True,
+        ),
     ],
 )
-def test_data_pattern(lookup, data, expected):
-    request = httpx.Request("POST", "https://foo.bar/", data=data)
-    match = Data(data, lookup=lookup).match(request)
+def test_data_pattern(lookup, data, request_data, expected):
+    request_with_data = httpx.Request(
+        "POST",
+        "https://foo.bar/",
+        data=request_data or data,
+    )
+    request_with_data_and_files = httpx.Request(
+        "POST",
+        "https://foo.bar/",
+        data=request_data or data,
+        files={"upload-file": ("report.xls", b"<...>", "application/vnd.ms-excel")},
+    )
+
+    match = Data(data, lookup=lookup).match(request_with_data)
+    assert bool(match) is expected
+
+    match = Data(data, lookup=lookup).match(request_with_data_and_files)
     assert bool(match) is expected