【问题标题】:Mocking a Sqlalchemy session for pytest为 pytest 模拟 Sqlalchemy 会话
【发布时间】:2019-07-25 02:44:17
【问题描述】:

我不知道这是否可以做到,但我正在尝试模拟我的 db.session.save。

我正在使用烧瓶和烧瓶炼金术。

db.py

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

单元测试

def test_post(self):
    with app.app_context():
        with app.test_client() as client:
            with mock.patch('models.db.session.save') as mock_save:
                with mock.patch('models.db.session.commit') as mock_commit:

                    data = self.gen_legend_data()
                    response = client.post('/legends', data=json.dumps([data]), headers=access_header)

                    assert response.status_code == 200
                    mock_save.assert_called()
                    mock_commit.assert_called_once()

以及方法:

def post(cls):
    legends = schemas.Legends(many=True).load(request.get_json())

    for legend in legends:
        db.session.add(legend)

    db.session.commit()

    return {'message': 'legends saved'}, 200

我正在尝试模拟 db.session.add 和 db.session.commit。我试过db.session.savelegends.models.db.session.savemodels.db.session.save。他们都返回了保存错误:

ModuleNotFoundError: No module named 'models.db.session'; 'models.db' is not a package

我没有收到错误,我不知道如何解决它。

或者我想要模拟 db.session 是不是在做一些完全错误的事情?

谢谢。 戴斯蒙德

【问题讨论】:

  • 当您的代码中没有调用此类方法时,您为什么要模拟 Session.save()? (而Session没有这种方法)

标签: python sqlalchemy mocking flask-sqlalchemy pytest


【解决方案1】:

您在这里遇到的问题最好通过重组代码来更好地解决,使其更易于测试,而不是模拟出每个组件,或者进行(非常)缓慢的集成测试。如果您养成以这种方式编写测试的习惯,那么随着时间的推移,您最终会得到一个缓慢的构建,需要很长时间才能运行,并且您最终会得到脆弱的测试(关于为什么快速测试很重要here)。

我们来看看这条路线:

def post(cls):
    legends = schemas.Legends(many=True).load(request.get_json())

    for legend in legends:
        db.session.add(legend)

    db.session.commit()

    return {'message': 'legends saved'}, 200

...分解它:

import typing
from flask import jsonify

class LegendsPostService:

    def __init__(self, json_args, _session=None) -> None:
        self.json_args = json_args
        self.session = _session or db.session

    def _get_legends(self) -> Legend:
        return schemas.Legends(many=True).load(self.json_args)

    def post(self) -> typing.List[typing.Dict[str, typing.Any]]:
        legends = self._get_legends()

        for legend in legends:
            self.session.add(legend)

        self.session.commit()
        return schemas.Legends(many=True).dump(legends)

def post(cls):
    service = LegendsPostService(json_args=request.get_json())
    service.post()
    return jsonify({'message': 'legends saved'})

请注意,我们如何将几乎所有故障点从 post 隔离到 LegendsPostService,此外,我们还从中删除了所有烧瓶内部(没有浮动的全局请求对象等)。如果我们需要稍后进行测试,我们甚至赋予它模拟 session 的能力。

我建议您将测试工作重点放在为LegendsPostService 编写测试用例上。一旦您对LegendsPostService 进行了出色的测试,请确定您是否相信即使更多 测试覆盖率也会增加价值。如果你这样做了,那么考虑为 post() 编写一个简单的集成测试,将它们结合在一起。

您需要考虑的下一件事情是您想如何考虑测试中的 SQLAlchemy 对象。我建议只使用 FactoryBoy 为您自动创建“模拟”模型。这是一个完整的应用示例,说明如何以这种方式设置 flask / sqlalchemy / factory-boy:How do I produce nested JSON from database query with joins? Using Python / SQLAlchemy

这是我为LegendsPostService 编写测试的方式(抱歉,这有点仓促,并不能完全代表您要执行的操作 - 但您应该能够根据您的用例调整这些测试):


from factory.alchemy import SQLAlchemyModelFactory

class ModelFactory(SQLAlchemyModelFactory):
    class Meta:
        abstract = True
        sqlalchemy_session = db.session

# setup your factory for Legends:
class LegendsFactory(ModelFactory):
    logo_url = factory.Faker('image_url')
    class Meta(ModelFactory.Meta):
        model = Legends


from unittest.mock import MagicMock, patch


# neither of these tests even need a database connection!
# so you should be able to write HUNDREDS of similar tests
# and you should be able to run hundreds of them in seconds (not minutes)

def test_LegendsPostService_can_init():
    session = MagicMock()
    service = LegendsPostService(json_args={'foo': 'bar'}, _session=session)
    assert service.session is session
    assert service.json_args['foo'] == 'bar'


def test_LegendsPostService_can_post():
    session = MagicMock()
    service = LegendsPostService(json_args={'foo': 'bar'}, _session=session)

    # let's make some fake Legends for our service!
    legends = LegendsFactory.build_batch(2)

    with patch.object(service, '_get_legends') as _get_legends:
        _get_legends.return_value = legends
        legends_post_json = service.post()

    # look, Ma! No database connection!
    assert legends_post_json[0]['image_url'] == legends[0].image_url

希望对你有帮助!

【讨论】:

  • 谢谢斯蒂芬。这有很大帮助。只是一个问题,行 session = MagicMock() 正确地模拟了 db.session。所以当代码到达 self.session.add() 时,它只会使用 MagicMock() 对象而不调用实际的会话。
  • 是的,没错。当您模拟session 时,您正在编写单元测试而不是系统/集成测试。如果您觉得需要进行集成测试以确保所有 sqlalchemy / 数据库行为交互都按预期工作,您可以为此编写集成测试。
  • 我建议尽可能少地编写集成测试(即那些不模拟 sqlalchemy 并与数据库对话的测试)。为什么?有几个原因,但其中一个更微妙的原因是您可能要到项目进行几年后才会注意到。考虑运行 1000 个单元测试应该花费不到 30 秒的时间。然而,运行 1000 个集成测试可能需要 30 分钟以上
猜你喜欢
  • 2016-12-25
  • 2016-02-24
  • 2018-06-08
  • 2020-05-27
  • 1970-01-01
  • 2015-07-23
  • 1970-01-01
  • 1970-01-01
  • 2022-07-04
相关资源
最近更新 更多