Skip to content

Instantly share code, notes, and snippets.

@GaretJax
Last active August 29, 2015 14:05
Show Gist options
  • Star 1 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save GaretJax/b78bd97434992a8ebf26 to your computer and use it in GitHub Desktop.
Save GaretJax/b78bd97434992a8ebf26 to your computer and use it in GitHub Desktop.
after_flush_postexec events skipped when removing handlers
python -m unittest test_events
.F...F
======================================================================
FAIL: testDoubleRemove (sqlalchemy_mptt.tests.test_flush.FlushingTest)
----------------------------------------------------------------------
Traceback (most recent call last):
File "sqlalchemy_mptt/tests/test_flush.py", line 60, in testDoubleRemove
self.assertEqual(self.calls, 2)
AssertionError: 1 != 2
======================================================================
FAIL: testTripleRemove (sqlalchemy_mptt.tests.test_flush.FlushingTest)
----------------------------------------------------------------------
Traceback (most recent call last):
File "sqlalchemy_mptt/tests/test_flush.py", line 67, in testTripleRemove
self.assertEqual(self.calls, 3)
AssertionError: 2 != 3
----------------------------------------------------------------------
Ran 6 tests in 0.046s
FAILED (failures=2)
import unittest
from sqlalchemy import create_engine, Column, Integer, String, event
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class TestModel(Base):
__tablename__ = 'test_model'
id = Column(Integer, primary_key=True)
name = Column(String, unique=True)
class Receiver(object):
def __init__(self, instance, remove):
self.instance = instance
self.remove = remove
def __call__(self, session, context):
self.instance.calls += 1
if self.remove:
event.remove(self.instance.session, 'after_flush_postexec', self)
@classmethod
def register(cls, instance, remove):
self = cls(instance, remove=remove)
event.listen(instance.session, 'after_flush_postexec', self)
class FlushingTest(unittest.TestCase):
base = Base
model = TestModel
def setUp(self):
self.engine = create_engine('sqlite:///:memory:')
Session = sessionmaker(bind=self.engine)
self.session = Session()
self.base.metadata.create_all(self.engine)
self.calls = 0
def tearDown(self):
self.base.metadata.drop_all(self.engine)
def flush(self):
self.session.add(self.model(name='test'))
self.session.flush()
def testSingleRemove(self):
Receiver.register(self, remove=True)
self.flush()
self.assertEqual(self.calls, 1)
def testDoubleRemove(self):
Receiver.register(self, remove=True)
Receiver.register(self, remove=True)
self.flush()
self.assertEqual(self.calls, 2)
def testTripleRemove(self):
Receiver.register(self, remove=True)
Receiver.register(self, remove=True)
Receiver.register(self, remove=True)
self.flush()
self.assertEqual(self.calls, 3)
def testSingleNoRemove(self):
Receiver.register(self, remove=False)
self.flush()
self.assertEqual(self.calls, 1)
def testDoubleNoRemove(self):
Receiver.register(self, remove=False)
Receiver.register(self, remove=False)
self.flush()
self.assertEqual(self.calls, 2)
def testTripleNoRemove(self):
Receiver.register(self, remove=False)
Receiver.register(self, remove=False)
Receiver.register(self, remove=False)
self.flush()
self.assertEqual(self.calls, 3)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment