-2

我是 pytest 的新手，想为单元测试覆盖添加以下 3 种方法，而不是实际使用真正的 mongo db 实例，而是模拟它。可以尝试使用真正的数据库实例，但不推荐。请求关于如何模拟 mongodb 客户端并获取文档的示例

import os
import logging
import urllib.parse
from dotenv import load_dotenv
from pymongo import MongoClient
from logger import *

load_dotenv()


def getMongoConnection():
    userName = urllib.parse.quote_plus(os.getenv("USER_NAME"))
    password = urllib.parse.quote_plus(os.getenv("PASSWORD"))
    hostName1_port = os.getenv("HOST_NAME1")
    hostName2_port = os.getenv("HOST_NAME2")
    hostName3_port = os.getenv("HOST_NAME3")
    authSourceDatabase = os.getenv("AUTH_SOURCE_DATABASE")
    replicaSet = os.getenv("REPLICA_SET")
    connectTimeoutMS = "1000"
    socketTimeoutMS = "30000"
    maxPoolSize = "100"

    try:
        client = MongoClient('mongodb://'+userName+':'+password+'@'+hostName1_port+','+hostName2_port+','+hostName3_port+'/'+authSourceDatabase+'?ssl=true&replicaSet='+replicaSet +
                             '&authSource='+authSourceDatabase+'&retryWrites=true&w=majority&connectTimeoutMS='+connectTimeoutMS+'&socketTimeoutMS='+socketTimeoutMS+'&maxPoolSize='+maxPoolSize)
        return client
    except Exception as e:
        logging.error("Error while connecting to mongoDB.")
        return False


def connectToDBCollection(client, databaseName, collectionName):
    db = client[databaseName]
    collection = db[collectionName]
    return collection


def getDoc(bucketName, databaseName, collectionName):
    try:
        client = getMongoConnection()
        if client != False:
            collection = connectToDBCollection(
                client, databaseName, collectionName)
            return collection.find_one({'bucket': bucketName})
    except Exception as e:
        logging.error("An exception occurred while fetching doc, error is ", e)

编辑：（尝试使用下面的代码并且能够涵盖大多数情况但看到错误）

def test_mongo():
    db_conn = mongomock.MongoClient()
    assert isinstance(getMongoConnection(), MongoClient)


def test_connect_mongo():
    return connectToDBCollection(mongomock.MongoClient(), "sampleDB", "sampleCollection")


//trying to cover exception block for getMongoConnection()
def test_exception():
    with pytest.raises(Exception) as excinfo:
        getMongoConnection()
    assert str(excinfo.value) == False


def test_getDoc():
    collection = mongomock.MongoClient().db.collection
    stored_obj = collection.find_one({'_id': 1})
    assert stored_obj == getDoc("bucket", "db", "collection")


def test_createDoc():
    collection = mongomock.MongoClient().db.collection
    stored_obj = collection.insert_one({'_id': 1})
    assert stored_obj == createDoc("bucket", "db", "collection")


def test_updateDoc():
    collection = mongomock.MongoClient().db.collection
    stored_obj = collection.replace_one({'_id': 1}, {'_id': 2})
    assert stored_obj == updateDoc(
        {'_id': 1}, {'$set': {'_id': 2}}, "db", "collection")

错误：test_exception - 失败：DID NOT RAISE <class 'Exception'> test_createDoc - TypeError：在字符串格式化期间并非所有参数都转换 AssertionError：断言 <pymongo.results.UpdateResult object at 0x7fc0e835a400> == <pymongo.results.UpdateResult object at 0x7fc0e8211900 >

4

2 回答 2

1

看起来 MongoClient 是一个嵌套的字典，带有databaseName和collectionName或用一个键访问器实现。

你可以先模拟客户端

import unittest
mocked_collection = unittest.mock.MagicMock()
# mock the find_one method
mocked_collection.find_one.return_value = {'data': 'collection_find_one_result'}
mocked_client = unittest.mock.patch('pymongo.MongoClient').start()
mocked_client.return_value = {
    'databaseName': {'collectionname': mocked_collection}
}
于 2021-06-03T17:51:37.987 回答
0

也许尝试像MongoMock这样的专门模拟库？

特别是最后一个使用的示例@mongomock.patch看起来可能与您的代码相关。

于 2021-06-03T18:07:35.673 回答