在写 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()
声明：未经允许禁止转载 东东东 陈煜东的博客 文章，谢谢。如经授权，转载请注明： 转载自东东东 陈煜东的博客
近期评论