Python:如何模拟 SQLAlchemy 事件处理程序(使用 mock.unittest)
Python: How to mock SQLAlchemy event handlers (using mock.unittest)
所以我有一个带有事件侦听器的 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()
我有一个单元测试需要 update/insert 模型
@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 (it is calling this function right?) and I have tried to look through the SQLAlchemy source code here 以找到要修补的适当模块(类似于 @patch('sqlalchemy.event.registrat._listen_fn')
),但无济于事。
有没有人遇到过这种情况?
我找到了在单元测试中禁用事件的解决方法
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)
也许有一个上下文管理器 class 可以删除侦听器并在退出时再次添加它可以解决问题。
类似于:
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 ...
所以我有一个带有事件侦听器的 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()
我有一个单元测试需要 update/insert 模型
@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 (it is calling this function right?) and I have tried to look through the SQLAlchemy source code here 以找到要修补的适当模块(类似于 @patch('sqlalchemy.event.registrat._listen_fn')
),但无济于事。
有没有人遇到过这种情况?
我找到了在单元测试中禁用事件的解决方法
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)
也许有一个上下文管理器 class 可以删除侦听器并在退出时再次添加它可以解决问题。
类似于:
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 ...