python - 使用 pytest 测试 aiohttp 和 mongo

Question

我有一个简单的协程register，它接受登录名和密码 作为后置参数，然后它进入数据库等等。 我遇到的问题是我不知道如何测试协程。

我遵循了以下示例 https://aiohttp.readthedocs.io/en/latest/testing.html .

一切看起来都很简单，直到我开始自己编写测试。

test_register.py的代码

from main import make_app
pytest_plugins = 'aiohttp.pytest_plugin'


@pytest.fixture
def cli(loop, test_client):
    return loop.run_until_complete(test_client(make_app))

async def test_register(cli):
    resp = await cli.post('/register', data={'login': 'emil', 'password': 'qwerty'})
    assert resp.status == 200
    text = await resp.text()    

和register.py

from settings import db

async def register(request):
    post_data = await request.post()
    print('Gotta: ', post_data)
    login, password = post_data['login'], post_data['password']
    matches = await db.users.find({'login': login}).count()
    ...

main.py

from aiohttp import web
from routes import routes


def make_app(loop=None):
    app = web.Application(loop=loop)
    for route in routes:
        app.router.add_route(route.method, route.url, route.handler)
    return app


def main():
    web.run_app(make_app())


if __name__ == "__main__":
    main()

settings.py

from motor.motor_asyncio import AsyncIOMotorClient
DBNAME = 'testdb'
db = AsyncIOMotorClient()[DBNAME]

然后我运行了py.test test_register.py，它卡在了数据库操作上 matches = wait db.users.find({'login': login}).count()

Answer 1

问题的根源是全局变量的使用。

我建议进行以下更改:

from aiohttp import web
from motor.motor_asyncio import AsyncIOMotorClient
from routes import routes

def make_app(loop=None):
    app = web.Application(loop=loop)
    DBNAME = 'testdb'
    mongo = AsyncIOMotorClient(io_loop=loop)
    db = mongo[DBNAME]
    app['db'] = db

    async def cleanup(app):
        mongo.close()

    app.on_cleanup.append(cleanup)

    for route in routes:
        app.router.add_route(route.method, route.url, route.handler)
    return app

注册.py

async def register(request):
    post_data = await request.post()
    print('Gotta: ', post_data)
    login, password = post_data['login'], post_data['password']
    matches = await request.app['db'].users.find(
        {'login': login}).count()
    ...

将常用对象推送到应用程序的存储中是处理数据库连接等的一种受欢迎的方式。