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 函数并且尚未成功模拟。我尝试阅读如何修补 here (它是在调用这个函数吗？)并且我试图查看 SQLAlchemy 源代码 here找到要修补的适当模块(类似于 @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 ...