Python:如何模拟SQLAlchemy事件处理程序(使用mock.unittest)

Question

所以我有一个SQLAlchemy模型，它有一个事件监听器：

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
@event.listens_for(User, "after_insert")
@event.listens_for(User, "after_update")
def do_something(mapper, connection, self):
    foo = SomeClass(self)
    foo.do_something_to_database()

我有一个单元测试，需要更新/插入型号

@patch('my_package.user.do_something')
def test_user(mock_do_something):
    user = User() # This insert will invoke 'do_something' via the event listener.
    assertSomething(user)

然而，我的测试失败了，因为do_something函数似乎仍在被调用，并且没有被成功地模拟。我试着在这里阅读如何进行修补（它调用这个函数对吗？），并试图在这里查看SQLAlchemy源代码，以找到合适的模块进行修补（类似于@patch('sqlalchemy.event.registrat._listen_fn')），但没有成功。

以前有人遇到过这种情况吗？

Answer 1

也许有一个上下文管理器类可以删除侦听器并在退出时再次添加它。

类似于：

class SADeListener(object):
    def __init__(self, class_, event, callable_):
        self.class_ = class_
        self.event = event
        self.callable_ = callable_
    def __enter__(self):
        sqlalchemy.event.remove(self.class_, self.event, self.callable_)
    def __exit__(self, type_, value, tb):
        sqlalchemy.event.listen(self.class_, self.event, self.callable_)

然后使用它：

with SADeListener(User, "after_insert", do_something),
   SADeListener(User, "after_update", do_something):
    .. code ...

Answer 2

我找到了一个在单元测试上禁用事件的解决方法

import sqlalchemy as sa
from unittest import TestCase
from mypackage.models.user import User
class TestUser(TestCase):
    def setUp(self):
        super(TestUser, self).setUp()
        sa.event.remove(User, "after_insert", do_something)
        sa.event.remove(User, "after_update", do_something)
    def tearDown(self):
        super(TestUser, self).tearDown()
        sa.event.listen(User, "after_insert", do_something)
        sa.event.listen(User, "after_update", do_something)
    @patch('my_package.user.do_something')
    def test_user(mock_do_something):
        user = User() 
        assertSomething(user)