def create_connect_args(self, url):
opts = url.translate_connect_args(['host', 'database', 'user', 'password', 'port'])
opts.update(url.query)
+ if opts.has_key('auto_identity_insert'):
+ self.auto_identity_insert = bool(int(opts['auto_identity_insert']))
return make_connect_string(opts)
def create_execution_context(self):
class TLTransactionTest(testbase.PersistTest):
def setUpAll(self):
global users, metadata, tlengine
- tlengine = create_engine(testbase.db_uri, strategy='threadlocal', **testbase.db_opts)
+ tlengine = create_engine(testbase.db_uri, strategy='threadlocal')
metadata = MetaData()
users = Table('query_users', metadata,
Column('user_id', INT, primary_key = True),
assert s.query(Address).selectone().address_id == a.address_id
assert s.query(User).selectfirst() is None
- def test_fetchid(self):
- # this is necessary to ensure the test fails on old versions of mssql
- if hasattr(autoseq.columns['autoseq_id'], 'sequence'):
- del autoseq.columns['autoseq_id'].sequence
-
- mapper(Autoseq, autoseq)
- s = create_session()
- u = Autoseq()
- s.save(u)
- s.flush()
- assert u.autoseq_id is not None
- s.clear()
-
-
class OrphanDeletionTest(AssertMixin):
# Column('foo', Boolean, default=True)
)
-autoseq = Table('autoseq', metadata,
- Column('autoseq_id', Integer, primary_key = True),
- Column('name', String)
-)
-
def create():
metadata.create_all()
def drop():
def __repr__(self):
return "Keyword: %s/%s" % (repr(getattr(self, 'keyword_id', None)),repr(self.name))
-class Autoseq(object):
- def __init__(self):
- self.autoseq_id = None
-
user_result = [{'user_id' : 7}, {'user_id' : 8}, {'user_id' : 9}]
user_address_result = [
db = None
metadata = None
db_uri = None
-db_opts = {}
echo = True
# redefine sys.stdout so all those print statements go to the echo func
def parse_argv():
# we are using the unittest main runner, so we are just popping out the
# arguments we need instead of using our own getopt type of thing
- global db, db_uri, db_opts, metadata
+ global db, db_uri, metadata
DBTYPE = 'sqlite'
PROXY = False
elif options.db:
DBTYPE = param = options.db
- if DBTYPE == 'mssql':
- db_opts['auto_identity_insert'] = True
-
+ opts = {}
if (None == db_uri):
if DBTYPE == 'sqlite':
db_uri = 'sqlite:///:memory:'
db_uri = 'oracle://scott:tiger@127.0.0.1:1521'
elif DBTYPE == 'oracle8':
db_uri = 'oracle://scott:tiger@127.0.0.1:1521'
- db_opts = {'use_ansi':False, 'auto_setinputsizes':True}
+ opts = {'use_ansi':False, 'auto_setinputsizes':True}
elif DBTYPE == 'mssql':
db_uri = 'mssql://scott:tiger@SQUAWK\\SQLEXPRESS/test'
elif DBTYPE == 'firebird':
with_coverage = options.coverage
if options.enginestrategy is not None:
- db_opts['strategy'] = options.enginestrategy
+ opts['strategy'] = options.enginestrategy
if options.mockpool:
- db = engine.create_engine(db_uri, poolclass=pool.AssertionPool, **db_opts)
+ db = engine.create_engine(db_uri, poolclass=pool.AssertionPool, **opts)
else:
- db = engine.create_engine(db_uri, **db_opts)
+ db = engine.create_engine(db_uri, **opts)
db = EngineAssert(db)
if options.topological:
sys.exit(not result.wasSuccessful())
+