responses 0.25.3

Record Responses to files

You can perform real requests to the server and responses will automatically record the output to the file. Recorded data is stored in YAML format.

Apply @responses._recorder.record(file_path="out.yaml") decorator to any function where you perform requests to record responses to out.yaml file.

Following code

import requests
from responses import _recorder


def another():
    rsp = requests.get("https://httpstat.us/500")
    rsp = requests.get("https://httpstat.us/202")


@_recorder.record(file_path="out.yaml")
def test_recorder():
    rsp = requests.get("https://httpstat.us/404")
    rsp = requests.get("https://httpbin.org/status/wrong")
    another()

will produce next output:

responses:
- response:
    auto_calculate_content_length: false
    body: 404 Not Found
    content_type: text/plain
    method: GET
    status: 404
    url: https://httpstat.us/404
- response:
    auto_calculate_content_length: false
    body: Invalid status code
    content_type: text/plain
    method: GET
    status: 400
    url: https://httpbin.org/status/wrong
- response:
    auto_calculate_content_length: false
    body: 500 Internal Server Error
    content_type: text/plain
    method: GET
    status: 500
    url: https://httpstat.us/500
- response:
    auto_calculate_content_length: false
    body: 202 Accepted
    content_type: text/plain
    method: GET
    status: 202
    url: https://httpstat.us/202

Replay responses (populate registry) from files

You can populate your active registry from a yaml file with recorded responses. (See Record Responses to files to understand how to obtain a file). To do that you need to execute responses._add_from_file(file_path="out.yaml") within an activated decorator or a context manager.

The following code example registers a patch response, then all responses present in out.yaml file and a post response at the end.

import responses


@responses.activate
def run():
    responses.patch("http://httpbin.org")
    responses._add_from_file(file_path="out.yaml")
    responses.post("http://httpbin.org/form")


run()

Main Interface

• responses.add(Response or Response args) - allows either to register Response object or directly provide arguments of Response object. See Response Parameters

import responses
import requests


@responses.activate
def test_simple():
    # Register via 'Response' object
    rsp1 = responses.Response(
        method="PUT",
        url="http://example.com",
    )
    responses.add(rsp1)
    # register via direct arguments
    responses.add(
        responses.GET,
        "http://twitter.com/api/1/foobar",
        json={"error": "not found"},
        status=404,
    )

    resp = requests.get("http://twitter.com/api/1/foobar")
    resp2 = requests.put("http://example.com")

    assert resp.json() == {"error": "not found"}
    assert resp.status_code == 404

    assert resp2.status_code == 200
    assert resp2.request.method == "PUT"

If you attempt to fetch a url which doesn’t hit a match, responses will raise a ConnectionError:

import responses
import requests

from requests.exceptions import ConnectionError


@responses.activate
def test_simple():
    with pytest.raises(ConnectionError):
        requests.get("http://twitter.com/api/1/foobar")

Shortcuts

Shortcuts provide a shorten version of responses.add() where method argument is prefilled

• responses.delete(Response args) - register DELETE response

• responses.get(Response args) - register GET response

• responses.head(Response args) - register HEAD response

• responses.options(Response args) - register OPTIONS response

• responses.patch(Response args) - register PATCH response

• responses.post(Response args) - register POST response

• responses.put(Response args) - register PUT response

import responses
import requests


@responses.activate
def test_simple():
    responses.get(
        "http://twitter.com/api/1/foobar",
        json={"type": "get"},
    )

    responses.post(
        "http://twitter.com/api/1/foobar",
        json={"type": "post"},
    )

    responses.patch(
        "http://twitter.com/api/1/foobar",
        json={"type": "patch"},
    )

    resp_get = requests.get("http://twitter.com/api/1/foobar")
    resp_post = requests.post("http://twitter.com/api/1/foobar")
    resp_patch = requests.patch("http://twitter.com/api/1/foobar")

    assert resp_get.json() == {"type": "get"}
    assert resp_post.json() == {"type": "post"}
    assert resp_patch.json() == {"type": "patch"}

Responses as a context manager

Instead of wrapping the whole function with decorator you can use a context manager.

import responses
import requests


def test_my_api():
    with responses.RequestsMock() as rsps:
        rsps.add(
            responses.GET,
            "http://twitter.com/api/1/foobar",
            body="{}",
            status=200,
            content_type="application/json",
        )
        resp = requests.get("http://twitter.com/api/1/foobar")

        assert resp.status_code == 200

    # outside the context manager requests will hit the remote server
    resp = requests.get("http://twitter.com/api/1/foobar")
    resp.status_code == 404

Matching Request Body Contents

When adding responses for endpoints that are sent request data you can add matchers to ensure your code is sending the right parameters and provide different responses based on the request body contents. responses provides matchers for JSON and URL-encoded request bodies.

import responses
import requests
from responses import matchers


@responses.activate
def test_calc_api():
    responses.post(
        url="http://calc.com/sum",
        body="4",
        match=[matchers.urlencoded_params_matcher({"left": "1", "right": "3"})],
    )
    requests.post("http://calc.com/sum", data={"left": 1, "right": 3})

import responses
import requests
from responses import matchers


@responses.activate
def test_calc_api():
    responses.post(
        url="http://example.com/",
        body="one",
        match=[
            matchers.json_params_matcher({"page": {"name": "first", "type": "json"}})
        ],
    )
    resp = requests.request(
        "POST",
        "http://example.com/",
        headers={"Content-Type": "application/json"},
        json={"page": {"name": "first", "type": "json"}},
    )

Query Parameters Matcher

import responses
import requests
from responses import matchers


@responses.activate
def test_calc_api():
    url = "http://example.com/test"
    params = {"hello": "world", "I am": "a big test"}
    responses.get(
        url=url,
        body="test",
        match=[matchers.query_param_matcher(params)],
    )

    resp = requests.get(url, params=params)

    constructed_url = r"http://example.com/test?I+am=a+big+test&hello=world"
    assert resp.url == constructed_url
    assert resp.request.url == constructed_url
    assert resp.request.params == params

import requests
import responses
from responses import matchers


@responses.activate
def my_func():
    responses.get(
        "https://httpbin.org/get",
        match=[matchers.query_string_matcher("didi=pro&test=1")],
    )
    resp = requests.get("https://httpbin.org/get", params={"test": 1, "didi": "pro"})


my_func()

Request Keyword Arguments Matcher

To validate request arguments use the matchers.request_kwargs_matcher function to match against the request kwargs.

Only following arguments are supported: timeout, verify, proxies, stream, cert.

Note, only arguments provided to matchers.request_kwargs_matcher will be validated.

import responses
import requests
from responses import matchers

with responses.RequestsMock(assert_all_requests_are_fired=False) as rsps:
    req_kwargs = {
        "stream": True,
        "verify": False,
    }
    rsps.add(
        "GET",
        "http://111.com",
        match=[matchers.request_kwargs_matcher(req_kwargs)],
    )

    requests.get("http://111.com", stream=True)

    # >>>  Arguments don't match: {stream: True, verify: True} doesn't match {stream: True, verify: False}

Request multipart/form-data Data Validation

To validate request body and headers for multipart/form-data data you can use matchers.multipart_matcher. The data, and files parameters provided will be compared to the request:

import requests
import responses
from responses.matchers import multipart_matcher


@responses.activate
def my_func():
    req_data = {"some": "other", "data": "fields"}
    req_files = {"file_name": b"Old World!"}
    responses.post(
        url="http://httpbin.org/post",
        match=[multipart_matcher(req_files, data=req_data)],
    )
    resp = requests.post("http://httpbin.org/post", files={"file_name": b"New World!"})


my_func()
# >>> raises ConnectionError: multipart/form-data doesn't match. Request body differs.

Request Fragment Identifier Validation

To validate request URL fragment identifier you can use matchers.fragment_identifier_matcher. The matcher takes fragment string (everything after # sign) as input for comparison:

import requests
import responses
from responses.matchers import fragment_identifier_matcher


@responses.activate
def run():
    url = "http://example.com?ab=xy&zed=qwe#test=1&foo=bar"
    responses.get(
        url,
        match=[fragment_identifier_matcher("test=1&foo=bar")],
        body=b"test",
    )

    # two requests to check reversed order of fragment identifier
    resp = requests.get("http://example.com?ab=xy&zed=qwe#test=1&foo=bar")
    resp = requests.get("http://example.com?zed=qwe&ab=xy#foo=bar&test=1")


run()

Request Headers Validation

When adding responses you can specify matchers to ensure that your code is sending the right headers and provide different responses based on the request headers.

import responses
import requests
from responses import matchers


@responses.activate
def test_content_type():
    responses.get(
        url="http://example.com/",
        body="hello world",
        match=[matchers.header_matcher({"Accept": "text/plain"})],
    )

    responses.get(
        url="http://example.com/",
        json={"content": "hello world"},
        match=[matchers.header_matcher({"Accept": "application/json"})],
    )

    # request in reverse order to how they were added!
    resp = requests.get("http://example.com/", headers={"Accept": "application/json"})
    assert resp.json() == {"content": "hello world"}

    resp = requests.get("http://example.com/", headers={"Accept": "text/plain"})
    assert resp.text == "hello world"

Because requests will send several standard headers in addition to what was specified by your code, request headers that are additional to the ones passed to the matcher are ignored by default. You can change this behaviour by passing strict_match=True to the matcher to ensure that only the headers that you’re expecting are sent and no others. Note that you will probably have to use a PreparedRequest in your code to ensure that requests doesn’t include any additional headers.

import responses
import requests
from responses import matchers


@responses.activate
def test_content_type():
    responses.get(
        url="http://example.com/",
        body="hello world",
        match=[matchers.header_matcher({"Accept": "text/plain"}, strict_match=True)],
    )

    # this will fail because requests adds its own headers
    with pytest.raises(ConnectionError):
        requests.get("http://example.com/", headers={"Accept": "text/plain"})

    # a prepared request where you overwrite the headers before sending will work
    session = requests.Session()
    prepped = session.prepare_request(
        requests.Request(
            method="GET",
            url="http://example.com/",
        )
    )
    prepped.headers = {"Accept": "text/plain"}

    resp = session.send(prepped)
    assert resp.text == "hello world"

Creating Custom Matcher

If your application requires other encodings or different data validation you can build your own matcher that returns Tuple[matches: bool, reason: str]. Where boolean represents True or False if the request parameters match and the string is a reason in case of match failure. Your matcher can expect a PreparedRequest parameter to be provided by responses.

Note, PreparedRequest is customized and has additional attributes params and req_kwargs.

Default Registry

By default, responses will search all registered Response objects and return a match. If only one Response is registered, the registry is kept unchanged. However, if multiple matches are found for the same request, then first match is returned and removed from registry.

Ordered Registry

In some scenarios it is important to preserve the order of the requests and responses. You can use registries.OrderedRegistry to force all Response objects to be dependent on the insertion order and invocation index. In following example we add multiple Response objects that target the same URL. However, you can see, that status code will depend on the invocation order.

import requests

import responses
from responses.registries import OrderedRegistry


@responses.activate(registry=OrderedRegistry)
def test_invocation_index():
    responses.get(
        "http://twitter.com/api/1/foobar",
        json={"msg": "not found"},
        status=404,
    )
    responses.get(
        "http://twitter.com/api/1/foobar",
        json={"msg": "OK"},
        status=200,
    )
    responses.get(
        "http://twitter.com/api/1/foobar",
        json={"msg": "OK"},
        status=200,
    )
    responses.get(
        "http://twitter.com/api/1/foobar",
        json={"msg": "not found"},
        status=404,
    )

    resp = requests.get("http://twitter.com/api/1/foobar")
    assert resp.status_code == 404
    resp = requests.get("http://twitter.com/api/1/foobar")
    assert resp.status_code == 200
    resp = requests.get("http://twitter.com/api/1/foobar")
    assert resp.status_code == 200
    resp = requests.get("http://twitter.com/api/1/foobar")
    assert resp.status_code == 404

Custom Registry

Built-in registries are suitable for most of use cases, but to handle special conditions, you can implement custom registry which must follow interface of registries.FirstMatchRegistry. Redefining the find method will allow you to create custom search logic and return appropriate Response

Example that shows how to set custom registry

import responses
from responses import registries


class CustomRegistry(registries.FirstMatchRegistry):
    pass


print("Before tests:", responses.mock.get_registry())
""" Before tests: <responses.registries.FirstMatchRegistry object> """


# using function decorator
@responses.activate(registry=CustomRegistry)
def run():
    print("Within test:", responses.mock.get_registry())
    """ Within test: <__main__.CustomRegistry object> """


run()

print("After test:", responses.mock.get_registry())
""" After test: <responses.registries.FirstMatchRegistry object> """

# using context manager
with responses.RequestsMock(registry=CustomRegistry) as rsps:
    print("In context manager:", rsps.get_registry())
    """ In context manager: <__main__.CustomRegistry object> """

print("After exit from context manager:", responses.mock.get_registry())
"""
After exit from context manager: <responses.registries.FirstMatchRegistry object>
"""

Responses as a pytest fixture

@pytest.fixture
def mocked_responses():
    with responses.RequestsMock() as rsps:
        yield rsps


def test_api(mocked_responses):
    mocked_responses.get(
        "http://twitter.com/api/1/foobar",
        body="{}",
        status=200,
        content_type="application/json",
    )
    resp = requests.get("http://twitter.com/api/1/foobar")
    assert resp.status_code == 200

Add default responses for each test

When run with unittest tests, this can be used to set up some generic class-level responses, that may be complemented by each test. Similar interface could be applied in pytest framework.

class TestMyApi(unittest.TestCase):
    def setUp(self):
        responses.get("https://example.com", body="within setup")
        # here go other self.responses.add(...)

    @responses.activate
    def test_my_func(self):
        responses.get(
            "https://httpbin.org/get",
            match=[matchers.query_param_matcher({"test": "1", "didi": "pro"})],
            body="within test",
        )
        resp = requests.get("https://example.com")
        resp2 = requests.get(
            "https://httpbin.org/get", params={"test": "1", "didi": "pro"}
        )
        print(resp.text)
        # >>> within setup
        print(resp2.text)
        # >>> within test

RequestMock methods: start, stop, reset

responses has start, stop, reset methods very analogous to unittest.mock.patch. These make it simpler to do requests mocking in setup methods or where you want to do multiple patches without nesting decorators or with statements.

class TestUnitTestPatchSetup:
    def setup(self):
        """Creates ``RequestsMock`` instance and starts it."""
        self.r_mock = responses.RequestsMock(assert_all_requests_are_fired=True)
        self.r_mock.start()

        # optionally some default responses could be registered
        self.r_mock.get("https://example.com", status=505)
        self.r_mock.put("https://example.com", status=506)

    def teardown(self):
        """Stops and resets RequestsMock instance.

        If ``assert_all_requests_are_fired`` is set to ``True``, will raise an error
        if some requests were not processed.
        """
        self.r_mock.stop()
        self.r_mock.reset()

    def test_function(self):
        resp = requests.get("https://example.com")
        assert resp.status_code == 505

        resp = requests.put("https://example.com")
        assert resp.status_code == 506

Assert based on Response object

Each Response object has call_count attribute that could be inspected to check how many times each request was matched.

@responses.activate
def test_call_count_with_matcher():
    rsp = responses.get(
        "http://www.example.com",
        match=(matchers.query_param_matcher({}),),
    )
    rsp2 = responses.get(
        "http://www.example.com",
        match=(matchers.query_param_matcher({"hello": "world"}),),
        status=777,
    )
    requests.get("http://www.example.com")
    resp1 = requests.get("http://www.example.com")
    requests.get("http://www.example.com?hello=world")
    resp2 = requests.get("http://www.example.com?hello=world")

    assert resp1.status_code == 200
    assert resp2.status_code == 777

    assert rsp.call_count == 2
    assert rsp2.call_count == 2

Assert based on the exact URL

Assert that the request was called exactly n times.

import responses
import requests


@responses.activate
def test_assert_call_count():
    responses.get("http://example.com")

    requests.get("http://example.com")
    assert responses.assert_call_count("http://example.com", 1) is True

    requests.get("http://example.com")
    with pytest.raises(AssertionError) as excinfo:
        responses.assert_call_count("http://example.com", 1)
    assert (
        "Expected URL 'http://example.com' to be called 1 times. Called 2 times."
        in str(excinfo.value)
    )


@responses.activate
def test_assert_call_count_always_match_qs():
    responses.get("http://www.example.com")
    requests.get("http://www.example.com")
    requests.get("http://www.example.com?hello=world")

    # One call on each url, querystring is matched by default
    responses.assert_call_count("http://www.example.com", 1) is True
    responses.assert_call_count("http://www.example.com?hello=world", 1) is True

Environment Configuration

Responses uses several linting and autoformatting utilities, so it’s important that when submitting patches you use the appropriate toolchain:

Clone the repository:

git clone https://github.com/getsentry/responses.git

Create an environment (e.g. with virtualenv):

virtualenv .env && source .env/bin/activate

Configure development requirements:

make develop

Tests and Code Quality Validation

The easiest way to validate your code is to run tests via tox. Current tox configuration runs the same checks that are used in GitHub Actions CI/CD pipeline.

Please execute the following command line from the project root to validate your code against:

• Unit tests in all Python versions that are supported by this project

• Type validation via mypy

• All pre-commit hooks

tox

Alternatively, you can always run a single test. See documentation below.

tox -e py37
tox -e py310

pytest

pytest -k '<test_function_name>'

tox -e mypy

mypy --config-file=./mypy.ini -p responses

tox -e precom

pre-commit run --all-files