diff --git a/CHANGELOG.md b/CHANGELOG.md index 3584e33..111a750 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/pytest_httpx/_request_matcher.py b/pytest_httpx/_request_matcher.py index b73a01f..1b4bbac 100644 --- a/pytest_httpx/_request_matcher.py +++ b/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 diff --git a/tests/test_multipart_binary.py b/tests/test_multipart_binary.py new file mode 100644 index 0000000..51edb8d --- /dev/null +++ b/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")}) + == [] + ) diff --git a/tests/test_multipart_boundaries.py b/tests/test_multipart_boundaries.py new file mode 100644 index 0000000..3920cf5 --- /dev/null +++ b/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)