Python: Wie SQLAlchemy Event-Handler (mit mock.unittest)

Question

So habe ich ein SQLAlchemy Modell verspotten, die ein Ereignis-Listener hat:

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() 

Und ich habe eine Unit-Test, der das Modell aktualisieren muss/einfügen

@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) 

jedoch nicht meine Tests, weil es wie die do_something Funktion scheint immer noch genannt wird, und nicht erfolgreich verspottet worden. Ich versuchte zu lesen, wie Patching here (es ruft diese Funktion richtig?) Und ich habe versucht, durch den SQLAlchemy-Quellcode here zu suchen, um das entsprechende Modul zu Patch (etwas wie @patch('sqlalchemy.event.registrat._listen_fn')) zu finden, aber ohne Erfolg.

Hat jemand schon einmal das schon einmal erlebt?

Answer 1

Vielleicht eine Context-Manager-Klasse mit, die den Hörer und fügen Sie es wieder an der Ausfahrt entfernt könnte den Trick.

Etwas wie:

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_) 

Und dann einfach verwenden:

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

Answer 2

Ich habe eine Abhilfe gefunden Ereignisse am Gerät deaktivieren testet

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)