requests-mock:如何匹配模拟端点中的 POSTed 有效负载

requests-mock: how can I match POSTed payload in a mocked endpoint

我做了什么

我已经编写了一个身份验证 class,用于使用应用程序的 API 密钥[=从 Twitter 获取应用程序的 不记名令牌 43=] 及其 API 密钥秘密 the Twitter developer docs.

所示

我已经使用 requests_mock 以这种方式模拟了适当的端点:

@pytest.fixture
def mock_post_bearer_token_endpoint(
    requests_mock, basic_auth_string, bearer_token
):
    requests_mock.post(
        "https://api.twitter.com/oauth2/token",
        request_headers={
            "Authorization": f"Basic {basic_auth_string}",
            "Content-Type": "application/x-www-form-urlencoded;charset=UTF-8",
        },
        json={"token_type": "bearer", "access_token": f"{bearer_token}"},
    )

而我的测试方法是:

@pytest.mark.usefixtures("mock_post_bearer_token_endpoint")
def test_basic_auth(api_key, api_key_secret, bearer_token):
    response = requests.post(
        'https://api.twitter.com/oauth2/token',
        data={"grant_type": "client_credentials"},
        auth=TwitterBasicAuth(api_key, api_key_secret),
    )
    assert response.json()['access_token'] == bearer_token

(其中 TwitterBasicAuth 是我写的身份验证 class,fixture basic_auth_string 是一个硬编码字符串,可以通过转换 fixture api_keyapi_key_secret适当)。

而且有效。

问题

但我对模拟端点不检查负载这一事实感到非常困扰。在这种特殊情况下,有效载荷对于获得不记名令牌至关重要。

我已经梳理了 requests_mock(和 responses)的文档,但还没有弄清楚如何让端点仅在正确的有效负载为已发布。

请帮忙。

更新答案

我使用 gold_cy's comment 并编写了一个 自定义匹配器 接受请求和 returns 如果请求具有正确的url 路径,headers 和 json 负载。它 returns 否则是 403 响应,正如我对 Twitter API.

的期望
@pytest.fixture
def mock_post_bearer_token_endpoint(
    requests_mock, basic_auth_string, bearer_token
):
    def matcher(req):
        if req.path != "/oauth2/token":
            # no mock address
            return None
        if req.headers.get("Authorization") != f"Basic {basic_auth_string}":
            return create_forbidden_response()
        if (
            req.headers.get("Content-Type")
            != "application/x-www-form-urlencoded;charset=UTF-8"
        ):
            return create_forbidden_response()
        if req.json().get("grant_type") != "client_credentials":
            return create_forbidden_response()

        resp = requests.Response()
        resp._content = json.dumps(
            {"token_type": "bearer", "access_token": f"{bearer_token}"}
        ).encode()
        resp.status_code = 200

        return resp

    requests_mock._adapter.add_matcher(matcher)
    yield

def create_forbidden_response():
    resp = requests.Response()
    resp.status_code = 403
    return resp

旧答案

我使用 gold_cy's comment 并编写了一个 附加匹配器 来接收请求并检查有效负载中是否存在感兴趣的数据。

@pytest.fixture(name="mock_post_bearer_token_endpoint")
def fixture_mock_post_bearer_token_endpoint(
    requests_mock, basic_auth_string, bearer_token
):
    def match_grant_type_in_payload(request):
        if request.json().get("grant_type") == "client_credentials":
            return True
        resp = Response()
        resp.status_code = 403
        resp.raise_for_status()

    requests_mock.post(
        "https://api.twitter.com/oauth2/token",
        request_headers={
            "Authorization": f"Basic {basic_auth_string}",
            "Content-Type": "application/x-www-form-urlencoded;charset=UTF-8",
        },
        json={"token_type": "bearer", "access_token": f"{bearer_token}"},
        additional_matcher=match_grant_type_in_payload,
    )

我选择引发 Http403 错误(而不是仅仅返回 False)以减少确定异常引发原因的认知负担——返回 False 将导致 requests_mock.exceptions.NoMockAddress提出,我认为在这种情况下描述性不够。

我仍然认为有更好的解决方法,我会继续寻找它。

我认为这里的误解是您需要将所有内容都放在匹配器中,然后让 NoMatchException 告诉您是否正确。

匹配器可能是为了 return 正确响应所需的最简单的东西,然后您可以将所有 request/response 检查作为正常单元测试处理的一部分。

例如,如果您需要根据请求的主体切换响应值,

additional_matchers 很有用,通常 true/false 就足够了。

例如,我没有尝试为此查找 Twitter 授权:

import requests
import requests_mock

class TwitterBasicAuth(requests.auth.AuthBase):

    def __init__(self, api_key, api_key_secret):
        self.api_key = api_key
        self.api_key_secret = api_key_secret

    def __call__(self, r):
        r.headers['x-api-key'] = self.api_key
        r.headers['x-api-key-secret'] = self.api_key_secret
        return r


with requests_mock.mock() as m:
    api_key = 'test'
    api_key_secret = 'val'

    m.post(
        "https://api.twitter.com/oauth2/token",
        json={"token_type": "bearer", "access_token": "token"},
    )

    response = requests.post(
        'https://api.twitter.com/oauth2/token',
        data={"grant_type": "client_credentials"},
        auth=TwitterBasicAuth(api_key, api_key_secret),
    )

    assert response.json()['token_type'] == "bearer"
    assert response.json()['access_token'] == "token"
    assert m.last_request.headers['x-api-key'] == api_key
    assert m.last_request.headers['x-api-key-secret'] == api_key_secret

https://requests-mock.readthedocs.io/en/latest/history.html