import threading, time
from sqlalchemy import pool, interfaces, create_engine
import testlib.sa as tsa
-from testlib import TestBase, gc, testing
+from testlib import TestBase, testing
+from testlib.compat import gc_collect
mcid = 1
c2 = None
del c1
del c2
- gc.collect()
+ gc_collect()
assert p.checkedout() == 0
c3 = p.connect()
assert c3 is not None
import testenv; testenv.configure_for_tests()
+import time
import weakref
from testlib.sa import select, MetaData, Table, Column, Integer, String, pool
import testlib.sa as tsa
-from testlib import TestBase, testing, engines, gc
-import time
+from testlib import TestBase, testing, engines
+from testlib.compat import gc_collect
+
class MockDisconnect(Exception):
pass
assert id(db.pool) != pid
# ensure all connections closed (pool was recycled)
- gc.collect()
+ gc_collect()
assert len(dbapi.connections) == 0
conn =db.connect()
pass
# assert was invalidated
- gc.collect()
+ gc_collect()
assert len(dbapi.connections) == 0
assert not conn.closed
assert conn.invalidated
assert conn.invalidated
# ensure all connections closed (pool was recycled)
- gc.collect()
+ gc_collect()
assert len(dbapi.connections) == 0
# test reconnects
from sqlalchemy.orm.collections import collection
from sqlalchemy.ext.associationproxy import *
from testlib import *
+from testlib.compat import gc_collect
class DictCollection(dict):
add_child('p1', 'c1')
- gc.collect()
+ gc_collect()
add_child('p1', 'c2')
session.flush()
p.kids.extend(['c1', 'c2'])
p_copy = copy.copy(p)
del p
- gc.collect()
+ gc_collect()
assert set(p_copy.kids) == set(['c1', 'c2']), p.kids
from sqlalchemy.orm.interfaces import AttributeExtension
from sqlalchemy import exc as sa_exc
from testlib import *
+from testlib.compat import gc_collect
from testlib.testing import eq_
from orm import _base
f.bar = "foo"
assert state.dict == {'bar':'foo', state.manager.STATE_ATTR:state}
del f
- gc.collect()
+ gc_collect()
assert state.obj() is None
assert state.dict == {}
"""Attribute/instance expiration, deferral of attributes, etc."""
import testenv; testenv.configure_for_tests()
-from testlib import gc, sa, testing
+from testlib import sa, testing
+from testlib.compat import gc_collect
from testlib.sa import Table, Column, Integer, String, ForeignKey, exc as sa_exc
from testlib.sa.orm import mapper, relation, create_session, attributes, deferred
from orm import _base, _fixtures
assert self.static.user_address_result == userlist
assert len(list(sess)) == 9
sess.expire_all()
- gc.collect()
+ gc_collect()
assert len(list(sess)) == 4 # since addresses were gc'ed
userlist = sess.query(User).order_by(User.id).all()
import inspect
import pickle
from sqlalchemy.orm import create_session, sessionmaker, attributes
-from testlib import engines, gc, sa, testing, config
+from testlib import engines, sa, testing, config
+from testlib.compat import gc_collect
from testlib.sa import Table, Column, Integer, String, Sequence
from testlib.sa.orm import mapper, relation, backref
from testlib.testing import eq_
u = sess.query(User).get(u.id)
q = sess.query(Address).filter(Address.user==u)
del u
- gc.collect()
+ gc_collect()
eq_(q.one(), Address(email_address='foo'))
user = s.query(User).one()
del user
- gc.collect()
+ gc_collect()
assert len(s.identity_map) == 0
user = s.query(User).one()
user.name = 'fred'
del user
- gc.collect()
+ gc_collect()
assert len(s.identity_map) == 1
assert len(s.dirty) == 1
assert None not in s.dirty
s.flush()
- gc.collect()
+ gc_collect()
assert not s.dirty
assert not s.identity_map
user = s.query(User).one()
user = None
print s.identity_map
- gc.collect()
+ gc_collect()
assert len(s.identity_map) == 1
user = s.query(User).one()
self.assert_(len(s.identity_map) == 0)
self.assert_(s.prune() == 0)
s.flush()
- gc.collect()
+ gc_collect()
self.assert_(s.prune() == 9)
self.assert_(len(s.identity_map) == 1)
from sqlalchemy import exc as sa_exc
from sqlalchemy.orm import *
from testlib import *
+from testlib.compat import gc_collect
from testlib.fixtures import *
+
class TransactionTest(FixtureTest):
keep_mappers = True
session = sessionmaker()
assert u1_state not in s.identity_map.all_states()
assert u1_state not in s._deleted
del u1
- gc.collect()
+ gc_collect()
assert u1_state.obj() is None
s.rollback()
import testenv; testenv.simple_setup()
from sqlalchemy.orm import attributes
-from testlib import gc
import time
manage_attributes = True
attributes.manage(a)
a.email = 'foo@bar.com'
u.addresses.append(a)
-# gc.collect()
# print len(managed_attributes)
# managed_attributes.clear()
total = time.time() - now
import random, string
from sqlalchemy.orm import attributes
-from testlib import gc
+from testlib.compat import gc_collect
# with this test, run top. make sure the Python process doenst grow in size arbitrarily.
a.user = u
print "clearing"
#managed_attributes.clear()
- gc.collect()
+ gc_collect()
query = sess.query(Item)
for x in range (1,NUM/100):
# this is not needed with cpython which clears non-circular refs immediately
- #gc.collect()
+ #gc_collect()
l = query.filter(items.c.item_id.between(x*100 - 100 + 1, x*100)).all()
assert len(l) == 100
print "loaded ", len(l), " items "
import testenv; testenv.configure_for_tests()
+import gc
import types
from sqlalchemy import *
from sqlalchemy.orm import *
import time, resource
from sqlalchemy import *
from sqlalchemy.orm import *
-from testlib import gc
+from testlib.compat import gc_collect
db = create_engine('sqlite://')
usage.snap = lambda stats=None: setattr(
usage, 'last', stats or resource.getrusage(resource.RUSAGE_SELF))
- gc.collect()
+ gc_collect()
usage.snap()
t = time.clock()
sqlite_select(RawPerson)
t2 = time.clock()
usage('sqlite select/native')
- gc.collect()
+ gc_collect()
usage.snap()
t = time.clock()
sqlite_select(Person)
t2 = time.clock()
usage('sqlite select/instrumented')
- gc.collect()
+ gc_collect()
usage.snap()
t = time.clock()
sql_select(RawPerson)
t2 = time.clock()
usage('sqlalchemy.sql select/native')
- gc.collect()
+ gc_collect()
usage.snap()
t = time.clock()
sql_select(Person)
t2 = time.clock()
usage('sqlalchemy.sql select/instrumented')
- gc.collect()
+ gc_collect()
usage.snap()
t = time.clock()
orm_select()
from sqlalchemy import *
from sqlalchemy.orm import *
from testlib import *
+from testlib.compat import gc_collect
+
NUM = 100
session = create_session()
- gc.collect()
+ gc_collect()
usage.snap()
t = time.clock()
people = orm_select(session)
t2 = time.clock()
usage('load objects')
- gc.collect()
+ gc_collect()
usage.snap()
t = time.clock()
update_and_flush(session, people)
from sqlalchemy import *
from sqlalchemy.orm import *
from testlib import *
+from testlib.compat import gc_collect
from testlib import fixtures
# in this test we are specifically looking for time spent in the attributes.InstanceState.__cleanup() method.
sess.close()
del sess
- gc.collect()
+ gc_collect()
@profiling.profiled('dirty', report=True)
def test_session_dirty(self):
t2.c2 = 'this is some modified text'
del t1s
- gc.collect()
+ gc_collect()
sess.close()
del sess
- gc.collect()
+ gc_collect()
@profiling.profiled('noclose', report=True)
def test_session_noclose(self):
t1s[index].t2s
del sess
- gc.collect()
-
+ gc_collect()
if __name__ == '__main__':
from sqlalchemy.orm import mapper, relation, create_session, clear_mappers, sessionmaker
from sqlalchemy.orm.mapper import _mapper_registry
from sqlalchemy.orm.session import _sessions
+import gc
import operator
-from testlib import gc, testing
+from testlib import testing
+from testlib.compat import gc_collect
from testlib.sa import MetaData, Table, Column, Integer, String, ForeignKey, PickleType
import sqlalchemy as sa
from sqlalchemy.sql import column
# run the test 50 times. if length of gc.get_objects()
# keeps growing, assert false
def profile(*args):
- gc.collect()
+ gc_collect()
samples = [0 for x in range(0, 50)]
for x in range(0, 50):
func(*args)
- gc.collect()
+ gc_collect()
samples[x] = len(gc.get_objects())
print "sample gc sizes:", samples
def assert_no_mappers():
clear_mappers()
- gc.collect()
+ gc_collect()
assert len(_mapper_registry) == 0
class EnsureZeroed(_base.ORMTest):
testing.eq_(len(session.identity_map._mutable_attrs), 12)
testing.eq_(len(session.identity_map), 12)
obj = None
- gc.collect()
+ gc_collect()
testing.eq_(len(session.identity_map._mutable_attrs), 0)
testing.eq_(len(session.identity_map), 0)
from testlib.orm import mapper
import testlib.profiling as profiling
import testlib.engines as engines
-import testlib.sa_gc as gc
import testlib.requires as requires
from testlib.compat import _function_named
'rowset',
'TestBase', 'AssertsExecutionResults', 'ORMTest',
'AssertsCompiledSQL', 'ComparesTables',
- 'profiling', 'engines', 'gc',
+ 'profiling', 'engines',
'_function_named')
-import sys, types, __builtin__
+import gc
+import sys
+import time
+import types
+import __builtin__
+
__all__ = '_function_named', 'callable', 'py3k', 'jython'
else:
callable = __builtin__.callable
+if sys.platform.startswith('java'):
+ def gc_collect(*args):
+ gc.collect()
+ time.sleep(0.1)
+ gc.collect()
+ gc.collect()
+ return 0
+else:
+ gc_collect = gc.collect
"""Profiling support for unit and performance tests."""
import os, sys
-from testlib.compat import _function_named
+from testlib.compat import _function_named, gc_collect
import testlib.config
__all__ = 'profiled', 'function_call_count', 'conditional_call_count'
import cProfile, gc, pstats, time
load_stats = lambda: pstats.Stats(filename)
- gc.collect()
+ gc_collect()
began = time.time()
cProfile.runctx('result = fn(*args, **kw)', globals(), locals(),
import gc, hotshot, hotshot.stats, time
load_stats = lambda: hotshot.stats.load(filename)
- gc.collect()
+ gc_collect()
prof = hotshot.Profile(filename)
began = time.time()
prof.start()
+++ /dev/null
-"""Cross platform garbage collection utility"""
-import gc
-import sys
-import time
-
-if sys.platform.startswith('java'):
- def collect(*args):
- gc.collect()
- time.sleep(0.1)
- gc.collect()
- gc.collect()
- return 0
-else:
- collect = gc.collect
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
