Python 用 Aiohttp 模拟不是模拟
Python Mock with Aiohttp isn't mocking
我正在使用 AioHttp 来实现工作中的服务,在我的测试期间,我正在模拟一个方法,问题是对该方法的调用是调用真实方法而不是模拟方法。
@unittest_run_loop
@patch('export_api.main.add_job_to_db')
async def test_view_job(self, mocked_method):
json = {
"edl": "somedata"
}
response = await self.client.request("PUT", "/v1/job", json=json)
mocked_method.assert_called_once_with()
assert response.status == 200
所以我在模拟的断言中得到了这个错误:
msg = "Expected 'add_job_to_db' to be called once. Called 0 times."
我的方法main.py:
async def __call__(self, request):
"""Faz post do Job na fila do Render"""
data = await request.json()
job_id = uuid.uuid4()
job = Jobs(
job_id=str(job_id),
body=data
)
try:
add_job_to_db(self.app['db'], job)
return web.Response(status=200)
except DatabaseError as e:
print(e)
return web.Response(status=500)
是的,是 class 中的一个可调用方法。没有模拟,测试工作正常。但是我需要模拟对数据库的调用,到目前为止我运气不好。
有什么想法吗?
清单:
- 使用asynctest包
- 不要忘记在单元测试周围使用
@asyncio.coroutine
装饰器
- 在使用它的地方修补对象,在本例中它
export_api.main.add_job_to_db
我是 mocket 的作者,几天前我发布了完全支持 asyncio 的版本 2.0.0 ]/aiohttp.
下面是在 HTTP 和 HTTPS 上模拟 URL 的相同代码示例:
import aiohttp
import asyncio
import async_timeout
from unittest import TestCase
from mocket.mocket import mocketize
from mocket.mockhttp import Entry
class AioHttpEntryTestCase(TestCase):
@mocketize
def test_http_session(self):
url = 'http://httpbin.org/ip'
body = "asd" * 100
Entry.single_register(Entry.GET, url, body=body, status=404)
Entry.single_register(Entry.POST, url, body=body*2, status=201)
async def main(l):
async with aiohttp.ClientSession(loop=l) as session:
with async_timeout.timeout(3):
async with session.get(url) as get_response:
assert get_response.status == 404
assert await get_response.text() == body
with async_timeout.timeout(3):
async with session.post(url, data=body * 6) as post_response:
assert post_response.status == 201
assert await post_response.text() == body * 2
loop = asyncio.get_event_loop()
loop.set_debug(True)
loop.run_until_complete(main(loop))
@mocketize
def test_https_session(self):
url = 'https://httpbin.org/ip'
body = "asd" * 100
Entry.single_register(Entry.GET, url, body=body, status=404)
Entry.single_register(Entry.POST, url, body=body*2, status=201)
async def main(l):
async with aiohttp.ClientSession(loop=l) as session:
with async_timeout.timeout(3):
async with session.get(url) as get_response:
assert get_response.status == 404
assert await get_response.text() == body
with async_timeout.timeout(3):
async with session.post(url, data=body * 6) as post_response:
assert post_response.status == 201
assert await post_response.text() == body * 2
loop = asyncio.get_event_loop()
loop.set_debug(True)
loop.run_until_complete(main(loop))
来源:https://github.com/mindflayer/python-mocket/blob/master/tests/tests35/test_http_aiohttp.py
我正在使用 AioHttp 来实现工作中的服务,在我的测试期间,我正在模拟一个方法,问题是对该方法的调用是调用真实方法而不是模拟方法。
@unittest_run_loop
@patch('export_api.main.add_job_to_db')
async def test_view_job(self, mocked_method):
json = {
"edl": "somedata"
}
response = await self.client.request("PUT", "/v1/job", json=json)
mocked_method.assert_called_once_with()
assert response.status == 200
所以我在模拟的断言中得到了这个错误:
msg = "Expected 'add_job_to_db' to be called once. Called 0 times."
我的方法main.py:
async def __call__(self, request):
"""Faz post do Job na fila do Render"""
data = await request.json()
job_id = uuid.uuid4()
job = Jobs(
job_id=str(job_id),
body=data
)
try:
add_job_to_db(self.app['db'], job)
return web.Response(status=200)
except DatabaseError as e:
print(e)
return web.Response(status=500)
是的,是 class 中的一个可调用方法。没有模拟,测试工作正常。但是我需要模拟对数据库的调用,到目前为止我运气不好。 有什么想法吗?
清单:
- 使用asynctest包
- 不要忘记在单元测试周围使用
@asyncio.coroutine
装饰器 - 在使用它的地方修补对象,在本例中它
export_api.main.add_job_to_db
我是 mocket 的作者,几天前我发布了完全支持 asyncio 的版本 2.0.0 ]/aiohttp.
下面是在 HTTP 和 HTTPS 上模拟 URL 的相同代码示例:
import aiohttp
import asyncio
import async_timeout
from unittest import TestCase
from mocket.mocket import mocketize
from mocket.mockhttp import Entry
class AioHttpEntryTestCase(TestCase):
@mocketize
def test_http_session(self):
url = 'http://httpbin.org/ip'
body = "asd" * 100
Entry.single_register(Entry.GET, url, body=body, status=404)
Entry.single_register(Entry.POST, url, body=body*2, status=201)
async def main(l):
async with aiohttp.ClientSession(loop=l) as session:
with async_timeout.timeout(3):
async with session.get(url) as get_response:
assert get_response.status == 404
assert await get_response.text() == body
with async_timeout.timeout(3):
async with session.post(url, data=body * 6) as post_response:
assert post_response.status == 201
assert await post_response.text() == body * 2
loop = asyncio.get_event_loop()
loop.set_debug(True)
loop.run_until_complete(main(loop))
@mocketize
def test_https_session(self):
url = 'https://httpbin.org/ip'
body = "asd" * 100
Entry.single_register(Entry.GET, url, body=body, status=404)
Entry.single_register(Entry.POST, url, body=body*2, status=201)
async def main(l):
async with aiohttp.ClientSession(loop=l) as session:
with async_timeout.timeout(3):
async with session.get(url) as get_response:
assert get_response.status == 404
assert await get_response.text() == body
with async_timeout.timeout(3):
async with session.post(url, data=body * 6) as post_response:
assert post_response.status == 201
assert await post_response.text() == body * 2
loop = asyncio.get_event_loop()
loop.set_debug(True)
loop.run_until_complete(main(loop))
来源:https://github.com/mindflayer/python-mocket/blob/master/tests/tests35/test_http_aiohttp.py