diff --git a/CHANGES b/CHANGES index fcab1798..9fab64a4 100644 --- a/CHANGES +++ b/CHANGES @@ -11,6 +11,9 @@ Unreleased include the ``params`` and ``req_kwargs`` attributes added by ``responses``. * Fixed non-strict `json_params_matcher` raising `TypeError` when an actual nested object is compared with an expected scalar value. +* Fixed `json_params_matcher` raising `UnicodeDecodeError` or `BadGzipFile` + for a request body that is neither UTF-8 nor gzip (e.g. a binary upload) + instead of reporting a mismatch and letting other responses match. 0.26.3 ------ diff --git a/responses/matchers.py b/responses/matchers.py index 15df1884..1a1641e9 100644 --- a/responses/matchers.py +++ b/responses/matchers.py @@ -1,6 +1,7 @@ import gzip import json as json_module import re +import zlib from json.decoder import JSONDecodeError from typing import Any from typing import Callable @@ -167,6 +168,13 @@ def match(request: PreparedRequest) -> Tuple[bool, str]: reason = ( "request.body doesn't match: JSONDecodeError: Cannot parse request.body" ) + except (UnicodeDecodeError, OSError, EOFError, zlib.error): + # body is neither UTF-8 text nor gzip-compressed UTF-8 text + valid = False + reason = ( + "request.body doesn't match: Cannot decode request.body " + "as UTF-8 or gzip-compressed UTF-8" + ) return valid, reason diff --git a/responses/tests/test_matchers.py b/responses/tests/test_matchers.py index 4f67810c..8a3cad4a 100644 --- a/responses/tests/test_matchers.py +++ b/responses/tests/test_matchers.py @@ -272,6 +272,34 @@ def test_json_params_matcher_body_is_gzipped(): assert result == (True, "") +def test_json_params_matcher_body_not_utf8_or_gzip(): + @responses.activate + def run(): + url = "http://example.com/upload" + responses.post(url, match=[matchers.json_params_matcher({"name": "a.png"})]) + responses.post(url, body="uploaded") + + # a binary body that is neither UTF-8 nor gzip must be reported as a + # mismatch so the next registered response can match + resp = requests.post(url, data=b"\x89PNG\r\n\x1a\n\xff\x00") + assert_response(resp, "uploaded") + + run() + assert_reset() + + # gzip-compressed body whose content is not UTF-8 + mock_request = Mock(body=gzip.compress(b"\xff\xfe\x00")) + valid, reason = matchers.json_params_matcher({"foo": 42})(mock_request) + assert not valid + assert "Cannot decode request.body" in reason + + # gzip header followed by corrupt deflate data raises zlib.error + mock_request = Mock(body=b"\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\xff\xff\xff") + valid, reason = matchers.json_params_matcher({"foo": 42})(mock_request) + assert not valid + assert "Cannot decode request.body" in reason + + def test_urlencoded_params_matcher_blank(): @responses.activate def run():