使用 mongomock 对 PyMongo Flask 应用程序进行单元测试

简介

这是一个利基指南。本指南的原因是在一个项目期间,我需要找到一种方法来测试一个简单的 Flask PyMongo CRUD 应用程序。在尝试测试时,我发现很少有关于如何将 Mongo 作为单元测试的一部分进行模拟的在线文档,并且在 StackOverflow 上没有找到可以轻松运行的答案。

我找到了使用 PyTest 的指南,以及用于使用 MongoClient 和 Connection 的 Flask 应用程序的指南。两者都不满足我使用的设置。也有使用 mockupdb 的测试解决方案,但它们似乎过于复杂。

看到通过 Flask 向许多开发人员介绍了 Web 开发,我认为这个小指南有一天可能会对某人有所帮助。请注意,本指南跳过了一些实践以使代码库更小,因此我们可以专注于手头的事情。

我正在使用的特定堆栈是:

• 蟒蛇3

• 烧瓶

• PyMongo(使用 init_app)

• 单元测试

申请

我们打算创建的应用程序是一个简单的 CRUD 应用程序,用 Flask 制作,带有一个 MongoDB 数据库。我们将公开用于创建、检索和删除“文章”资源的端点。这些资源的内容只是“作者”、“内容”和“标签”列表。

该项目的所有代码都可以在 GitHub 存储库中找到:https://github.com/reritom/Flask-PyMongo-Unittest-Guide。

目录布局

我们将遵循关于 repo 设置的最佳实践。我们的源代码将有一个src目录,我们的测试将有一个tests目录,并且应用程序名义上将通过运行python main.py来部署。

我们的目录将如下所示(忽略 README、requirements.txt 和其他杂项文件):

.
├── main.py
├── src
│   ├── __init__.py
│   ├── application.py
│   ├── database.py
│   └── controllers
│       ├── __init__.py
│       └── article_controller.py
└── tests
    ├── __init__.py
    └── articles_test.py

安装运行

安装

如果您有兴趣测试代码。创建虚拟环境,运行pip install -r requirements.txt。

运行测试

要运行测试,您可以使用python -m unittest discover -p '*_test.py'。

来源

如果你以前用过 Flask,你会知道通常有一个application.py(或者有时你把它放在 src 的__init__.py中)。然后,您创建一个database.py将包含您的数据库对象。当你创建你的第一个 Flask 应用程序时,通常你会将数据库对象放在你的application.py中,但是当你开始拆分应用程序时,如果数据库存在于你的application.py中,你会遇到循环导入问题。所以我们创建了两个文件:database.py,application.py。

database.py很简单。它开始看起来像这样:

# src/database.py
from flask_pymongo import PyMongo

mongo = PyMongo()

每当我们想访问我们的数据库时,我们都会导入这个单个PyMongo实例并与之交互,该实例分配给变量mongo。

应用程序将遵循应用程序工厂方法,我们将跳过配置对象,因为在这种情况下它们无关紧要。所以我们可以将数据库地址传递给它。

application.py将如下所示:

# src/application.py
from flask import Flask
from src.database import mongo
from src.controllers import ArticleController

def create_app(db_uri: str) -> Flask:
    app = Flask(__name__)
    app.config["MONGO_URI"] = db_uri
    mongo.init_app(app)

    # Add the articles collection if it doesn't already exist
    if not 'articles' in mongo.db.list_collection_names():
        articles_collection = mongo.db['articles']

    # Register the article routes
    app.add_url_rule("/articles", methods=["POST"], view_func=ArticleController.create_article)
    app.add_url_rule("/articles", methods=["GET"], view_func=ArticleController.get_articles)
    app.add_url_rule("/articles/<uuid:article_id>", methods=["GET"], view_func=ArticleController.get_article)
    app.add_url_rule("/articles/<uuid:article_id>", methods=["DELETE"], view_func=ArticleController.delete_article)

    return app

您注意到我们在控制器子目录中有一个名为ArticleController的类。我们可以直接从控制器模块导入它,而不是像from src.controllers.article_controller import ArticleController那样导入它的原因是因为我们已经将控制器导入到src/controllers/__init__.py中。

ArticleController 现在只关注 create_article 方面。对于创建文章,我们不会进行任何验证。相反,我们只是从请求中获取内容,将其存储在我们的 mongo 集合中,然后从请求中返回数据以及 mongo 文档 ID,如下所示:

# src/controllers/article_controller.py
import uuid
from flask import request, jsonify
from src.database import mongo

class ArticleController:
    @staticmethod
    def create_article():
        """
        Take the article from the request and deposit it directly into our mongo collection
        """
        data = request.get_json(force=True)
        mongo.db.articles.insert_one(data)
        data["_id"] = str(data["_id"]) # The mongo-added id isn't serialisable, so we convert it to a string
        return jsonify(data), 201

    @staticmethod
    def get_article(article_id: uuid.UUID):
        ...

    @staticmethod
    def get_articles():
        ...

    @staticmethod
    def delete_article(article_id: uuid.UUID):
        ...

最后,为了运行这个应用程序,我们创建了main.py,在这种情况下它可能只有几行。

# main.py
from src import create_app

if __name__=="__main__":
    db_uri = "mongodb://'127.0.0.1:27017/mydatabase"
    app = create_app(db_uri)
    app.run("0.0.0.0", port=5000, debug=False)

再次注意,create_app可以直接从 src 导入,因为在src/__init__.py中我添加了from src.application import create_app。这主要是为了方便项目增长和变得更加嵌套。

现在,只需这四个文件,我们就可以运行我们的应用程序并通过将数据发布到/articles端点来创建文章,假设您在后台运行 mongodb,并将正确的 db_uri 传递给 create_app 函数。

测试

所以我们可以有一个可以运行的应用程序。现在我们要测试它。测试的原因有很多,包括确保应用程序符合预期的行为,以及确保没有回归(错误)被引入到代码中。

在测试的时候,我们专注于代码,我们的测试应该和mongod解耦,这样测试可以更容易运行,不依赖mongod服务。

我们为此采取的方法称为“模拟”。我们想要模拟 Mongo,这类似于创建一个假的 mongo 实例,该实例的行为方式与真实的 Mongo 相同(尽管很明显,相似之处通常是肤浅的)。

从我们的代码的角度来看,我们希望能够将文档插入到我们模拟的 mongo 中,并能够在以后的请求中检索。

为此,我们将使用一个名为mongomock的库,它提供一个名为MongoClient的类,它在大多数名义情况下的行为与PyMongo.MongoClient相同。

修补

修补是一种用其他东西替换程序命名空间中的对象的方法。这通常用于修补环境或修补 API 调用。假设您有一个脚本,如果os.environ["FLAG"] == True以一种方式运行,如果os.environ["FLAG"] == False以另一种方式运行。您需要创建两个测试,每个案例一个,然后修补 os.environ 以将 FLAG 设置为每个测试的正确值。

修补时要注意的重要一点是,您可以修补对象的属性,或者修补正在使用该对象的模块的命名空间中的对象(它们实际上是相同的)。这是什么意思?好吧,当您使用 os.environ 时,您导入os。实际上,这意味着在您的模块中现在有一个名为os的模块对象,这就是您要修补的内容,因为这是您的代码在使用时消耗的内容os.environ 稍后。

实际上是这样的:

# dummy.py
import os
def print_flag():
  print(os.environ.get("FLAG"))

# dummy_test.py
from unittest.mock import patch
from dummy import print_flag

# If we patch "os", we are patching it in the wrong namespace, so we can't control what will be printed.
with patch("os") as dummy_os:
  print_flag()

with patch("dummy.os") as dummy_os:
  dummy_os.environ.get.return_value = True
  print_flag() # This will print True, because we have patched the os in the namespace of dummy, and explicitly told the mock object (dummy_os) to return True when os.environ.get(...) is called.

在这种情况下,在测试时,我们可以修补从app.database导入mongo的所有情况。在每个使用者的命名空间中,我们可以将mongo(MongoClient) 对象替换为我们的mongomock.MongoClient。

但是,随着应用程序的增长,无论何时使用 mongo 客户端,您都需要不断添加更多补丁。

看到所有的数据库消费者都从 app.database 导入 mongo,如果我们可以在app.database模块中修补mongo会很方便。然后所有的消费者都可以在不知情的情况下继续导入这个对象。在我们的测试中,我们只需要确保数据库对象是模拟的,这意味着随着我们的应用程序的增长,我们的测试仍然有效。

然后我们可以考虑当我们的src.database导入PyMongo时,我们可以在src.database的命名空间中模拟PyMongo。

在您的测试中,您可以尝试以下操作:

import unittest
from unittest.mock import patch
from src import create_app
import mongomock

class TestApplication(unittest.TestCase):
  def test_application(self):
    with patch("src.database.PyMongo", side_effect=mongomock.MongoClient):
      # Create the app and run the tests
      ...

现在上面的代码会模拟PyMongo来引用mongomock.MongoClient,但你的测试仍然会失败。这是因为在运行测试之前已经加载了src.database模块。所以是的,PyMongo现在指的是mongomock.MongoClient,但是您的mongo变量被分配给PyMongo的实例,因为它是在模拟之前运行的。所以你在嘲笑班级,但为时已晚。

然后,您可以考虑尝试预先模拟src.database模块,或者使用我们的mongomock.MongoClient实例修补src.database.mongo对象。

如果我们考虑后者,我们可以通过对代码进行一些更改来做到这一点。我们想要的是在src.database中模拟mongo对象,因此它现在不是引用PyMongo的实例,而是引用mongomock.MongoClient的实例。现在我们需要记住命名空间。在src/application.py和src/controllers/article_controllers.py中,我们导入mongo。这意味着在每个命名空间中,它们已经有一个mongo的引用。因此,如果我们在src.database模块中修补mongo,它不会反映在存在的mongo对象中在这两个命名空间中。所以我们需要做的代码更改是不要将mongo导入这两个模块,而是导入app.database模块,并使用app.database.mongo访问 mongo。

这些变化会是这样的:

# src/application.py
from flask import Flask
from src.controllers import ArticleController
import src.database

def create_app(db_uri: str) -> Flask:
    app = Flask(__name__)
    app.config["MONGO_URI"] = db_uri
    src.database.mongo.init_app(app)

    # Add the articles collection if it doesn't already exist
    if not 'articles' in src.database.mongo.db.list_collection_names():
        articles_collection = src.database.mongo.db['articles']

    # Register the article routes
    ...

    return app

和

# src/controllers/article_controller.py
from flask import request, jsonify
import src.database

class ArticleController:
    @staticmethod
    def create_article():
        data = request.get_json(force=True)
        src.database.mongo.db.articles.insert_one(data)
        data["_id"] = str(data["_id"]) # The mongo-added id isn't serialisable, so we convert it to a string
        return jsonify(data), 201
    ...

在我们的测试中,我们可以修补app.database模块对象,以便mongo引用我们的mongomock.MongoClient实例,而不是PyMongo。

# tests/articles_test.py
import unittest
from unittest.mock import patch
from src import create_app
import src.database
import mongomock

class TestApplication(unittest.TestCase):
  def test_application(self):
    with patch.object(src.database, "mongo", mongomock.MongoClient()):
      # Create the app and run the tests
      ...

此时,我们正在修补正确命名空间中的正确对象,而mongo的消费者正在获取我们修补的资源。但是,flask_pymongo.PyMongo和mongomock.MongoClient引用的对象类型不同。PyMongo是MongoClient的超类。所以你会得到这个错误:

Traceback (most recent call last):
  File "/Users/***/projects/flask-pymongo-unittest-guide/tests/articles_test.py", line 20, in test_create_article
    app = create_app("mongodb://localhost:27017/mydatabase").test_client()
  File "/Users/***/projects/flask-pymongo-unittest-guide/src/application.py", line 8, in create_app
    src.database.mongo.init_app(app)
TypeError: 'Database' object is not callable

或者,如果您使用mongomock.MongoClient而不是mongomock.MongoClient()进行修补,则会收到此错误。

Traceback (most recent call last):
  File "/Users/***/projects/flask-pymongo-unittest-guide/tests/articles_test.py", line 20, in test_create_article
    app = create_app("mongodb://localhost:27017/mydatabase").test_client()
  File "/Users/***/projects/flask-pymongo-unittest-guide/src/application.py", line 8, in create_app
    src.database.mongo.init_app(app)
AttributeError: type object 'MongoClient' has no attribute 'init_app'

后一个错误是您的代码中的错误,一旦您修复它,您将得到第一个错误。

为了处理这个问题,我们可以创建一个具有init_app方法的虚拟超类,我们可以用它来修补 mongo:

# tests/articles_test.py
import unittest
from unittest.mock import patch
from src import create_app
import src.database
from mongomock import MongoClient

class PyMongoMock(MongoClient):
    def init_app(self, app):
        return super().__init__()

class TestApplication(unittest.TestCase):
  def test_application(self):
    with patch.object(src.database, "mongo", PyMongoMock()):
      # Create the app and run the tests
      ...

再次注意,我们使用PyMongoMock的实例修补src.database.mongo,而不是类。

此时,您的测试将能够使用mongo的模拟实例成功运行。您可以在 repo 中查看我为本指南编写的具体测试。

修补是 Python 和 unittest 框架中非常强大的工具,编写强大的、自包含的单元测试是您保证应用程序按预期运行的方式。修补和命名空间的组合可能会令人困惑,我见过很多测试环境被错误地修补,导致测试通过,但在其他机器上可能会失败,所以它是 Python 学习的一个非常重要的部分关于。

我希望本指南能够帮助遇到这个测试障碍的一两个开发人员。