mongodb 如何在Python中实现对使用mongoengine连接的函数的测试?

x33g5p2x  于 2023-06-29  发布在  Go
关注(0)|答案(2)|浏览(134)

我有一个Mongo数据库的快速API,我试图实现我的一些功能,需要mongo连接测试这里我的集合用户及其字段:

from mongoengine import Document, EmailField, StringField, EnumField 
class User(Document):  
    meta = {"collection": "users"}
    email = EmailField(unique=True)
    hashed_password = StringField()
    role = EnumField(UserRole)

    def username(self):
        return self.email.split('@')[0]

    def __str__(self):
        return f'{self.email} ({self.role})

这里我的函数

class UserService:

    @staticmethod
    def existing_user(email: str) -> Optional[User]:

        try:
             user = User.objects(email=email).first()
        except ServerSelectionTimeoutError as e:
            raise e

        return user

测试这个函数的最好方法是什么(使用pytest)?我想嘲笑我的MongoDB,但我没有使用mongomock或直接使用monkeypatch管理
非常感谢
我的测试文件

from domain.entities.user import UserRole, User
 from mongomock import MongoClient
 class TestUserServiceMethods:

    email="john@doe.fr"

    # User example
    user = User(email=email,
                  role=UserRole.admin,
                  hashed_password="hashed_pwd")

    add_user = {
      "email": email,
      "role": UserRole.admin,
      "hashed_password": "hashed_pwd"
    }

    

    user_service = UserService()

    client = MongoClient()

    mock_db = client['ml-db']
    user_collection = mock_db['users']
    user_collection.insert_one(add_user)


    def test_existing_user(self, monkeypatch):
        with monkeypatch.context() as m:
            m.setattr('pymongo.MongoClient', lambda *args, **kwargs: self.client)

            result = self.user_service.existing_user(self.email)

             assert result == self.user

看起来模拟没有成功,因为我得到了错误:

mongoengine.connection.ConnectionFailure: You have not defined a default connection

我应该直接使用Fast app和test_client吗?

56lgkhnf

56lgkhnf1#

在执行任何操作之前,您需要使用db_name(默认别名为“default”)和connection_string从mogoengine运行connect方法。
例如:

from mongoengine import connect

connect("db_name", host="connection_string", mongo_client_class=client)

代码中的客户端来自monfoengine,测试中的客户端来自mongomock

yquaqz18

yquaqz182#

UPDATE正如Ozs所说,最好的方法是创建一个Fake数据库,并在最后删除它。这里有一个解决我的问题的例子

from unittest import TestCase

from mongoengine import connect

import UserService, password_context 
import UserRole, User

class TestUserServiceMethods(TestCase):

    @classmethod
    def setUpClass(cls):
        cls.connection = connect('mongoenginetest', host='mongomock://localhost/ml-db')

    def setUp(cls):
        cls.data = {
            "id": "6499594c58ebb74dd4985e59",
            "email": "mail@test.fr",
            "pwd": "test",
            "hashed_pwd": password_context.hash("test"),
            "role": UserRole.admin
        }

        cls.doc = {
            "email": cls.data['email'],
            "role": cls.data['role'],
            "hashed_password": cls.data['hashed_pwd']
        }
      

        cls.user = User(
            email=cls.data['email'],
            role=cls.data['role'],
            hashed_password=cls.data['hashed_pwd'])

        cls.user_service = UserService()

        cls.user.save()

    @classmethod
    def tearDownClass(cls):
        cls.connection.drop_database('mongoenginetest')

    def test_existing_user(self):
        user = self.user_service.existing_user(self.data['email'])
        self.assertIsNotNone(user)

        user_ko = self.user_service.existing_user("error@mail.fr")
        self.assertIsNone(user_ko)

相关问题