pool.Pool.dispatch._clear()
class DeprecatedPoolListenerTest(PoolTestBase):
+ @testing.requires.predictable_gc
@testing.uses_deprecated(r".*Use event.listen")
def test_listeners(self):
class InstrumentingListener(object):
still_opened = len([c for c in strong_refs if not c.closed])
eq_(still_opened, 2)
+ @testing.requires.predictable_gc
def test_weakref_kaboom(self):
p = self._queuepool_fixture(pool_size=3,
max_overflow=-1, use_threadlocal=True)
class ClsRegistryTest(fixtures.TestBase):
+ __requires__ = 'predictable_gc',
+
def test_same_module_same_name(self):
base = weakref.WeakValueDictionary()
f1 = MockClass(base, "foo.bar.Foo")
], "savepoints not supported")(fn)
def denormalized_names(fn):
- """Target database must have 'denormalized', i.e. UPPERCASE as case insensitive names."""
+ """Target database must have 'denormalized', i.e.
+ UPPERCASE as case insensitive names."""
return skip_if(
lambda: not testing.db.dialect.requires_name_normalize,
- "Backend does not require denomralized names."
+ "Backend does not require denormalized names."
)(fn)
def schemas(fn):
- """Target database must support external schemas, and have one named 'test_schema'."""
+ """Target database must support external schemas, and have one
+ named 'test_schema'."""
return skip_if([
"sqlte",
"""Target database must support two-phase transactions."""
return skip_if([
- no_support('access', 'not supported by database'),
+ no_support('access', 'two-phase xact not supported by database'),
no_support('firebird', 'no SA implementation'),
- no_support('maxdb', 'not supported by database'),
- no_support('mssql', 'FIXME: guessing, needs confirmation'),
- no_support('oracle', 'no SA implementation'),
- no_support('drizzle', 'not supported by database'),
- no_support('sqlite', 'not supported by database'),
- no_support('sybase', 'FIXME: guessing, needs confirmation'),
+ no_support('maxdb', 'two-phase xact not supported by database'),
+ no_support('mssql', 'two-phase xact not supported by drivers'),
+ no_support('oracle', 'two-phase xact not implemented in SQLA/oracle'),
+ no_support('drizzle', 'two-phase xact not supported by database'),
+ no_support('sqlite', 'two-phase xact not supported by database'),
+ no_support('sybase', 'two-phase xact not supported by drivers/SQLA'),
no_support('postgresql+zxjdbc',
'FIXME: JDBC driver confuses the transaction state, may '
'need separate XA implementation'),
- exclude('mysql', '<', (5, 0, 3), 'not supported by database'),
+ exclude('mysql', '<', (5, 0, 3),
+ 'two-phase xact not supported by database'),
])(fn)
def views(fn):
self.assert_(o4.mt2[0].a == 'abcde')
self.assert_(o4.mt2[0].b is None)
+ @testing.requires.predictable_gc
def test_state_gc(self):
"""test that InstanceState always has a dict, even after host
object gc'ed."""
f = Foo()
state = attributes.instance_state(f)
f.bar = "foo"
- assert state.dict == {'bar':'foo', state.manager.STATE_ATTR:state}
+ eq_(state.dict, {'bar': 'foo', state.manager.STATE_ATTR: state})
del f
gc_collect()
assert state.obj() is None
self._assert_not_hasparent(a1)
+ @testing.requires.predictable_gc
def test_stale_state_positive_gc(self):
User = self.classes.User
s, u1, a1 = self._fixture()
self._assert_not_hasparent(a1)
+ @testing.requires.predictable_gc
def test_stale_state_positive_pk_change(self):
"""Illustrate that we can't easily link a
stale state to a fresh one if the fresh one has
self._assert_not_hasparent(a1)
#self._assert_hasparent(a1)
+ @testing.requires.predictable_gc
def test_stale_state_negative(self):
User = self.classes.User
s, u1, a1 = self._fixture()
assert user not in s
assert s.query(User).count() == 0
-
- def test_weak_ref(self):
+ @testing.requires.predictable_gc
+ def test_weakref(self):
"""test the weak-referencing identity map, which strongly-
references modified items."""
assert user.name == 'fred'
assert s.identity_map
- def test_weak_ref_pickled(self):
+ @testing.requires.predictable_gc
+ def test_weakref_pickled(self):
users, User = self.tables.users, pickleable.User
s = create_session()
sa.orm.attributes.instance_state(u2))
+ @testing.requires.predictable_gc
def test_weakref_with_cycles_o2m(self):
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
user = s.query(User).options(joinedload(User.addresses)).one()
eq_(user, User(name="ed", addresses=[Address(email_address="ed2")]))
+ @testing.requires.predictable_gc
def test_weakref_with_cycles_o2o(self):
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
assert u1 in s
assert u1 not in s.deleted
+ @testing.requires.predictable_gc
def test_gced_delete_on_rollback(self):
User, users = self.classes.User, self.tables.users