在写 Python Mock 的 requests测试用例,需要mock对方服务的http返回包行为,方便进行调试。而如果对方还没准备好,或者有副作用,又不敢直接请求。因此我们需要一个 Mock 的功能。

以前使用 Python 自带的 Mock 类库来进行,感觉写起来总感觉有点不顺畅。

在网上找到了一个类库 requests_mock ,用起来很方便,这里给大家分享一下。

requests 类库有一个可插件化的传输层适配器,并且允许你根据不同的url或者协议注册自己的handler。requests-mock的核心就是一个简单的被提前加载的传输层适配器。

安装 requests_mock

pip install requests_mock

使用 Mocker

通过 Context Manager方法使用

>>> import requests
>>> import requests_mock

>>> with requests_mock.Mocker() as m:
...     m.get('http://test.com', text='resp')
...     requests.get('http://test.com').text
...
'resp'

通过装饰器使用

>>> @requests_mock.Mocker()
... def test_function(m):
...     m.get('http://test.com', text='resp')
...     return requests.get('http://test.com').text
...
>>> test_function()
'resp'

这种方法需要在参数的最后一个位置进行声明,会传递进来这个对象。

如果有冲突,可以提前指定好。例如

>>> @requests_mock.Mocker(kw='mock')
... def test_kw_function(**kwargs):
...     kwargs['mock'].get('http://test.com', text='resp')
...     return requests.get('http://test.com').text
...
>>> test_kw_function()
'resp'

类装饰器

>>> requests_mock.Mocker.TEST_PREFIX = 'foo'
>>>
>>> @requests_mock.Mocker()
... class Thing(object):
...     def foo_one(self, m):
...        m.register_uri('GET', 'http://test.com', text='resp')
...        return requests.get('http://test.com').text
...     def foo_two(self, m):
...         m.register_uri('GET', 'http://test.com', text='resp')
...         return requests.get('http://test.com').text
...
>>>
>>> Thing().foo_one()
'resp'
>>> Thing().foo_two()
'resp'

类似 unitest 一样寻找测试函数开头前缀, requests_mock 也是类似的寻找测试函数,不过不想test前缀开头,可以用requests_mock.Mocker.TEST_PREFIX 来指定。

请求真实的HTTP服务

通过real_http关键字,可以请求真实的 HTTP 服务。

>>> with requests_mock.Mocker(real_http=True) as m:
...     m.register_uri('GET', 'http://test.com', text='resp')
...     print(requests.get('http://test.com').text)
...     print(requests.get('http://www.google.com').status_code)  
...
'resp'
200

或者

>>> with requests_mock.Mocker() as m:
...     m.register_uri('GET', 'http://test.com', text='resp')
...     m.register_uri('GET', 'http://www.google.com', real_http=True)
...     print(requests.get('http://test.com').text)
...     print(requests.get('http://www.google.com').status_code)  
...
'resp'
200

Mock url 的方法

其实上面已经简单提到了,主要是用

adapter.register_uri('GET', url, ...)

匹配具体的url地址

只要协议和url地址都匹配上,才有效。

.. >>> adapter.register_uri('GET', 'mock://test.com/path', text='resp')
.. >>> session.get('mock://test.com/path').text
.. 'resp'

url中的path 路径匹配

比如不管协议,主要域名匹配上就行。

.. >>> adapter.register_uri('GET', '//test.com/', text='resp')
.. >>> session.get('mock://test.com/').text
.. 'resp'

或者主要url中的path匹配就行

.. >>> adapter.register_uri('GET', '/path', text='resp')
.. >>> session.get('mock://test.com/path').text
.. 'resp'
.. >>> session.get('mock://another.com/path').text
.. 'resp'

匹配url中的 query 参数部分

>>> adapter.register_uri('GET', '/7?a=1', text='resp')
>>> session.get('mock://test.com/7?a=1&b=2').text
'resp'

匹配任意 http method,比如get、 post

>>> adapter.register_uri(requests_mock.ANY, 'mock://test.com/8', text='resp')
>>> session.get('mock://test.com/8').text
'resp'
>>> session.post('mock://test.com/8').text
'resp'

下面是无脑什么url地址,直接返回指定的response

>>> adapter.register_uri(requests_mock.ANY, requests_mock.ANY, text='resp')
>>> session.get('mock://whatever/you/like').text
'resp'
>>> session.post('mock://whatever/you/like').text
'resp'

更多的匹配url的方法,参考 https://requests-mock.readthedocs.io/en/latest/matching.html

动态返回 response

通过上面的动态匹配 url 地址,接下来可能多次请求一个url地址需要返回不同的数据。

主要还是利用上面的 requests_mock.Adapter.register_uri() 这个函数来支持。

注册 Reponses

register_uri 用于模拟http的请求。 主要有参数控制 response的一些 header 信息。

status_code:    The HTTP status response to return. Defaults to 200.
reason:         The reason text that accompanies the Status (e.g. ‘OK’ in ‘200 OK’)
headers:    A dictionary of headers to be included in the response.
cookies:    A CookieJar containing all the cookies to add to the response.

还有控制body的的参数

json:       A python object that will be converted to a JSON string.
text:       A unicode string. This is typically what you will want to use for regular textual content.
content:    A byte string. This should be used for including binary data in responses.
body:       A file like object that contains a .read() function.
raw:        A prepopulated urllib3.response.HTTPResponse to be returned.
exc:        An exception that will be raised instead of returning a response.

这些参数都是 requests.Response 对象的成员变量。

使用示例

>>> adapter.register_uri('GET', 'mock://test.com/1', json={'a': 'b'}, status_code=200)
>>> resp = session.get('mock://test.com/1')
>>> resp.json()
{'a': 'b'}

>>> adapter.register_uri('GET', 'mock://test.com/2', text='Not Found', status_code=404)
>>> resp = session.get('mock://test.com/2')
>>> resp.text
'Not Found'
>>> resp.status_code
404

动态响应

requests_mock 提供了一个回调函数,用于进行动态判断。

def callback(request, context):

request请求的对象和,context是返回的response对象

request:    The requests.Request object that was provided.
context:    An object containing the collected known data about this response.

使用示例

>>> def text_callback(request, context):
...     context.status_code = 200
...     context.headers['Test1'] = 'value1'
...     return 'response'
...
>>> adapter.register_uri('GET',
...                      'mock://test.com/3',
...                      text=text_callback,
...                      headers={'Test2': 'value2'},
...                      status_code=400)
>>> resp = session.get('mock://test.com/3')
>>> resp.status_code, resp.headers, resp.text
(200, {'Test1': 'value1', 'Test2': 'value2'}, 'response')

按指定list结果返回

提前写好需要返回的 response 列表。依次返回。

>>> adapter.register_uri('GET', 'mock://test.com/4', [{'text': 'resp1', 'status_code': 300},
...                                                   {'text': 'resp2', 'status_code': 200}])
>>> resp = session.get('mock://test.com/4')
>>> (resp.status_code, resp.text)
(300, 'resp1')
>>> resp = session.get('mock://test.com/4')
>>> (resp.status_code, resp.text)
(200, 'resp2')
>>> resp = session.get('mock://test.com/4')
>>> (resp.status_code, resp.text)
(200, 'resp2')

demo 1 : 需要请求获取任务id,然后查询运行中,然后成功的场景

import json
import time
from unittest import TestCase
from threading import Thread

import requests_mock

def task_result_success(m):
    """想要成功的时候的返回包"""
    time.sleep(0.05)
    resp = {"data": {},
            "code": "OK"}

    m.register_uri(requests_mock.ANY, requests_mock.ANY, text=json.dumps(resp))


@requests_mock.Mocker()
def test(m):
    resp = {"code": "OK", 
            "data": {"task_id": "xxxxx"}
                }
    # 注册结果,直接返回需要的任务id
    m.register_uri(requests_mock.ANY, requests_mock.ANY, text=json.dumps(resp))

    Thread(target=task_result_success, args=(m,)).start()   # 过一会后,mock对象的返回包会被替代。

    your.dosomething()  # 自己的业务逻辑,这里会请求获得一个任务ID,然后查询直接返回成功。

这里通过 pytest 运行了 test() 函数,先对http请求返回一个任务id的返回包。然后开启一个线程,过0.05s后,返回一个成功的请求包。

因为执行了 your.dosomething() 会阻塞主线程,所以通过一个子线程来修改 Mocker 对象的返回。想了好久想到的这么一个方法。

demo 2: 也是轮训taskid,然后返回的running,然后success

# 定义返回的顺序
response_list = [
    {'json': get_task_id_resp(), },
    {'json': get_task_query_running_resp(), },
    {'json': get_task_query_success_resp(), }
]

# 注册
m.register_uri(requests_mock.ANY, requests_mock.ANY, response_list)

# 执行调用代码开始测试,观察log
your.dosomething()

声明:未经允许禁止转载 东东东 陈煜东的博客 文章,谢谢。如经授权,转载请注明: 转载自东东东 陈煜东的博客

本文链接地址: Python mock http requests 单元测试用例测试 – https://www.chenyudong.com/archives/python-mock-http-request.html