Fix multipart matching with custom boundaries

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Explicit support for python `3.15`.
 
 ### Fixed
+- Multipart requests with custom boundaries are now matched by `match_files` (and `match_data`), including when retrieving requests.
 - `match_params` now preserves `unittest.mock.ANY` inside tuple values, as it already does for lists.
 - Sending a non UTF8 body while having a `json` matcher registered will not raise a `UnicodeDecodeError` anymore.
 - `match_params` is now handling `bool`, `int` and `float` values in addition to `str`. Only str values were previously expected.

pytest_httpx/_request_matcher.py

@@ -213,7 +213,9 @@ def _content_match(self, request: httpx.Request) -> bool:
 
         if self.files:
             if not (
-                boundary_matched := re.match(b"^--([0-9a-f]*)\r\n", request.content)
+                boundary_matched := re.match(
+                    b"^--([^\r\n\x80-\xff]+)\r\n", request.content
+                )
             ):
                 return False
             # Ensure we re-use the same boundary for comparison

tests/test_multipart_binary.py

@@ -0,0 +1,81 @@
+import httpx
+import pytest
+
+from pytest_httpx import HTTPXMock
+
+
+@pytest.fixture(params=[b"\xff", b"caf\xc3\xa9", b"ascii\x80suffix"])
+def body(request: pytest.FixtureRequest) -> bytes:
+    return b"--" + request.param + b"\r\nopaque binary document"
+
+
+def test_binary_response_fallback(httpx_mock: HTTPXMock, body: bytes) -> None:
+    httpx_mock.add_response(
+        match_files={"upload": ("sample.txt", b"payload")},
+        status_code=400,
+        is_optional=True,
+    )
+    httpx_mock.add_response(match_content=body, status_code=202)
+
+    with httpx.Client() as client:
+        response = client.post(
+            "https://test_url",
+            content=body,
+            headers={"Content-Type": "application/octet-stream"},
+        )
+    assert response.status_code == 202
+
+
+@pytest.mark.asyncio
+async def test_binary_response_fallback_async(
+    httpx_mock: HTTPXMock, body: bytes
+) -> None:
+    httpx_mock.add_response(
+        match_files={"upload": ("sample.txt", b"payload")},
+        status_code=400,
+        is_optional=True,
+    )
+    httpx_mock.add_response(match_content=body, status_code=202)
+
+    async with httpx.AsyncClient() as client:
+        response = await client.post(
+            "https://test_url",
+            content=body,
+            headers={"Content-Type": "application/octet-stream"},
+        )
+    assert response.status_code == 202
+
+
+def test_binary_requests_retrieval(httpx_mock: HTTPXMock, body: bytes) -> None:
+    httpx_mock.add_response()
+
+    with httpx.Client() as client:
+        response = client.post(
+            "https://test_url",
+            content=body,
+            headers={"Content-Type": "application/octet-stream"},
+        )
+    assert httpx_mock.get_requests(match_content=body) == [response.request]
+    assert (
+        httpx_mock.get_requests(match_files={"upload": ("sample.txt", b"payload")})
+        == []
+    )
+
+
+@pytest.mark.asyncio
+async def test_binary_requests_retrieval_async(
+    httpx_mock: HTTPXMock, body: bytes
+) -> None:
+    httpx_mock.add_response()
+
+    async with httpx.AsyncClient() as client:
+        response = await client.post(
+            "https://test_url",
+            content=body,
+            headers={"Content-Type": "application/octet-stream"},
+        )
+    assert httpx_mock.get_requests(match_content=body) == [response.request]
+    assert (
+        httpx_mock.get_requests(match_files={"upload": ("sample.txt", b"payload")})
+        == []
+    )

tests/test_multipart_boundaries.py

@@ -0,0 +1,114 @@
+import httpx
+import pytest
+
+from pytest_httpx import HTTPXMock
+
+pytestmark = [
+    pytest.mark.parametrize(
+        "headers",
+        [
+            {},
+            {"Content-Type": "multipart/form-data; boundary=test-boundary"},
+            {"Content-Type": "multipart/form-data; boundary=ABCDEF012345"},
+            {
+                "Content-Type": 'multipart/form-data; boundary="test:boundary with space"'
+            },
+        ],
+        ids=["default", "custom", "uppercase", "quoted"],
+    ),
+    pytest.mark.parametrize("data", [{}, {"field": "value"}], ids=["files", "data"]),
+]
+
+
+def mismatched_bodies(data: dict[str, str]) -> list[dict]:
+    return [
+        {"match_files": {"other": ("file_name", b"File content")}, "match_data": data},
+        {"match_files": {"name": ("other", b"File content")}, "match_data": data},
+        {"match_files": {"name": ("file_name", b"other")}, "match_data": data},
+        {
+            "match_files": {"name": ("file_name", b"File content")},
+            "match_data": {"field": "other"},
+        },
+    ]
+
+
+def register_responses(httpx_mock: HTTPXMock, data: dict[str, str]) -> None:
+    for mismatch in mismatched_bodies(data):
+        httpx_mock.add_response(**mismatch, status_code=400, is_optional=True)
+    httpx_mock.add_response(
+        match_files={"name": ("file_name", b"File content")},
+        match_data=data,
+        status_code=201,
+    )
+
+
+def assert_requests(
+    httpx_mock: HTTPXMock, response: httpx.Response, data: dict[str, str]
+) -> None:
+    assert httpx_mock.get_requests(
+        match_files={"name": ("file_name", b"File content")}, match_data=data
+    ) == [response.request]
+    for mismatch in mismatched_bodies(data):
+        assert httpx_mock.get_requests(**mismatch) == []
+
+
+def test_response_matching(
+    httpx_mock: HTTPXMock, headers: dict[str, str], data: dict[str, str]
+) -> None:
+    register_responses(httpx_mock, data)
+
+    with httpx.Client() as client:
+        response = client.put(
+            "https://test_url",
+            headers=headers,
+            files={"name": ("file_name", b"File content")},
+            data=data,
+        )
+    assert response.status_code == 201
+
+
+@pytest.mark.asyncio
+async def test_response_matching_async(
+    httpx_mock: HTTPXMock, headers: dict[str, str], data: dict[str, str]
+) -> None:
+    register_responses(httpx_mock, data)
+
+    async with httpx.AsyncClient() as client:
+        response = await client.put(
+            "https://test_url",
+            headers=headers,
+            files={"name": ("file_name", b"File content")},
+            data=data,
+        )
+    assert response.status_code == 201
+
+
+def test_requests_retrieval(
+    httpx_mock: HTTPXMock, headers: dict[str, str], data: dict[str, str]
+) -> None:
+    httpx_mock.add_response()
+
+    with httpx.Client() as client:
+        response = client.put(
+            "https://test_url",
+            headers=headers,
+            files={"name": ("file_name", b"File content")},
+            data=data,
+        )
+    assert_requests(httpx_mock, response, data)
+
+
+@pytest.mark.asyncio
+async def test_requests_retrieval_async(
+    httpx_mock: HTTPXMock, headers: dict[str, str], data: dict[str, str]
+) -> None:
+    httpx_mock.add_response()
+
+    async with httpx.AsyncClient() as client:
+        response = await client.put(
+            "https://test_url",
+            headers=headers,
+            files={"name": ("file_name", b"File content")},
+            data=data,
+        )
+    assert_requests(httpx_mock, response, data)