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

のコード

そして私はpy.test test_register.pyを実行し、それがあなたの問題の根はグローバル変数の使用あるデータベース操作 matches = await 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 

register.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() 
    ... 

は