使用Motor AsyncIO和Pytest测试MongoDB功能

如何解决使用Motor AsyncIO和Pytest测试MongoDB功能

因此，我试图编写一些测试来测试使用异步MongoDB连接的功能。为了连接到MongoDB，我将Motor与asyncio一起使用。我需要模拟马达连接的帮助。

我的代码：

commons.py

mongo = None

blacklist.py

import commons

class Blacklist(object):
    async def check_if_blacklisted(self,word: str):
        blacklisted = False
        if await commons.mongo.blacklist.find_one({'word': word}):
            blacklisted = True
        return blacklisted

main.py

import asyncio
from blacklist import Blacklist
from motor.motor_asyncio import AsyncIOMotorClient

async def run():
    commons.mongo = AsyncIOMotorClient("mongodb://localhost",io_loop=asyncio.get_event_loop())
    blacklist_checker = Blacklist()
    result = await blacklist_checker.check_if_blacklisted(word="should_be_false")
    print(result)
    # > False

    result = await blacklist_checker.check_if_blacklisted(word="should_be_true")
    print(result)
    # > True

loop = asyncio.get_even_loop()
loop.run_until_complete(run())
loop.close()

我现在想通过模拟马达连接来测试blacklist.py，但似乎无法正常运行测试。这是我尝试过的代码：

test_blacklist.py

import pytest
from blacklist import Blacklist

class TestBlacklist(object):

@pytest.fixture
async def motor(self,event_loop):
    # I know I'm not mocking the Motor Connection here,# but just wanted to show you the output using this fixture.
    commons.mongo = motor.motor_asyncio.AsyncIOMotorClient(io_loop=event_loop)
    yield commons.mongo
    commons.mongo.close()

@pytest.mark.asyncio
async def test_check_if_blacklisted(self):
    blacklist_checker = Blacklist()
    blacklisted = await blacklist_checker.check_if_blacklisted(word="should_be_false")
    assert blacklisted == False
    # > AttributeError: 'NoneType' object has no attribute 'blacklist'

pytest-mongodb：

import pytest
from unittest.mock import patch
from blacklist import Blacklist

class TestBlacklist(object):

@pytest.mark.asyncio
async def test_check_if_blacklisted(self,mongodb):
    with patch("blacklist.commons.mongo") as db:
        db = mongodb
        blacklist_checker = Blacklist()
        blacklisted = await blacklist_checker.check_if_blacklisted(word="should_be_false")
    assert blacklisted == False
    # > TypeError: object MagicMock can't be used in 'await' expression

我尝试在线搜索，但找不到合适的线程来模拟异步的Motor连接，从而帮助我执行测试。此外，如果您认为我要进行测试的方向不正确，请告诉我，因为我是编写测试的新手，尤其是异步数据库连接。

注意： blacklist.py具有要求MongoDB功能的各种功能，因此，如果在我的test_blacklist.py中我可以一次初始化commons.db，然后所有后续测试都使用它，那就太好了。 / p>

解决方法

您可以使用 pytest-async-mongodb 模拟异步 MongoDB 数据库，但请记住它已过时且存在依赖项错误，因此您必须按以下方式修复依赖项版本：

mongomock==3.12.0
pyyaml==3.13
pytest-asyncio==0.10.0
pytest==3.6.4

使用 pytest-async-mongodb，您可以通过添加一个名为 async_mongodb 的参数来获取测试中的模拟数据库。我会让你知道代码和结构。

project
  -app
    __init__.py
    blacklist.py
    commons.py
  -test
    -fixtures
      blacklist.json
    __init__.py
    test_blacklist.py
  main.py
  pytest.ini

main.py

import asyncio
from app.blacklist import Blacklist
from app.commons import get_database,set_client
from motor.motor_asyncio import AsyncIOMotorClient


async def run():
    set_client(
        AsyncIOMotorClient("mongodb://localhost",io_loop=asyncio.get_event_loop())
    )
    db = await get_database()
    blacklist_checker = Blacklist()
    result = await blacklist_checker.check_if_blacklisted(db,word="should_be_false")
    print(result)
    # > False

    result = await blacklist_checker.check_if_blacklisted(db,word="should_be_true")
    print(result)
    # > True


loop = asyncio.get_event_loop()
loop.run_until_complete(run())
loop.close()

blacklist.py

from motor.motor_asyncio import AsyncIOMotorDatabase


class Blacklist(object):
    async def check_if_blacklisted(self,db: AsyncIOMotorDatabase,word: str):
        blacklisted = False
        if await db.blacklist.find_one({"word": word}):
            blacklisted = True
        return blacklisted

commons.py

from motor.motor_asyncio import AsyncIOMotorClient,AsyncIOMotorDatabase


class DataBase:
    client: AsyncIOMotorClient = None


db = DataBase()


async def get_database() -> AsyncIOMotorDatabase:
    return db.client["dbtest"]


def set_client(client):
    db.client = client

test_blacklist.py

import pytest
from app import blacklist


@pytest.mark.asyncio
async def test_should_be_false(async_mongodb):
    blacklist_checker = blacklist.Blacklist()
    blacklisted = await blacklist_checker.check_if_blacklisted(
        async_mongodb,word="should_be_false"
    )
    assert blacklisted == False


@pytest.mark.asyncio
async def test_should_be_true(async_mongodb):
    blacklist_checker = blacklist.Blacklist()
    blacklisted = await blacklist_checker.check_if_blacklisted(
        async_mongodb,word="should_be_true"
    )
    assert blacklisted == True

pytest.ini

[pytest]
async_mongodb_fixture_dir =
  test/fixtures

async_mongodb_fixtures =
  blacklist

blacklist.json

[
    {
      "_id": {"$oid": "60511d158f80a8d34986e2b0"},"word" : "should_be_true"
    }
]

夹具也可以是 .yaml 并且可以定义您想要的数量。阅读package documentation了解更多信息。

由于它已经过时，我创建了一个 fork 来更新它并使用新功能对其进行改进。邀请您看一看，如果您愿意，可以使用它。

使用Motor AsyncIO和Pytest测试MongoDB功能

如何解决使用Motor AsyncIO和Pytest测试MongoDB功能

解决方法

相关推荐