我试图使用python模拟包来模拟python请求模块。让我在下面的场景中工作的基本调用是什么?
在views.py中,我有一个函数,它每次都以不同的响应进行各种request .get()调用
def myview(request):
res1 = requests.get('aurl')
res2 = request.get('burl')
res3 = request.get('curl')
在我的测试类中,我想做类似的事情,但不能确定确切的方法调用
步骤1:
# Mock the requests module
# when mockedRequests.get('aurl') is called then return 'a response'
# when mockedRequests.get('burl') is called then return 'b response'
# when mockedRequests.get('curl') is called then return 'c response'
步骤2:
调用我的视图
步骤3:
验证响应包含'a response', 'b response', 'c response'
我如何完成第1步(模拟请求模块)?
我使用requests-mock为单独的模块编写测试:
# module.py
import requests
class A():
def get_response(self, url):
response = requests.get(url)
return response.text
测试:
# tests.py
import requests_mock
import unittest
from module import A
class TestAPI(unittest.TestCase):
@requests_mock.mock()
def test_get_response(self, m):
a = A()
m.get('http://aurl.com', text='a response')
self.assertEqual(a.get_response('http://aurl.com'), 'a response')
m.get('http://burl.com', text='b response')
self.assertEqual(a.get_response('http://burl.com'), 'b response')
m.get('http://curl.com', text='c response')
self.assertEqual(a.get_response('http://curl.com'), 'c response')
if __name__ == '__main__':
unittest.main()
我将添加这些信息,因为我很难弄清楚如何模拟异步api调用。
以下是我模拟异步调用所做的工作。
这是我想测试的函数
async def get_user_info(headers, payload):
return await httpx.AsyncClient().post(URI, json=payload, headers=headers)
您仍然需要MockResponse类
class MockResponse:
def __init__(self, json_data, status_code):
self.json_data = json_data
self.status_code = status_code
def json(self):
return self.json_data
添加MockResponseAsync类
class MockResponseAsync:
def __init__(self, json_data, status_code):
self.response = MockResponse(json_data, status_code)
async def getResponse(self):
return self.response
下面是测试。这里重要的是,我之前创建了响应,因为init函数不能是异步的,对getResponse的调用是异步的,所以它都签出了。
@pytest.mark.asyncio
@patch('httpx.AsyncClient')
async def test_get_user_info_valid(self, mock_post):
"""test_get_user_info_valid"""
# Given
token_bd = "abc"
username = "bob"
payload = {
'USERNAME': username,
'DBNAME': 'TEST'
}
headers = {
'Authorization': 'Bearer ' + token_bd,
'Content-Type': 'application/json'
}
async_response = MockResponseAsync("", 200)
mock_post.return_value.post.return_value = async_response.getResponse()
# When
await api_bd.get_user_info(headers, payload)
# Then
mock_post.return_value.post.assert_called_once_with(
URI, json=payload, headers=headers)
如果你有更好的方法告诉我,不过我觉得这样很干净。
对于那些不想为pytest安装额外库的人,这里有一个例子。我将在这里复制一些扩展,基于上面的例子:
import datetime
import requests
class MockResponse:
def __init__(self, json_data, status_code):
self.json_data = json_data
self.status_code = status_code
self.elapsed = datetime.timedelta(seconds=1)
# mock json() method always returns a specific testing dictionary
def json(self):
return self.json_data
def test_get_json(monkeypatch):
# Any arguments may be passed and mock_get() will always return our
# mocked object, which only has the .json() method.
def mock_get(*args, **kwargs):
return MockResponse({'mock_key': 'mock_value'}, 418)
# apply the monkeypatch for requests.get to mock_get
monkeypatch.setattr(requests, 'get', mock_get)
# app.get_json, which contains requests.get, uses the monkeypatch
response = requests.get('https://fakeurl')
response_json = response.json()
assert response_json['mock_key'] == 'mock_value'
assert response.status_code == 418
assert response.elapsed.total_seconds() == 1
============================= test session starts ==============================
collecting ... collected 1 item
test_so.py::test_get_json PASSED [100%]
============================== 1 passed in 0.07s ===============================
为了避免安装其他依赖项,您应该创建一个假响应。这个FakeResponse可以是Response的子类(我认为这是一个很好的方法,因为它更现实),或者只是一个具有您需要的属性的简单类。
简单的假类
class FakeResponse:
status_code = None
def __init__(self, *args, **kwargs):
self.status_code = 500
self.text = ""
回应之子
class FakeResponse(Response):
encoding = False
_content = None
def __init__(*args, **kwargs):
super(FakeResponse).__thisclass__.status_code = 500
# Requests requires to be not be None, if not throws an exception
# For reference: https://github.com/psf/requests/issues/3698#issuecomment-261115119
super(FakeResponse).__thisclass__.raw = io.BytesIO()