responses 0.25.3

将响应记录到文件中

您可以对服务器执行真实请求，并且responses将自动将输出记录到文件中。记录的数据以YAML格式存储。

将@responses._recorder.record(file_path="out.yaml")装饰器应用于执行请求以记录响应到out.yaml文件的任何函数。

以下代码

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

将产生以下输出

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

从文件中重新播放响应（填充注册表）

您可以从包含记录响应的yaml文件中填充您的活动注册表。（有关如何获取文件，请参阅将响应记录到文件）。为此，您需要在激活的装饰器或上下文管理器中执行responses._add_from_file(file_path="out.yaml")。

以下代码示例注册了一个patch响应，然后是out.yaml文件中存在的所有响应以及最后的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.add(Response或Response args) - 允许注册Response对象或直接提供Response对象的参数。请参阅响应参数

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"

如果您尝试获取一个没有匹配的URL，responses 将抛出 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")

快捷方式

快捷方式提供了 responses.add() 的简短版本，其中方法参数已预先填充。

• responses.delete(Response args) - 注册 DELETE 响应

• responses.get(Response args) - 注册 GET 响应

• responses.head(Response args) - 注册 HEAD 响应

• responses.options(Response args) - 注册 OPTIONS 响应

• responses.patch(Response args) - 注册 PATCH 响应

• responses.post(Response args) - 注册 POST 响应

• responses.put(Response args) - 注册 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"}

响应作为上下文管理器

您不必用装饰器包裹整个函数，可以使用上下文管理器。

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

匹配请求体内容

对于发送请求数据的端点，添加响应时，您可以添加匹配器以确保您的代码正在发送正确的参数，并根据请求体内容提供不同的响应。responses 提供了用于JSON和URL编码请求体的匹配器。

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

查询参数匹配器

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

请求关键字参数匹配器

要验证请求参数，请使用matchers.request_kwargs_matcher函数与请求kwargs进行匹配。

仅支持以下参数：timeout、verify、proxies、stream、cert。

请注意，只有提供给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)

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

请求multipart/form-data数据验证

要验证multipart/form-data数据请求正文和头部，您可以使用matchers.multipart_matcher。提供的data和files参数将与请求进行比较

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.

请求片段标识符验证

要验证请求URL片段标识符，您可以使用matchers.fragment_identifier_matcher。匹配器将片段字符串（#符号之后的所有内容）作为比较的输入

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

请求头验证

在添加响应时，您可以指定匹配器以确保您的代码发送正确的头部，并根据请求头部提供不同的响应。

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"

因为requests将发送一些标准头部，而不仅仅是您的代码中指定的，所以默认情况下，匹配器之外的请求头部将被忽略。您可以通过将strict_match=True传递给匹配器来更改此行为，以确保仅发送您期望的头部，而不发送其他头部。请注意，您可能需要在使用requests时使用PreparedRequest来确保不包含任何额外的头部。

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"

创建自定义匹配器

如果您的应用程序需要其他编码或不同的数据验证，您可以构建自己的匹配器，该匹配器返回Tuple[matches: bool, reason: str]。其中布尔值表示请求参数是否匹配，字符串是在匹配失败时提供的原因。您的匹配器可以期望由responses提供的PreparedRequest参数。

请注意，PreparedRequest已自定义，并具有额外的属性params和req_kwargs。

默认注册表

默认情况下，responses将搜索所有已注册的Response对象并返回一个匹配项。如果只有一个Response被注册，则注册表保持不变。然而，如果在同一请求中找到多个匹配项，则返回第一个匹配项并将其从注册表中删除。

有序注册表

在某些场景中，保留请求和响应的顺序很重要。您可以使用registries.OrderedRegistry来强制所有Response对象依赖于插入顺序和调用索引。在以下示例中，我们添加了多个针对相同URL的Response对象。然而，您可以看到，状态码将取决于调用顺序。

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

自定义注册表

内置的registries适用于大多数用例，但要处理特殊条件，您可以实现自定义注册表，该注册表必须遵循registries.FirstMatchRegistry接口。重新定义find方法将允许您创建自定义搜索逻辑并返回适当的Response

示例：展示如何设置自定义注册表

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

作为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

为每个测试添加默认响应

在运行 unittest 测试时，这可以用来设置一些通用的类级别响应，这些响应可能由每个测试补充。类似的接口也可以应用于 pytest 框架。

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方法：start、stop、reset

responses 有 start，stop，reset 方法，与 unittest.mock.patch 非常相似。这使得在 setup 方法或需要执行多个补丁而无需嵌套装饰器或语句的地方进行请求模拟变得更容易。

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

基于响应对象进行断言

每个 Response 对象都有一个 call_count 属性，可以用来检查每个请求匹配了多少次。

@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

基于精确URL进行断言

断言请求恰好被调用 n 次。

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

环境配置

Responses 使用多个代码检查和自动格式化工具，因此提交补丁时使用适当的工具链非常重要。

克隆仓库

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

创建环境（例如使用 virtualenv）

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

配置开发需求

make develop

测试和代码质量验证

验证代码的最简单方法是通过 tox 运行测试。当前的 tox 配置运行与 GitHub Actions CI/CD 管道中使用的相同的检查。

请从项目根目录执行以下命令行以验证您的代码

• 支持的所有 Python 版本的单元测试

• 使用 mypy 进行类型验证

• 所有 pre-commit 钩子

tox

或者，您始终可以运行单个测试。请参阅以下文档。

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