getsentry / responses

A utility for mocking out the Python Requests library.
Apache License 2.0
4.08k stars 347 forks source link

Cannot match request with `data=filehandle` #719

Open jwodder opened 1 month ago

jwodder commented 1 month ago

Describe the bug

When mocking responses to a request where an open filehandle is passed as the data argument to the request function/method, the prepared request passed to the matcher has its body attribute set to the filehandle rather than to the filehandle's contents (as would be sent in an actual request).

Additional context

No response

Version of responses

0.25.0

Steps to Reproduce

Run pytest on the following file:

from __future__ import annotations
from pathlib import Path
import requests
import responses

@responses.activate
def test_post_file(tmp_path: Path) -> None:
    CONTENT = b"This is test text.\n"

    def match_body(req: requests.PreparedRequest) -> tuple[bool, str]:
        if req.body == CONTENT:
            return (True, "")
        else:
            return (False, f"Request body is not the expected content: {req.body!r}")

    responses.post(
        "http://example.nil/endpoint",
        status=200,
        json={"success": True},
        match=[match_body],
    )
    p = tmp_path / "foo.txt"
    p.write_bytes(CONTENT)
    with requests.Session() as s:
        with p.open("rb") as fp:
            assert s.post("http://example.nil/endpoint", data=fp) == {"success": True}

Expected Result

Matcher should successfully match the request body against the expected bytes, resulting in the test passing without error.

Actual Result

============================= test session starts ==============================
platform darwin -- Python 3.12.3, pytest-8.2.1, pluggy-1.5.0
rootdir: /Users/jwodder/work/dev/tmp/responses-bug
collected 1 item

test.py F                                                                [100%]

=================================== FAILURES ===================================
________________________________ test_post_file ________________________________

tmp_path = PosixPath('/private/var/folders/l7/wrkq93d133d8zpn36fmqrq0r0000gn/T/pytest-of-jwodder/pytest-91/test_post_file0')

    @responses.activate
    def test_post_file(tmp_path: Path) -> None:
        CONTENT = b"This is test text.\n"

        def match_body(req: requests.PreparedRequest) -> tuple[bool, str]:
            if req.body == CONTENT:
                return (True, "")
            else:
                return (False, f"Request body is not the expected content: {req.body!r}")

        responses.post(
            "http://example.nil/endpoint",
            status=200,
            json={"success": True},
            match=[match_body],
        )
        p = tmp_path / "foo.txt"
        p.write_bytes(CONTENT)
        with requests.Session() as s:
            with p.open("rb") as fp:
>               assert s.post("http://example.nil/endpoint", data=fp) == {"success": True}

test.py:27: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../../../.local/virtualenvwrapper/venvs/tmp-ca431dc82c2b8f1/lib/python3.12/site-packages/requests/sessions.py:637: in post
    return self.request("POST", url, data=data, json=json, **kwargs)
../../../../.local/virtualenvwrapper/venvs/tmp-ca431dc82c2b8f1/lib/python3.12/site-packages/requests/sessions.py:589: in request
    resp = self.send(prep, **send_kwargs)
../../../../.local/virtualenvwrapper/venvs/tmp-ca431dc82c2b8f1/lib/python3.12/site-packages/requests/sessions.py:703: in send
    r = adapter.send(request, **kwargs)
../../../../.local/virtualenvwrapper/venvs/tmp-ca431dc82c2b8f1/lib/python3.12/site-packages/responses/__init__.py:1173: in send
    return self._on_request(adapter, request, **kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <responses.RequestsMock object at 0x10b0ffad0>
adapter = <requests.adapters.HTTPAdapter object at 0x10b0dbef0>
request = <PreparedRequest [POST]>, retries = None
kwargs = {'cert': None, 'proxies': OrderedDict(), 'stream': False, 'timeout': None, ...}
request_url = 'http://example.nil/endpoint', match = None
match_failed_reasons = ["Request body is not the expected content: <_io.BufferedReader name='/private/var/folders/l7/wrkq93d133d8zpn36fmqrq0r0000gn/T/pytest-of-jwodder/pytest-91/test_post_file0/foo.txt'>"]
resp_callback = None
error_msg = "Connection refused by Responses - the call doesn't match any registered mock.\n\nRequest: \n- POST http://example.nil...name='/private/var/folders/l7/wrkq93d133d8zpn36fmqrq0r0000gn/T/pytest-of-jwodder/pytest-91/test_post_file0/foo.txt'>\n"

    def _on_request(
        self,
        adapter: "HTTPAdapter",
        request: "PreparedRequest",
        *,
        retries: Optional["_Retry"] = None,
        **kwargs: Any,
    ) -> "models.Response":
        # add attributes params and req_kwargs to 'request' object for further match comparison
        # original request object does not have these attributes
        request.params = self._parse_request_params(request.path_url)  # type: ignore[attr-defined]
        request.req_kwargs = kwargs  # type: ignore[attr-defined]
        request_url = str(request.url)

        match, match_failed_reasons = self._find_match(request)
        resp_callback = self.response_callback

        if match is None:
            if any(
                [
                    p.match(request_url)
                    if isinstance(p, Pattern)
                    else request_url.startswith(p)
                    for p in self.passthru_prefixes
                ]
            ):
                logger.info("request.allowed-passthru", extra={"url": request_url})
                return self._real_send(adapter, request, **kwargs)  # type: ignore

            error_msg = (
                "Connection refused by Responses - the call doesn't "
                "match any registered mock.\n\n"
                "Request: \n"
                f"- {request.method} {request_url}\n\n"
                "Available matches:\n"
            )
            for i, m in enumerate(self.registered()):
                error_msg += "- {} {} {}\n".format(
                    m.method, m.url, match_failed_reasons[i]
                )

            if self.passthru_prefixes:
                error_msg += "Passthru prefixes:\n"
                for p in self.passthru_prefixes:
                    error_msg += f"- {p}\n"

            response = ConnectionError(error_msg)
            response.request = request

            self._calls.add(request, response)
>           raise response
E           requests.exceptions.ConnectionError: Connection refused by Responses - the call doesn't match any registered mock.
E           
E           Request: 
E           - POST http://example.nil/endpoint
E           
E           Available matches:
E           - POST http://example.nil/endpoint Request body is not the expected content: <_io.BufferedReader name='/private/var/folders/l7/wrkq93d133d8zpn36fmqrq0r0000gn/T/pytest-of-jwodder/pytest-91/test_post_file0/foo.txt'>

../../../../.local/virtualenvwrapper/venvs/tmp-ca431dc82c2b8f1/lib/python3.12/site-packages/responses/__init__.py:1100: ConnectionError
=========================== short test summary info ============================
FAILED test.py::test_post_file - requests.exceptions.ConnectionError: Connect...
============================== 1 failed in 0.18s ===============================