Skip to content

Latest commit

 

History

History
1560 lines (1110 loc) · 47.8 KB

File metadata and controls

1560 lines (1110 loc) · 47.8 KB

Responses

https://img.shields.io/pypi/dm/responses

Una biblioteca utilitaria para simular (mock) la biblioteca Python requests.

Note

Responses requiere Python 3.8 o superior, y requests >= 2.30.0

Contents

pip install responses

Aquí encontrarás una lista de funcionalidades deprecadas y la ruta de migración para cada una. Asegúrate de actualizar tu código según las indicaciones.

Deprecaciones y Migración
Funcionalidad Deprecada Deprecada en Versión Ruta de Migración
responses.json_params_matcher 0.14.0 responses.matchers.json_params_matcher
responses.urlencoded_params_matcher 0.14.0 responses.matchers.urlencoded_params_matcher
argumento stream en Response y CallbackResponse 0.15.0 Usa el argumento stream directamente en la solicitud.
argumento match_querystring en Response y CallbackResponse. 0.17.0 Usa responses.matchers.query_param_matcher o responses.matchers.query_string_matcher
responses.assert_all_requests_are_fired, responses.passthru_prefixes, responses.target 0.20.0 Usa responses.mock.assert_all_requests_are_fired, responses.mock.passthru_prefixes, responses.mock.target en su lugar.

El núcleo de responses consiste en registrar respuestas simuladas y envolver la función de prueba con el decorador responses.activate. responses ofrece una interfaz similar a requests.

  • responses.add(Response o argumentos de Response) — permite registrar un objeto Response o proporcionar directamente los argumentos del objeto Response. Ver Parámetros de Response
import responses
import requests


@responses.activate
def test_simple():
    # Registro mediante objeto 'Response'
    rsp1 = responses.Response(
        method="PUT",
        url="http://example.com",
    )
    responses.add(rsp1)
    # Registro mediante argumentos directos
    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"

Si intentas acceder a una URL que no coincide con ninguna registrada, responses lanzará un 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")

Los atajos ofrecen una versión abreviada de responses.add() donde el argumento del método ya viene predefinido.

  • responses.delete(argumentos de Response) — registra una respuesta DELETE
  • responses.get(argumentos de Response) — registra una respuesta GET
  • responses.head(argumentos de Response) — registra una respuesta HEAD
  • responses.options(argumentos de Response) — registra una respuesta OPTIONS
  • responses.patch(argumentos de Response) — registra una respuesta PATCH
  • responses.post(argumentos de Response) — registra una respuesta POST
  • responses.put(argumentos de Response) — registra una respuesta PUT
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"}

En lugar de envolver toda la función con un decorador, puedes usar un 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

    # fuera del context manager, las solicitudes llegan al servidor real
    resp = requests.get("http://twitter.com/api/1/foobar")
    resp.status_code == 404

Los siguientes atributos pueden pasarse a un mock de Response:

method (str)
El método HTTP (GET, POST, etc.).
url (str o expresión regular compilada)
La URL completa del recurso.
match_querystring (bool)

DEPRECADO: Usa responses.matchers.query_param_matcher o responses.matchers.query_string_matcher

Incluye la cadena de consulta (query string) al comparar solicitudes. Activado por defecto si la URL de respuesta contiene una cadena de consulta; desactivado si no la contiene o si la URL es una expresión regular.

body (str o BufferedReader o Exception)
El cuerpo de la respuesta. Lee más en Excepción como cuerpo de Response
json
Un objeto Python que representa el cuerpo de la respuesta en formato JSON. Configura automáticamente el Content-Type apropiado.
status (int)
El código de estado HTTP.
content_type (content_type)
Por defecto es text/plain.
headers (dict)
Cabeceras de la respuesta.
stream (bool)
DEPRECADO: usa el argumento stream directamente en la solicitud.
auto_calculate_content_length (bool)
Desactivado por defecto. Calcula automáticamente la longitud de un cuerpo de tipo cadena o JSON.
match (tuple)

Un iterable (se recomienda tuple) de callbacks para comparar solicitudes en función de sus atributos. El módulo proporciona múltiples comparadores que puedes utilizar para verificar:

  • contenido del cuerpo en formato JSON
  • contenido del cuerpo en formato URL-encoded
  • parámetros de consulta de la solicitud
  • cadena de consulta de la solicitud (similar a los parámetros de consulta, pero acepta una cadena como entrada)
  • kwargs proporcionados a la solicitud, por ejemplo stream, verify
  • contenido y cabeceras de tipo multipart/form-data en la solicitud
  • cabeceras de la solicitud
  • identificador de fragmento de la solicitud

El usuario también puede crear un comparador personalizado. Lee más en Comparación de Solicitudes

Puedes pasar una Exception como cuerpo para provocar un error en la solicitud:

import responses
import requests


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

Al agregar respuestas para endpoints que reciben datos en la solicitud, puedes añadir comparadores para asegurarte de que tu código envía los parámetros correctos y para ofrecer respuestas diferentes según el contenido del cuerpo. responses ofrece comparadores para cuerpos de solicitud en formato JSON y URL-encoded.

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})

La comparación de datos codificados en JSON se realiza con matchers.json_params_matcher().

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"}},
    )

Puedes usar la función matchers.query_param_matcher para comparar contra el parámetro params de la solicitud. Utiliza el mismo diccionario que usarías en el argumento params de request.

Nota: no incluyas los parámetros de consulta como parte de la URL. Evita usar el argumento deprecado match_querystring.

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

Por defecto, el comparador valida que todos los parámetros coincidan estrictamente. Para validar que solo los parámetros especificados en el comparador estén presentes en la solicitud original, usa strict_match=False.

Como alternativa, puedes usar el valor de la cadena de consulta en matchers.query_string_matcher para comparar los parámetros de consulta de tu solicitud.

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()

Para validar los argumentos de la solicitud, usa la función matchers.request_kwargs_matcher para comparar contra los kwargs de la solicitud.

Solo se admiten los siguientes argumentos: timeout, verify, proxies, stream, cert.

Nota: solo se validarán los argumentos proporcionados a matchers.request_kwargs_matcher.

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)

    # >>>  Los argumentos no coinciden: {stream: True, verify: True} no coinciden con {stream: True, verify: False}

Para validar el cuerpo y las cabeceras de una solicitud con datos multipart/form-data, puedes usar matchers.multipart_matcher. Los parámetros data y files proporcionados se compararán con la solicitud:

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()
# >>> genera ConnectionError: multipart/form-data no coincide. El cuerpo (body) del request es diferente

Para validar el identificador de fragmento de la URL de la solicitud, puedes usar matchers.fragment_identifier_matcher. El comparador toma como entrada la cadena de fragmento (todo lo que aparece después del signo #):

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",
    )

    # dos solicitudes para verificar el orden inverso del identificador de fragmento
    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()

Al agregar respuestas puedes especificar comparadores para asegurarte de que tu código envía las cabeceras correctas y para ofrecer respuestas diferentes según las cabeceras de la solicitud.

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"})],
    )

    # ¡solicitudes en orden inverso al que fueron agregadas!
    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"

Dado que requests enviará varias cabeceras estándar además de las especificadas por tu código, las cabeceras adicionales a las pasadas al comparador se ignoran por defecto. Puedes cambiar este comportamiento pasando strict_match=True al comparador para asegurarte de que solo se envíen exactamente las cabeceras esperadas. Ten en cuenta que probablemente necesitarás usar un PreparedRequest en tu código para evitar que requests incluya cabeceras adicionales.

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)],
    )

    # esto fallará porque requests agrega sus propias cabeceras
    with pytest.raises(ConnectionError):
        requests.get("http://example.com/", headers={"Accept": "text/plain"})

    # una solicitud preparada donde se sobreescriben las cabeceras antes del envío sí funcionará
    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"

Si tu aplicación requiere otras codificaciones o una validación de datos diferente, puedes crear tu propio comparador que devuelva Tuple[matches: bool, reason: str]. El booleano indica True o False según si los parámetros de la solicitud coinciden, y la cadena contiene la razón en caso de fallo. Tu comparador puede esperar recibir un parámetro PreparedRequest proporcionado por responses.

Nota: PreparedRequest está personalizado y tiene atributos adicionales params y req_kwargs.

Por defecto, responses buscará entre todos los objetos Response registrados y devolverá la primera coincidencia. Si solo hay un Response registrado, el registro permanece sin cambios. Sin embargo, si se encuentran múltiples coincidencias para la misma solicitud, se devuelve la primera coincidencia y se elimina del registro.

En algunos escenarios es importante preservar el orden de las solicitudes y respuestas. Puedes usar registries.OrderedRegistry para forzar que todos los objetos Response dependan del orden de inserción y del índice de invocación. En el siguiente ejemplo se agregan múltiples objetos Response que apuntan a la misma URL. Sin embargo, verás que el código de estado depende del orden de invocación.

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

Los registries incluidos son adecuados para la mayoría de los casos de uso, pero para manejar condiciones especiales puedes implementar un registro personalizado que siga la interfaz de registries.FirstMatchRegistry. Redefinir el método find te permitirá crear una lógica de búsqueda personalizada y devolver el Response apropiado.

Ejemplo que muestra cómo establecer un registro personalizado:

import responses
from responses import registries


class CustomRegistry(registries.FirstMatchRegistry):
    pass


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


# usando decorador de función
@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> """

# usando 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>
"""

Puedes usar callbacks para proporcionar respuestas dinámicas. El callback debe devolver una tupla de (status, headers, body).

import json

import responses
import requests


@responses.activate
def test_calc_api():
    def request_callback(request):
        payload = json.loads(request.body)
        resp_body = {"value": sum(payload["numbers"])}
        headers = {"request-id": "728d329e-0e86-11e4-a748-0c84dc037c13"}
        return (200, headers, json.dumps(resp_body))

    responses.add_callback(
        responses.POST,
        "http://calc.com/sum",
        callback=request_callback,
        content_type="application/json",
    )

    resp = requests.post(
        "http://calc.com/sum",
        json.dumps({"numbers": [1, 2, 3]}),
        headers={"content-type": "application/json"},
    )

    assert resp.json() == {"value": 6}

    assert len(responses.calls) == 1
    assert responses.calls[0].request.url == "http://calc.com/sum"
    assert responses.calls[0].response.text == '{"value": 6}'
    assert (
        responses.calls[0].response.headers["request-id"]
        == "728d329e-0e86-11e4-a748-0c84dc037c13"
    )

También puedes pasar una expresión regular compilada a add_callback para que coincida con múltiples URLs:

import re, json

from functools import reduce

import responses
import requests

operators = {
    "sum": lambda x, y: x + y,
    "prod": lambda x, y: x * y,
    "pow": lambda x, y: x**y,
}


@responses.activate
def test_regex_url():
    def request_callback(request):
        payload = json.loads(request.body)
        operator_name = request.path_url[1:]

        operator = operators[operator_name]

        resp_body = {"value": reduce(operator, payload["numbers"])}
        headers = {"request-id": "728d329e-0e86-11e4-a748-0c84dc037c13"}
        return (200, headers, json.dumps(resp_body))

    responses.add_callback(
        responses.POST,
        re.compile("http://calc.com/(sum|prod|pow|unsupported)"),
        callback=request_callback,
        content_type="application/json",
    )

    resp = requests.post(
        "http://calc.com/prod",
        json.dumps({"numbers": [2, 3, 4]}),
        headers={"content-type": "application/json"},
    )
    assert resp.json() == {"value": 24}


test_regex_url()

Si quieres pasar argumentos adicionales al callback, por ejemplo para reutilizar una función callback con un resultado ligeramente diferente, puedes usar functools.partial:

from functools import partial


def request_callback(request, id=None):
    payload = json.loads(request.body)
    resp_body = {"value": sum(payload["numbers"])}
    headers = {"request-id": id}
    return (200, headers, json.dumps(resp_body))


responses.add_callback(
    responses.POST,
    "http://calc.com/sum",
    callback=partial(request_callback, id="728d329e-0e86-11e4-a748-0c84dc037c13"),
    content_type="application/json",
)

Usa el paquete pytest-responses para exportar responses como una fixture de pytest.

pip install pytest-responses

Luego puedes acceder a ella en un script de pytest de la siguiente manera:

import pytest_responses


def test_api(responses):
    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

Al ejecutar con pruebas de unittest, responses puede usarse para definir respuestas genéricas a nivel de clase que cada prueba puede complementar. Se puede aplicar una interfaz similar en el framework pytest.

class TestMyApi(unittest.TestCase):
    def setUp(self):
        responses.get("https://example.com", body="within setup")
        # aquí van otros 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

responses tiene los métodos start, stop, reset, muy análogos a unittest.mock.patch. Esto simplifica el uso de mocks de solicitudes en métodos setup o cuando se quieren aplicar múltiples parches sin anidar decoradores ni sentencias with.

class TestUnitTestPatchSetup:
    def setup(self):
        """Crea una instancia de ``RequestsMock`` y la inicia."""
        self.r_mock = responses.RequestsMock(assert_all_requests_are_fired=True)
        self.r_mock.start()

        # opcionalmente se pueden registrar algunas respuestas por defecto
        self.r_mock.get("https://example.com", status=505)
        self.r_mock.put("https://example.com", status=506)

    def teardown(self):
        """Detiene y reinicia la instancia de RequestsMock.

        Si ``assert_all_requests_are_fired`` está en ``True``, se lanzará un error
        si algunas solicitudes no fueron procesadas.
        """
        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

Cuando se usa como context manager, Responses lanzará por defecto un error de aserción si una URL fue registrada pero no fue accedida. Esto puede desactivarse pasando el valor assert_all_requests_are_fired:

import responses
import requests


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

Cuando assert_all_requests_are_fired=True y ocurre una excepción dentro del gestor de contexto, las aserciones sobre las solicitudes no ejecutadas se lanzarán igualmente. Esto proporciona contexto valioso sobre qué solicitudes simuladas fueron o no invocadas al depurar fallos en las pruebas.

import responses
import requests


def test_with_exception():
    with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
        rsps.add(responses.GET, "http://example.com/users", body="test")
        rsps.add(responses.GET, "http://example.com/profile", body="test")
        requests.get("http://example.com/users")
        raise ValueError("Something went wrong")

    # Salida:
    # ValueError: Something went wrong
    #
    # During handling of the above exception, another exception occurred:
    #
    # AssertionError: Not all requests have been executed [('GET', 'http://example.com/profile')]

Cada objeto Response tiene el atributo call_count que puede inspeccionarse para comprobar cuántas veces fue invocada cada solicitud.

@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

Verifica que la solicitud fue invocada exactamente n veces.

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")

    # Una llamada por cada URL; la cadena de consulta se compara por defecto
    responses.assert_call_count("http://www.example.com", 1) is True
    responses.assert_call_count("http://www.example.com?hello=world", 1) is True

El objeto Request tiene una lista calls cuyos elementos corresponden a objetos Call en la lista global del Registry. Esto puede ser útil cuando el orden de las solicitudes no está garantizado pero necesitas verificar su corrección, por ejemplo en aplicaciones multihilo.

import concurrent.futures
import responses
import requests


@responses.activate
def test_assert_calls_on_resp():
    rsp1 = responses.patch("http://www.foo.bar/1/", status=200)
    rsp2 = responses.patch("http://www.foo.bar/2/", status=400)
    rsp3 = responses.patch("http://www.foo.bar/3/", status=200)

    def update_user(uid, is_active):
        url = f"http://www.foo.bar/{uid}/"
        response = requests.patch(url, json={"is_active": is_active})
        return response

    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        future_to_uid = {
            executor.submit(update_user, uid, is_active): uid
            for (uid, is_active) in [("3", True), ("2", True), ("1", False)]
        }
        for future in concurrent.futures.as_completed(future_to_uid):
            uid = future_to_uid[future]
            response = future.result()
            print(f"{uid} updated with {response.status_code} status code")

    assert len(responses.calls) == 3  # total de llamadas

    assert rsp1.call_count == 1
    assert rsp1.calls[0] in responses.calls
    assert rsp1.calls[0].response.status_code == 200
    assert json.loads(rsp1.calls[0].request.body) == {"is_active": False}

    assert rsp2.call_count == 1
    assert rsp2.calls[0] in responses.calls
    assert rsp2.calls[0].response.status_code == 400
    assert json.loads(rsp2.calls[0].request.body) == {"is_active": True}

    assert rsp3.call_count == 1
    assert rsp3.calls[0] in responses.calls
    assert rsp3.calls[0].response.status_code == 200
    assert json.loads(rsp3.calls[0].request.body) == {"is_active": True}

También puedes agregar múltiples respuestas para la misma URL:

import responses
import requests


@responses.activate
def test_my_api():
    responses.get("http://twitter.com/api/1/foobar", status=500)
    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 == 500
    resp = requests.get("http://twitter.com/api/1/foobar")
    assert resp.status_code == 200

En el siguiente ejemplo puedes ver cómo crear una cadena de redirección y agregar una excepción personalizada que se lanzará durante la ejecución y contendrá el historial de redirecciones.

A -> 301 redirect -> B
B -> 301 redirect -> C
C -> connection issue
import pytest
import requests

import responses


@responses.activate
def test_redirect():
    # crea múltiples objetos Response donde los dos primeros contienen cabeceras de redirección
    rsp1 = responses.Response(
        responses.GET,
        "http://example.com/1",
        status=301,
        headers={"Location": "http://example.com/2"},
    )
    rsp2 = responses.Response(
        responses.GET,
        "http://example.com/2",
        status=301,
        headers={"Location": "http://example.com/3"},
    )
    rsp3 = responses.Response(responses.GET, "http://example.com/3", status=200)

    # registra los objetos Response generados en el módulo ``responses``
    responses.add(rsp1)
    responses.add(rsp2)
    responses.add(rsp3)

    # realiza la primera solicitud para generar una respuesta genuina de ``requests``
    # este objeto contendrá atributos reales de la respuesta, como ``history``
    rsp = requests.get("http://example.com/1")
    responses.calls.reset()

    # personaliza la excepción con el atributo ``response``
    my_error = requests.ConnectionError("custom error")
    my_error.response = rsp

    # actualiza el cuerpo de la 3ª respuesta con una excepción; esta se lanzará durante la ejecución
    rsp3.body = my_error

    with pytest.raises(requests.ConnectionError) as exc_info:
        requests.get("http://example.com/1")

    assert exc_info.value.args[0] == "custom error"
    assert rsp1.url in exc_info.value.response.history[0].url
    assert rsp2.url in exc_info.value.response.history[1].url

Si usas las características de Retry de urllib3 y quieres cubrir escenarios que pongan a prueba tus límites de reintentos, también puedes hacerlo con responses. El mejor enfoque es usar un Registro Ordenado.

import requests

import responses
from responses import registries
from urllib3.util import Retry


@responses.activate(registry=registries.OrderedRegistry)
def test_max_retries():
    url = "https://example.com"
    rsp1 = responses.get(url, body="Error", status=500)
    rsp2 = responses.get(url, body="Error", status=500)
    rsp3 = responses.get(url, body="Error", status=500)
    rsp4 = responses.get(url, body="OK", status=200)

    session = requests.Session()

    adapter = requests.adapters.HTTPAdapter(
        max_retries=Retry(
            total=4,
            backoff_factor=0.1,
            status_forcelist=[500],
            method_whitelist=["GET", "POST", "PATCH"],
        )
    )
    session.mount("https://", adapter)

    resp = session.get(url)

    assert resp.status_code == 200
    assert rsp1.call_count == 1
    assert rsp2.call_count == 1
    assert rsp3.call_count == 1
    assert rsp4.call_count == 1

Si usas procesamiento personalizado en requests mediante subclases o mixins, o si tienes herramientas que interactúan con requests a bajo nivel, puede que necesites agregar procesamiento extendido al objeto Response simulado para simular completamente el entorno de tus pruebas. Se puede usar un response_callback, que será envuelto antes de devolverse al invocador. El callback acepta una response como único argumento y se espera que devuelva un único objeto response.

import responses
import requests


def response_callback(resp):
    resp.callback_processed = True
    return resp


with responses.RequestsMock(response_callback=response_callback) as m:
    m.add(responses.GET, "http://example.com", body=b"test")
    resp = requests.get("http://example.com")
    assert resp.text == "test"
    assert hasattr(resp, "callback_processed")
    assert resp.callback_processed is True

En algunos casos puede ser necesario permitir que ciertas solicitudes pasen a través de responses y lleguen a un servidor real. Esto se puede hacer con los métodos add_passthru:

import responses


@responses.activate
def test_my_api():
    responses.add_passthru("https://percy.io")

Esto permitirá que cualquier solicitud que coincida con ese prefijo, y que no esté registrada como respuesta simulada, pase usando el comportamiento estándar.

Los endpoints de paso pueden configurarse con patrones de expresión regular si necesitas permitir que todo un dominio o subárbol de ruta envíe solicitudes:

responses.add_passthru(re.compile("https://percy.io/\\w+"))

Por último, puedes usar el argumento passthrough del objeto Response para forzar que una respuesta se comporte como paso directo.

# Habilitar passthrough para una sola respuesta
response = Response(
    responses.GET,
    "http://example.com",
    body="not used",
    passthrough=True,
)
responses.add(response)

# Usar PassthroughResponse
response = PassthroughResponse(responses.GET, "http://example.com")
responses.add(response)

Las responses registradas están disponibles como método público de la instancia RequestMock. A veces es útil para depuración ver la pila de responses registradas, a la que se puede acceder mediante responses.registered().

La función replace permite modificar una response previamente registrada. La firma del método es idéntica a add. Las response s se identifican por method y url. Solo se reemplaza la primera response que coincide.

import responses
import requests


@responses.activate
def test_replace():
    responses.get("http://example.org", json={"data": 1})
    responses.replace(responses.GET, "http://example.org", json={"data": 2})

    resp = requests.get("http://example.org")

    assert resp.json() == {"data": 2}

La función upsert permite modificar una response previamente registrada al igual que replace. Si la response no está registrada, la función upsert la registrará como add.

remove acepta un argumento method y url y eliminará todas las responses coincidentes de la lista de registradas.

Por último, reset reiniciará todas las responses registradas.

responses admite tanto corrutinas como multihilo de forma nativa. Ten en cuenta que responses bloquea el hilo en el objeto RequestMock, permitiendo que solo un hilo acceda a él a la vez.

async def test_async_calls():
    @responses.activate
    async def run():
        responses.get(
            "http://twitter.com/api/1/foobar",
            json={"error": "not found"},
            status=404,
        )

        resp = requests.get("http://twitter.com/api/1/foobar")
        assert resp.json() == {"error": "not found"}
        assert responses.calls[0].request.url == "http://twitter.com/api/1/foobar"

    await run()

A continuación encontrarás una lista de funcionalidades BETA. Aunque intentaremos mantener la compatibilidad con versiones anteriores de la API con la versión publicada, nos reservamos el derecho de cambiar estas APIs antes de que sean consideradas estables. Comparte tu opinión a través de GitHub Issues.

Puedes realizar solicitudes reales al servidor y responses grabará automáticamente la salida en un archivo. Los datos grabados se almacenan en formato YAML.

Aplica el decorador @responses._recorder.record(file_path="out.yaml") a cualquier función donde realices solicitudes para grabar las responses en el archivo out.yaml.

El siguiente código:

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()

producirá la siguiente salida:

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

Si estás en el REPL, también puedes activar el grabador para todas las responses siguientes:

import requests
from responses import _recorder

_recorder.recorder.start()

requests.get("https://httpstat.us/500")

_recorder.recorder.dump_to_file("out.yaml")

# puedes detener o reiniciar el grabador
_recorder.recorder.stop()
_recorder.recorder.reset()

Puedes poblar tu registro activo desde un archivo yaml con responses grabadas. (Consulta Grabar Responses en Archivos para entender cómo obtener el archivo.) Para ello necesitas ejecutar responses._add_from_file(file_path="out.yaml") dentro de un decorador activado o un context manager.

El siguiente ejemplo registra una respuesta patch, luego todas las responses presentes en el archivo out.yaml y finalmente una respuesta post.

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()

Responses usa varias utilidades de linting y autoformateo, por lo que es importante que al enviar parches utilices la cadena de herramientas adecuada:

Clona el repositorio:

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

Crea un entorno (por ejemplo con virtualenv):

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

Configura los requisitos de desarrollo:

make develop

La forma más sencilla de validar tu código es ejecutar las pruebas mediante tox. La configuración actual de tox ejecuta las mismas verificaciones que se usan en el pipeline de CI/CD de GitHub Actions.

Ejecuta el siguiente comando desde la raíz del proyecto para validar tu código:

  • Pruebas unitarias en todas las versiones de Python admitidas por este proyecto
  • Validación de tipos mediante mypy
  • Todos los hooks de pre-commit
tox

También puedes ejecutar una sola prueba en cualquier momento. Consulta la documentación a continuación.

Responses usa Pytest para las pruebas. Puedes ejecutar todas las pruebas con:

tox -e py37
tox -e py310

O activando manualmente la versión de Python requerida y ejecutando:

pytest

Y ejecutar una sola prueba con:

pytest -k '<test_function_name>'

Para verificar el cumplimiento de type, ejecuta el linter mypy:

tox -e mypy

O bien:

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

Para verificar y reformatear el estilo del código, ejecuta:

tox -e precom

O bien:

pre-commit run --all-files